Producer Consumer using custom blocking queue

Question

After the feedback I have got from this question, I tried to to re-implement the classic Producer Consumer problem using multithreading.

The requirements are:

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

It's running and seems to be giving correct output. However, I have a small concern. In my code, if a producer is producing an item, consumer won't be able to consume simultaneously. Is there a way to break this dependency? i.e. A consumer should be able to consume item (provided the queue is not empty) irrespective of whether the producer is producing item or not and vice versa.

Custom Blocking Queue

public class CustomQueue<E> {

    private Queue<E> queue;
    private int limit;

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

    public  synchronized void enQueue(E e) throws InterruptedException{
        while(queue.size()==limit){
            System.out.println("Queue is full."+ Thread.currentThread().getName() + " is waiting....");
            wait();
        }

        System.out.println("Queue is not full. "+ Thread.currentThread().getName() + " is producing "+ e);
        queue.add(e);
        notifyAll();
    }

    public synchronized E deQueue() throws InterruptedException{
        while(queue.size()==0){
            System.out.println("Queue is empty."+ Thread.currentThread().getName() + " is waiting....");
            wait();
        }
        E e = queue.remove();
        System.out.println("Queue is not empty. "+ Thread.currentThread().getName() + " is consuming "+ e);
        notifyAll();
        return e;
    }

}

Producer Code

public class Producer implements Runnable{

    private CustomQueue<Integer> sharedQueue;

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

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

    private void produceItems() {           
        for(int i=0; i<10; i++){
            try {
                sharedQueue.enQueue(i);
            } catch (InterruptedException e) {
                System.out.println("Exception in producer."+ e);
            }
        }       
    }

}

Consumer Code

public class Consumer implements Runnable {

    private CustomQueue<Integer> sharedQueue;

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

    @Override
    public void run() {
        while(true){
            int item = consumeItem();   
        }
    }

    private int consumeItem() {
        int item;
        try {
            item = sharedQueue.deQueue();
        } catch (InterruptedException e) {
            item = Integer.MIN_VALUE;
            System.out.println("Exception in consumer."+ e);
        }
        return item;
    }

}

Test Code

public class Test {


    public static void main(String args[]){
        CustomQueue<Integer> sharedQueue = new CustomQueue<>(10);
        Producer p1 = new Producer(sharedQueue);
        Producer p2 = new Producer(sharedQueue);
        Consumer c1 = new Consumer(sharedQueue);
        Consumer c2 = new Consumer(sharedQueue);

        Thread pt1 = new Thread(p1);
        Thread ct1 = new Thread(c1);
        Thread pt2 = new Thread(p2);
        Thread ct2 = new Thread(c2);

        pt1.setName("Producer 1");
        pt2.setName("Producer 2");
        ct1.setName("Consumer 1");
        ct2.setName("Consumer 2");

        ct1.start();
        pt2.start();
        ct2.start();
        pt1.start();


    }
}

Output snippet

• Queue is not empty. Consumer 1 is consuming 5
• Queue is not empty. Consumer 1 is consuming 6
• Queue is not empty. Consumer 1 is consuming 7
• Queue is not empty. Consumer 1 is consuming 8
• Queue is empty.Consumer 1 is waiting....
• Queue is empty.Consumer 2 is waiting....
• Queue is not full. Producer 2 is producing 6
• Queue is not full. Producer 2 is producing 7

Answer 1

You implemented your producers and consumers as Runnable -- that's good. You are also, judging from your sample output, using a thread naming mechanism that makes it easier to identify what's going on -- that's very good.

Not bad. Buckle up.

There are a couple immediate problems with your producer logic. First, your producers are generating identical messages in the same sequence. That isn't particularly realistic, and is going to make it harder to really see what messages are going where. Yeah, this is a toy problem, but it should be a toy problem with a purpose. You might solve this, for example, by having one producer emit positive integers, and the other emit negative integers. Or one emits powers of 2, and the other powers of 3.

Second, your production lines are infinite; which means that your code completely ignores the problem of recognizing the difference between the consumers exhausting the queue vs the consumers exhausting the producers.

