Path: csiph.com!news.swapon.de!aioe.org!.POSTED!not-for-mail
From: =?UTF-8?Q?Arne_Vajh=c3=b8j?= <arne@vajhoej.dk>
Newsgroups: comp.lang.java.programmer
Subject: Re: To BlockingQueue or not to BlockingQueue?
Date: Sat, 14 Nov 2015 19:13:29 -0500
Organization: Aioe.org NNTP Server
Lines: 102
Message-ID: <n28ind$ep7$1@speranza.aioe.org>
References: <n24r35$f4s$1@dont-email.me> <n266vk$g9n$1@speranza.aioe.org> <n28c6k$act$1@dont-email.me>
NNTP-Posting-Host: 1lM8B9/qdtKBc/uIRK2Gjw.user.speranza.aioe.org
Mime-Version: 1.0
Content-Type: text/plain; charset=utf-8; format=flowed
Content-Transfer-Encoding: 8bit
X-Complaints-To: abuse@aioe.org
User-Agent: Mozilla/5.0 (Windows NT 6.1; WOW64; rv:38.0) Gecko/20100101 Thunderbird/38.3.0
X-Notice: Filtered by postfilter v. 0.8.2
Xref: csiph.com comp.lang.java.programmer:33131

On 11/14/2015 5:24 PM, Eric Sosman wrote:
> On 11/13/2015 9:40 PM, Arne Vajhøj wrote:
>> 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<Result> results;
>>      private ConcurrentLinkedQueue<Token> tokens;
>>      public ErSoQueueDQImpl(int noworkers, int maxpending) {
>>          es = Executors.newFixedThreadPool(noworkers);
>>          results = new LinkedBlockingQueue<Result>(maxpending);
>>          tokens = new ConcurrentLinkedQueue<Token>();
>>      }
>>      @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.

Oops. That should be a blocking put.

I am assuming that it is a requirement to identify not completed tasks.

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

There were submitted a token back when the task was submitted.

Due to interrupt the real response will not show up, so
we send am interrupted result.

But there should still be a 1:1 correlation between tokens
and responses.

Unless I have missed something. Which could obvious be
the case.

>      By the way, couldn't the queue of Tokens just be a Semaphore?

Probably yes.

Arne