Enforce printing sequence using wait() and notify()

Question

This is a follow-up answer "implementation" to a question I posted on SO. For the sake of convenience, I will repeat my intent: I want one thread (called sub-thread) to print 10 times under outer-loop with 2 iterations; then another (boss-thread) to print 100 times under outer-loop with 2 iterations provided that sub-thread goes first. It will look something like this:

Sub Thread- iter = 1
Sub Thread- iter = 2
...
Sub Thread- iter = 10
Boss Thread- iter = 1
Boss Thread- iter = 2
...
Boss Thread- iter = 100

This sub-thread and boss-thread printing sequence will continue for 2 times (outer-loop).

Here's my code:

public class InterThCom {
    // flag default to false for checking if sub-thread 
    // gets the lock first
    private boolean isTh2RunFirst = false; 

    public static void main(String[] args) {
        InterThCom itc = new InterThCom(); 

        Thread t1 = new Thread(itc.new Th1(), "Boss-thread-"); 
        Thread t2 = new Thread(itc.new Th2(), "Sub-thread-");

        t1.start();
        t2.start();
    }

    private class Th1 implements Runnable {

        @Override
        public void run() { 
            for (int i = 0; i < 2; i++) { 
                synchronized (InterThCom.class) { // lock up inner-loop

                    // boss-thread gets the lock first 
                    // wait for sub-thread and let it run;
                    // otherwise, skip this check
                    if (isTh2RunFirst == false) {
                        // wait for sub-thread, if boss-thread gets the lock first 
                        try {
                            InterThCom.class.wait();
                        } catch (InterruptedException e1) { 
                            e1.printStackTrace();
                        }
                    } 

                    // print iteration 100 times 
                    for (int j = 0; j < 100; j++) {
                        System.out.println(Thread.currentThread().getName() + " iter-" + (j + 1));
                    }
                    // done printing 100 times

                    // sub-thread should run already at this point 
                    isTh2RunFirst = true;

                    // This print helps split boss-th and sub-th prints
                    System.out.println(Thread.currentThread().getName() + " outer-loop iter:" + (i + 1));

                    // wake up sub-thread 
                    InterThCom.class.notify();

                    // wait for sub-thread 
                    try {
                        InterThCom.class.wait();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } 
                } 
            }
        } 
    }

    private class Th2 implements Runnable {

        @Override
        public void run() {
            for (int i = 0; i < 2; i++) { 
                synchronized (InterThCom.class) {
                    // print iteration 10 times 
                    for (int j = 0; j < 10; j++) {
                        System.out.println(Thread.currentThread().getName() + " iter-" + (j + 1));
                    }
                    // done printing 10 times

                    // sub-thread already prints j iteration
                    isTh2RunFirst = true; 

                    // This print helps split boss-th and sub-th prints
                    System.out.println(Thread.currentThread().getName() + " outer-loop iter:" + (i + 1));

                    // wake up boss-thread 
                    InterThCom.class.notify();

                    // wait for boss-thread 
                    try {
                        InterThCom.class.wait();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } 
                } 
            }
            synchronized (InterThCom.class) {
                // boss-thread is waiting at the last iteration, so wake it up
                InterThCom.class.notify();
            } 
        } 
    }

} 

Things I would like help with:

• Did I use "synchronized" block in an efficient way that aligns with conventional approach?

• Are there other locking approach that will make my code less cluttered and cleaner?

  • My initial thought was using a separate class called PrintStmt to wrap all the statements inside the run and then called it in the run method then lock the invocation. That way, run only has the invocation and the lock.

• Also, my wait and notify pairs are all over, is there a better way to "organize" them in a way that looks better? E.g. one of my notify call is outside of the double for-loop in the sub-thread Th2 class. This is an edge case but I am having trouble to integrate that inside the double-loops.

• I am new to multi-threading. So, I am grateful for any other addresses and corrections to my implementation of the inter-communication for two threads with some extra requirement. Or other suggestions on implementing thread-communication will be appreciated.

Answer 1

Multi-threading isn't easy, and the original Java constructs - threads and monitors - provided only limited support.

These days if I have to do significant management of threads, such as making them wait on each other, I am most likely to make use of the more modern java.util.concurrent constructs, such as latches and barriers.

However, this task can reasonably be achieved using the traditional methods, so let's discuss some of the issues with your approach.

1. Synchronizing on classes is not normally what you want to do. You should synchronize on objects, and a lot of the time it is simpler and clearer to have separate objects for separate synchronizations. In your code, you use the same "object" - InterThCom.class - both for the boss to wait for the sub and for the sub to wait for the boss. This is, in my view, not an ideal approach.

2. InterruptedException on wait() shouldn't be handled simply by printing a stack trace and proceeding as if you've been notified. Unless you are using Thread.interrupt() as an out-of-band signalling mechanism, you should probably retry the wait. A common approach is to have a separate boolean to indicate that the wait()ed for event has occurred.

3. The "isTh2RunFirst" field is, in my view, not clear in its purpose. There are probably clearer approaches - if you use more than one monitor, then this need for this field should disappear.

