How to wait for all threads to complete

Question

I created some workflow how to wait for all thread which I created. This example works in 99 % of cases but sometimes method waitForAllDone is finished sooner then all thread are completed. I know it because after waitForAllDone I am closing stream which is using created thread so then occurs exception

Caused by: java.io.IOException: Stream closed

my thread start with:

  @Override
  public void run() {
    try {
      process();
    } finally {
      Factory.close(this);
    }
  }

closing:

  protected static void close(final Client client) {
    clientCount--;
  }

when I creating thread I call this:

  public RobWSClient getClient() {
    clientCount++;
    return new Client();
  }

and clientCount variable inside factory:

  private static volatile int clientCount = 0;

wait:

  public void waitForAllDone() {
    try {
      while (clientCount > 0) {
        Thread.sleep(10);
      }

    } catch (InterruptedException e) {
      LOG.error("Error", e);
    }
  }

Check out the doc of Semaphores in java. The example is nearly the same, what you need: http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html — Jordi Laforge, Nov 20 '12 at 12:28

John B · Accepted Answer · 2012-11-20T12:27:06.603

5

You need to protect the modification and reading of clientCount via synchronized. The main issue is that clientCount-- and clientCount++ are NOT an atomic operation and therefore two threads could execute clientCount-- / clientCount++ and end up with the wrong result.

Simply using volatile as you do above would ONLY work if ALL operations on the field were atomic. Since they are not, you need to use some locking mechanism. As Anton states, AtomicInteger is an excellent choice here. Note that it should be either final or volatile to ensure it is not thread-local.

That being said, the general rule post Java 1.5 is to use a ExecutorService instead of Threads. Using this in conjuction with Guava's Futures class could make waiting for all to complete to be as simple as:

Future<List<?>> future = Futures.successfulAsList(myFutureList);
future.get();
// all processes are complete

Futures.successfulAsList

edited Nov 20 '12 at 12:27

answered Nov 20 '12 at 12:21

John B

32,493
6
77
98

but I have just one thread which is creating other threads. So who else can inc this value ? – hudi Nov 20 '12 at 12:33
You can increment the value from any number of threads you want, so long as you put the appropriate protections in place. – John B Nov 20 '12 at 13:01
It is challenging to write unit tests to test threading. Usually you write a test and they cause it to repeat a large number of times. You can also add a library which will cause random thread context switching to help aid the discovery of thread bugs. See the concurrency section of `Clean Code` for a discussion of this. – John B Nov 20 '12 at 13:05

score 3 · Answer 2 · answered Nov 20 '12 at 12:22

3

I'm not sure that the rest of your your code has no issues, but you can't increment volatile variable like this - clientCount++; Use AtomicInteger instead

answered Nov 20 '12 at 12:22

Anton

5,831
3
35
45

why I cant increment it. Can you post some documentation ? – hudi Nov 20 '12 at 12:23
Cause it not atomic operation. During A = A + 1 value of A could have changed by another thread – Anton Nov 20 '12 at 12:24
4

@hudi consider ordering http://www.amazon.com/Java-Concurrency-Practice-Brian-Goetz/dp/0321349601 from amazon - it's nearly impossible to write a thread-safe code with the trial and error approach – Boris Treukhov Nov 20 '12 at 12:26
but I have just one thread which is creating other threads. So who else can inc this value ? – hudi Nov 20 '12 at 12:33

score 1 · Answer 3 · answered Nov 20 '12 at 12:52

The best way to wait for threads to terminate, is to use one of the high-level concurrency facilities. In this case, the easiest way would be to use an ExecutorService.

You would 'offer' a new task to the executor in this way:

...
ExecutorService executor = Executors.newFixedThreadPool(POOL_SIZE);
...

Client client = getClient(); //assuming Client implements runnable
executor.submit(client);
...

public void waitForAllDone() {
    executor.awaitTermination(30, TimeUnit.SECOND) ; wait termination of all threads for 30 secs
... 
}

In this way, you don't waste valuable CPU cycles in busy waits or sleep/awake cycles. See ExecutorService docs for details.

How to wait for all threads to complete

3 Answers3

Linked