ThreadPoolExecutor with blocking execute?

wrote:
> wrote:
> > > Yeah, but that code will basically be in a busy wait loop. It will
> > > constantly have an exception thrown, recaught and retried. By using a
> > > blocking put on the queue (my first post), the thread can yield itself
> > > until it can actually do something. I think your example would eat the
> > > CPU and is more complex than the first example I had.
> >
> > Huh?
> >
> > There is no exception thrown, recaught and retried. Not sure where you
> > got this idea from. My code uses the blocking methods of the
> > BlockingQueue (which the executor does not). The CPU will not run hot
> > and an exception is not thrown unless the thread is interupted (which
> > it is not under normal operation).
> >
>
>
> Sorry, I misread the code, I thought one of the comments indicated that
> "more code should be written..." (And I didn't mean to sound like a
> jerk - sorry)
>
> So I ran your code, but it still doesn't actually limit the main
> producing thread. The queue you used for the ThreadPool (not the other
> one) is unbounded, so that's why you didn't get any exceptions. So the
> problem I had with an unbounded queue, is that I could fill up memory.
> If you put a print statement right after your for loop which puts items
> on, you will see it gets through the loop very fast. The extra thread
> and ArrayBlockingQueue don't help in limiting throughput.


You are absolutely right, sorry, must have been experiencing a mental
haywire when I wrote that code yesterday. It is probably best ignored.

> > Personally, I don't the idea of managing rejected execution
> > retroactively (as per your example). If you are happy with this and
> > prefer this approach then great. It is fairly subjective after all.
> >
>
> I don't know that I like it either, but it seems like that was the
> model intended by the API.

I dont think it was the model intended for implementing blocking
executors. It probably is the model intended to inform a process that a
task it provided cannot be executed due to heavy load (all threads
busy, and pool at maximum).

> > Dan's solution is nice. I was under the (incorrect) impression that the
> > offer method was documented as failing when queue is full, but it seems
> > it isn't.
>
> It is documented:
> http://java.sun.com/j2se/1.5.0/docs/...e.html#offer(E)

Sure, but there is nothing there that says explicitally that 'offer'
cannot block, Dan mentioned this before. The ThreadPoolExecutor uses
queue.offer rather than queue.put when execute is called, so you need
offer to block. You could write an ArrayBlockingQueue subclass to
redirect all calls to 'offer' to 'put' which would have the desired
effect. I think I would prefer something like this....

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.Collection;

public class DefaultedOfferTimeoutBlockingQueue<E> extends
ArrayBlockingQueue<E>
{
private long defaultOfferTimeoutDuration;
private TimeUnit defaultOfferTimeUnit;

public DefaultedOfferTimeoutBlockingQueue(int capacity, long
defaultOfferTimeoutDuration, TimeUnit defaultOfferTimeUnit)
{
super(capacity);
this.defaultOfferTimeoutDuration = defaultOfferTimeoutDuration;
this.defaultOfferTimeUnit = defaultOfferTimeUnit;
}

public DefaultedOfferTimeoutBlockingQueue(int capacity, boolean
fair, long defaultOfferTimeoutDuration, TimeUnit defaultOfferTimeUnit)
{
super(capacity, fair);
this.defaultOfferTimeoutDuration = defaultOfferTimeoutDuration;
this.defaultOfferTimeUnit = defaultOfferTimeUnit;
}

public DefaultedOfferTimeoutBlockingQueue(int capacity, boolean
fair, Collection<? extends E> initialElements, long
defaultOfferTimeoutDuration, TimeUnit defaultOfferTimeUnit)
{
super(capacity, fair, initialElements);
this.defaultOfferTimeoutDuration = defaultOfferTimeoutDuration;
this.defaultOfferTimeUnit = defaultOfferTimeUnit;
}


public boolean offer(E element)
{
try
{
return offer(element, defaultOfferTimeoutDuration,
defaultOfferTimeUnit);
}
catch (InterruptedException e)
{
//todo: probably should log something here
return false;
}
}
}

Which is a simple blocking queue that lets you create a default timeout
for the basic 'offer' method (rather than it failing instantly).

With this class in place you can simply do this...

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolExecutorBlockTest
{
public static void main(String[] args)
{
Executor executor = new ThreadPoolExecutor(5, 10, 10,
TimeUnit.SECONDS, new DefaultedOfferTimeoutBlockingQueue<Runnable>(5,
86400, TimeUnit.SECONDS));

for(int i = 0; i < 50; i++)
{
executor.execute(new Printer(i));
System.out.println("Task " + i + " added");
}
}

private static class Printer implements Runnable
{
private int number;

public Printer(int number)
{
this.number = number;
}

public void run()
{
System.out.println("Running task: " + number);
try
{
Thread.sleep(10000);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
}
}

This seems to work very well for me.

Note though, items are removed from the queue (freeing up space) before
the task is run and not after, so there will always be (queueSize +
threadPoolSize) tasks that are sitting in the executor.

wesley.hall@gmail.com

wesley.h...@gmail.com wrote:
> wrote:
> > wrote:
<snip>
> > > Personally, I don't the idea of managing rejected execution
> > > retroactively (as per your example). If you are happy with this and
> > > prefer this approach then great. It is fairly subjective after all.
> > >
> >
> > I don't know that I like it either, but it seems like that was the
> > model intended by the API.
>
> I dont think it was the model intended for implementing blocking
> executors. It probably is the model intended to inform a process that a
> task it provided cannot be executed due to heavy load (all threads
> busy, and pool at maximum).
>

But there is a class CallerRunsPolicy (it implements
RejectedExecutionHandler), which handles the runnable by executing it
in the current thread. Using this class, you pretty much get a
blocking executor. So I think the RejectedExecutionHandler was meant
to be used for a variety of things, including processing the Runnable.


http://java.sun.com/j2se/1.5.0/docs/...unsPolicy.html

Anyway, there are 2 solutions here. I'm just left wondering why there
isn't direct support for a blocking ThreadPoolExecutor in the API.
(The documentation for BlockingQueue has a single producer, multiple
consumers example.) I think the ThreadPoolExecutor would be the
perfect place to have that code already in place. Its always such a
pain writing your own, with proper shutdown etc....



<snip>

castillo.bryan@gmail.com