Title: Thread Wait Notify
Course: CPS 261: Advanced Java Concepts ~ DL & MM MASTER

Thread Wait & Notify



Techniques for Accessing/updating data in a threaded environment.

I am going to illustrate the basic techniques needed in the real world with a simplified program. Keep in mind that the program would be more complicated if I showed you a real world problem, however, the concepts would be the same.

BusyLoop

Imagine that you have a bucket of water, and you have one Thread who has the job of filling the bucket when it is empty (call this Thread the Producer). You have another Thread who has the job of emptying the bucket of water(call this Thread the Consumer). Pretend like we are fighting fires in the old days and we have one guy who can fill a bucket and another guy who can empty a bucket. Don't worry about the fact that it would be useful having more than one bucket ... we will add that later.

1. We create a Bucket class and provide a put method for filling the bucket. We provide a get routine for emptying the bucket. A put call can only succeed if the bucket is empty, and a get call can only succeed when the bucket is full.
2. To avoid conflicts, we synchronize the get and put methods. The code shows up below.
3. The code works, but there is a very inefficient thing happening. The Consumer needs to make a lot of unproductive calls to the get method, because get returns a fail boolean when the bucket is empty.
4. There must be a better way ----> Move on to the next example.


class Bucket {
    private int contents;
    private boolean fullFlag = false;

    public synchronized boolean get(int consumerNumber) {
        if (fullFlag)
        {
            System.out.println("Consumer " + consumerNumber +
                " Emptying Bucket " + contents);
            fullFlag = false;
            return true; // successful
        }
        return false; // unsuccessful
    }

    public synchronized boolean put(int value) {
        if(!fullFlag)
        {
            System.out.println("Filling Bucket " + value);
            contents = value;
            fullFlag = true;
            return true; // successful
        }
        return false; // unsuccessful
    }
}
// ******************************************************
class Producer extends Thread {
    private Bucket bucket;
    private int number;

    public Producer(Bucket c, int number) {
        bucket = c;
        this.number = number;
    }

    public void run() {
        int numSuccess = 0;
        int numTries = 0;
        int fillValue;

        do
        {
            try {
                sleep((int)(Math.random() * 2000));
            } catch (InterruptedException e) { }

            fillValue = numSuccess + number*100;
            numTries += 1;
            if (bucket.put(fillValue))
                numSuccess += 1;
         } while (numSuccess < 10);
        System.out.println(number + "****Terminating Producer " +
            numTries + " numTries");
    }
}

//********************************************************
class Consumer extends Thread {
    private Bucket bucket;
    private int number;

    public Consumer(Bucket c, int number) {
        bucket = c;
        this.number = number;
    }

    public void run() {
        int numSuccess = 0;
        int numTries = 0;
        do
        {
            try {
                   sleep((int)(Math.random() * 20));
                }
            catch (InterruptedException e) { }

            numTries += 1;
            if (bucket.get(number))
                numSuccess += 1;
        } while (numSuccess < 10);

        System.out.println(number + "****Terminating Consumer " +
            numTries + " numTries");
    }
}

//**********************************************************

public class BusyLoop {
    public static void main(String[] args) {
        Bucket bucket = new Bucket();
        Producer p1 = new Producer(bucket,1);
        Consumer c1 = new Consumer(bucket,1);

        c1.start();
        p1.start();
        System.out.println("BusyLoop main will exit now");
    }
}



Solution to the Busy Loop

Java provides a very elegant solution to the above problem. The object class has 3 key methods that are available to any Java class:

- wait - The effect of this call is to put the calling thread on a queue of waiting threads. Once this thread is on the queue, the object lock becomes available and other threads can enter a synchronized function. The thread that calls wait is removed from the waiting queue, when some other thread calls either notify or notifyall. The wait call has a timeout feature that allows the thread to also exit the queue because a certain amount of time has elapsed.
wait(number_of_milliseconds);
- notify -The effect of this call is to remove one of the threads on the suspension queue and to have it rescheduled to return from it's wait call.
- notifyall - The effect of this call is to schedule all of the threads on the suspension queue.
- To call these 3 routines, you must be executing inside of a synchronized section for the object instance of interest (i.e. Bucket class instance in the above example)
- Take a look at the online help for the Object class to see these methods. You will see that certain exceptions can be thrown for these 3 methods if you don't follow the rules. Most of these exceptions are derived from RuntimeException which means the compiler doesn't force you to catch them. However, the wait method can throw an InterruptedException and this exception is not a RuntimeException so you have deal with it.

WaitNotify.java


class Bucket {
    private int contents;
    private boolean fullFlag = false;

    public synchronized int get(int consumerNumber) {
        while (!fullFlag) {
            try {
                wait();
            } catch (InterruptedException e) { }
        }
        System.out.println("Consumer " + consumerNumber +
                " Emptying Bucket " + contents);
        fullFlag = false;
        notify(); // Wake up any Producer waiting for an empty bucket
        return contents;
    }

    public synchronized void put(int value) {
        while (fullFlag) {
            try {
                wait();
            } catch (InterruptedException e) { }
        }
        System.out.println("Filling Bucket " + value);
        contents = value;
        fullFlag = true;
        notify(); // Wake up any Consumer waiting for a full bucket
    }
}
// ******************************************************
class Producer extends Thread {
    private Bucket bucket;
    private int number;

    public Producer(Bucket c, int number) {
        bucket = c;
        this.number = number;
    }

    public void run() {
        int fillValue;
        for (int i = 0; i < 10; i++) {
            try {
                sleep((int)(Math.random() * 2000));
            }
            catch (InterruptedException e) { }

            fillValue = i + number*100;
            bucket.put(fillValue);
        }
        System.out.println("****Terminating Producer " + number);
    }
}

//********************************************************
class Consumer extends Thread {
    private Bucket bucket;
    private int number;

    public Consumer(Bucket c, int number) {
        bucket = c;
        this.number = number;
    }

    public void run() {
        int value = 0;
        for (int i=0; i < 10; i++) {
            value = bucket.get(number);
        }
        System.out.println("****Terminating Consumer " + number);
    }
}

//**********************************************************

public class WaitNotify {
    public static void main(String[] args) {
        Bucket bucket = new Bucket();
        Producer p1 = new Producer(bucket,1);
        Consumer c1 = new Consumer(bucket,1);

        c1.start();
        p1.start();
        System.out.println("WaitNotify will exit now");
    }
}




Getting more parallel activity.

If you want to double your fire fighting capability, add the lines of code shown below:


 Bucket bucket2 = new Bucket();
 Producer p2 = new Producer(bucket2,2);
 Consumer c2 = new Consumer(bucket2,2);
 c2.start();
 p2.start();


Note that I can have consumer c2 inside of the get method of bucket2 at the same time as consumer c1 is inside of the get method of bucket1. Remember synchronization only prevents 2 threads inside of a synchronized method of the same object .

Fire Fighting Becomes popular

Suppose that we want a Queue of Buckets and we want to allow as many producers and consumers as desired.

1. We can now have M Producers and N Consumers. In the previous example, producing and consuming was very balanced. So when a producer or a consumer accessed the bucket successfully 10 times, it voluntarily exitted. For example, you can go with 4 Producers and 5 Consumers with the following call:
java WaitNotify2 4 5
2. Now we would like the main routine to detect when the producers are done ( they still finish after 10 successful bucket accesses). When the producers are done, we would like to have the main routine tell the consumers to quit.
3. Our main routine periodically looks at each Producer thread and calls the Thread method isAlive() to determine if the Thread is still around.
4. Once all producers have terminated, the main routine calls a new method in the bucket class called stop. The stop method sets a flag and calls notifyAll(). Since it is possible to have multiple Consumers who are waiting, we use the notifyAll routine to schedule all of them. Each Consumer will wake up and sense the flag which will result in terminating the Thread.
5. Note that notify reschedules the next Thread in a waiting queue. The notifyAll routine reschedules ALL Threads in the waiting queue.
6. I have a more practical version of this kind of logic where I have a Reader thread, Compute thread, and a Writer thread passing buffers between each other in the following link (note this is a more complicated and only for those who have a curiosity and extra time) : Parallel IO

WaitNotify2.java


class Bucket
{
    private int contents;
    public int getContents()
    {
        return contents;
    }
    public void setContents(int contents){
        this.contents = contents;
    }
}

class BucketCircularQueue {
    static final int CQUEUE_SIZE = 8;
    private Bucket[] cQueue = new Bucket[CQUEUE_SIZE];
    private int getIndex=0, putIndex=0, count=0;
    boolean timeToStop = false;

