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

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

Re: To BlockingQueue or not to BlockingQueue?

Show all headers | View raw

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());
			}
		}
	}
}


Arne

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