Build a bounded blocking queue

Why build a blocking queue by hand when queue.Queue exists?

Because the round is asking whether you can write the synchronisation yourself, and queue.Queue hides all of it. Say in your first minute that queue.Queue is what you would ship, show it in a few lines, then build the parts: one lock guarding a deque, a condition that producers wait on while the queue is full, a condition that consumers wait on while it is empty, and a way to close it. Every one of those is a decision the interviewer wants to hear you make out loud.

Why must Condition.wait sit inside a while loop and not an if?

Because waking up means you hold the lock again, not that the thing you waited for is still there. Between the notify and the moment the woken thread gets the lock back, another thread can take the item or fill the slot. An if checks once on the way in and then acts on stale information: a consumer pops an empty deque, or a producer overfills the queue. A while checks again after every wake-up. Condition.wait_for(predicate) is the same loop written for you.

Should a blocking queue use notify or notify_all?

With one condition shared by producers and consumers, notify_all, because a single notify can wake a thread of the wrong side: a put wakes another producer, which finds the queue still full and goes back to sleep, while the consumers that could use the item never wake. Every thread can end up asleep with work in the queue. With two conditions, one for room and one for items, each waiter on a condition wants the same thing, so one notify per put or get is enough and wakes nobody for nothing.

How do you shut down a blocking queue with producers and consumers still running?

Add close(). Under the lock it sets a closed flag and calls notify_all on both conditions, so every blocked thread wakes and looks again. A put after close raises a specific exception, so a producer is refused loudly rather than having its item dropped. A get keeps handing out what is already queued, and once the queue is closed and empty it raises too, which is how a consumer learns to stop. Python 3.13 added the same behaviour to queue.Queue as shutdown() and the ShutDown exception.

How do you test a blocking queue with real threads without a flaky test?

Assert something no interleaving can change. Start at least two producers and two consumers, let each producer put a known set of items, join the producers, close the queue, join the consumers, and assert that the sorted list of items that came out equals the items that went in. That holds on every run if the queue is correct. Never assert an order, a timing, or that a race happened; those depend on the scheduler, and a test built on them passes on your machine and fails somewhere else.