Design a Thread-Safe Bounded Blocking Queue

Problem Statement

Design and implement a thread-safe bounded blocking queue for multi-producer and multi-consumer scenarios. The queue maintains a strict FIFO (First-In, First-Out) ordering and has a fixed maximum capacity.

Your class should implement the following methods:

• __init__(capacity: int): Initializes the queue with a maximum capacity. Raises ValueError if capacity None: Inserts an element into the queue. If the queue is full, the calling thread blocks until space becomes available, the specified timeout (in seconds) expires, or the queue is closed. If timeout is None, it blocks indefinitely.
• take(timeout: Optional[float] = None) -> Any: Removes and returns the head element of the queue. If the queue is empty, the calling thread blocks until an element is available, the specified timeout (in seconds) expires, or the queue is closed. If timeout is None, it blocks indefinitely.
• close() -> None: Closes the queue. All currently blocked put calls must unblock and fail. Future put calls must fail immediately. Threads waiting in take should be allowed to consume any remaining elements; once the queue is empty, subsequent or currently waiting take calls must fail.

Define and use custom exceptions QueueClosedError and QueueTimeoutError for failed operations.

Interface

from typing import Any, Optional

class QueueClosedError(Exception):
    pass

class QueueTimeoutError(Exception):
    pass

class BoundedBlockingQueue:
    def __init__(self, capacity: int):
        pass

    def put(self, item: Any, timeout: Optional[float] = None) -> None:
        pass

    def take(self, timeout: Optional[float] = None) -> Any:
        pass

    def close(self) -> None:
        pass

Example Usage

q = BoundedBlockingQueue(capacity=2)
q.put(1)
q.put(2)
# q is now full; a third put without timeout would block.

q.take()  # Returns 1
q.put(3)  # Succeeds; queue contains [2, 3]

q.close()
# q.put(4) raises QueueClosedError
q.take()  # Returns 2 (draining buffered items)
q.take()  # Returns 3
# q.take() raises QueueClosedError

Constraints and Edge Cases

• 1 <= capacity <= 10^5.
• Multiple producer and consumer threads may call methods simultaneously.
• Handle spurious wakeups correctly.
• Handle timeout=0 (immediate non-blocking check) and negative timeouts gracefully (treat negative values as non-blocking or raise ValueError).
• Ensure no memory leaks or thread starvation when threads time out or when close() is invoked.