How to Design a Design Bounded Blocking Queue in Python using Threading Library?

python

Implement a thread safe bounded blocking queue that has the following methods:

• BoundedBlockingQueue(int capacity) The constructor initializes the queue with a maximum capacity.
• void enqueue(int element) Adds an element to the front of the queue. If the queue is full, the calling thread is blocked until the queue is no longer full.
• int dequeue() Returns the element at the rear of the queue and removes it. If the queue is empty, the calling thread is blocked until the queue is no longer empty.
• int size() Returns the number of elements currently in the queue.

Your implementation will be tested using multiple threads at the same time. Each thread will either be a producer thread that only makes calls to the enqueue method or a consumer thread that only makes calls to the dequeue method. The size method will be called after every test case.

Please do not use built-in implementations of bounded blocking queue as this will not be accepted in an interview.

Example 1:
Input:
1
1

1 2	["BoundedBlockingQueue","enqueue","dequeue","dequeue","enqueue","enqueue","enqueue","enqueue","dequeue"] [[2],[1],[],[],[0],[2],[3],[4],[]]

["BoundedBlockingQueue","enqueue","dequeue","dequeue","enqueue","enqueue","enqueue","enqueue","dequeue"]
[[2],[1],[],[],[0],[2],[3],[4],[]]

Output:

1

[1,0,2,2]

[1,0,2,2]

Explanation:
Number of producer threads = 1
Number of consumer threads = 1

1 2 3 4 5 6 7 8 9 10 11

BoundedBlockingQueue queue = new BoundedBlockingQueue(2);   // initialize the queue with capacity = 2.   queue.enqueue(1);   // The producer thread enqueues 1 to the queue. queue.dequeue();    // The consumer thread calls dequeue and returns 1 from the queue. queue.dequeue();    // Since the queue is empty, the consumer thread is blocked. queue.enqueue(0);   // The producer thread enqueues 0 to the queue. The consumer thread is unblocked and returns 0 from the queue. queue.enqueue(2);   // The producer thread enqueues 2 to the queue. queue.enqueue(3);   // The producer thread enqueues 3 to the queue. queue.enqueue(4);   // The producer thread is blocked because the queue's capacity (2) is reached. queue.dequeue();    // The consumer thread returns 2 from the queue. The producer thread is unblocked and enqueues 4 to the queue. queue.size();       // 2 elements remaining in the queue. size() is always called at the end of each test case.

BoundedBlockingQueue queue = new BoundedBlockingQueue(2);   // initialize the queue with capacity = 2.

queue.enqueue(1);   // The producer thread enqueues 1 to the queue.
queue.dequeue();    // The consumer thread calls dequeue and returns 1 from the queue.
queue.dequeue();    // Since the queue is empty, the consumer thread is blocked.
queue.enqueue(0);   // The producer thread enqueues 0 to the queue. The consumer thread is unblocked and returns 0 from the queue.
queue.enqueue(2);   // The producer thread enqueues 2 to the queue.
queue.enqueue(3);   // The producer thread enqueues 3 to the queue.
queue.enqueue(4);   // The producer thread is blocked because the queue's capacity (2) is reached.
queue.dequeue();    // The consumer thread returns 2 from the queue. The producer thread is unblocked and enqueues 4 to the queue.
queue.size();       // 2 elements remaining in the queue. size() is always called at the end of each test case.

Example 2:

Input:
3
4

1 2	["BoundedBlockingQueue","enqueue","enqueue","enqueue","dequeue","dequeue","dequeue","enqueue"] [[3],[1],[0],[2],[],[],[],[3]]

["BoundedBlockingQueue","enqueue","enqueue","enqueue","dequeue","dequeue","dequeue","enqueue"]
[[3],[1],[0],[2],[],[],[],[3]]

Output:

1

[1,0,2,1]

[1,0,2,1]

Explanation:
Number of producer threads = 3
Number of consumer threads = 4

1 2 3 4 5 6 7 8 9 10

BoundedBlockingQueue queue = new BoundedBlockingQueue(3);   // initialize the queue with capacity = 3.   queue.enqueue(1);   // Producer thread P1 enqueues 1 to the queue. queue.enqueue(0);   // Producer thread P2 enqueues 0 to the queue. queue.enqueue(2);   // Producer thread P3 enqueues 2 to the queue. queue.dequeue();    // Consumer thread C1 calls dequeue. queue.dequeue();    // Consumer thread C2 calls dequeue. queue.dequeue();    // Consumer thread C3 calls dequeue. queue.enqueue(3);   // One of the producer threads enqueues 3 to the queue. queue.size();       // 1 element remaining in the queue.

BoundedBlockingQueue queue = new BoundedBlockingQueue(3);   // initialize the queue with capacity = 3.

queue.enqueue(1);   // Producer thread P1 enqueues 1 to the queue.
queue.enqueue(0);   // Producer thread P2 enqueues 0 to the queue.
queue.enqueue(2);   // Producer thread P3 enqueues 2 to the queue.
queue.dequeue();    // Consumer thread C1 calls dequeue.
queue.dequeue();    // Consumer thread C2 calls dequeue.
queue.dequeue();    // Consumer thread C3 calls dequeue.
queue.enqueue(3);   // One of the producer threads enqueues 3 to the queue.
queue.size();       // 1 element remaining in the queue.

Since the number of threads for producer/consumer is greater than 1, we do not know how the threads will be scheduled in the operating system, even though the input seems to imply the ordering. Therefore, any of the output [1,0,2] or [1,2,0] or [0,1,2] or [0,2,1] or [2,0,1] or [2,1,0] will be accepted.

A Semaphore is like a mutex lock except that you can specify the total capacity. Thus, it is perfect to be used when designing a bounded thread-safe queue in Python. To construct a semaphore object, we need to import the threading library.

The constructor of the threading.Semaphore takes a parameter (counter) which is default to 1. The acquire method will decrement the counter and return true immediately if the internal counter is larger than 0. Otherwise (when the counter is zero), it will block current thread until being awoken by release(). When being awoken, it will then again decrement the counter(0 and return true immediately.

The release() on the threading.Semaphore object will increment the internal counter and if the counter was zero on entry then will become one, which will wake another thread that is being blocked.

We import the collections library and use the collections.deque – the double-ended queue that allows us to push and pull on both ends of the queue.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

import threading
import collections
 
class BoundedBlockingQueue(object):
    def __init__(self, capacity: int):
        self.pushing = threading.Semaphore(capacity)
        self.pulling = threading.Semaphore(0)
        self.data = collections.deque()
 
    def enqueue(self, element: int) -> None:
        self.pushing.acquire()
        self.data.append(element)
        self.pulling.release()
 
    def dequeue(self) -> int:
        self.pulling.acquire()
        self.pushing.release()
        return self.data.popleft()
        
    def size(self) -> int:
        return len(self.data)

import threading
import collections

class BoundedBlockingQueue(object):
    def __init__(self, capacity: int):
        self.pushing = threading.Semaphore(capacity)
        self.pulling = threading.Semaphore(0)
        self.data = collections.deque()

    def enqueue(self, element: int) -> None:
        self.pushing.acquire()
        self.data.append(element)
        self.pulling.release()

    def dequeue(self) -> int:
        self.pulling.acquire()
        self.pushing.release()
        return self.data.popleft()
        
    def size(self) -> int:
        return len(self.data)

The threading.Semaphore in Python is a great way to synchronize all the threads, and the above Bounded Blocking Queue is thus thread-safe.

–EOF (The Ultimate Computing & Technology Blog) —