Blocking Queue | Bounded Buffer | Consumer Producer

A blocking queue is defined as a queue which blocks the caller of the enqueue method if there's no more capacity to add the new item being enqueued. Similarly, the queue blocks the dequeue caller if there are no items in the queue. Also, the queue notifies a blocked enqueuing thread when space becomes available and a blocked dequeuing thread when an item becomes available in the queue.

The queue.Queue module already provides a synchronized queue out of the box. We'll be creating one from scratch that can be used by multiple consumers and producers.

Our queue will have a finite size that is passed in via the constructor. Additionally, we'll use a list as the backing data structure for our queue. We can leverage the pop() and append() method on the list to emulate the operation of a queue. appends() adds an element to the front of the list and pop() retrieves the last element of the list, freeing us from keeping track of the front and back of the queue.

We'll need two variables: one to keep track of the maximum size of the queue and another to track the current size of the queue. Moreover, we'll also need a condition variable that either the producer or the consumer can wait on if the queue is full or empty respectively.

Initially, our class will look like as follows:

class BlockingQueue:

    def __init__(self, max_size):
        self.max_size = max_size
        self.curr_size = 0
        self.cond = Condition()
        self.q = []

    def enqueue(self):
        pass

    def dequeue(self):
        pass

Let's start with the enqueue() method, which takes in an item to be enqueued to the back of the queue. If the current size of the queue == capacity then we know we'll need to block the invoker of the method. We can do so by calling wait() on the condition variable wrapped with a while loop. The while loop is conditioned on the size of the queue being equal to the maximum capacity of the queue. The loop's predicate would become false as soon as another thread performs a dequeue.

Note that whenever we test for the value of the curr_size variable, we also need to make sure that no other thread is manipulating the curr_size variable. This can be achieved by acquiring the lock associated with the condition variable at the start of each method.

The enqueue method would look like as follows:

    def enqueue(self, item):

        self.cond.acquire()
        while self.curr_size == self.max_size:
            self.cond.wait()

        self.q.append(item)
        self.curr_size += 1

        self.cond.notifyAll()
        self.cond.release()

Note that in the end we are calling notifyAll() method. Since we just added an item to the queue, it is possible that a consumer thread might be blocked in the dequeue() method of the queue class waiting for an item to become available. So it's necessary we send a signal to wake up any waiting threads.

If no thread is waiting, then the signal will simply go unnoticed and be ignored, which wouldn't affect the correct working of our class. This would be an instance of a missed signal.

Now let's design the dequeue() method. Similar to the enqueue() method, we need to block the caller of the dequeue method if there's nothing to dequeue i.e. self.curr_size == 0. When this condition is true a consumer thread needs to wait on the condition variable.

After consuming an item, remember to call notifyAll() since if the queue were full then there might be producer threads blocked in the enqueue() method. This logic in code appears as below:

    def dequeue(self):

        self.cond.acquire()
        while self.curr_size == 0:
            self.cond.wait()

        item = self.q.pop(0)
        self.curr_size -= 1
        
        self.cond.notifyAll()
        self.cond.release()

        return item

We see the dequeue method is analogous to enqueue method. The complete code appears in the code widget below along with an example.