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

Groups > comp.lang.java.programmer > #33121 > unrolled thread

To BlockingQueue or not to BlockingQueue?

Back to article view | Back to comp.lang.java.programmer

Contents

  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

#33121 — To BlockingQueue or not to BlockingQueue?

     Short version: I wrote a simple submit-tasks-and-reap-results
class based on a ThreadPoolExecutor.  Initially it seemed I could
use an ExecutorCompletionService to manage the flow of work, but
that didn't pan out.  Then I tried having the tasks place their own
results on a BlockingQueue explicitly, but that didn't pan out either.
Eventually, I retrogressed all the way to using an ordinary Queue and
an old-fashioned synchronized/wait/notifyAll scheme.  I'm hoping for
some hints about how to use the java.util.concurrent classes better,
since they're more modern and fashionable.

     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?

-- 
esosman@comcast-dot-net.invalid
"Don't be afraid of work. Make work afraid of you." -- TLM

[toc] | [next] | [standalone]

#33122

On Friday, November 13, 2015 at 9:14:24 AM UTC-5, Eric Sosman wrote:
> Short version: I wrote a simple submit-tasks-and-reap-results
> class based on a ThreadPoolExecutor.  Initially it seemed I could
> use an ExecutorCompletionService to manage the flow of work, but
> that didn't pan out.  Then I tried having the tasks place their own
> results on a BlockingQueue explicitly, but that didn't pan out either.
> Eventually, I retrogressed all the way to using an ordinary Queue and
> an old-fashioned synchronized/wait/notifyAll scheme.  I'm hoping for
> some hints about how to use the java.util.concurrent classes better,
> since they're more modern and fashionable.

I'm certainly no expert on concurrency, and the only queue I've had to reference was EventQueue.invokeLater() to kick off a GUI screen (appears to do the same as SwingUtilities.invokeLater()?), but I did recently write a multithreaded method if it helps.

I needed to call a method from the Google API, which goes out to the web so it could take a minute, without the GUI thread showing unresponsive.  So I created a Semaphore and assigned it 1 permit, then acquired a permit, then created a SwingWorker which creates a new thread on the doInBackground() which releases a permit when it's done.  Then I call the worker.execute(), then the semaphore.tryAcquire(), then the worker.get().  I don't know if this is proper or most efficient but it appears to be working.  You can set a timeout on the acquire methods, and they throw InterruptedException if your current thread is aborted.

[toc] | [prev] | [next] | [standalone]

#33123

On 11/13/2015 10:29 AM, Eric Douglas wrote:
> On Friday, November 13, 2015 at 9:14:24 AM UTC-5, Eric Sosman wrote:
>> Short version: I wrote a simple submit-tasks-and-reap-results
>> class based on a ThreadPoolExecutor.  Initially it seemed I could
>> use an ExecutorCompletionService to manage the flow of work, but
>> that didn't pan out.  Then I tried having the tasks place their own
>> results on a BlockingQueue explicitly, but that didn't pan out either.
>> Eventually, I retrogressed all the way to using an ordinary Queue and
>> an old-fashioned synchronized/wait/notifyAll scheme.  I'm hoping for
>> some hints about how to use the java.util.concurrent classes better,
>> since they're more modern and fashionable.
>
> I'm certainly no expert on concurrency, and the only queue I've had to reference was EventQueue.invokeLater() to kick off a GUI screen (appears to do the same as SwingUtilities.invokeLater()?), but I did recently write a multithreaded method if it helps.
>
> I needed to call a method from the Google API, which goes out to the web so it could take a minute, without the GUI thread showing unresponsive.  So I created a Semaphore and assigned it 1 permit, then acquired a permit, then created a SwingWorker which creates a new thread on the doInBackground() which releases a permit when it's done.  Then I call the worker.execute(), then the semaphore.tryAcquire(), then the worker.get().  I don't know if this is proper or most efficient but it appears to be working.  You can set a timeout on the acquire methods, and they throw InterruptedException if your current thread is aborted.

     Thanks for your ideas.  I don't think they help my particular
