From 6a52d7b1a2b772b29ba82822640fa8c430ed2a12 Mon Sep 17 00:00:00 2001 From: "Antonio S. Musumeci" Date: Thu, 9 Apr 2026 20:21:36 -0400 Subject: [PATCH] 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) --- vendored/libfuse/include/bounded_queue.hpp | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/vendored/libfuse/include/bounded_queue.hpp b/vendored/libfuse/include/bounded_queue.hpp index 5e401594..74675286 100644 --- a/vendored/libfuse/include/bounded_queue.hpp +++ b/vendored/libfuse/include/bounded_queue.hpp @@ -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 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(item_)); }