2
\$\begingroup\$

I need to run several identical threads. All threads must wait to do their job, until all threads are running (alive). I use wait/notify to synchronize: when a thread is created it waits until it gets a notify to continue.

It works, but not sure it's the best way. Is it efficient? I read from Java 5, is it better to use concurrency package. Is it true?

public class MainTest {

    public static void main(String[] args) throws InterruptedException { 

        TestTH[] ths = new TestTH[5];

        System.out.println("== Start == ");

        for (int i = 0; i < ths.length; i++) {
            ths[i] = new TestTH("Name_" + i, new Object()); //<- Creates runnable class and lock object that will be used to synchronize 
            ths[i].start(); //<- Creates and starts thread 
        }

        Thread.sleep(100L); // <- All threads are running

        for (int i = ths.length -1 ; i >= 0; i--) {
            Object lock  = ths[i].getLock(); 
            synchronized (lock) {
                lock.notify();  //<- Thread i, can continue; 
            }
            ths[i].getThread().join(); // Wait until thread i, finishes 
        }

        System.out.println("== End == ");
    }

}

class TestTH implements Runnable {
    private String name; 
    private Object lock; 
    private Thread t; 

    public TestTH(String name_, Object lock_) { 
        this.name = name_; 
        this.lock = lock_;
        System.out.println("Create thread " + this.name);
    }

    public Object getLock() { 
        return this.lock;
    }

    public Thread getThread () {
        return t; 
    }

    public void start() {
        if (t == null) {
            t = new Thread(this, name); 
            t.start();
        }
    }

