Path: csiph.com!eternal-september.org!feeder.eternal-september.org!mx02.eternal-september.org!.POSTED!not-for-mail From: Steven Simpson Newsgroups: comp.lang.java.programmer Subject: Re: To BlockingQueue or not to BlockingQueue? Date: Sun, 15 Nov 2015 10:37:41 +0000 Organization: A noiseless patient Spider Lines: 266 Message-ID: References: Mime-Version: 1.0 Content-Type: text/plain; charset=utf-8; format=flowed Content-Transfer-Encoding: 7bit Injection-Info: mx02.eternal-september.org; posting-host="09ca1bb8bf88c847a16f76dfef96405f"; logging-data="21760"; mail-complaints-to="abuse@eternal-september.org"; posting-account="U2FsdGVkX18WH1qJCuItAoT/asewhbm+Zd+6TU77z+Q=" User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:38.0) Gecko/20100101 Thunderbird/38.3.0 In-Reply-To: Cancel-Lock: sha1:WOYmkFj4Fbz8X8nILPJ5h4SaVL8= Xref: csiph.com comp.lang.java.programmer:33132 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 implements CompletionService { private final Semaphore sem; private final ExecutorCompletionService ecs; private int submitted = 0; public ErSoECS(Executor e, int permits) { this.sem = new Semaphore(permits); this.ecs = new ExecutorCompletionService<>(e); } public Future submit(Callable task) { synchronized (this) { submitted++; return ecs.submit(new Wrapper(task)); } } public Future submit(Runnable task, T result) { return submit(() -> { task.run(); return result; }); } public Future take() throws InterruptedException { synchronized (this) { if (submitted == 0) return null; Future result = ecs.take(); submitted--; sem.release(); return result; } } public Future poll() { synchronized (this) { Future result = ecs.poll(); if (result != null) { submitted--; sem.release(); } return result; } } public Future poll(long timeout, TimeUnit unit) throws InterruptedException { synchronized (this) { if (submitted == 0) return null; Future result = ecs.poll(timeout, unit); if (result != null) { submitted--; sem.release(); } return result; } } class Wrapper implements Callable { private final Callable base; public Wrapper(Callable 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 { 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 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 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