Fix semaphore drift from enqueue_unbounded in BoundedQueue

enqueue_unbounded bypasses the semaphore so control messages
(poison pills, thread removal tasks) can never block on a full
queue. But wait_dequeue always signals a slot after consuming,
so each unbounded item that gets dequeued inflated the semaphore
by +1, permanently raising the effective max queue depth.

Fix by calling tryWait() in enqueue_unbounded before enqueueing.
When the queue has room (common case), this consumes a slot and
the dequeue-side signal restores balance perfectly. When the queue
is genuinely full (tryWait fails), the item still enqueues but
temporarily exceeds max_depth by one until consumed - a transient
overshoot bounded by control messages in flight.

A CAS-based approach that absorbed extra signals on the dequeue
side was also considered but rejected: it can starve blocked
bounded producers by swallowing the signal they are waiting for.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Antonio S. Musumeci 2026-04-09 20:21:36 -04:00
parent cb6f444e45
commit 6a52d7b1a2

View file

@ -29,6 +29,12 @@
// - Producers wait (decrement) before enqueuing, blocking when full.
// - Consumers signal (increment) after dequeuing, freeing a slot.
// This co-locates the invariant so the pairing cannot drift apart.
//
// enqueue_unbounded bypasses the semaphore for control messages that must
// never block. Each such item that is later dequeued via wait_dequeue
// causes a +1 semaphore inflation, slightly raising the effective max
// depth. This is bounded by the number of control messages in flight
// (thread removals, shutdown pills) and is acceptable for correctness.
template<typename T, typename Traits = moodycamel::ConcurrentQueueDefaultTraits>
class BoundedQueue
{
@ -127,6 +133,10 @@ public:
void
enqueue_unbounded(U &&item_)
{
// Try to consume a slot so the dequeue-side signal stays balanced.
// If the queue is full (tryWait fails), enqueue anyway - the item
// will temporarily exceed max_depth by one until it is consumed.
_slots.tryWait();
_queue.enqueue(std::forward<U>(item_));
}