    @Override
    public void run() {
        System.out.println("Init thread  " + this.name);

        synchronized (lock) {
            try {
                lock.wait(); // <- Once started, it waits current until notify. 
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        // doStuff(); // <- do stuff when lock is released

        System.out.println("End thread " + this.name);
    } 

}

Any advice, will be very welcome.

\$\endgroup\$
6
\$\begingroup\$

Concurrent programming adds a lot of complexity only by the fact that it is concurrent - this alone is reason enough to use concurrency API. One of the rules of good programming is to not reinvent a wheel - which means

always use available libraries instead of coding yourself unless you have a very good reason to do it

This applies above all to JDK, in your case to the java.util.concurrent package.

Here is a very simple solution, which is so simple only because of concurrent package in JDK - it uses lambda expression so you'll need Java 8, but it's easy to replace is with a Runnable:

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadStartTest {

  public static void main(String[] args) {
    int count = 5;

    CountDownLatch latch = new CountDownLatch(count);
    CyclicBarrier barrier = new CyclicBarrier(count);
    ExecutorService pool = Executors.newFixedThreadPool(count);

    Random random = new Random();
    for (int i = 0; i < count; i++) {
      pool.execute(() -> {
        try {
          System.out.println("Thread starts");
          // startup
          Thread.sleep(random.nextInt(100));
          // wait for other threads
          System.out.println("Thread waits");
          barrier.await();
          // process whaterver should be processed
          System.out.println("Thread works");
          Thread.sleep(random.nextInt(1000));
          System.out.println("Thread finished");
        } catch (Exception e) {
          System.err.println("Worker thread inrerrupted");
        } finally {
          latch.countDown();
        }
      });
    }

    try {
      // wait for the threads to be done
      latch.await();
      System.out.println("== End == ");
    } catch (InterruptedException e) {
      System.err.println("Starting inrerrupted");
    }
    pool.shutdown();
  }

}

It does all you need without the need to cope with complex lock/notify sequences with possible deadlocks - just using what the language offers.

If you increase the number of threads you'll notice how the threads start to wait before all even got started, yet all wait for other before starting processing.

Explanation

First let me stress that the concurrent package has very powerful asynchronous concepts that make concurrent programming almost easy :-), so I would suggest trying to get familiar with it and refrain from complex and hard to debug state machines with lock/notify.

Now here we have some constructs from the java.util.concurrent package playing together:

ExecutorService
CyclicBarrier
CountDownLatch

Simple speaking the ExecutorService manages a pool of threads and distributes work between them – it is not guaranteed that each work unit gets a new thread though. Anyway it frees you from creating threads on your own, additionaly you can not only execute Runnable without a return value but also Callable instances, where you can get a return value using Futures. This is powerful, istn't it - submit code to be executed asynchronously and retrieve the result at some time in the future - all that in a very short and simple and expressive code.

In this case the ExecutorService gets created with Executors.newFixedThreadPool(count), so we have exactly the same number of threads as work units, which is important - more about it later.

From Javadoc:

CyclicBarrier is a synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point. CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other.

This is exactly the case here – you wish that all your threads wait for each other before continuing. The parameter to the constructor defines how many threads should wait – in this case exactly the same value as the number of started threads, so each of the threads waits for the count threads get to to that point by calling barrier.await(). After calling the barrier.await() count times the barrier and thus all the threads are released and can continue.

The CyclicBarrier can be reused, this is why it is called “Cyclic” – just call reset() and you can repeat the whole operation.

Again from the JavaDoc:

CountDownLatch is a synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.

In this case the main thread waits in latch.await until all the threads finish the work and let the latch’s counter get down to zero by calling latch.countDown().

Actually we could make a version without the CountDownLatch – which is then even simpler and uses the ExecutorService awaitTermination method – here the ExecutorService waits until all its threads are finished, so that they do not need the latch. The one drawback of this strategy is that we have to actively cope with timeouts, since the awaitTermination method requires a timeout parameter:

Version without CountDownLatch

public static void main(String[] args) {
  int count = 50;

  CyclicBarrier barrier = new CyclicBarrier(count);
  ExecutorService pool = Executors.newFixedThreadPool(count);

  Random random = new Random();
  for (int i = 0; i < count; i++) {
    pool.execute(() -> {
      try {
        System.out.println("Thread starts");
        // startup
        Thread.sleep(random.nextInt(10));
        // wait for other threads
        System.out.println("Thread waits");
        barrier.await();
        // process whaterver should be processed
        System.out.println("Thread works");
        Thread.sleep(random.nextInt(1000));
        System.out.println("Thread finished");
      } catch (Exception e) {
        System.err.println("Worker thread inrerrupted");
      }
    });
  }

  pool.shutdown();
  try {
    // wait 1 minute for the threads to be done
    pool.awaitTermination(1, TimeUnit.MINUTES);
    System.out.println("== End == ");
  } catch (InterruptedException e) {
    System.err.println("Starting inrerrupted");
  }
}

Now this solution has its own disadvantage: if the thread pool's size is smaller than the number of threads it will dead lock in barrier.await(), because all running threads are waiting for more to come yet no more threads can get executed since the pool has no more available threads. You can try it yourself:

ExecutorService pool = Executors.newFixedThreadPool(count - 1);

And it hangs!

Now I have a 3rd version for you, which doesn’t deadlock – to remove the factor limiting the number of threads I’ll give on ExecutorService up and create Threads manually, in this case I have to revert back to CountDownLatch to be able to wait for the threads to finish:

Version without ExecutorService

public static void main(String[] args) {
  int count = 50;

  CountDownLatch latch = new CountDownLatch(count);
  CyclicBarrier barrier = new CyclicBarrier(count);

  Random random = new Random();
  for (int i = 0; i < count; i++) {
    new Thread(() -> {
      try {
        System.out.println("Thread starts");
        // startup
        Thread.sleep(random.nextInt(10));
        // wait for other threads
        System.out.println("Thread waits");
        barrier.await();
        // process whaterver should be processed
        System.out.println("Thread works");
        Thread.sleep(random.nextInt(1000));
        System.out.println("Thread finished");
      } catch (Exception e) {
        System.err.println("Worker thread inrerrupted");
      } finally {
        latch.countDown();
      }
    }).start();
  }

  try {
    // wait for the threads to be done
    latch.await();
    System.out.println("== End == ");
  } catch (InterruptedException e) {
    System.err.println("Starting inrerrupted");
  }
}  

Even in this version it remains pretty short and simple thanks to lambda in Java 8.

\$\endgroup\$
  • \$\begingroup\$ Really compact. The only thing I did not get is what "latch.countDown();" in the finally block should handle. May you explain it? \$\endgroup\$ – oopexpert Jul 4 '16 at 15:22
  • 1
    \$\begingroup\$ It decreases the latch by one in a thread safe manner, the same the main thread is waiting on in latch.await(), so that it continues only after all threads are finished. You can think of it as a join() on all threads with one call, preventing the app from exiting until it is ready to do so. \$\endgroup\$ – Tomasz Stanczak Jul 4 '16 at 17:21
  • \$\begingroup\$ Can you explain a little how this works? It looks nice, but I don't understand how works. \$\endgroup\$ – Albert Jul 5 '16 at 6:30
  • \$\begingroup\$ I'll add an explanation and two more versions of the code to show the differences. \$\endgroup\$ – Tomasz Stanczak Jul 5 '16 at 9:43
  • \$\begingroup\$ I've been reading this article to understand your answer, but it doesn't explain CyclicBarrier or CountDownLatch. Thanks for your explanation. \$\endgroup\$ – Albert Jul 5 '16 at 10:25
2
\$\begingroup\$

Monitor

I would start with formulating the Monitor. The Monitor is synchronizing the things of interest:

