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

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

Re: To BlockingQueue or not to BlockingQueue?

Show all headers | View raw

On 13/11/15 14:13, Eric Sosman wrote:
>     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.

Do you want them to pause what they are doing, or just not start any new 
tasks?  I'm assuming the latter.

>     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).

Can you not wrap/extend ECS so that it wraps its incoming Callables so 
that they can detect when the submitted Callable has completed? Then you 
can keep track of how many went in, how many are now ready but 
unclaimed, etc.  The Callable wrapper could semaphore.acquire() when the 
wrapped Caller completes, and your take() could semaphore.release().  
The wrapper also calls acquire then release just before entering the 
wrapped Caller, so it won't start doing more work if the client is not 
consuming results.

In the code below, I've omitted doing anything special for shutting down.

import java.util.concurrent.*;

public class ErSoECS<T> implements CompletionService<T> {
     private final Semaphore sem;
     private final ExecutorCompletionService<T> ecs;

     private int submitted = 0;

     public ErSoECS(Executor e, int permits) {
         this.sem = new Semaphore(permits);
         this.ecs = new ExecutorCompletionService<>(e);
     }

     public Future<T> submit(Callable<T> task) {
         synchronized (this) {
             submitted++;
             return ecs.submit(new Wrapper(task));
         }
     }

     public Future<T> submit(Runnable task, T result) {
         return submit(() -> { task.run(); return result; });
     }

     public Future<T> take() throws InterruptedException {
         synchronized (this) {
             if (submitted == 0) return null;
             Future<T> result = ecs.take();
             submitted--;
             sem.release();
             return result;
         }
     }

     public Future<T> poll() {
         synchronized (this) {
             Future<T> result = ecs.poll();
             if (result != null) {
                 submitted--;
                 sem.release();
             }
             return result;
         }
     }

     public Future<T> poll(long timeout, TimeUnit unit)
         throws InterruptedException {
         synchronized (this) {
             if (submitted == 0) return null;
             Future<T> result = ecs.poll(timeout, unit);
             if (result != null) {
                 submitted--;
                 sem.release();
             }
             return result;
         }
     }

     class Wrapper implements Callable<T> {
         private final Callable<T> base;

         public Wrapper(Callable<T> base) {
             this.base = base;
         }

         public T call() throws Exception {
             try {
                 // Attempt to briefly claim and release a permit, so
                 // we block if the semaphore is exhausted.
                 for ( ; ; ) {
                     try {
                         sem.acquire();
                         break;
                     } catch (InterruptedException ex) {
                         // Go round again.
                     }
                 }
                 sem.release();
                 return base.call();
             } finally {
                 // Claim the semaphore, as if we're consuming space in
                 // an output queue.
                 for ( ; ; ) {
                     try {
                         sem.acquire();
                         break;
                     } catch (InterruptedException ex) {
                         // Go round again.
                     }
                 }
             }
         }
     }

     /***** Test code *****/

     static class MyTask implements Callable<Integer> {
         private final int value, delay;

         MyTask(int value, int delay) {
             this.value = value;
             this.delay = delay;
         }

         public Integer call() {
             System.out.printf("%4.1f Starting on %d for %ds%n",
                               now(), value, delay);
             try {
                 Thread.sleep(delay * 1000);
             } catch (InterruptedException ex) {
                 // Don't care right now.
             }
             System.out.printf("%4.1f Completed %d for %ds%n",
                               now(), value, delay);
             return value;
         }
     }

     static double now() {
         return (System.currentTimeMillis() - start) / 1000.0;
     }

     static final long start = System.currentTimeMillis();

     public static void main(String[] args) throws Exception {
         ExecutorService exec = Executors.newFixedThreadPool(10);
         CompletionService<Integer> ecs = new ErSoECS<>(exec, 5);

         // Check that taking when no tasks are present does not block.
         ecs.take();
         System.out.printf("%4.1f Pre-submission take has returned%n", now());

         // Submit some jobs.
         for (int i = 0; i < 20; i++) {
             final int j = i;
             ecs.submit(new MyTask(i, 2 + i % 5));
         }
         System.out.printf("%4.1f Submitted jobs%n", now());

         // Refuse to accept results for a while.
         Thread.sleep(15 * 1000);
         System.out.printf("%4.1f Starting consumption%n", now());

         // Now start consuming results.
         for (Future<Integer> f = ecs.take(); f != null; f = ecs.take()) {
             System.out.printf("%4.1f Received %d%n", now(), f.get());
         }
         
         exec.shutdown();
     }
}



Output:

  0.0 Pre-submission take has returned
  0.0 Starting on 0 for 2s
  0.0 Starting on 4 for 6s
  0.0 Starting on 3 for 5s
  0.0 Starting on 2 for 4s
  0.0 Starting on 1 for 3s
  0.0 Starting on 7 for 4s
  0.0 Starting on 6 for 3s
  0.0 Starting on 5 for 2s
  0.0 Submitted jobs
  0.0 Starting on 9 for 6s
  0.0 Starting on 8 for 5s
  2.0 Completed 0 for 2s
  2.0 Starting on 10 for 2s
  2.0 Completed 5 for 2s
  2.0 Starting on 11 for 3s
  3.0 Completed 1 for 3s
  3.0 Starting on 12 for 4s
  3.0 Completed 6 for 3s
  3.0 Starting on 13 for 5s
  4.0 Completed 2 for 4s
  4.0 Completed 10 for 2s
  4.0 Completed 7 for 4s
  5.0 Completed 3 for 5s
  5.0 Completed 11 for 3s
  5.0 Completed 8 for 5s
  6.0 Completed 4 for 6s
  6.0 Completed 9 for 6s
  7.0 Completed 12 for 4s
  8.0 Completed 13 for 5s
15.0 Starting consumption
15.0 Received 0
15.0 Received 5
15.0 Received 1
15.0 Received 6
15.0 Received 2
15.0 Received 10
15.0 Received 7
15.0 Received 3
15.0 Received 11
15.0 Received 8
15.0 Received 4
15.0 Starting on 14 for 6s
15.0 Received 9
15.0 Starting on 19 for 6s
15.0 Starting on 18 for 5s
15.0 Starting on 17 for 4s
15.0 Starting on 16 for 3s
15.0 Starting on 15 for 2s
15.0 Received 12
15.0 Received 13
17.0 Completed 15 for 2s
17.0 Received 15
18.0 Completed 16 for 3s
18.0 Received 16
19.0 Completed 17 for 4s
19.0 Received 17
20.0 Completed 18 for 5s
20.0 Received 18
21.0 Completed 14 for 6s
21.0 Received 14
21.0 Completed 19 for 6s
21.0 Received 19







-- 
ss at comp dot lancs dot ac dot uk

Back to comp.lang.java.programmer | Previous | Next — Previous in thread | Next 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