1
\$\begingroup\$

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();
            }
        }
    }
}
\$\endgroup\$
  • \$\begingroup\$ Let's see if I understood it correctly. Scenario 1: Step 1: producer puts 1 element, Step 2 : all readers read? How does the producer know when there is no more readers? This should be avoided in general, but it's doable as long there is one extra operation where a thread says that it is the last reader. Scenario 2: Step 1: readers wait for one element. Step 2: producer puts element Step 3: readers awake. On this scenario there are no readers able to read after the element is produced. Unless there is one extra operation as well. Which is it? \$\endgroup\$ – Bruno Costa Oct 19 '16 at 13:17
  • \$\begingroup\$ Does Blocking Queue solve your problem? If not, can you explain why? \$\endgroup\$ – Bruno Costa Oct 19 '16 at 13:23
  • \$\begingroup\$ The producer puts data to queue and waits till all consumers have finished before adding more data. So effectively producers are only able to produce 1 element at a time? \$\endgroup\$ – Bruno Costa Oct 19 '16 at 13:28
  • \$\begingroup\$ Let me rephrase the problem statement. The producer puts 5 elements in queue and waits till each consumer consumes 1 element. The cycle repeats till 100 elements have been added and consumed. \$\endgroup\$ – AnjuG Oct 19 '16 at 14:21
  • \$\begingroup\$ Could you edit the question accordingly? \$\endgroup\$ – Bruno Costa Oct 19 '16 at 14:29
1
\$\begingroup\$

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.

\$\endgroup\$
1
\$\begingroup\$

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();
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.