  1. All Threads that should run must be known by the Monitor (before any Thread was started) because The Monitor should decide when all Threads are alive to give the signal to all threads to continue.
  2. Threads notify the Monitor if they are running ("notifyIAmRunning"), so the Monitor can remove the Thread from the list of not running threads. The notifying thread pu itself to "waitUntilAllOtherThreadsRunning".
  3. if the last Thread is removed the Monitor gives the signal "continue" to all Threads.

The Monitor looks like this:

public class Monitor {


    private Set<Thread> notRunningThreads;


    public Monitor() {
        this.notRunningThreads = new HashSet<>();
    }


    public synchronized void waitUntilAllOtherThreadsRunning() {

        while (notRunningThreads.size() > 0) {
            try {

                this.wait();

            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

    }


    public synchronized void notifyIAmRunning() {

        this.notRunningThreads.remove(Thread.currentThread());

        if (this.notRunningThreads.isEmpty()) {
            this.notifyAll();
        }

    }


    public void registerThreadToBeRunning(TestTH myThread) {
        this.notRunningThreads.add(myThread);
    }


}

Thread

Of course you can distinguish between "Runnable" and "Thread". But you should not instantiate a thread within its runnable. The whole concept is compromised as you introduce a bi-directional dependency between those classes. A Thread my know a Runnable. But the Runnable shouldn't know the Thread it is running in.

For simplicity I left away the Runnable in my suggestion. But it would be cleaner to distinguish beetwen "code to run" and "thread to run the code in".

public class TestTH extends Thread {


    private Monitor monitor;


    public TestTH(String name, Monitor monitor) {
        super(name);
        this.monitor = monitor;
        System.out.println("Create thread " + this.getName());
    }


    @Override
    public void run() {

        System.out.println("Init thread " + this.getName());

        monitor.notifyIAmRunning();
        monitor.waitUntilAllOtherThreadsRunning();

        System.out.println("End thread " + this.getName());

    }

}

Controller

Now following this structure your Main-class (I named it ThreadController) looks like this:

public class ThreadController {


    public static void main(String[] args) {

        new ThreadController().start();

    }


    public ThreadController() {
    }


    public void start() {

        Monitor monitor = new Monitor();

        System.out.println("== Start == ");

        TestTH[] ths = createThreads(monitor);

        registerThreadsInMonitor(monitor, ths);
        startThreads(ths);
        waitForThreadsEnding(ths);

        System.out.println("== End == ");

    }


    private void waitForThreadsEnding(TestTH[] ths) {
        for (int i = ths.length -1 ; i >= 0; i--) {
            try {
                ths[i].join();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } // Wait until thread i, finishes 
        }
    }


    private void startThreads(TestTH[] ths) {
        for (int i = 0; i < ths.length; i++) {
            ths[i].start(); //<- Creates and starts thread 
        }
    }


    private void registerThreadsInMonitor(Monitor monitor, TestTH[] ths) {
        for (int i = 0; i < ths.length; i++) {
            monitor.registerThreadToBeRunning(ths[i]);
        }
    }


    private TestTH[] createThreads(Monitor monitor) {
        TestTH[] ths = new TestTH[5];

        for (int i = 0; i < ths.length; i++) {
            ths[i] = new TestTH("Name_" + i, monitor); //<- Creates runnable class and lock object that will be used to synchronize 
        }
        return ths;
    }


}

Conclusion

You do not always need a library X or a special API. Simple language mechanics and expressive names for methods and classes will mostly do the job as well.

Of course the need for the concurrency API rises by the "complexity" of your requirements. In this case I do not see any need.

\$\endgroup\$
  • \$\begingroup\$ Nice. It's more or less what I did, but much better organized. I'll try to adapt to my needs. Thanks. \$\endgroup\$ – Albert Jul 5 '16 at 6:35

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.