One producer and multiple consumers wherein the producer has to wait until all consumers finish before adding more data

Question

I am trying to solve the scenario involving 1 producer and 5 consumers. The producer puts data to queue and waits till all consumers have finished before adding more data.

Please review and let me know if there is room for improvement.

import java.util.LinkedList;
import java.util.Queue;

public class ReadUpdate {

   static int cntr;
   static volatile int writeLck = 1;
   static int numSupplied;
   Queue queue = new LinkedList<Node>();

    public static void main(String arg[]) {
        ReadUpdate rU = new ReadUpdate();
        Producer p = new Producer(rU.queue);
        p.start();
        Consumer c1 = new Consumer(rU.queue, 1);
        c1.setName("1");
        c1.start();
        Consumer c2 = new Consumer(rU.queue, 2);
        c2.setName("2");
        c2.start();
        Consumer c3 = new Consumer(rU.queue, 3);
        c3.setName("3");
        c3.start();
        Consumer c4 = new Consumer(rU.queue, 4);
        c4.setName("4");
        c4.start();
        Consumer c5 = new Consumer(rU.queue, 0);
        c5.setName("5");
        c5.start();
    }
 }

class Consumer extends Thread {
    Queue queue;
    int turn;

    Consumer(Queue queue, int turn) {
        this.queue = queue;
        this.turn = turn;
    }

    public void run() {
        while (ReadUpdate.numSupplied <= 100) {
            synchronized (ReadUpdate.class) {
                while (queue.isEmpty() || ReadUpdate.cntr % 5 != this.turn) {
                    try {
                        ReadUpdate.class.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println("THREAD " + Thread.currentThread().getName()
                        + " GOT " + queue.poll());
                if (ReadUpdate.cntr % 5 == 0) {
                    ReadUpdate.writeLck = 1;
                } else {
                    ReadUpdate.cntr++;
                }
                ReadUpdate.class.notifyAll();
            }
        }
    }
}

class Producer extends Thread {
    Queue queue;

    Producer(Queue queue) {
        this.queue = queue;
        ReadUpdate.numSupplied++;
    }

    public void run() {
        while (ReadUpdate.numSupplied < 100) {
            synchronized (ReadUpdate.class) {
                while (ReadUpdate.writeLck == 0) {
                    try {
                        ReadUpdate.class.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                for (int i = 0; i < 5; i++) {
                    queue.offer(ReadUpdate.numSupplied++);
                }
                ReadUpdate.writeLck = 0;
                ReadUpdate.cntr++;
                ReadUpdate.class.notifyAll();
            }
        }
    }
}

Answer 1

It's not recommended to inherit from Thread. Instead implement Runnable and pass it to a new Thread which you store in a member and provide accesses for what you need. This also lets you pass a ThreadFactory for some customization options for the threads used.

class Producer extends Runnable{

    private Thread thr;

    Producer(Queue queue, ThreadFactory fact) {
        if(fact == null)
            thr = new Thread(this, "producer thread");
        else
            thr = fact.newThread(this);
        //...
    }

    public void start(){
        thr.start();
    }

    //...
}

As for synchronizing if you only needed the consumers to run once you could have used a CountDownLatch but that cannot be reset. So you'll have to use a CyclicBarrier.

With a cyclic barrier you can make the producer the Runnable that gets invoked when the last thread calls its await. Then you can be sure all but one consumer threads are waiting for the producer code to finish.

Answer 2

You are mixing two different concerns into one place.

You have a synchronization problem to solve. But you are bringing that problem to the code of each thread. Instead of doing that you should prefer to have a class that solves the problem, and the threads just use that class so they are synchronized.

The implementation that I am providing is not fair (that means that consumers will not have an equal chance to consume items).

public class BlockingBoundQueue{
    private int limit;
    private LinkedList<Node> elems = new LinkedList<Node>();
    private boolean canTake;

    public BlockingBoundQueue(int limit){
        this.limit = limit;
    }

    public void put(Node elem){
        synchronized (this){
            if(elems.size() < limit){
                elems.add(elem);
                this.notifyAll();
                return;
            }
            canTake = true;
            //the producer needs to wait that all consumers remove an element from the list
            //when all consumers are done the list will be empty
            while(!elems.isEmpty()){
                this.wait();
            }
            canTake = false;
            elems.add(elem);
            this.notifyAll();
        }
    }

    public T take(){
        synchronized (this){
            if(canTake && !elems.isEmpty()){
                Node node = elems.pop();
                if(elems.isEmpty()){
                    //I will notify the producer that I am the last consumer
                    this.notifyAll();
                }
                return node;
            }
            while(!canTake || elems.isEmpty()){
                this.wait();
            }
            Node node = elems.pop();
            this.notifyAll();
            return node;
        }
    }
}

The sample program would be something in the lines of:

BlockingBoundQueue queue = new BlockingBoundQueue();

Runnable consume = new Runnable() { 
    public void run() { 
        for(int i = 0; i < 20; ++i){
            queue.take();
        }
    }
};

Runnable produce = new Runnable() { 
    public void run() { 
        for(int i = 0; i < 100; ++i){
            queue.put(new Node());
        }
    }
};

Thread[] consumers = new Thread[]{
    new Thread(consume),
    new Thread(consume),
    new Thread(consume),
    new Thread(consume),
    new Thread(consume),
};

Thread producer = new Thread(produce);
producer.Start();

for(Thread c : consumers){
    c.start();
}