From af2419ff7100e616b3fd8643c38d0717845d16c7 Mon Sep 17 00:00:00 2001 From: "Antonio S. Musumeci" Date: Thu, 9 Apr 2026 23:50:54 -0400 Subject: [PATCH] Extract _process_func to deduplicate worker loop logic start_routine and start_routine_autoscale shared ~30 lines of identical post-dequeue logic: poison-pill detection, exception-guarded execution, func destruction, tasks_completed bookkeeping, and the _tl_should_exit exit protocol. Any bug fix to this logic had to be applied in both places. Extract the shared code into _process_func which returns a WorkerAction enum (CONTINUE / BREAK / EXIT). Both worker loops now call this helper, reducing the maintenance surface and eliminating the duplication. Also document the BoundedQueue::try_enqueue no-move-on-failure contract in enqueue_work_autoscale, since the correctness of the try-then-block pattern depends on it. Co-Authored-By: Claude Opus 4.6 (1M context) --- vendored/libfuse/include/thread_pool.hpp | 149 ++++++++++------------- 1 file changed, 67 insertions(+), 82 deletions(-) diff --git a/vendored/libfuse/include/thread_pool.hpp b/vendored/libfuse/include/thread_pool.hpp index 40736b4a..79d54ea4 100644 --- a/vendored/libfuse/include/thread_pool.hpp +++ b/vendored/libfuse/include/thread_pool.hpp @@ -80,11 +80,14 @@ public: ThreadPool(ThreadPool&&) = delete; ThreadPool& operator=(ThreadPool&&) = delete; + enum WorkerAction { CONTINUE, BREAK, EXIT }; + private: static void *start_routine(void *arg_); static void *start_routine_autoscale(void *arg_); static void *monitor_routine(void *arg_); + WorkerAction _process_func(Func &func_); int _add_thread_scaling_locked(const std::string &name_); int _remove_thread_scaling_locked(); void _remove_self(pthread_t t_); @@ -327,6 +330,60 @@ ThreadPool::~ThreadPool() } +// Shared post-dequeue logic for both worker loops. +// Returns BREAK for poison pill, EXIT for _tl_should_exit, CONTINUE otherwise. +inline +ThreadPool::WorkerAction +ThreadPool::_process_func(Func &func_) +{ + // Empty func is a poison pill - exit cleanly. + // If _stop is set, destructor owns the join - don't self-remove. + // Otherwise this is a set_threads/remove_thread shrink - detach. + if(!func_) + { + if(!_stop.load(std::memory_order_acquire)) + _remove_self(pthread_self()); + return BREAK; + } + + try + { + func_(); + } + catch(const std::exception &e) + { + syslog(LOG_CRIT, + "threadpool (%s): uncaught exception caught by worker - %s", + _name.c_str(), + e.what()); + } + catch(...) + { + syslog(LOG_CRIT, + "threadpool (%s): uncaught non-standard exception caught by worker", + _name.c_str()); + } + + // Force destruction to release resources + 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; + } + + return CONTINUE; +} + + // Standard worker loop: blocks indefinitely on dequeue. inline void* @@ -342,49 +399,9 @@ ThreadPool::start_routine(void *arg_) { q.wait_dequeue(ctok,func); - // Empty func is a poison pill - exit cleanly. - // If _stop is set, destructor owns the join - don't self-remove. - // Otherwise this is a set_threads/remove_thread shrink - detach. - if(!func) - { - if(!btp->_stop.load(std::memory_order_acquire)) - btp->_remove_self(pthread_self()); - break; - } - - try - { - func(); - } - catch(const std::exception &e) - { - syslog(LOG_CRIT, - "threadpool (%s): uncaught exception caught by worker - %s", - btp->_name.c_str(), - e.what()); - } - catch(...) - { - syslog(LOG_CRIT, - "threadpool (%s): uncaught non-standard exception caught by worker", - btp->_name.c_str()); - } - - // Force destruction to release resources - func = {}; - - btp->_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 nullptr; - } + WorkerAction action = btp->_process_func(func); + if(action != CONTINUE) + break; } return nullptr; @@ -438,45 +455,9 @@ ThreadPool::start_routine_autoscale(void *arg_) continue; } - // Empty func is a poison pill - exit cleanly. - // If _stop is set, destructor owns the join - don't self-remove. - // Otherwise this is a set_threads/remove_thread shrink - detach. - if(!func) - { - if(!btp->_stop.load(std::memory_order_acquire)) - btp->_remove_self(pthread_self()); - break; - } - - try - { - func(); - } - catch(const std::exception &e) - { - syslog(LOG_CRIT, - "threadpool (%s): uncaught exception caught by worker - %s", - btp->_name.c_str(), - e.what()); - } - catch(...) - { - syslog(LOG_CRIT, - "threadpool (%s): uncaught non-standard exception caught by worker", - btp->_name.c_str()); - } - - // Force destruction to release resources - func = {}; - - btp->_tasks_completed.fetch_add(1,std::memory_order_relaxed); - - // See comment in start_routine for the _tl_should_exit protocol. - if(_tl_should_exit) - { - _tl_should_exit = false; - return nullptr; - } + WorkerAction action = btp->_process_func(func); + if(action != CONTINUE) + break; } return nullptr; @@ -920,6 +901,9 @@ ThreadPool::enqueue_work_autoscale(ThreadPool::PToken &ptok_, { // Materialize into a Func up front so we own the value across // the try/block-enqueue boundary without a double std::forward. + // + // Safe: BoundedQueue::try_enqueue checks the semaphore before + // forwarding, so f is not moved-from on failure. Func f(std::forward(func_)); if(_queue.try_enqueue(std::move(f))) @@ -941,6 +925,7 @@ inline void ThreadPool::enqueue_work_autoscale(FuncType &&func_) { + // See ptok_ overload for safety note on try_enqueue + move. Func f(std::forward(func_)); if(_queue.try_enqueue(std::move(f)))