Simulation of Multiple Producer Consumer

Question

I am trying to implement the Multiple Producer-Consumer problem. I have made the buffer as the lock to maintain consistency of the system. Does this system actually implement multithreading? How can I improve the system using wait and notify?

Producer.java

import java.util.*;
import java.io.*;

class Producer extends Thread  
{
    private Buffer buffer;
    private String name;

    Producer(Buffer buffer,String name)
    {
        this.buffer = buffer;
        this.name = name;
    }
    @Override
    public void run()
    {
        while(true)
        {
            Random r = new Random();
            int x = r.nextInt(100);
            produce(x);
            try 
            {
                //(r.nextInt(6)+1)*
                Thread.sleep(100);
            } 
            catch (Exception e) 
            {
                e.printStackTrace();
            }
        }
    }
    public void produce(int x)
    {
        synchronized(buffer)
        {

            if(buffer.size()==buffer.cap)
            {
                System.out.println("QF");
            }
            else
            {
                buffer.add(x);
                System.out.println(name+": "+x);
            }
        }
    }
}

Buffer.java

class Buffer
{
    private Vector<Integer> buffer;
    int cap;
    Buffer(int cap)
    {
        buffer = new Vector<Integer>();
        this.cap = cap;
    }
    int size()
    {
        return buffer.size();
    }
    boolean isEmpty()
    {
        return buffer.isEmpty();
    }
    void add(int x)
    {
        buffer.add(x);
    }
    int remove()
    {
        if(buffer.isEmpty())
            return -1;
        int x = buffer.firstElement();
        buffer.remove(buffer.firstElement());
        return x;
    }
}

Consumer.java

class Consumer extends Thread
{
    private Buffer buffer;
    private String name;

    Consumer(Buffer buffer,String name)
    {
        this.buffer = buffer;
        this.name   = name;
    }
    @Override
    public void run()
    {
        while(true)
        {
            consume();
            try
            {
                Random r = new Random();    
                Thread.sleep(100);
            } 
            catch (Exception e) 
            {
                e.printStackTrace();
            }
        }
    }
    public void consume() 
    {
        synchronized(buffer)
        {
            int x = buffer.remove();
            if(x==-1)
            {
                System.out.println("QE");
            }
            else
            {
                System.out.println(name+": "+x);
            }
        }
    }
}

App.java

public class App
{
    public static void main(String args[]){
        Buffer b = new Buffer(5);
        Producer p = new Producer(b,"P1");
        Producer p2 = new Producer(b,"P2");
        Consumer c = new Consumer(b,"C1");
        p.start();
        p2.start();
        c.start();
    }
}

Answer 1

The buffer is the resource. So it should take care about the access to itself by itself or another monitor. This monitor should not be the consumer nor the producer. So synchronization should NOT be located in the consumer or the producer.

If you want to omit effort (that will be technical debts) the first approach is to make the resource (in this case the Buffer) a monitor by introducing synchronized as the keyword in the method signatures of add() and remove(). size() and isEmpty() should not be available for clients, make them private. Now the buffer will be thread-safe.

To make use of wait and notify, you have to exactly know what is happening. wait() will suspend the current thread WITHIN a monitor.

A consumer-thread therefore may be suspended with wait() in the method remove() if nothing is in the buffer. A producer may be suspended in the method add() with wait() if he want wants to produce something but the cap is reached.

Both the consumer and the producer should notify all other suspended threads with notifyAll() after they changed the state of the resource. A simple notify() is not applicable here because in your current construct you do not distinguish between locks for consumers and locks for producers. After notifying the "notifiers" must leave the monitor by waiting or exiting the synchronized method (this will be maybe your solution).

A notified thread should first check, what has happend during its absence. So a consumer should check if something is in the buffer. If not then again wait(), consume() otherwise. A producer should check if the cap is reached. If not then produce(), wait() otherwise.

I see two further improvements:

First separate the locking responsibility from the resource. Introduce a "BufferMonitor" class for example delegating add() and remove() to the encapsulated real buffer and make its methods synchronized.

Second introduce additional locks for consumers and producers to be able to distinguish when to wake up one consumer or one producer. notifyAll() on the BufferMonitor will then become a notify() on the specific lock. Because I haven't implemented this scenario yet I do not know how much effort it is. After all it should be possible. But recognize: The BufferMonitor will not be obselete.

One sentence to the remove()-method. I suggest to somehow avoid returning -1 even it SEEMS to be impossible. A solution that does not need to return -1 is the most elegant one and I admit the most difficult. You may need something like a consumption announcement mechanism before calling "remove()". Maybe you take the challenge. But be careful: this is often called "developing gold taps". ;-)

Finally you should think about proper shutdown mechanisms of your threads. So waiting threads should be notified on shutdown and evaluate this shutdown request on their own.