situation, but I appreciate the input.

     FWIW, I think your Semaphore might be unnecessary. SwingWorker
has a done() method that is called (on the EDT) when the work finishes,
and you can call get() from inside that method -- but then, I may not
have understood exactly what you're doing.

     Thanks again!

-- 
esosman@comcast-dot-net.invalid
"Don't be afraid of work. Make work afraid of you." -- TLM

[toc] | [prev] | [next] | [standalone]

#33124

On Friday, November 13, 2015 at 10:41:38 AM UTC-5, Eric Sosman wrote:
> On 11/13/2015 10:29 AM, Eric Douglas wrote:
> > On Friday, November 13, 2015 at 9:14:24 AM UTC-5, Eric Sosman wrote:
> >> Short version: I wrote a simple submit-tasks-and-reap-results
> >> class based on a ThreadPoolExecutor.  Initially it seemed I could
> >> use an ExecutorCompletionService to manage the flow of work, but
> >> that didn't pan out.  Then I tried having the tasks place their own
> >> results on a BlockingQueue explicitly, but that didn't pan out either.
> >> Eventually, I retrogressed all the way to using an ordinary Queue and
> >> an old-fashioned synchronized/wait/notifyAll scheme.  I'm hoping for
> >> some hints about how to use the java.util.concurrent classes better,
> >> since they're more modern and fashionable.
> >
> > I'm certainly no expert on concurrency, and the only queue I've had to reference was EventQueue.invokeLater() to kick off a GUI screen (appears to do the same as SwingUtilities.invokeLater()?), but I did recently write a multithreaded method if it helps.
> >
> > I needed to call a method from the Google API, which goes out to the web so it could take a minute, without the GUI thread showing unresponsive.  So I created a Semaphore and assigned it 1 permit, then acquired a permit, then created a SwingWorker which creates a new thread on the doInBackground() which releases a permit when it's done.  Then I call the worker.execute(), then the semaphore.tryAcquire(), then the worker.get().  I don't know if this is proper or most efficient but it appears to be working.  You can set a timeout on the acquire methods, and they throw InterruptedException if your current thread is aborted.
> 
>      Thanks for your ideas.  I don't think they help my particular
> situation, but I appreciate the input.
> 
>      FWIW, I think your Semaphore might be unnecessary. SwingWorker
> has a done() method that is called (on the EDT) when the work finishes,
> and you can call get() from inside that method -- but then, I may not
> have understood exactly what you're doing.

done() is if you need that SwingWorker to directly access a component to do something on the EDT.  I just needed to run a background method from an application that has GUI components on the EDT and have it wait for results to pass back to the application without locking the EDT.  The result I'm returning from the get() doesn't have anything to do with the GUI, at least not directly.  The application is locked for the user before calling this method, with a popup window which has a message and an indeterminate JProgressBar.  I just don't want Windows to show the window as unresponsive.

In this particular case all my threads are from a single method (I also create a Timer thread within the doInBackground thread for a retry) but it would seem similar logic should work for a service handling multiple object instances requesting values.

[toc] | [prev] | [next] | [standalone]

#33125

In article <n24r35$f4s$1@dont-email.me>,
 Eric Sosman <esosman@comcast-dot-net.invalid> wrote:

>      Short version: I wrote a simple submit-tasks-and-reap-results
> class based on a ThreadPoolExecutor.  Initially it seemed I could
> use an ExecutorCompletionService to manage the flow of work, but
> that didn't pan out.  Then I tried having the tasks place their own
> results on a BlockingQueue explicitly, but that didn't pan out either.
> Eventually, I retrogressed all the way to using an ordinary Queue and
> an old-fashioned synchronized/wait/notifyAll scheme.  I'm hoping for
> some hints about how to use the java.util.concurrent classes better,
> since they're more modern and fashionable.
> 
>      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?

