0
\$\begingroup\$

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;
}
\$\endgroup\$
  • \$\begingroup\$ Have you tried to test this code? If so, how? \$\endgroup\$ – rolfl Oct 28 '15 at 10:20
  • \$\begingroup\$ @rolfl I have not tested the code while posting. Now only trying to run and got the issues \$\endgroup\$ – real gadha Oct 28 '15 at 11:30
  • \$\begingroup\$ Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers. \$\endgroup\$ – Vogel612 Oct 28 '15 at 11:44
  • \$\begingroup\$ Question: is it mandatory to implement the queue yourself? because if it isn't I got some more points than now. \$\endgroup\$ – Vogel612 Oct 28 '15 at 11:46
  • \$\begingroup\$ @Vogel612 Yes. We are asked not to use library function as won't learn if we dont do it by our own. This is not for production purpose, for learning purpose only \$\endgroup\$ – real gadha Oct 28 '15 at 12:26
3
\$\begingroup\$

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.

\$\endgroup\$
2
\$\begingroup\$

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

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


}
\$\endgroup\$
  • \$\begingroup\$ @Vogel612 could you please review this piece of code? I have one doubt. I didn't want to synchronize the whole enqueue and dequeue methods, because, if I do, then if a producer is in enqueue method, the consumer has to wait for producer to finish the task. I dont want this dependency, but dont know how to do that. \$\endgroup\$ – real gadha Oct 28 '15 at 12:22
  • \$\begingroup\$ @Mat, same request for you too.. \$\endgroup\$ – real gadha Oct 28 '15 at 12:22
  • 2
    \$\begingroup\$ Mentioning us in these comments here doesn't work as you seem to intend. Additionally this is an answer. Answers are not subject to review. If you want a review of new / updated code, you should How to Ask a new question. Please test your code beforehand and link back to this question if you ask a new one. \$\endgroup\$ – Vogel612 Oct 28 '15 at 12:30

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