Problem Description

This problem asks you to implement a bounded blocking queue that is thread-safe and can handle concurrent operations from multiple threads.

The queue has a fixed maximum capacity and must support the following operations:

1. BoundedBlockingQueue(int capacity): Initialize the queue with a maximum capacity limit.

2. enqueue(int element): Add an element to the queue. The key requirement is that if the queue is already at full capacity, the calling thread must block (wait) until space becomes available in the queue.

3. dequeue(): Remove and return an element from the queue. If the queue is empty, the calling thread must block (wait) until an element becomes available.

4. size(): Return the current number of elements in the queue.

The "blocking" behavior is crucial here - when a producer thread tries to enqueue to a full queue, it should pause execution until a consumer thread dequeues an element. Similarly, when a consumer thread tries to dequeue from an empty queue, it should pause until a producer thread enqueues an element.

The solution uses two semaphores to coordinate this blocking behavior:

• s1 (initialized to capacity): Tracks available space in the queue. Producer threads acquire this semaphore before enqueuing.
• s2 (initialized to 0): Tracks available elements in the queue. Consumer threads acquire this semaphore before dequeuing.

The implementation uses a deque as the underlying data structure, with append() for enqueue operations and popleft() for dequeue operations, maintaining FIFO order. The semaphores ensure thread-safety and proper blocking when the queue is full or empty.

Intuition

The key insight is recognizing that we need to coordinate two types of threads with opposite concerns: producers want to know if there's space to add elements, while consumers want to know if there are elements to remove.

Think of it like a parking garage with a fixed number of spaces. Cars (producers) can only enter if there are empty spaces, and cars (consumers) can only leave if there are cars present. We need two counters: one for empty spaces and one for occupied spaces.

This naturally leads us to use semaphores, which are perfect for managing limited resources across threads. A semaphore is essentially a thread-safe counter that can block threads when the count reaches zero.

We need two semaphores because we're tracking two different resources:

1. Available capacity - starts at capacity and decreases when we enqueue
2. Available elements - starts at 0 and increases when we enqueue

The brilliant part is how these semaphores create the blocking behavior automatically:

• When enqueue() is called, we first try to acquire from the "capacity" semaphore (s1). If the queue is full, s1 is at 0, so the thread blocks until someone dequeues and releases a capacity permit.
• When dequeue() is called, we first try to acquire from the "elements" semaphore (s2). If the queue is empty, s2 is at 0, so the thread blocks until someone enqueues and releases an element permit.

After successfully acquiring the appropriate semaphore, we perform the actual queue operation, then release the opposite semaphore to signal the change to waiting threads. This acquire-operate-release pattern ensures that the queue never exceeds capacity and dequeue never operates on an empty queue, all while maintaining thread safety through the semaphores' built-in synchronization.

Solution Approach

The implementation uses two semaphores and a deque to create a thread-safe bounded blocking queue.

Data Structures Used:

• deque: A double-ended queue for efficient O(1) operations at both ends
• Two Semaphore objects for synchronization

Implementation Details:

1. Constructor (__init__):

   self.s1 = Semaphore(capacity)  # Tracks available space
   self.s2 = Semaphore(0)         # Tracks available elements
   self.q = deque()                # The actual queue

   • s1 starts with capacity permits, representing empty slots in the queue
   • s2 starts with 0 permits, representing no elements initially
   • q is the underlying queue structure

2. Enqueue Operation:

   def enqueue(self, element: int) -> None:
       self.s1.acquire()        # Wait for available space
       self.q.append(element)   # Add to queue
       self.s2.release()        # Signal element available

   • s1.acquire() decrements the available space counter. If it's 0 (queue full), the thread blocks
   • Once space is available, add the element to the queue
   • s2.release() increments the element counter, potentially unblocking a waiting consumer

3. Dequeue Operation:

   def dequeue(self) -> int:
       self.s2.acquire()        # Wait for available element
       ans = self.q.popleft()   # Remove from queue
       self.s1.release()        # Signal space available
       return ans

   • s2.acquire() decrements the element counter. If it's 0 (queue empty), the thread blocks
   • Once an element is available, remove it from the front of the queue
   • s1.release() increments the space counter, potentially unblocking a waiting producer

4. Size Operation:

   def size(self) -> int:
       return len(self.q)

   • Simply returns the current queue length
   • Note: This doesn't need synchronization as Python's len() on deque is thread-safe for reading

Why This Works:

• The semaphores act as gatekeepers: producers can't proceed when s1 = 0 (full), consumers can't proceed when s2 = 0 (empty)
• The acquire-modify-release pattern ensures atomicity of operations
• The sum s1.value + s2.value always equals capacity, maintaining the invariant
• No explicit locks are needed because semaphores handle synchronization internally

Example Walkthrough

Let's walk through a concrete example with a bounded blocking queue of capacity 2.

Initial State:

• Queue capacity: 2
• s1 (space semaphore): 2 permits (2 empty slots available)
• s2 (element semaphore): 0 permits (0 elements in queue)
• Queue: []

Step 1: Thread A calls enqueue(5)

• s1.acquire() → s1 decrements from 2 to 1 (claiming one empty slot)
• Add 5 to queue: [5]
• s2.release() → s2 increments from 0 to 1 (signaling one element available)
• State: s1=1, s2=1, Queue=[5]

Step 2: Thread B calls enqueue(10)

• s1.acquire() → s1 decrements from 1 to 0 (claiming the last empty slot)
• Add 10 to queue: [5, 10]
• s2.release() → s2 increments from 1 to 2 (signaling two elements available)
• State: s1=0, s2=2, Queue=[5, 10] (FULL)

