I would like a pool of threads where a producer puts work items on the
queue, and there are a pool of threads pulling things off and doing
them.  Standard sort of producer/consumer pattern I would have thought.  
The work items can take between 10ms and 10minutes to complete.  
java.util.concurrent is tempting, but I cannot work out how to make it
work for me.

I first off tried an unbounded queue.  But it quickly used up all memory
when the producer out-ran the consumer threads.

I then tried a bounded queue, but by being bounded it means it throws an
exception if you give it too much work to do.  (It does not block the
producer it terms of making it wait - it blocks the producer by throwing
an exception!)

There is a rejection policy you can use which is to let the producer do
the work if all threads in the pool are busy.  But this is not ideal.  
The work items are of different sizes, so if a large work item comes
along and the producer does the work itself, all the other threads in
the pool will run out of work.

I do like the way the java.util.concurrent thread pool stuff nicely
shuts down threads etc when idle.  However, not being able to do a
simple producer/consumer pattern with the library just seems bizarre.  
Surely I am just not understanding how to use it??!!  Otherwise I guess
I roll my own code (which is not hard - just trying to avoid it).

http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/LinkedBl...

   put() and take() block under the expected conditions.

      - Bret -

On Aug 13, 2:34 am, Alan Kent <ALAN.J.K...@saic.com> wrote:

I tried defining a subclass of ArrayBlockingQueue which made offer()
call put() to block, but the end result was only a single thread would
run - very strange.  It never created more than one thread.  As soon as
I put it back, multiple threads would run.  Looking in the
ArrayBlockingQueue source it looked fine, so I suspect the problem is in
the ThreadPoolExecutor code - it just is not designed that way.

It took me a while to realize that "blocking" in the
java.util.concurrent terminology meant "stop it from happening by
throwing an exception", not cause the caller to wait until resources
were free.

I can go off and write my own thread pool - I was just wondering first
if I had overlooked something obvious.

package zinger.util;

import java.util.*;

import zinger.util.recycling.*;

public class ThreadLimiter
{
    public static class RunnableException extends RuntimeException
    {
        private final Exception runnableException;

        public RunnableException(final Exception runnableException)
        {
            this.runnableException = runnableException;
        }

        public Exception getException()
        {
            return this.runnableException;
        }
    }

    protected static final ObjectGenerator COUNTER_GENERATOR = new
ArrayGenerator(int.class, 1)
    {
        public boolean prepareObject(final Object obj, final Object
arg) throws IllegalArgumentException
        {
            try
            {
                ((int[])obj)[0] = 0;
                return true;
            }
            catch(final ClassCastException ex)
            {
                throw new IllegalArgumentException();
            }
        }
    };

    protected final int maxThreadCount;
    protected final long threadTimeout;
    protected final int maxThreadWaits;

    protected final Map threads = new HashMap();
    protected final ObjectRecycler counterRecycler;

    public ThreadLimiter(final int maxThreadCount, final long
threadTimeout, final int maxThreadWaits)
    {
        this.maxThreadCount = maxThreadCount;
        this.threadTimeout = threadTimeout;
        this.maxThreadWaits = maxThreadWaits;

        this.counterRecycler = new
CappedObjectRecycler(ThreadLimiter.COUNTER_GENERATOR,
this.maxThreadCount);
    }

    protected String getCategoryName()
    {
        return this.getClass().getName();
    }

    public final void run(final Runnable toRun) throws
TimeoutException, InterruptedException, Exception
    {
        final Thread thread = Thread.currentThread();
        int[] threadCounter;
        synchronized(this.threads)
        {
            threadCounter = (int[])this.threads.get(thread);
            if(threadCounter == null)
            {
                for(int i = 0; this.threads.size() >= maxThreadCount; +
+i)
                {
                    if(i >= this.maxThreadWaits)
                    {
                        throw new TimeoutException();
                    }
                    this.threads.wait(this.threadTimeout);
                }
                threadCounter =
(int[])this.counterRecycler.getObject();
                this.threads.put(thread, threadCounter);
            }
        }
        ++threadCounter[0];
        try
        {
            toRun.run();
        }
        catch(final ThreadLimiter.RunnableException ex)
        {
            throw ex.getException();
        }
        finally
        {
            if((--threadCounter[0]) <= 0)
            {
                synchronized(this.threads)
                {
                    this.threads.remove(thread);
                    this.threads.notify();
                }
            }
        }
    }

    public StringBuffer status(StringBuffer sb)
    {
        if(sb == null)
        {
            sb = new StringBuffer();
        }

        sb.append("maxThreadCount
\t").append(this.maxThreadCount).append('\n');

        sb.append("threadTimeout
\t").append(this.threadTimeout).append('\n');

        sb.append("maxThreadWaits
\t").append(this.maxThreadWaits).append('\n');

        sb.append("active threads:");
        synchronized(this.threads)
        {
            if(this.threads.isEmpty())
            {
                sb.append(" none\n");
            }
            else
            {
                sb.append('\n');
                Map.Entry entry;
                for(final Iterator iterator =
this.threads.entrySet().iterator(); iterator.hasNext();)
                {
                    entry = (Map.Entry)iterator.next();
                    sb.append(entry.getKey())
                      .append(" (")
                      .append(((int[])entry.getValue())[0])
                      .append(")\n");
                }
            }
        }

        return sb;
    }

You will probably need to wrap the task in another runnable inside the
submitTask method and call run yourself in order to have a run method
that can see the semaphore to release the permit.

Hope that makes sense...

Paul

On Aug 13, 8:34 am, Alan Kent <ALAN.J.K...@saic.com> wrote:

We do something similar to this with a LinkedBlockingQueue. We have a
number of threads (manually created at the moment) that add items to a
BlockingQueue. A separate thread takes items off the queue and
processes them.

The Java threaded pool routines are just a way of creating threads.
You are correct in thinking that its up to you to build a framework
for this - it won't be hard.

I agree with you that this sort of Producer/Consumer is so prevalent
that it should have been in the Java 5 libraries. Even Brian Goetz's
book doesn't really give you a complete solution for this.

There are a number of scenarios.

1. Multiple threads add items to a BlockingQueue and a single thread
takes off items
2. A Single thread adds items to a BlockinqQueue and multiple threads
take off items
3. A Single thread adds items to a BlockQueue and another thread takes
off items.

Arguably these are all variants on a core pattern but there are
implementation reasons why you might want to treat them separately.

We have done something with item 1 and it works very well indeed. We
used Generics to create a class to handle this, it also defines JMX
methods to enable the queue to be monitored. I can't post the code
though because I work for a corporate!

On Aug 13, 8:34 am, Alan Kent <ALAN.J.K...@saic.com> wrote: