Problem with Thread Interaction

[Pradeep Bhat]

Hi,
I was reading about Thread Interaction and came up with the following program that has a single producer and multiple consumer threads all working on a common 'queue' list. If the list is empty, the consumer threads have to wait till the queue is full. The produces fills the queue, notifies everyone and goes to sleep.
---------------------

package chapter9_Threads;
 
import java.util.ArrayList;
import java.util.List;
 

class Producer implements Runnable {

List<String> queue;

public Producer (List<String> myList){
this.queue = myList;
}

/**
 * This method is used to fill items in the queue.
 */
public void fill() {
queue.add("apple");
queue.add("cherry");
queue.add("litchie");
System.out.println(queue);
}

/**
 * Locks the queue. Then goes on filling it. 
 * Once it is full, it notifys all
 * other threads.
 */
public void produce() {
synchronized (queue) {
System.out.println(Thread.currentThread().getName() 
+ " produces.");
this.fill();
queue.notifyAll();
}
}

/**
 * This method is ensures that the queue is filled. 
 * Once the queue is filled,
 * it waits for some time.
 */
public void run() {
while(true) {
produce();
try {
Thread.sleep(2000);
} catch (InterruptedException e) {}
}
}
}
 
class Consumer implements Runnable {
List<String> queue;

public Consumer (List<String> myList){
this.queue = myList;
}

/**
 * Locks the queue. Consumes one item from the queue.
 * If at all the queue is empty, it waits for the 
 * queue to be full again by calling wait on the
 * locked queue.
 */
public void consume() {
synchronized (queue) {
if(queue.isEmpty()) {
try {
queue.wait();
} catch (InterruptedException e) {}
}
System.out.println(Thread.currentThread().getName() 
+ " consumes " + queue.get(0));
queue.remove(0);
System.out.println("Pending Items : " + queue);
}
}

public void run() {
while(true) {
consume();
}
}
}
 
/**
 * This method tests thread interaction.
 */
public class NotifyAndWait {
public static void main(String[] args) {
// Common data that will be used by all threads
List<String> foodList = new ArrayList<String>();

Producer pr = new Producer(foodList);
Consumer con = new Consumer(foodList);
Consumer con1 = new Consumer(foodList);
Consumer con2 = new Consumer(foodList);

Thread t1 = new Thread(pr,"Producer");
Thread t2 = new Thread(con,"Consumer1");
Thread t3 = new Thread(con1,"Consumer2");
Thread t4 = new Thread(con2,"Consumer3");

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

------------------------------------

Is this code proper ?

Also the output appears as follows

------------------------------------

Producer produces.
[apple, cherry, litchie]
Consumer1 consumes apple
Pending Items : [cherry, litchie]
Consumer1 consumes cherry
Pending Items : [litchie]
Consumer1 consumes litchie
Pending Items : []
Producer produces.
[apple, cherry, litchie]
[B]Exception in thread "Consumer2" java.lang.IndexOutOfBoundsException: Index: 0, Size: 0
at java.util.ArrayList.RangeCheck(Unknown Source)
at java.util.ArrayList.get(Unknown Source)
at chapter9_Threads.Consumer.consume(NotifyAndWait.java:75)
at chapter9_Threads.Consumer.run(NotifyAndWait.java:83)
at java.lang.Thread.run(Unknown Source)
Exception in thread "Consumer1" java.lang.IndexOutOfBoundsException: Index: 0, Size: 0
at java.util.ArrayList.RangeCheck(Unknown Source)
at java.util.ArrayList.get(Unknown Source)
at chapter9_Threads.Consumer.consume(NotifyAndWait.java:75)[/B]
Consumer3 consumes apple
Pending Items : [cherry, litchie]
Consumer3 consumes cherry
Pending Items : [litchie]
Consumer3 consumes litchie
Pending Items : []
 
at chapter9_Threads.Consumer.run(NotifyAndWait.java:83)
at java.lang.Thread.run(Unknown Source)
Producer produces.
[apple, cherry, litchie]

------------------------------------

The consumer thread should wait in case the queue is empty and wait for the notification. Then why is the index out of bound exception thrown ?

[Carey Evans]

The simplest solution would be to use a LinkedBlockingQueue<String> instead of a List. But to answer your question...

When the producer calls notifyAll(), every waiting consumer wakes up. One of them gets the lock, empties the list, and releases the lock. Then another gets the lock, tries to empty the list, and fails. Finally, the third consumer does the same.

Making this worse is that wait() can return without notifyAll() being called at all.

To do this properly, you can use a while loop instead of an if statement:

public void consume() throws InterruptedException {
    synchronized (queue) {
        while (queue.isEmpty()) {
            queue.wait();
        }
        System.out.println(Thread.currentThread() + " : " + queue.remove(0));
    }
}

Or use a BlockingQueue:

public void consume() throws InterruptedException {
    System.out.println(Thread.currentThread() + " : " + queue.take());
}

[ July 29, 2008: Message edited by: Carey Evans ]