Path: csiph.com!eternal-september.org!feeder.eternal-september.org!mx02.eternal-september.org!.POSTED!not-for-mail From: Eric Sosman Newsgroups: comp.lang.java.programmer Subject: Re: To BlockingQueue or not to BlockingQueue? Date: Sat, 14 Nov 2015 17:24:20 -0500 Organization: A noiseless patient Spider Lines: 189 Message-ID: References: Mime-Version: 1.0 Content-Type: text/plain; charset=utf-8; format=flowed Content-Transfer-Encoding: 8bit Injection-Date: Sat, 14 Nov 2015 22:22:12 -0000 (UTC) Injection-Info: mx02.eternal-september.org; posting-host="858a9d65d95f4ea4d355bd116aa2cdbf"; logging-data="10653"; mail-complaints-to="abuse@eternal-september.org"; posting-account="U2FsdGVkX19cFnQ7lv0j/2KyQwIc+7PE" User-Agent: Mozilla/5.0 (Windows NT 6.1; WOW64; rv:38.0) Gecko/20100101 Thunderbird/38.3.0 In-Reply-To: Cancel-Lock: sha1:QQZwJvAqVbs/zgR9B/M4WwTgL5g= Xref: csiph.com comp.lang.java.programmer:33130 On 11/13/2015 9:40 PM, Arne Vajhøj wrote: > On 11/13/2015 9:13 AM, 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. >> >> 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? > > I am sure that there are multiple ways to attack this problem. > > My initial thought is that you need a queue that is blocking and > also non-blocking, so you really need two queues. > > That thought resulted in the following code: > > import java.util.concurrent.Callable; > > public interface ErSoQueue { > public static class Work implements Callable{ > // dummy implementation > @Override > public Result call() throws Exception { > Thread.sleep(1000); > return new Result(true); > } > } > public static class Result { > // dummy implementation > private boolean status; > public Result(boolean status) { > this.status = status; > } > public boolean isStatus() { > return status; > } > } > public static class ErSoException extends Exception { > private static final long serialVersionUID = 1L; > public ErSoException(String msg) { > super(msg); > } > public ErSoException(Exception ex) { > super(ex); > } > } > public void submitWork(Work w); > public Result getResult() throws ErSoException; > public void stop(boolean wait); > } > > and: > > import java.util.concurrent.BlockingQueue; > import java.util.concurrent.ConcurrentLinkedQueue; > import java.util.concurrent.ExecutorService; > import java.util.concurrent.Executors; > import java.util.concurrent.LinkedBlockingQueue; > > public class ErSoQueueDQImpl implements ErSoQueue { > private static class Token { > } > private static class InterruptedResult extends Result { > public InterruptedResult() { > super(false); > } > } > private static class ErSoExceptionNothing extends ErSoException { > private static final long serialVersionUID = 1L; > public ErSoExceptionNothing() { > super("Nothing to wait for"); > } > } > private ExecutorService es; > private BlockingQueue results; > private ConcurrentLinkedQueue tokens; > public ErSoQueueDQImpl(int noworkers, int maxpending) { > es = Executors.newFixedThreadPool(noworkers); > results = new LinkedBlockingQueue(maxpending); > tokens = new ConcurrentLinkedQueue(); > } > @Override > public void submitWork(final Work w) { > tokens.add(new Token()); > es.submit(new Runnable() { > @Override > public void run() { > try { > results.put(w.call()); > } catch (InterruptedException e) { > // this is expected in case of stop(false) > results.add(new InterruptedResult()); > } catch (Exception e) { > // not specified what action should be done here > e.printStackTrace(); > } > } > }); > } > @Override > public Result getResult() throws ErSoException { > try { > if(tokens.poll() != null) { > return results.take(); > } else { > throw new ErSoExceptionNothing(); > } > } catch (InterruptedException e) { > throw new ErSoException(e); > } > } > @Override > public void stop(boolean wait) { > if(wait) { > es.shutdown(); > } else { > for(@SuppressWarnings("unused") Runnable r : > es.shutdownNow()) { > results.add(new InterruptedResult()); > } > } > } > } Thank you, Arne. I may have misunderstood your sketch, but something seems amiss inside the Runnable in submitWork(). If results.put() throws InterruptedException while waiting for the results queue to become non-full, the following results.add() probably won't have any better luck and will throw IllegalStateException. That's all right from the perspective of the submitWork() caller, but look: The results queue is now out of step with the tokens queue. I don't think this can be fixed by plucking the extra token out of its queue again in the catch block, because some other thread may already have removed it with a call to getResult() ... I'll study this some more. By the way, couldn't the queue of Tokens just be a Semaphore? Thanks again for your help. -- esosman@comcast-dot-net.invalid "Don't be afraid of work. Make work afraid of you." -- TLM