4. Java style is to use long, clear names. InterThCom is perhaps better expressed as InterThreadCommunicator (or something like that), etc...

5. My personal preference is to keep synchronized blocks short.

6. I think it's clearer if the purpose of the synchronization is more explicit, so I'm inclined to use methods with meaningful names. You have operations like "allow sub to proceed", "wait for sub to finish work", and so on, so why not extract them into little methods?

Here's your code reworked a little in line with the points above:

public class InterThreadCommunicator {

  public static void main(String[] args) {
    InterThreadCommunicator interThreadCommunicator = new InterThreadCommunicator();

    interThreadCommunicator.runTheThreads();
  }

  private void runTheThreads() {

    BossRunnable bossRunnable = new BossRunnable();
    Thread bossThread = new Thread(bossRunnable, "Boss-thread-");
    SubRunnable subRunnable = new SubRunnable(bossRunnable);
    Thread subThead = new Thread(subRunnable, "Sub-thread-");
    bossRunnable.setSub(subRunnable);

    bossThread.start();
    subThead.start();
  }

  private class BossRunnable implements Runnable {

    private SubRunnable subRunnable;

    private boolean subThreadComplete = false;

    @Override
    public void run() {
      for (int i = 0; i < 2; i++) {

        subRunnable.allowToProceed();

        waitForSubThreadToCompleteLoop();

        // print iteration 100 times
        for (int j = 0; j < 100; j++) {
          System.out.println(Thread.currentThread().getName() + " iter-" + (j + 1));
        }
        // done printing 100 times

        // This print helps split boss-th and sub-th prints
        System.out.println(Thread.currentThread().getName() + " outer-loop iter:" + (i + 1));
      }

      subRunnable.terminate();
    }

    private synchronized void waitForSubThreadToCompleteLoop() {
      while (!subThreadComplete) {
        try {
          wait();
        }
        catch (InterruptedException e) {
          // ignore this, and retry the wait
        }
      }
      subThreadComplete = false; // reset the latch
    }

    private synchronized void notifySubThreadCompletion() {
      subThreadComplete = true;
      notify();
    }

    public void setSub(SubRunnable subRunnable) {
      this.subRunnable = subRunnable;
    }
  }

  private class SubRunnable implements Runnable {

    volatile boolean shouldRun = true; // Not always accessed under lock, so use volatile
    boolean shouldProceed = false; // accessed under lock, so no need for volatile
    private BossRunnable bossRunnable;

    public SubRunnable(BossRunnable bossRunnable) {
      this.bossRunnable = bossRunnable;
    }

    @Override
    public void run() {

      waitToProceed();

      int loopCount = 1;
      while (shouldRun) {
        // print iteration 10 times
        for (int j = 0; j < 10; j++) {
          System.out.println(Thread.currentThread().getName() + " iter-" + (j + 1));
        }
        // This print helps split boss-th and sub-th prints
        System.out.println(Thread.currentThread().getName() + " outer-loop iter:" + loopCount++);

        bossRunnable.notifySubThreadCompletion();
        waitToProceed();
      }
    }

    private synchronized void waitToProceed() {
      while (!shouldProceed) {
        try {
          wait();
        }
        catch (InterruptedException e) {
          // ignore this, and retry the wait
        }
      }
      shouldProceed = false; // reset the "latch"
    }

    public synchronized void allowToProceed() {
      shouldProceed = true;
      notify();
    }

    public void terminate() {
      shouldRun = false;
      allowToProceed();
    }

  }

}

Answer 2

One completely different approach would be to use RxJava to abstract away any manual thread handling. RxJava offers the class Observable. For our purpose here you can think of it as a stream of data that begins to emit values once you call subscribe() on it. Observable#onNext gets called every time the next value gets emitted from the stream. This way we can create two Observables from a stream of integers (100 for boss, 10 for sub) and arrange them in the desired way:

public static void main(String[] args) throws InterruptedException {
    Observable<Integer> bossObservable = Observable.range(1, 100) // emit integers from 1 to 100
            .subscribeOn(Schedulers.io()) // choose thread pool to run on
            .doOnNext(i -> System.out.println(Thread.currentThread() + " Boss " + i)); // gets executed for each
                                                                                        // value in the stream
    Observable<Integer> subObservable = Observable.range(1, 10) // emit integers from 1 to 10
            .subscribeOn(Schedulers.io()) // choose thread pool to run on
            .doOnNext(i -> System.out.println(Thread.currentThread() + " Sub " + i)); // gets executed for each
                                                                                        // value in the stream

    subObservable.concatWith(bossObservable) // append bossObservable to subObservable
                                        // -> bossObservable runs after subObservable has finished
            .repeat(2) // we repeat this two times, like the outer for loop in your solution
            .subscribe(); // this starts the execution

    while (true) {
        Thread.sleep(100); // keep main thread alive
    }
}

The example code uses RxJava 2.2.3. As you can probably tell, the code is quite concise but also hard to grasp if you're not familiar with Rx, because the framework takes care of the thread handling.

If you want to learn more about reactive programming you can read the excellent introduction by Andre Staltz: https://gist.github.com/staltz/868e7e9bc2a7b8c1f754