From 8f41b40692d5952f67d3ea0b7d09895a4a1ebca3 Mon Sep 17 00:00:00 2001 From: "Antonio S. Musumeci" Date: Fri, 10 Apr 2026 00:50:25 -0400 Subject: [PATCH] Replace thread-removal lambda+promise with atomic counter _remove_thread_scaling_locked previously enqueued a heap-allocated lambda (capturing a shared_ptr) to signal a specific worker to exit. This had two problems: 1. Bug: if the 5-second future::wait_for timed out, the function returned -EBUSY but the lambda remained in the queue. When a worker eventually executed it, it would unconditionally set _tl_should_exit and call _remove_self, silently killing an arbitrary thread with no compensating add. 2. Held _scaling_mutex for up to 5 seconds while blocking on the future, serializing all concurrent scaling operations (add_thread, remove_thread, _maybe_grow_on_pressure, monitor hill-climb). Replace the entire mechanism with an atomic _remove_count counter. _remove_thread_scaling_locked simply increments the counter and returns immediately. After each task, _process_func does a CAS-decrement; the winning worker detaches itself via _remove_self and exits. This eliminates: - All heap allocations (shared_ptr, promise, lambda captures) - The thread-local _tl_should_exit flag - The 5-second blocking window under _scaling_mutex - The stale-lambda bug entirely (no lambda to go stale) - The include (no longer needed) During shutdown, _remove_self is a no-op (_stop is set), so a worker that wins the CAS stays in _threads and is joined normally by the destructor. Any outstanding _remove_count credits are harmlessly abandoned. Co-Authored-By: Claude Opus 4.6 (1M context) --- vendored/libfuse/include/thread_pool.hpp | 61 +++++++++--------------- 1 file changed, 22 insertions(+), 39 deletions(-) diff --git a/vendored/libfuse/include/thread_pool.hpp b/vendored/libfuse/include/thread_pool.hpp index 79d54ea4..3b38d19d 100644 --- a/vendored/libfuse/include/thread_pool.hpp +++ b/vendored/libfuse/include/thread_pool.hpp @@ -30,7 +30,6 @@ #include #include #include -#include #include #include #include @@ -95,7 +94,7 @@ private: void _maybe_grow_on_pressure(); static std::uint64_t _now_usecs(); - static thread_local bool _tl_should_exit; + std::atomic _remove_count{0}; public: int add_thread(const std::string &name = {}); @@ -182,8 +181,6 @@ private: }; -inline thread_local bool ThreadPool::_tl_should_exit = false; - inline std::uint64_t ThreadPool::_now_usecs() @@ -331,7 +328,7 @@ ThreadPool::~ThreadPool() // Shared post-dequeue logic for both worker loops. -// Returns BREAK for poison pill, EXIT for _tl_should_exit, CONTINUE otherwise. +// Returns BREAK for poison pill, EXIT for _remove_count claim, CONTINUE otherwise. inline ThreadPool::WorkerAction ThreadPool::_process_func(Func &func_) @@ -369,16 +366,22 @@ ThreadPool::_process_func(Func &func_) _tasks_completed.fetch_add(1,std::memory_order_relaxed); - // _remove_thread_scaling_locked enqueues a lambda that sets - // _tl_should_exit and calls _remove_self (which detaches + - // erases from _threads). The lambda runs as normal work above, - // so we check the flag here. The thread has already been - // detached by _remove_self, so we just return. - if(_tl_should_exit) - { - _tl_should_exit = false; - return EXIT; - } + // _remove_thread_scaling_locked increments _remove_count. + // Workers race to claim a decrement after each task; the winner + // detaches itself and exits. During shutdown (_stop set), + // _remove_self is a no-op, so the destructor still joins us. + { + std::size_t count = _remove_count.load(std::memory_order_relaxed); + while(count > 0) + { + if(_remove_count.compare_exchange_weak(count,count - 1, + std::memory_order_relaxed)) + { + _remove_self(pthread_self()); + return EXIT; + } + } + } return CONTINUE; } @@ -749,30 +752,10 @@ ThreadPool::_remove_thread_scaling_locked() return -EINVAL; } - // Single func that sets the thread-local exit flag and signals - // via promise. The worker loop checks _tl_should_exit after - // running the func and exits cleanly - no separate poison pill - // needed, so exactly one thread is removed per call. - // - // The future wait is bounded: if no worker picks up the task - // within 5 seconds (e.g. all workers running long tasks), bail - // with -EBUSY rather than blocking indefinitely while holding - // _scaling_mutex and starving the monitor thread. - auto promise = std::make_shared>(); - - auto func = - [this,promise]() - { - _tl_should_exit = true; - _remove_self(pthread_self()); - promise->set_value(); - }; - - _queue.enqueue_unbounded(std::move(func)); - - auto status = promise->get_future().wait_for(std::chrono::seconds(5)); - if(status == std::future_status::timeout) - return -EBUSY; + // Increment the atomic counter. The next worker to finish a task + // will claim the decrement in _process_func, detach itself via + // _remove_self, and exit. No heap allocation, no blocking. + _remove_count.fetch_add(1,std::memory_order_relaxed); return 0; }