Implementing multiple Producer Consumer using custom Blocking queue [closed]

Question

I am trying to implement multiple producer multiple consumer problem using a custom blocking queue. The requirement is that:

1. If the queue is not full, all the producers should be able to produce products independently.
2. If the queue is full, the producers have to wait.
3. If the queue is not empty, the consumer can consume products independently.
4. If the queue is empty, the consumers have to wait.

Here is the code I have implemented. Could you please review my code and let me know if there is any flaw?

Custom Blocking Queue

public class CustomQueue<T> {

private static final Object LOCK = new Object();
private int queueSize = 100; // default size
private List<T> queue;

public CustomQueue(int size){
    this.queueSize = size;
    this.queue = new LinkedList<>();
}

public void produce(T t){
    if(queue.size()==queueSize){ // queue is full. Producer need to wait
        synchronized(LOCK){
            while(queue.size()==queueSize){
                try {
                    LOCK.wait();
                } catch (InterruptedException e) {
                    System.out.println("Exception in produce() while waiting. Reason: "+e.getCause());
                }
            }
        }
    }else{ // queue is not full. Producer can produce items
        queue.add(t);
        LOCK.notifyAll();
    }
}

public T consume(){
    if(queue.size()==0){
        while(queue.size()==0){// queue is empty. Consumer need to wait
            try {
                LOCK.wait();
            } catch (InterruptedException e) {
                System.out.println("Exception in consume() while waiting. Reason: "+e.getCause());
            }
        }
    }else{ // queue is not empty. Consumers can consume item
        T t = queue.get(0);
        return t;
    }
    return null;
}

Producer code

public class Producer<T> implements Runnable{

private CustomQueue<T> sharedQueue;

public Producer(CustomQueue<T> sharedQueue){
    this.sharedQueue = sharedQueue;
}

@Override
public void run() {
    produceItems();     
}

private void produceItems() {
    List<T> items = getItems();     
    for(T t: items){
        sharedQueue.produce(t);
    }       
}

private List<T> getItems() {
    //dummy method to get list of items
    return null;
}

Consumer code

public class Consumer<T> implements Runnable {

private CustomQueue<T> sharedQueue;

public Consumer(CustomQueue<T> sharedQueue){
    this.sharedQueue = sharedQueue;
}

@Override
public void run() {
    T t = consumeItem();        
}

private T consumeItem() {
    T t = sharedQueue.consume();
    return t;
}

Answer 1

Your produce and consume methods are both buggy for the same reason: you've got a time of check/time of use bug.

Imagine a queue with size 1. Start two producer threads at the same time. They'll both see queue.size() == 0 in produce() and they'll both happily proceed to queue.add() without a lock, overflowing your queue (and causing other problems if that add() method isn't thread-safe itself).

Same thing exactly on the consumer side. If there's one item in the queue and two threads enter consume() at the same time, both will try to extract an item and at least one will fail.

You need to hold a lock that covers the check and the produce/consume actions.

---

You can't call notifyAll() on an object unless you own the object's monitor. You'll get a IllegalMonitorStateException if you try to run your code.

Answer 2

Your CustomQueue has another bug over that mentioned in Mat's answer:

You are locking on a shared Object:

private static final Object LOCK = new Object();

This means, if one instance of your queue is locking on that object, all other instances also have to wait. It's simple to deadlock all queues by:

CustomQueue<Object> deadlock = new CustomQueue<>(1);
deadlock.produce(new Object());
deadlock.produce(new Object()); // locks #produce for all instances

This is because you synchronize on your shared lock and never release the lock for the synchronized block. This means other instances of the queue that ever become full will never enter the block to wait on the LOCK (since deadlock owns the monitor) and never unlock again :(

---

Additional nitpicks:

• You should indent your code.
• A dummy method getItems() is a really really bad idea for the Producer.
  If it's mandated, you should at least return Collections.emptyList();

Answer 3

Since the custom blocking queue that I wrote was buggy, I have tried to improve the code here.

public class LinkedBlockingQueue<E> implements BlockingQueue<E>{

private int size;
private Queue<E> queue = null;

public LinkedBlockingQueue(int size){
    this.size = size;
    this.queue = new LinkedList<E>();
}

@Override
public synchronized void enqueue(E e) {
    while(this.queue.size()== size){
        try {
            wait();
        } catch (InterruptedException ex) {             
            ex.printStackTrace();
        }
    }

    this.queue.add(e);

    if(this.queue.size()==0){
        notifyAll();
    }   

}

@Override
public synchronized E dequeue() {
    while(this.queue.size()==0){
        try {
            wait();
        } catch (InterruptedException e) {              
            e.printStackTrace();                
        }
    }

    E e = queue.remove();   

    if(this.queue.size()==size){
        notifyAll();
    }       

    return e;
}

@Override
public int size() {     
    return this.queue.size();
}


}