try {
    sharedQueue.enQueue(i);
} catch (InterruptedException e) {
    System.out.println("Exception in producer."+ e);
}

If you are writing multi threaded code, you need to understand what the InterruptedException is for. The short version, is that it is used to cancel a running thread. "Cancel" here is an important term of art; we aren't trying to kill or abort the thread, but rather to tell the thread that it is time to clean up and shut down gracefully. This is appropriate for use cases like the user hitting the "stop" button, or the JVM trying to shut down.

In java, the idiom for this is to set the thread's interrupt flag. Long running processes (especially looping constructs) are supposed to check the state of the flag every so often, and abandon their work (again - gracefully) if it is set.

So your producer code should really look more like

    while(! Thread.currentThread.isInterrupted()){
        produceItems(); 
    }

I've deliberately put the check out here, rather than inside your for loop, to emphasize that you aren't necesarily expected to check the interrupt on every iteration. Very fast loops will check more frequently than slow loops; very slow loops might check the status multiple times per iteration, depending on how responsive you want the app to be.

Turning our attention back to produceItems()....

try {
    sharedQueue.enQueue(i);
} catch (InterruptedException e) {
    System.out.println("Exception in producer."+ e);
}

The InterruptedException here means that somebody tried to interrupt the thread while Object.wait() was blocking. The documentation for Object.wait(long) has this nugget tucked into the fine print:

The interrupted status of the current thread is cleared when [InterruptedException] is thrown.

So if we only catch the Exception inside of processItems(), then the run() method will never know that the thread was interrupted. So the job still runs "forever". Not good. Unless you are deliberately designing your code to ignore cancellation attempts, it is your responsibility to either rethrow the exception or to call Thread.interrupt(), to reset the state of the flag in your thread.

Looking at your consumer code:

private int consumeItem() {
    int item;
    try {
        item = sharedQueue.deQueue();
    } catch (InterruptedException e) {
        item = Integer.MIN_VALUE;
        System.out.println("Exception in consumer."+ e);
    }
    return item;
}

This is awful - returning an invalid value because you caught an exception that you didn't want to handle? evil evil evil. You can't satisfy your contract (because you were interrupted waiting for a turn at the queue - the correct way to signal that a method can't currently meet its contract is to throw.

In my code, if a producer is producing an item, consumer won't be able to consume simultaneously. Is there a way to break this dependency?

OK, first a caution; because you are using a toy example, you are confusing two ideas. Let's rewrite slightly:

@Override
public void run() {
    while(true){
        consumeItem();   
    }
}

private int consumeItem() {
    try {
        int item = sharedQueue.deQueue();
        consume(item);
    } catch (InterruptedException e) {
        System.out.println("Exception in consumer."+ e);
    }
}

private void consume(int item) {
    // no-op
}

This should make it clearer that the producers aren't blocked because the consumers are consume()ing, they only get blocked when consumers are deQueue()ing items. In most problem domains, the latency of consume is going to be orders of magnitude larger than the contention issues in the queue, so don't worry about it.

On the other hand, if the Consumers can't keep up with the Producers, then your pipeline is broken.

If the Consumers sometimes fall behind the Producers, but will eventually catch up, then you need to decide what the appropriate strategy is when the pipeline is congested. You can, as you have here, block indefinitely, but it's really hard to meet an aggressive SLA if you wait in line for 20 minutes. So many pipelines will instead use backpressure - if the Producer can't put the item into the queue immediately, it will instead fail fast, rather than block.

But none of those are really the question you asked - you wanted to add and remove from the queue at the same time. This can be done, but you need the right data structures - I don't think you can do it with a List. My quick proof of this is that enqueue and dequeue both need to write to the list (in particular, they are both writing to List.size, in some fashion), and concurrent writes from multiple threads are a Bad Idea [tm].

The Disruptor Whitepaper goes very deep into explaining how you can prevent the consumers from blocking the consumers, using some simple data structures and advanced access patterns.