Design a Multi-Producer Multi-Consumer Circular Buffer

Design a Multi-Producer Multi-Consumer Circular Buffer

Implement a thread-safe, fixed-capacity circular buffer (bounded queue) that supports concurrent access by multiple producer threads and multiple consumer threads.

The queue must support blocking pushes and pops with optional timeouts, alongside a close() operation that prevents new items from being pushed while allowing existing items to be completely drained by consumers.

Interface

Implement the MPMCQueue class with the following methods:

class MPMCQueue:
    def __init__(self, capacity: int):
        """Initialize the queue with a fixed positive capacity."""
        pass

    def push(self, item: object, timeout_ms: float | None = None) -> bool:
        """
        Pushes an item into the queue. Blocks if the queue is full.
        Returns True if the item was successfully added.
        Returns False if timeout expires or the queue is closed before/during wait.
        """
        pass

    def pop(self, timeout_ms: float | None = None) -> object | None:
        """
        Removes and returns an item from the queue. Blocks if empty.
        Returns the item if available.
        Returns None if timeout expires, or if the queue is closed and completely drained.
        """
        pass

    def close(self) -> None:
        """
        Closes the queue. Prevents any new pushes.
        Wakes all waiting producers (which fail) and waiting consumers (to drain or finish).
        """
        pass

    def size(self) -> int:
        """Returns the current number of elements in the buffer."""
        pass

Requirements & Constraints

• 1 <= capacity <= 10^5
• The internal storage must use a fixed-size array/list of length capacity with head and tail pointers wrapping around via modulo arithmetic.
• Must be safe against race conditions, deadlocks, and spurious wakeups.
• Operations must provide proper publication safety: an item must be fully written to the slot before a consumer can read it.
• close() is idempotent: multiple calls have no additional effect.

Example Walkthrough

q = MPMCQueue(capacity=2)

q.push("A")        # Returns True, buffer = ["A"]
q.push("B")        # Returns True, buffer = ["A", "B"]
# Thread 1 calls q.push("C", timeout_ms=50) -> blocks because buffer is full.
# Thread 2 calls q.pop()                     -> returns "A", wakes Thread 1.
# Thread 1 unblocks and places "C"           -> returns True, buffer = ["B", "C"]

q.close()
q.push("D")        # Returns False immediately (queue closed)
q.pop()             # Returns "B"
q.pop()             # Returns "C"
q.pop()             # Returns None immediately (queue closed and empty)

Edge Cases

• Zero timeout: non-blocking attempt; fails immediately if full/empty.
• Multiple threads unblocked simultaneously when close() is called.
• High-contention churn where producers fill and consumers drain concurrently.