The good old synchronization block shouldn't be avoided just because 
it's not trendy.  When making complex decisions or performing actions 
that can't easily be undone, blocking is often the better choice.

Your problem here is the need for #3 and some sucky Sun code.  The Java 
queue/dequeue interfaces completely lack signaling.  There's no error 
signal, no end of stream signal, and no abort signal.  Attempting to 
pass those signals outside of the queue/dequeue results in race 
conditions.

-- 
I will not see posts from astraweb, theremailer, dizum, or google
because they host Usenet flooders.

[toc] | [prev] | [next] | [standalone]

#33129

On 11/13/2015 7:56 PM, Kevin McMurtrie wrote:
> In article <n24r35$f4s$1@dont-email.me>,
>   Eric Sosman <esosman@comcast-dot-net.invalid> wrote:
>> [...]
>>       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?
>
> The good old synchronization block shouldn't be avoided just because
> it's not trendy.  When making complex decisions or performing actions
> that can't easily be undone, blocking is often the better choice.
>
> Your problem here is the need for #3 and some sucky Sun code.  The Java
> queue/dequeue interfaces completely lack signaling.  There's no error
> signal, no end of stream signal, and no abort signal.  Attempting to
> pass those signals outside of the queue/dequeue results in race
> conditions.

     Thanks, Kevin.  My references to using "old hat" approaches
were a bit tongue-in-cheek, poking fun at my own inability to find
a java.util.concurrent solution to what seemed a simple problem.
If you, too, see no solution, then I feel less inadequate.  ;)

     "Sometimes the old ways are best."

-- 
esosman@comcast-dot-net.invalid
"Don't be afraid of work. Make work afraid of you." -- TLM

[toc] | [prev] | [next] | [standalone]

#33126

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

[toc] | [prev] | [next] | [standalone]

#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<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

[toc] | [prev] | [next] | [standalone]

#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

[toc] | [prev] | [next] | [standalone]

