Groups | Search | Server Info | Keyboard shortcuts | Login | Register [http] [https] [nntp] [nntps]

Groups > comp.lang.java.programmer > #33134

Re: To BlockingQueue or not to BlockingQueue?

Show all headers | View raw

Am 13.11.2015 15:13, schrieb Eric Sosman:
>
>      Short version: I wrote a simple submit-tasks-and-reap-results
> class based on a ThreadPoolExecutor.  Initially it seemed I could
> use an ExecutorCompletionService to manage the flow of work, but
> that didn't pan out.  Then I tried having the tasks place their own
> results on a BlockingQueue explicitly, but that didn't pan out either.
> Eventually, I retrogressed all the way to using an ordinary Queue and
> an old-fashioned synchronized/wait/notifyAll scheme.  I'm hoping for
> some hints about how to use the java.util.concurrent classes better,
> since they're more modern and fashionable.
>
>      Longer version: What I wanted was a multi-threaded server class
> to which clients could submit tasks and from which they could retrieve
> results as they became available.  The submissions and retrievals might
> happen in one client thread or in many, but they'd all be the same
> "kind" of job (thus, no need to deliver a result to the same client
> thread that submitted the task; any client could accept any result).
> That much is pretty simple, and an ExecutorCompletionService looks like
> a perfect match -- but there were a few wrinkles:
>
>      1) If the worker threads get too far ahead of the consuming clients
> and start clogging the system with too many unretrieved results, I want
> the workers to pause until the backlog diminishes.
>
>      2) If a client tries to retrieve a result and none is ready yet,
> the client should wait.  But if there are no tasks in the system, no
> results are forthcoming and the client should be sent away empty-handed
> (I used NoSuchElementException, but a null return would also work).
>
>      3) There should be a way to shut the whole thing down, either
> gracefully (finish all submitted tasks first) or forcibly (discard all
> tasks not yet started and try to cancel those in progress).
>
>      To implement #1 I first tried using a bounded BlockingQueue to hold
> the output from an ExecutorCompletionService.  Unfortunately, an ECS
> confronting a full BQ won't wait for space, but will just discard the
> result.  So I ditched the ECS and used the ThreadPoolExecutor directly,
> arranging for the tasks to put their own results on the BQ and to wait
> for space if necessary.  That looked promising, but ...
>
>      ... still didn't solve #2.  A client trying to retrieve a result
> from an empty BlockingQueue will wait, and that's fine.  But I wanted
> to distinguish between "Wait, something's coming" and "Nothing's in
> the pipeline."  The BQ could handle the first condition, but not the
> two conditions combined.
>
>      I considered abandoning #2 and placing the burden on the clients:
> They know how many tasks they submitted, so they should know how many
> results to reap, right?  But #3 makes this untenable: If you shut down
> a ThreadPoolExecutor forcibly, all the running worker threads get
> InterruptedExceptions.  If some of those threads were interrupted while
> waiting to put results on the BlockingQueue, those results get lost.
> That's no tragedy in the context of a forcible shutdown, but it throws
> off the clients' count of "How many results can we reap?" and you could
> wind up waiting forever.  Also, a forcible shutdown wouldn't awaken a
> client already waiting on an empty BQ, and he'd be stuck indefinitely.
>
>      Eventually I settled on a plain Queue with explicit synchronization:
> very old-fashioned and out of style, I know, but I couldn't think of a
> way to do things in a more modern mode.  It doesn't seem to me that my
> goals #1,2,3 are outrageously weird, and there *ought* to be a way to
> satisfy them without making myself look like a refugee from the last
> millennium.  Does anyone have suggestions?
>
Here's another idea: Put CompletableFutures in a BlockingQueue before 
you start the computation. This takes care of both #1 and #2. You don't 
need to do any sort of bookkeeping with a Semaphore. When the queue is 
full with unretrieved futures, no more results get added and hence no 
more computations get triggered. To address #3, you could shutdown the 
thread pool supplied from the outside.

The code below would make the task producers block when the workers have 
no queue space left to deposit unretrieved results, and the result 
receivers when they have retrieved a result (Future) that is not yet 
completed. Don't know if you can accept that. You speak of making the 
workers pause if there are too many unretrieved results, but that 
doesn't really make sense to me. If tasks can keep coming in, they'd 
just pile up in an unbounded queue.

To me, it seems that the clients should adjust their submit/receive 
rate, or that the thread pool ought to be tweaked to deliver results at 
a lower rate. You might use add() instead of put() and have the client 
deal with the IllegalArgumentException instead of being blocked until 
another client retrieves a result. This would avoid the possible 
deadlock when all clients continue to submit tasks and never retrieve them.

