04. Interthread Communication.md

InterThread Communication

• Java includes an elegant interprocess communication mechanism via the wait, notify and notifyAll methods.
• These methods are all implemented as final methods within Object.

final void wait() throws InterruptedException 
final void notify()
final void notifyAll()

• wait() tells the calling thread to give up the monitor and go to sleep until some other thread enters the same monitor and calls notify() or notifyAll().
• notify() wakes up a thread that called wait() on the same object.
• notifyAll() wakes up all the threads that called wait() on the same object. One of the threads will be granted access.

lets take an example where we have a class Queue with tasks. A Producer that puts a task in the Queue and a consumer that consumes task from the Queue.

class Queue{
    private int task;

    synchronized int getTask(){
        System.out.println("Got task "+ this.task);
        return this.task;
    }

    synchronized  void putTask(int task){
        this.task = task;
        System.out.println("Adding task: " + task );
    }
}

class Producer implements Runnable{
    private final Queue q;
    private final Thread t;
    Producer(Queue q){
        this.q = q;
        this.t = new Thread(this);
    }

    public Thread getThread(){
        return this.t;
    }

    public void run(){
        int i = 0;
        while(true){
            this.q.putTask(i++);
        }
    }
}

class Consumer implements Runnable{
    private final Queue q;
    private final Thread t;
    Consumer(Queue q){
        this.q = q;
        this.t = new Thread(this, "consumer");
    }

    public Thread getThread(){
        return this.t;
    }

    public void run(){
        while(true){
            this.q.getTask();
        }
    }
}

public class Main {

    public static void main(String[] args) {
        Queue q = new Queue();
        Consumer con = new Consumer(q);
        Producer prod = new Producer(q);
        con.getThread().start();
        prod.getThread().start();
    }
}

Although the get and put operation inside Queue are synchronized, nothing stops the producer from overrunning the consumer, not will anything stop the consumer from consuming the same task from the queue multiple times. On observing the output we can see that the producer produces a task, but the consumer consumes the same task multiple number of times.

The proper way to implement this program would be to use wait() and notify() to signal in both directions.

class Queue{
    private int task;
    boolean valueSet = false;

    synchronized int getTask(){
        while(!valueSet) {
            try {
                wait();
            } catch (InterruptedException e) {
                System.out.println("getTask: Queue interruption");
            }
        }
        System.out.println("Got task "+ this.task);
        valueSet = false;
        notify();
        return this.task;
    }

    synchronized  void putTask(int task){
        while (valueSet) {
            try {
                wait();
            } catch (InterruptedException e) {
                System.out.println("putTask: Queue interruption");
            }
        }
        this.task = task;
        valueSet = true;
        System.out.println("Adding task: " + task );
        notify();
    }

}

class Producer implements Runnable{
    private final Queue q;
    private final Thread t;
    Producer(Queue q){
        this.q = q;
        this.t = new Thread(this);
    }

    public Thread getThread(){
        return this.t;
    }

    public void run(){
        int i = 0;
        while(true){
            this.q.putTask(i++);
        }
    }
}

class Consumer implements Runnable{
    private final Queue q;
    private final Thread t;
    Consumer(Queue q){
        this.q = q;
        this.t = new Thread(this, "consumer");
    }

    public Thread getThread(){
        return this.t;
    }

    public void run(){
        while(true){
            this.q.getTask();
        }
    }
}

public class Main {

    public static void main(String[] args) {
        Queue q = new Queue();
        Consumer con = new Consumer(q);
        Producer prod = new Producer(q);
        con.getThread().start();
        prod.getThread().start();

        System.out.println("Press CTRL + C to stop");
    }
}