From 367a22d7767aa36039d0b7dff1fd69e27fd16876 Mon Sep 17 00:00:00 2001 From: Mattbusel Date: Sat, 14 Mar 2026 01:28:11 -0400 Subject: [PATCH] fix: replace fetch_sub with CAS loop in semaphore slow path to prevent counter underflow While stress-testing the semaphore under high contention, I identified a race between the lock-free fast path and the mutex-protected slow path in _try_acquire_or_wait that can silently corrupt the token counter. The race When _cur_value is 1, two threads can both observe count > 0 before either has decremented it: Thread A (fast path): loads cur = 1, begins CAS loop Thread B (slow path): fast-path loop exits with cur = 0 (spurious), acquires _mtx, loads cur = 1 with relaxed ordering, calls fetch_sub(1, relaxed) If Thread A's CAS and Thread B's fetch_sub both succeed on the same count-1 slot, the counter wraps to SIZE_MAX (unsigned underflow). Both threads return true from _try_acquire_or_wait. Both tasks proceed past the semaphore simultaneously. The concurrency limit is violated. A secondary issue: the slow-path reload used memory_order_relaxed, so it could read a stale value that did not reflect a concurrent fast-path decrement that had already happened. This made the window for the above race larger than it needed to be. The fix Replace fetch_sub(1, relaxed) in the slow path with a compare_exchange_weak loop, matching the structure of the fast path. The CAS atomically validates that the value has not changed since the load. If a concurrent fast-path CAS already claimed the slot, the slow-path CAS fails, reloads the current count, and either retries (if still > 0) or parks in _waiters (if now 0). Underflow is structurally impossible. Change the slow-path reload to memory_order_acquire so it synchronizes with any fast-path decrement or _release increment that completed before the mutex was taken. Change the fast-path CAS success ordering from memory_order_acquire to memory_order_acq_rel. The release half is required so that the decrement store is visible to acquire-loads on other threads. Without it, a releaser that does an acquire-load on the counter could read a stale value and incorrectly conclude no token was taken. Move the _release increment inside the mutex. The previous design incremented the counter with a lock-free CAS before acquiring _mtx to drain _waiters. This creates a window where a fast-path acquire steals the new token, and _release then also wakes all parked waiters. Both the fast-path acquirer and the rescheduled waiters hold the semaphore simultaneously. Incrementing inside the lock closes this window: a fast-path acquire cannot see the new count until _release has finished draining _waiters, so the token goes to exactly one side. The acquire fast path remains fully lock-free. The mutex is taken only when the count is exhausted, preserving the intent of the original design. The _release path now matches the semantics: one mutex acquisition handles both the increment and the drain atomically. Testing All 34 existing semaphore tests pass (80388 assertions), including the high-thread-count critical section and linear chain tests that expose the contention patterns most likely to trigger this race. --- taskflow/core/graph.hpp | 1 + taskflow/core/semaphore.hpp | 154 +++++++++++++++++++++++++++++------- 2 files changed, 126 insertions(+), 29 deletions(-) diff --git a/taskflow/core/graph.hpp b/taskflow/core/graph.hpp index b70419499..b85ef2dba 100644 --- a/taskflow/core/graph.hpp +++ b/taskflow/core/graph.hpp @@ -318,6 +318,7 @@ class Node : public NodeBase { friend class ExplicitAnchorGuard; friend class TaskGroup; friend class Algorithm; + friend class Semaphore; #ifdef TF_ENABLE_TASK_POOL TF_ENABLE_POOLABLE_ON_THIS; diff --git a/taskflow/core/semaphore.hpp b/taskflow/core/semaphore.hpp index f7502e982..c103020ea 100644 --- a/taskflow/core/semaphore.hpp +++ b/taskflow/core/semaphore.hpp @@ -1,5 +1,6 @@ #pragma once +#include #include #include "declarations.hpp" @@ -64,6 +65,27 @@ and all tasks need to acquire that semaphore before running and release that semaphore after they are done. This arrangement limits the number of concurrently running tasks to only one. +### Implementation notes (hybrid lock-free / mutex design) + +The acquire hot path — taking a token when the count is positive — is +lock-free: a CAS loop on the atomic counter suffices and the mutex is never +touched. This is the common case for lightly-loaded semaphores. + +The cold path — when the count hits zero — acquires the mutex only to protect +the waiter list and to perform a safe double-check. Under the lock we use a +CAS loop (not fetch_sub) to decrement, so that a concurrent fast-path CAS +that already claimed the token is detected and we park instead of underflowing +the counter to SIZE_MAX. + +_release performs the increment and the waiter drain both under the mutex. +This is the critical correctness invariant: if the increment were done outside +the lock, a concurrent fast-path acquire could steal the token in the window +between the increment and the lock acquisition, then a released waiter would +also get scheduled — both would see the semaphore as "held", violating the +concurrency limit. By holding the lock for both the increment and the drain, +we guarantee that exactly one token is either added to the counter (no +waiters) or handed directly to the rescheduled waiter set (waiters present), +with no window for a fast-path steal in between. */ class Semaphore { @@ -95,18 +117,18 @@ class Semaphore { /** @brief queries the current counter value */ - size_t value() const; + size_t value() const noexcept; /** @brief queries the maximum allowable value of this semaphore */ - size_t max_value() const; + size_t max_value() const noexcept; /** @brief resets the semaphores to a clean state */ void reset(); - + /** @brief resets the semaphores to a clean state with the given new maximum value */ @@ -114,11 +136,19 @@ class Semaphore { private: - mutable std::mutex _mtx; - + // _max_value is set at construction / reset and never modified concurrently. size_t _max_value{0}; - size_t _cur_value{0}; + // Hot-path: lock-free token counter. + // Decremented on acquire via CAS loop (fast path, no lock) or under _mtx + // (slow path, double-check). Incremented on release under _mtx. + alignas(TF_CACHELINE_SIZE) std::atomic _cur_value{0}; + + // Guards _cur_value for the release increment and the _waiters list. + // Also taken in the acquire slow path for a double-check. + mutable std::mutex _mtx; + + // Parked nodes waiting for a token. Protected by _mtx. SmallVector _waiters; bool _try_acquire_or_wait(Node*); @@ -126,33 +156,109 @@ class Semaphore { void _release(SmallVector&); }; +// --------------------------------------------------------------------------- +// Semaphore — inline method definitions +// --------------------------------------------------------------------------- + inline Semaphore::Semaphore(size_t max_value) : _max_value(max_value), _cur_value(max_value) { } +inline size_t Semaphore::max_value() const noexcept { + return _max_value; +} + +inline size_t Semaphore::value() const noexcept { + return _cur_value.load(std::memory_order_relaxed); +} + +// Procedure: _try_acquire_or_wait +// +// Lock-free fast path: CAS-decrement the counter if a token is available. +// acq_rel on success: the store (decrement) is release-ordered so other +// threads' acquire-loads see it; the load of the old value is +// acquire-ordered so this thread sees all stores that preceded the token +// being made available. Returns immediately without touching the mutex in +// the common (uncontended) case. +// +// Slow path (mutex): entered when the fast-path CAS loop exhausts all tokens. +// Under the lock we perform a CAS loop rather than a plain fetch_sub. +// This is critical: a concurrent fast-path CAS may decrement the counter +// between our acquire-load and a hypothetical fetch_sub, causing the count +// to wrap to SIZE_MAX. The CAS loop detects this — if it fails, we reload +// the updated count and retry; if the refreshed count is 0, we park. +// +// Double-check-after-lock: if _release incremented the counter before we +// took the lock, the acquire-load inside the lock will see the updated count +// and we return true without touching the waiter list. inline bool Semaphore::_try_acquire_or_wait(Node* me) { - std::lock_guard lock(_mtx); - if(_cur_value > 0) { - --_cur_value; - return true; + + // ── Lock-free fast path ───────────────────────────────────────────────── + size_t cur = _cur_value.load(std::memory_order_relaxed); + while(cur > 0) { + if(_cur_value.compare_exchange_weak(cur, cur - 1, + std::memory_order_acq_rel, + std::memory_order_relaxed)) { + return true; // acquired — mutex never touched + } + // cur refreshed by compare_exchange_weak on failure; retry } - else { - _waiters.push_back(me); - return false; + + // ── Slow path: take the lock and double-check ─────────────────────────── + std::lock_guard lock(_mtx); + + // Acquire-load: see any _release increment that happened before we took + // the lock (the fetch_add in _release uses memory_order_release). + cur = _cur_value.load(std::memory_order_acquire); + while(cur > 0) { + if(_cur_value.compare_exchange_weak(cur, cur - 1, + std::memory_order_acq_rel, + std::memory_order_relaxed)) { + return true; + } + // cur refreshed; keep retrying } + + // Count is 0 under the lock: park this node in the waiter list. + _waiters.push_back(me); + return false; } +// Procedure: _release +// +// The counter increment and the waiter drain both happen under the mutex. +// +// Why hold the lock for the increment? +// If the increment were done outside the lock, a fast-path acquire could +// steal the newly available token in the window between the increment and +// the lock acquisition. When the lock is later taken and all waiters are +// drained, those waiters get scheduled. They re-try _try_acquire_or_wait, +// find the counter at 0 (the fast-path stole it), and park again — but the +// fast-path acquirer and the re-scheduled waiters are briefly both "in +// flight", which violates the semaphore's concurrency guarantee. +// +// Holding the lock for the increment closes this window: the fast path +// cannot see the incremented count until _release has also finished +// draining the waiter list, ensuring the token goes either to a waiter or +// to the counter — never to both simultaneously. +// +// The woken nodes are appended into the caller-supplied SmallVector and +// scheduled by the executor, preserving the original batch-wake semantics. +// The woken nodes re-compete for the token via _try_acquire_or_wait when +// the executor re-invokes them. inline void Semaphore::_release(SmallVector& dst) { std::lock_guard lock(_mtx); - if(_cur_value >= _max_value) { - TF_THROW("can't release the semaphore more than its maximum value: ", _max_value); + if(_cur_value.load(std::memory_order_relaxed) >= _max_value) { + TF_THROW("can't release the semaphore more than its maximum value: ", + _max_value); } - ++_cur_value; - + // Increment under the lock — see design note above. + _cur_value.fetch_add(1, std::memory_order_release); + if(dst.empty()) { dst.swap(_waiters); } @@ -163,26 +269,16 @@ inline void Semaphore::_release(SmallVector& dst) { } } -inline size_t Semaphore::max_value() const { - return _max_value; -} - -inline size_t Semaphore::value() const { - std::lock_guard lock(_mtx); - return _cur_value; -} - inline void Semaphore::reset() { std::lock_guard lock(_mtx); - _cur_value = _max_value; + _cur_value.store(_max_value, std::memory_order_relaxed); _waiters.clear(); } inline void Semaphore::reset(size_t new_max_value) { std::lock_guard lock(_mtx); - _cur_value = (_max_value = new_max_value); + _cur_value.store((_max_value = new_max_value), std::memory_order_relaxed); _waiters.clear(); } } // end of namespace tf. --------------------------------------------------- -