You don't say if your tasks really need to be able to throw exceptions. 
If yes, with CompletableFuture you need to adapt your Callables to 
Suppliers. If not, you can skip that and just pass in Suppliers.

Here's some code (untested, except with a variant of Steven Simpson's 
test code that submits jobs in an extra thread).

public class EricsCompletor<T> {

     private final BlockingQueue<Future<T>> results;
     private final ExecutorService pool;

     public EricsCompletor(ExecutorService pool,
             int resultBufferSize) {
         this.pool = pool;
         this.results = new ArrayBlockingQueue<>(
                 resultBufferSize);
     }

     public CompletableFuture<T> submit(Callable<T> task)
             throws InterruptedException {
         return submit(toSupplier(task));
     }

     /**
      * Submits a task for asynchronous execution. Blocks if
      * there are too many unretrieved results.
      *
      * @param task computation to be executed
      * @return a future representing the result of executing
      *         the task
      * @throws InterruptedException if interrupted while
      *             waiting to queue the task
      */
     public CompletableFuture<T> submit(Supplier<T> task)
             throws InterruptedException {
         CompletableFuture<T> future = new CompletableFuture<>();
         results.put(future); // block when full. consider add()

         CompletableFuture<T> cf = CompletableFuture
                 .supplyAsync(task, pool); // submit task
         cf.whenComplete((v, ex) -> { // update result
             if (ex == null) future.complete(v);
             else future.completeExceptionally(ex);
         });

         // propagate cancel if returned future is cancelled
         future.whenComplete((v, ex) -> cf.cancel(true));
         return future;
     }

     /**
      * Retrieves and removes the Future representing the next
      * completed task or null if none are present.
      */
     public Future<T> poll() {
         return results.poll();
     }

     public boolean cannotSubmit() {
         return results.remainingCapacity() == 0;
     }

     // ---------- helpers

     /**
      * Exception hiding for Suppliers passed to
      * CompletableFutures, because these suppliers cannot
      * throw a checked exception (cf. {@link Supplier#get()}
      * ).
      *
      * @param callable exception-throwing code
      * @return supplier that returns the result of the
      *         callable and declares no exception. Any
      *         exception thrown by the Callable will be
      *         rethrown wrapped in a CompletionException
      * @throws CompletionException an exception with any
      *             Throwable thrown from callable as its cause
      */
     static <T> Supplier<T> toSupplier(Callable<T> callable) {
         return () -> {
             try {
                 return callable.call();
             }
             catch (Throwable t) {
                 throw new CompletionException(t);
             }
         };
     }

}

Back to comp.lang.java.programmer | Previous | Next — Previous in thread | Find similar | Unroll thread

Thread

To BlockingQueue or not to BlockingQueue? Eric Sosman <esosman@comcast-dot-net.invalid> - 2015-11-13 09:13 -0500
  Re: To BlockingQueue or not to BlockingQueue? Eric Douglas <e.d.programmer@gmail.com> - 2015-11-13 07:29 -0800
    Re: To BlockingQueue or not to BlockingQueue? Eric Sosman <esosman@comcast-dot-net.invalid> - 2015-11-13 10:41 -0500
      Re: To BlockingQueue or not to BlockingQueue? Eric Douglas <e.d.programmer@gmail.com> - 2015-11-13 07:57 -0800
  Re: To BlockingQueue or not to BlockingQueue? Kevin McMurtrie <mcmurtrie@pixelmemory.us> - 2015-11-13 16:56 -0800
    Re: To BlockingQueue or not to BlockingQueue? Eric Sosman <esosman@comcast-dot-net.invalid> - 2015-11-14 17:08 -0500
  Re: To BlockingQueue or not to BlockingQueue? Arne Vajhøj <arne@vajhoej.dk> - 2015-11-13 21:40 -0500
    Re: To BlockingQueue or not to BlockingQueue? Eric Sosman <esosman@comcast-dot-net.invalid> - 2015-11-14 17:24 -0500
      Re: To BlockingQueue or not to BlockingQueue? Arne Vajhøj <arne@vajhoej.dk> - 2015-11-14 19:13 -0500
  Re: To BlockingQueue or not to BlockingQueue? Steven Simpson <ss@my.domain.invalid> - 2015-11-15 10:37 +0000
  Re: To BlockingQueue or not to BlockingQueue? Sebastian <news@seyweiler.dyndns.org> - 2015-11-16 01:00 +0100

csiph-web