Replace thread-removal lambda+promise with atomic counter

_remove_thread_scaling_locked previously enqueued a heap-allocated
lambda (capturing a shared_ptr<promise>) 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 <memory> 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) <noreply@anthropic.com>
This commit is contained in:
Antonio S. Musumeci 2026-04-10 00:50:25 -04:00
parent af2419ff71
commit 8f41b40692

View file

@ -30,7 +30,6 @@
#include <csignal>
#include <cstring>
#include <future>
#include <memory>
#include <stdexcept>
#include <string>
#include <thread>
@ -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<std::size_t> _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<std::promise<void>>();
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;
}