rt/mergerfs
1
0
Fork 0
mirror of https://github.com/trapexit/mergerfs.git synced 2026-10-26 19:01:56 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Move _tasks_enqueued increment to after successful enqueue

Browse source 
The counter was incremented before the enqueue call, which can
block indefinitely for bounded queues. If tasks_pending() is used
as a drain-complete signal (pending == 0), the pre-increment
would overcount while a producer is blocked waiting for a slot.

Move the increment to after the enqueue returns in all paths:
enqueue_work, enqueue_work_autoscale, and enqueue_task. The
try_enqueue variants already incremented after success.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Antonio S. Musumeci 2026-04-09 20:39:32 -04:00
parent 3bcf66cdaa
commit 73b6bb3cdf
1 changed files with 13 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

22
vendored/libfuse/include/thread_pool.hpp
View file

@ -746,9 +746,9 @@ void
ThreadPool::enqueue_work(ThreadPool::PToken &ptok_,
FuncType &&func_)
{
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
_queue.enqueue(ptok_,
std::forward<FuncType>(func_));
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
}
template<typename FuncType>
@ -756,8 +756,8 @@ inline
void
ThreadPool::enqueue_work(FuncType &&func_)
{
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
_queue.enqueue(std::forward<FuncType>(func_));
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
}
@ -819,19 +819,21 @@ void
ThreadPool::enqueue_work_autoscale(ThreadPool::PToken &ptok_,
FuncType &&func_)
{
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
// Materialize into a Func up front so we own the value across
// the try/block-enqueue boundary without a double std::forward.
Func f(std::forward<FuncType>(func_));
if(_queue.try_enqueue(std::move(f)))
return;
{
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
return;
}
_maybe_grow_on_pressure();
// Block-enqueue (will succeed once a worker frees a slot)
_queue.enqueue(ptok_,std::move(f));
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
}
@ -840,17 +842,19 @@ inline
void
ThreadPool::enqueue_work_autoscale(FuncType &&func_)
{
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
Func f(std::forward<FuncType>(func_));
if(_queue.try_enqueue(std::move(f)))
return;
{
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
return;
}
_maybe_grow_on_pressure();
// Block-enqueue (will succeed once a worker frees a slot)
_queue.enqueue(std::move(f));
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
}
@ -888,8 +892,8 @@ ThreadPool::enqueue_task(FuncType&& func_)
}
};
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
_queue.enqueue(std::move(work));
_tasks_enqueued.fetch_add(1,std::memory_order_relaxed);
return future;
}