Step 3: Thread C calls enqueue(15) (Queue is FULL)

• s1.acquire() → s1 is 0, so Thread C BLOCKS waiting for space
• Thread C is suspended...

Step 4: Thread D calls dequeue()

• s2.acquire() → s2 decrements from 2 to 1 (claiming one element)
• Remove and get 5 from queue: [10]
• s1.release() → s1 increments from 0 to 1 (signaling one space available)
• This unblocks Thread C!
• Returns: 5
• State: s1=1, s2=1, Queue=[10]

Step 5: Thread C resumes (was blocked in Step 3)

• Now s1.acquire() succeeds → s1 decrements from 1 to 0
• Add 15 to queue: [10, 15]
• s2.release() → s2 increments from 1 to 2
• State: s1=0, s2=2, Queue=[10, 15]

Step 6: Thread E calls dequeue() twice

• First dequeue():
  • s2.acquire() → s2: 2→1
  • Remove 10: [15]
  • s1.release() → s1: 0→1
  • Returns: 10
• Second dequeue():
  • s2.acquire() → s2: 1→0
  • Remove 15: []
  • s1.release() → s1: 1→2
  • Returns: 15
• State: s1=2, s2=0, Queue=[] (EMPTY)

Step 7: Thread F calls dequeue() (Queue is EMPTY)

• s2.acquire() → s2 is 0, so Thread F BLOCKS waiting for an element
• Thread F will remain blocked until another thread enqueues an element

Key Observations:

• The sum s1 + s2 always equals capacity (2 in this example)
• Producers block when s1=0 (no space), consumers block when s2=0 (no elements)
• The semaphores automatically handle the blocking and unblocking of threads
• FIFO order is maintained: elements are dequeued in the order they were enqueued

Solution Implementation

Time and Space Complexity

Time Complexity:

• __init__(capacity): O(1) - Initializing two semaphores and an empty deque takes constant time.
• enqueue(element): O(1) - Acquiring a semaphore, appending to the deque, and releasing a semaphore are all constant time operations. The append() operation on a deque is O(1).
• dequeue(): O(1) - Acquiring a semaphore, removing from the left of the deque using popleft(), and releasing a semaphore are all constant time operations. The popleft() operation on a deque is O(1).
• size(): O(1) - Getting the length of a deque is a constant time operation as Python's deque maintains its size internally.

Space Complexity:

• Overall space complexity: O(n) where n is the capacity of the queue.
• The space is used by:
  • The deque self.q which can hold up to capacity elements: O(capacity)
  • Two semaphores self.s1 and self.s2: O(1) each
  • Since the maximum number of elements that can be stored is bounded by the capacity parameter, the space complexity is O(capacity).

Learn more about how to find time and space complexity quickly.

Common Pitfalls

1. Incorrect Semaphore Order Leading to Deadlock

A critical mistake is reversing the order of semaphore operations, which can cause deadlock:

Incorrect Implementation:

def enqueue(self, element: int) -> None:
    self.queue.append(element)  # Modifying queue BEFORE acquiring semaphore
    self.empty_slots.acquire()  # Wrong order!
    self.filled_slots.release()

Why it fails: Multiple threads could append to the queue simultaneously before any semaphore is acquired, breaking thread-safety and capacity constraints.

Correct approach: Always follow the pattern: acquire → modify → release

2. Using a Regular List Instead of Deque

Problematic Implementation:

def __init__(self, capacity: int):
    self.queue = []  # Using list instead of deque

def dequeue(self) -> int:
    self.filled_slots.acquire()
    element = self.queue.pop(0)  # O(n) operation!
    self.empty_slots.release()
    return element

Why it fails: list.pop(0) is O(n) because it shifts all remaining elements. Under high concurrency, this becomes a bottleneck.

Solution: Use collections.deque with popleft() for O(1) operations.

3. Attempting to Make size() "More Accurate" with Synchronization

Over-engineered Implementation:

def size(self) -> int:
    # Unnecessary synchronization attempt
    self.empty_slots.acquire()
    self.filled_slots.acquire()
    size = len(self.queue)
    self.filled_slots.release()
    self.empty_slots.release()
    return size

Why it fails:

• This can cause deadlock if the queue is empty (filled_slots = 0)
• The size is inherently a "snapshot" in concurrent systems - it may change immediately after reading
• Python's len() on deque is already thread-safe for reading

Solution: Keep it simple - just return len(self.queue)

4. Using Lock Instead of Semaphores

Alternative but Less Elegant Implementation:

def __init__(self, capacity: int):
    self.lock = Lock()
    self.not_full = Condition(self.lock)
    self.not_empty = Condition(self.lock)
    self.queue = deque()
    self.capacity = capacity

def enqueue(self, element: int):
    with self.lock:
        while len(self.queue) >= self.capacity:
            self.not_full.wait()  # More complex logic
        self.queue.append(element)
        self.not_empty.notify()

Why semaphores are better here:

• Semaphores naturally track counts (available space/elements)
• Cleaner code without explicit while loops and condition checks
• Semaphores handle the "permit" counting automatically

5. Not Handling Thread Interruption

In production code, you might want to handle interruption:

def enqueue(self, element: int, timeout: float = None) -> bool:
    if not self.empty_slots.acquire(timeout=timeout):
        return False  # Timed out
    try:
        self.queue.append(element)
        self.filled_slots.release()
        return True
    except:
        self.empty_slots.release()  # Restore semaphore state on error
        raise

Best Practice: Consider adding timeout parameters and exception handling for production use, though the basic implementation is correct for the problem requirements.