    private Bucket fetchNext()
    {
        if (count == 0)
            return null;
        Bucket b = cQueue[getIndex];
        getIndex = (getIndex+1) % CQUEUE_SIZE;
        count -= 1;
        System.out.println("fetchNext: count="+ count);
        return b;
    }
    private Bucket putNext()
    {
        if (count == CQUEUE_SIZE)
            return null;
        Bucket b = cQueue[putIndex];
        putIndex = (putIndex+1) % CQUEUE_SIZE;
        count += 1;

        System.out.println("putNext: count="+ count);
        return b;
    }

    BucketCircularQueue()
    {
        for (int i=0; i < CQUEUE_SIZE; i++)
            cQueue[i] = new Bucket();
    }


    public synchronized int get(int consumerNumber) {
        Bucket b = fetchNext();
        while (b==null) {
            if (timeToStop) return -1;
            try {
                wait();
            } catch (InterruptedException e) { }

            b = fetchNext();
        }
        int contents = b.getContents();
        System.out.println("Consumer " + consumerNumber +
                " Emptying Bucket " + contents);
        notifyAll();  // Need to wake up interested producer
                      // Note this could wake up other consumers
        return contents;
    }

    public synchronized void put(int value) {
        Bucket b = putNext();
        while (b==null) {
            try {
                wait();
            } catch (InterruptedException e) { }
            b = putNext();
        }
        System.out.println("Filling Bucket " + value);
        b.setContents(value);
        notifyAll(); // Wake up interested consumers
                  // Note this could wake up other producers
    }
    public synchronized void stop()
    {
        timeToStop = true;
        notifyAll();
    }
}
// ******************************************************
class Producer extends Thread {
    private BucketCircularQueue bucketCQueue;
    private int number;

    public Producer(BucketCircularQueue c, int number) {
        bucketCQueue = c;
        this.number = number;
    }

    public void run() {
        int fillValue;
        for (int i = 0; i < 10; i++) {
            try {
                sleep((int)(Math.random() * 2000));
            } catch (InterruptedException e) { }

            fillValue = i + number*100;
            bucketCQueue.put(fillValue);
        }
        System.out.println("****Terminating Producer " + number);
    }
}

//********************************************************
class Consumer extends Thread {
    private BucketCircularQueue bucketCQueue;
    private int number;

    public Consumer(BucketCircularQueue c, int number) {
        bucketCQueue = c;
        this.number = number;
    }

    public void run() {
        int value = 0;
        while (true) {
            value = bucketCQueue.get(number);
            if (value < 0) break; // Time to exit

            try {
                sleep((int)(Math.random() * 3000));
            } catch (InterruptedException e) { }

        }
        System.out.println("****Terminating Consumer " + number);
    }
}

//**********************************************************



public class WaitNotify2 {
    static boolean anyoneAlive(Thread[] t)
    {
        for (int i=0; i < t.length; i++)
        {
            if (t[i].isAlive())
                return true;
        }
        return false;
    }
    public static void main(String[] args) {
        BucketCircularQueue bucketCQueue = new BucketCircularQueue();
        int numProducers = 1;
        int numConsumers = 1;
        Producer p[];
        Consumer c[];

        if (args.length > 0)
            numProducers = Integer.parseInt(args[0]);
        if (args.length > 1)
            numConsumers = Integer.parseInt(args[1]);

        p = new Producer[numProducers];
        c = new Consumer[numConsumers];

      // Launch all of the Consumer Threads
        for (int i = 0; i < numConsumers; i++)
        {
           c[i] = new Consumer(bucketCQueue, i+1);
           c[i].start();
        }

     // Launch all of the Producer Threads
        for (int i = 0; i < numProducers; i++)
        {
           p[i] = new Producer(bucketCQueue, i+1);
           p[i].start();
        }

  // Monitor Status of Producers.
  // When the producers go away, get rid of the
  // Consumer.

        while (anyoneAlive(p))
        {
            try
            {
                Thread.sleep(1000);
            }
            catch (InterruptedException e) {}
         }

     // All the Producers are done
        System.out.println("Stopping Consumers");
        bucketCQueue.stop();
    }
}



More Examples - Other examples in case you need more

Last Updated: July 22, 2014 8:43 PM