#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<T> implements CompletionService<T> {
     private final Semaphore sem;
     private final ExecutorCompletionService<T> ecs;

     private int submitted = 0;

     public ErSoECS(Executor e, int permits) {
         this.sem = new Semaphore(permits);
         this.ecs = new ExecutorCompletionService<>(e);
     }

     public Future<T> submit(Callable<T> task) {
         synchronized (this) {
             submitted++;
             return ecs.submit(new Wrapper(task));
         }
     }

     public Future<T> submit(Runnable task, T result) {
         return submit(() -> { task.run(); return result; });
     }

     public Future<T> take() throws InterruptedException {
         synchronized (this) {
             if (submitted == 0) return null;
             Future<T> result = ecs.take();
             submitted--;
             sem.release();
             return result;
         }
     }

     public Future<T> poll() {
         synchronized (this) {
             Future<T> result = ecs.poll();
             if (result != null) {
                 submitted--;
                 sem.release();
             }
             return result;
         }
     }

     public Future<T> poll(long timeout, TimeUnit unit)
         throws InterruptedException {
         synchronized (this) {
             if (submitted == 0) return null;
             Future<T> result = ecs.poll(timeout, unit);
             if (result != null) {
                 submitted--;
                 sem.release();
             }
             return result;
         }
     }

     class Wrapper implements Callable<T> {
         private final Callable<T> base;

         public Wrapper(Callable<T> 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<Integer> {
         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<Integer> 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<Integer> 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

[toc] | [prev] | [next] | [standalone]

#33134

Am 13.11.2015 15:13, schrieb Eric Sosman:
>
>      Short version: I wrote a simple submit-tasks-and-reap-results
> class based on a ThreadPoolExecutor.  Initially it seemed I could
> use an ExecutorCompletionService to manage the flow of work, but
> that didn't pan out.  Then I tried having the tasks place their own
> results on a BlockingQueue explicitly, but that didn't pan out either.
> Eventually, I retrogressed all the way to using an ordinary Queue and
> an old-fashioned synchronized/wait/notifyAll scheme.  I'm hoping for
> some hints about how to use the java.util.concurrent classes better,
> since they're more modern and fashionable.
>
>      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?
>
Here's another idea: Put CompletableFutures in a BlockingQueue before 
you start the computation. This takes care of both #1 and #2. You don't 
need to do any sort of bookkeeping with a Semaphore. When the queue is 
full with unretrieved futures, no more results get added and hence no 
more computations get triggered. To address #3, you could shutdown the 
thread pool supplied from the outside.

The code below would make the task producers block when the workers have 
no queue space left to deposit unretrieved results, and the result 
receivers when they have retrieved a result (Future) that is not yet 
completed. Don't know if you can accept that. You speak of making the 
workers pause if there are too many unretrieved results, but that 
doesn't really make sense to me. If tasks can keep coming in, they'd 
just pile up in an unbounded queue.

To me, it seems that the clients should adjust their submit/receive 
rate, or that the thread pool ought to be tweaked to deliver results at 
a lower rate. You might use add() instead of put() and have the client 
deal with the IllegalArgumentException instead of being blocked until 
another client retrieves a result. This would avoid the possible 
deadlock when all clients continue to submit tasks and never retrieve them.

You don't say if your tasks really need to be able to throw exceptions. 
If yes, with CompletableFuture you need to adapt your Callables to 
Suppliers. If not, you can skip that and just pass in Suppliers.

Here's some code (untested, except with a variant of Steven Simpson's 
test code that submits jobs in an extra thread).

public class EricsCompletor<T> {

     private final BlockingQueue<Future<T>> results;
     private final ExecutorService pool;

     public EricsCompletor(ExecutorService pool,
             int resultBufferSize) {
         this.pool = pool;
         this.results = new ArrayBlockingQueue<>(
                 resultBufferSize);
     }

     public CompletableFuture<T> submit(Callable<T> task)
             throws InterruptedException {
         return submit(toSupplier(task));
     }

     /**
      * Submits a task for asynchronous execution. Blocks if
      * there are too many unretrieved results.
      *
      * @param task computation to be executed
      * @return a future representing the result of executing
      *         the task
      * @throws InterruptedException if interrupted while
      *             waiting to queue the task
      */
     public CompletableFuture<T> submit(Supplier<T> task)
             throws InterruptedException {
         CompletableFuture<T> future = new CompletableFuture<>();
         results.put(future); // block when full. consider add()

         CompletableFuture<T> cf = CompletableFuture
                 .supplyAsync(task, pool); // submit task
         cf.whenComplete((v, ex) -> { // update result
             if (ex == null) future.complete(v);
             else future.completeExceptionally(ex);
         });

         // propagate cancel if returned future is cancelled
         future.whenComplete((v, ex) -> cf.cancel(true));
         return future;
     }

     /**
      * Retrieves and removes the Future representing the next
      * completed task or null if none are present.
      */
     public Future<T> poll() {
         return results.poll();
     }

     public boolean cannotSubmit() {
         return results.remainingCapacity() == 0;
     }

     // ---------- helpers

     /**
      * Exception hiding for Suppliers passed to
      * CompletableFutures, because these suppliers cannot
      * throw a checked exception (cf. {@link Supplier#get()}
      * ).
      *
      * @param callable exception-throwing code
      * @return supplier that returns the result of the
      *         callable and declares no exception. Any
      *         exception thrown by the Callable will be
      *         rethrown wrapped in a CompletionException
      * @throws CompletionException an exception with any
      *             Throwable thrown from callable as its cause
      */
     static <T> Supplier<T> toSupplier(Callable<T> callable) {
         return () -> {
             try {
                 return callable.call();
             }
             catch (Throwable t) {
                 throw new CompletionException(t);
             }
         };
     }

}

[toc] | [prev] | [standalone]

Back to top | Article view | comp.lang.java.programmer

csiph-web