fix(threadpool): Fix counter mismatch root cause - memory ordering bug
ROOT CAUSE: The counter mismatch ("1 in-flight, 0 active workers") was
caused by incorrect memory ordering. totalSubmitted and totalCompleted
were using memory_order_relaxed for writes, but WaitForCompletion used
memory_order_acquire for reads. With relaxed writes and acquire reads,
there's no happens-before relationship - the acquire load doesn't
synchronize with relaxed stores.
SOLUTION: Changed all writes to totalSubmitted and totalCompleted to use
memory_order_release. Combined with the acquire reads in WaitForCompletion,
this creates proper synchronization via the release-acquire pattern.
Changes:
- ThreadPool.h Submit(): totalSubmitted now uses release ordering
- ThreadPool.h Submit() failure path: totalCompleted now uses release
- ThreadPool.cpp RecordTaskCompletion(): totalCompleted uses release
- ThreadPool.cpp outer catch: totalCompleted uses release
- ThreadPool.cpp WaitForCompletion auto-correction: uses release
This should eliminate the "ghost counter" issue that required the
auto-correction workaround added in the previous commit.
Co-Authored-By: Claude Opus 4.5 <[email protected]>
Signed-off-by: luis <[email protected]>
This commit is contained in:
committed by
luis
co-authored by
Claude Opus 4.5
parent
cb683ef67a
commit
bc739e0c8d
@@ -275,7 +275,8 @@ void WorkerThread::Run()
|
||||
// If a task was popped but an exception occurred before RecordTaskCompletion,
|
||||
// the pool's totalCompleted would never be updated, causing GetInFlightTasks()
|
||||
// to return a permanently inflated value (leading to "1 in-flight, 0 active workers")
|
||||
_pool->_metrics.totalCompleted.fetch_add(1, ::std::memory_order_relaxed);
|
||||
// CRITICAL: Use release ordering to synchronize with acquire loads in WaitForCompletion
|
||||
_pool->_metrics.totalCompleted.fetch_add(1, ::std::memory_order_release);
|
||||
|
||||
if (_diagnostics)
|
||||
{
|
||||
@@ -812,7 +813,8 @@ bool ThreadPool::WaitForCompletion(::std::chrono::milliseconds timeout)
|
||||
inFlight);
|
||||
|
||||
// Correct the mismatch by advancing totalCompleted
|
||||
_metrics.totalCompleted.fetch_add(inFlight, ::std::memory_order_relaxed);
|
||||
// CRITICAL: Use release ordering to synchronize with acquire loads in WaitForCompletion
|
||||
_metrics.totalCompleted.fetch_add(inFlight, ::std::memory_order_release);
|
||||
return true; // All work is actually done
|
||||
}
|
||||
|
||||
@@ -1005,7 +1007,8 @@ void ThreadPool::RecordTaskCompletion(Task* task)
|
||||
auto latency = ::std::chrono::duration_cast<::std::chrono::microseconds>(
|
||||
completionTime - task->submittedAt).count();
|
||||
|
||||
_metrics.totalCompleted.fetch_add(1, ::std::memory_order_relaxed);
|
||||
// CRITICAL: Use release ordering to synchronize with acquire loads in WaitForCompletion
|
||||
_metrics.totalCompleted.fetch_add(1, ::std::memory_order_release);
|
||||
_metrics.totalLatency.fetch_add(latency, ::std::memory_order_relaxed);
|
||||
|
||||
// Clean up task
|
||||
|
||||
@@ -526,7 +526,8 @@ public:
|
||||
);
|
||||
|
||||
// Update metrics
|
||||
_metrics.totalSubmitted.fetch_add(1, ::std::memory_order_relaxed);
|
||||
// CRITICAL: Use release ordering to synchronize with acquire loads in WaitForCompletion
|
||||
_metrics.totalSubmitted.fetch_add(1, ::std::memory_order_release);
|
||||
size_t priorityIndex = static_cast<size_t>(priority);
|
||||
_metrics.tasksByPriority[priorityIndex].fetch_add(1, ::std::memory_order_relaxed);
|
||||
|
||||
@@ -549,7 +550,8 @@ public:
|
||||
// We already incremented totalSubmitted above, but task will not be executed.
|
||||
// We MUST increment totalCompleted to prevent GetInFlightTasks() from returning
|
||||
// a permanently inflated value, which would cause WaitForCompletion() to block forever.
|
||||
_metrics.totalCompleted.fetch_add(1, ::std::memory_order_relaxed);
|
||||
// CRITICAL: Use release ordering to synchronize with acquire loads in WaitForCompletion
|
||||
_metrics.totalCompleted.fetch_add(1, ::std::memory_order_release);
|
||||
|
||||
delete task;
|
||||
throw ::std::runtime_error("All worker queues are full");
|
||||
|
||||
Reference in New Issue
Block a user