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

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

Re: To BlockingQueue or not to BlockingQueue?

Show all headers | View raw

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<Result>{
>          // 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<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.
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

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