diff --git a/tests/tests.cpp b/tests/tests.cpp index c8b8c264..f7b34d78 100644 --- a/tests/tests.cpp +++ b/tests/tests.cpp @@ -1700,7 +1700,7 @@ test_tp_remove_thread() int rv = tp.remove_thread(); TEST_CHECK(rv == 0); - TEST_CHECK(tp.threads().size() == 2); + TEST_CHECK(wait_until([&tp](){ return tp.thread_count() == 2; })); } void @@ -1739,7 +1739,7 @@ test_tp_set_threads_shrink() int rv = tp.set_threads(2); TEST_CHECK(rv == 0); - TEST_CHECK(tp.threads().size() == 2); + TEST_CHECK(wait_until([&tp](){ return tp.thread_count() == 2; })); } void @@ -2104,7 +2104,7 @@ test_tp_remove_thread_under_load() TEST_CHECK(rv1 == 0); TEST_CHECK(rv2 == 0); - TEST_CHECK(tp.threads().size() == 2); + TEST_CHECK(wait_until([&tp](){ return tp.thread_count() == 2; })); TEST_CHECK(wait_until([&counter, N](){ return counter.load() == N; }, 10000)); TEST_CHECK(counter.load() == N); @@ -2460,6 +2460,602 @@ test_tp_heavy_add_remove_churn_under_enqueue() TEST_CHECK(executed.load() == TOTAL); } +// ===================================================================== +// ThreadPool — auto-scaling and introspection tests +// ===================================================================== + +void +test_tp_construct_zero_queue_depth_throws() +{ + bool threw = false; + + try + { + ThreadPool tp(2, 0, "test.zqdepth"); + } + catch(const std::invalid_argument &) + { + threw = true; + } + + TEST_CHECK(threw); +} + +void +test_tp_construct_scaling_min_gt_max_throws() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 10; + cfg.max_threads = 5; + + bool threw = false; + + try + { + ThreadPool tp(2, 4, "test.mingtmax", cfg); + } + catch(const std::invalid_argument &) + { + threw = true; + } + + TEST_CHECK(threw); +} + +void +test_tp_construct_scaling_negative_interval_throws() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 2; + cfg.max_threads = 8; + cfg.sample_interval_usecs = 0; + + bool threw = false; + + try + { + ThreadPool tp(2, 4, "test.negival", cfg); + } + catch(const std::invalid_argument &) + { + threw = true; + } + + TEST_CHECK(threw); +} + +void +test_tp_construct_scaling_negative_cooldown_throws() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 2; + cfg.max_threads = 8; + cfg.cooldown_usecs = -1; + + bool threw = false; + + try + { + ThreadPool tp(2, 4, "test.negcool", cfg); + } + catch(const std::invalid_argument &) + { + threw = true; + } + + TEST_CHECK(threw); +} + +void +test_tp_construct_scaling_negative_idle_throws() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 2; + cfg.max_threads = 8; + cfg.idle_threshold_usecs = 0; + + bool threw = false; + + try + { + ThreadPool tp(2, 4, "test.negidle", cfg); + } + catch(const std::invalid_argument &) + { + threw = true; + } + + TEST_CHECK(threw); +} + +void +test_tp_construct_scaling_clamps_initial_count() +{ + // Request 1 thread, but min_threads is 3 — should clamp up to 3 + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 3; + cfg.max_threads = 6; + + ThreadPool tp(1, 8, "test.clampup", cfg); + + TEST_CHECK(tp.thread_count() >= 3); + + // Request 10 threads, but max_threads is 6 — should clamp down to 6 + ThreadPool::ScalingConfig cfg2; + + cfg2.min_threads = 2; + cfg2.max_threads = 6; + + ThreadPool tp2(10, 8, "test.clampdn", cfg2); + + TEST_CHECK(tp2.thread_count() <= 6); +} + +void +test_tp_introspection_thread_count() +{ + ThreadPool tp(3, 8, "test.itc"); + + TEST_CHECK(tp.thread_count() == 3); + + tp.add_thread(); + + TEST_CHECK(tp.thread_count() == 4); +} + +void +test_tp_introspection_queue_capacity() +{ + ThreadPool tp(2, 16, "test.iqcap"); + + TEST_CHECK(tp.queue_capacity() == 16); +} + +void +test_tp_introspection_tasks_enqueued_completed() +{ + ThreadPool tp(2, 64, "test.ienqcomp"); + + TEST_CHECK(tp.tasks_enqueued() == 0); + TEST_CHECK(tp.tasks_completed() == 0); + + const int N = 20; + std::atomic counter{0}; + + for(int i = 0; i < N; ++i) + tp.enqueue_work([&counter](){ counter.fetch_add(1); }); + + TEST_CHECK(wait_until([&counter, N](){ return counter.load() == N; })); + // Give a moment for tasks_completed to catch up + TEST_CHECK(wait_until([&tp, N]() + { + return tp.tasks_completed() == static_cast(N); + })); + + TEST_CHECK(tp.tasks_enqueued() == static_cast(N)); + TEST_CHECK(tp.tasks_completed() == static_cast(N)); +} + +void +test_tp_introspection_tasks_pending() +{ + ThreadPool tp(1, 4, "test.ipend"); + + std::atomic release{false}; + std::atomic started{false}; + + tp.enqueue_work([&release, &started]() + { + started.store(true); + while(!release.load()) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + }); + + TEST_CHECK(wait_until([&started](){ return started.load(); })); + + // Enqueue a couple more while the first is blocked + tp.enqueue_work([](){}); + tp.enqueue_work([](){}); + + // tasks_pending should be > 0 (at least the 2 queued + 1 running) + TEST_CHECK(tp.tasks_pending() > 0); + + release.store(true); + + TEST_CHECK(wait_until([&tp](){ return tp.tasks_pending() == 0; })); +} + +void +test_tp_introspection_queue_depth() +{ + ThreadPool tp(1, 8, "test.iqdepth"); + + std::atomic release{false}; + std::atomic started{false}; + + tp.enqueue_work([&release, &started]() + { + started.store(true); + while(!release.load()) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + }); + + TEST_CHECK(wait_until([&started](){ return started.load(); })); + + tp.enqueue_work([](){}); + tp.enqueue_work([](){}); + + // Queue depth should be approximately 2 + std::size_t depth = tp.queue_depth(); + TEST_CHECK(depth >= 1); + + release.store(true); + + TEST_CHECK(wait_until([&tp](){ return tp.queue_depth() == 0; }, 5000)); +} + +void +test_tp_enqueue_work_autoscale() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 2; + cfg.max_threads = 8; + cfg.idle_threshold_usecs = 500000; // 500ms + cfg.sample_interval_usecs = 100000; // 100ms + cfg.cooldown_usecs = 50000; // 50ms + + ThreadPool tp(2, 4, "test.eas", cfg); + + std::atomic counter{0}; + const int N = 50; + + for(int i = 0; i < N; ++i) + tp.enqueue_work_autoscale([&counter](){ counter.fetch_add(1); }); + + TEST_CHECK(wait_until([&counter, N](){ return counter.load() == N; }, 5000)); + TEST_CHECK(counter.load() == N); +} + +void +test_tp_enqueue_work_autoscale_with_ptoken() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 2; + cfg.max_threads = 8; + cfg.idle_threshold_usecs = 500000; + cfg.sample_interval_usecs = 100000; + cfg.cooldown_usecs = 50000; + + ThreadPool tp(2, 4, "test.easp", cfg); + + std::atomic counter{0}; + const int N = 50; + auto ptok = tp.ptoken(); + + for(int i = 0; i < N; ++i) + tp.enqueue_work_autoscale(ptok, [&counter](){ counter.fetch_add(1); }); + + TEST_CHECK(wait_until([&counter, N](){ return counter.load() == N; }, 5000)); + TEST_CHECK(counter.load() == N); +} + +void +test_tp_autoscale_grows_under_pressure() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 1; + cfg.max_threads = 8; + cfg.idle_threshold_usecs = 2000000; // 2s + cfg.sample_interval_usecs = 100000; // 100ms + cfg.cooldown_usecs = 10000; // 10ms - fast cooldown for test + + // Queue depth of 2 with 1 thread — pressure should trigger growth + ThreadPool tp(1, 2, "test.easgrow", cfg); + + TEST_CHECK(tp.thread_count() == 1); + + std::atomic counter{0}; + const int N = 100; + + // Saturate with slow tasks to trigger pressure growth + for(int i = 0; i < N; ++i) + tp.enqueue_work_autoscale([&counter]() + { + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + counter.fetch_add(1); + }); + + TEST_CHECK(wait_until([&counter, N](){ return counter.load() == N; }, 15000)); + TEST_CHECK(counter.load() == N); + + // Pool should have grown beyond 1 thread at some point + // (We can't reliably assert thread_count() > 1 at this instant + // since idle timeout may have already shrunk it back, but the work + // completing in reasonable time proves growth happened.) +} + +void +test_tp_autoscale_idle_self_exit() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 1; + cfg.max_threads = 8; + cfg.idle_threshold_usecs = 200000; // 200ms idle timeout + cfg.sample_interval_usecs = 100000; + cfg.cooldown_usecs = 10000; + + ThreadPool tp(1, 2, "test.easidle", cfg); + + // Manually grow + tp.add_thread(); + tp.add_thread(); + tp.add_thread(); + + TEST_CHECK(tp.thread_count() == 4); + + // Wait for idle threads to self-exit (200ms timeout + margin) + TEST_CHECK(wait_until([&tp]() + { + return tp.thread_count() <= 1; + }, + 5000)); + + std::size_t count = tp.thread_count(); + TEST_CHECK(count >= 1); + TEST_CHECK(count <= 2); // should have shrunk significantly +} + +void +test_tp_set_threads_clamps_to_scaling_bounds() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 2; + cfg.max_threads = 6; + + ThreadPool tp(3, 16, "test.stclamp", cfg); + + // Try to set below min — should clamp to 2 + tp.set_threads(1); + + TEST_CHECK(tp.thread_count() >= 2); + + // Try to set above max — should clamp to 6 + tp.set_threads(100); + + // Wait for threads to settle + TEST_CHECK(wait_until([&tp](){ return tp.thread_count() <= 6; })); + TEST_CHECK(tp.thread_count() <= 6); +} + +void +test_tp_enqueue_task_autoscale_path() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 2; + cfg.max_threads = 8; + cfg.idle_threshold_usecs = 500000; + cfg.sample_interval_usecs = 100000; + cfg.cooldown_usecs = 50000; + + ThreadPool tp(2, 4, "test.etauto", cfg); + + // enqueue_task uses the autoscale try-then-block path when dynamic scaling is enabled + std::vector> futures; + const int N = 30; + + for(int i = 0; i < N; ++i) + futures.push_back(tp.enqueue_task([i]() -> int { return i * 2; })); + + for(int i = 0; i < N; ++i) + TEST_CHECK(futures[i].get() == i * 2); +} + +void +test_tp_autoscale_stress_concurrent_enqueue() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 2; + cfg.max_threads = 16; + cfg.idle_threshold_usecs = 1000000; + cfg.sample_interval_usecs = 50000; + cfg.cooldown_usecs = 10000; + + ThreadPool tp(2, 8, "test.easconc", cfg); + + const int PRODUCERS = 6; + const int PER_PRODUCER = 500; + const int TOTAL = PRODUCERS * PER_PRODUCER; + std::atomic counter{0}; + + std::vector producers; + for(int p = 0; p < PRODUCERS; ++p) + { + producers.emplace_back([&tp, &counter]() + { + auto ptok = tp.ptoken(); + for(int i = 0; i < PER_PRODUCER; ++i) + tp.enqueue_work_autoscale(ptok, [&counter]() + { + counter.fetch_add(1); + }); + }); + } + + for(auto &t : producers) + t.join(); + + TEST_CHECK(wait_until([&counter, TOTAL](){ return counter.load() == TOTAL; }, 15000)); + TEST_CHECK(counter.load() == TOTAL); +} + +void +test_tp_autoscale_with_slow_tasks_stress() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 1; + cfg.max_threads = 8; + cfg.idle_threshold_usecs = 1000000; + cfg.sample_interval_usecs = 50000; + cfg.cooldown_usecs = 10000; + + ThreadPool tp(1, 4, "test.easslow", cfg); + + std::atomic counter{0}; + const int N = 80; + + for(int i = 0; i < N; ++i) + tp.enqueue_work_autoscale([&counter]() + { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + counter.fetch_add(1); + }); + + TEST_CHECK(wait_until([&counter, N](){ return counter.load() == N; }, 30000)); + TEST_CHECK(counter.load() == N); +} + +void +test_tp_autoscale_repeated_construct_destroy() +{ + const int ROUNDS = 50; + + for(int r = 0; r < ROUNDS; ++r) + { + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 1; + cfg.max_threads = 4; + cfg.idle_threshold_usecs = 200000; + cfg.sample_interval_usecs = 50000; + cfg.cooldown_usecs = 10000; + + std::atomic counter{0}; + + { + ThreadPool tp(2, 32, "test.easctor", cfg); + + for(int i = 0; i < 100; ++i) + tp.enqueue_work_autoscale([&counter]() + { + counter.fetch_add(1); + }); + } + + TEST_CHECK(counter.load() == 100); + } +} + +void +test_tp_remove_thread_refuses_below_scaling_min() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 3; + cfg.max_threads = 6; + + ThreadPool tp(4, 8, "test.rmscmin", cfg); + + TEST_CHECK(tp.thread_count() == 4); + + // Should succeed: 4 -> 3 + int rv = tp.remove_thread(); + TEST_CHECK(rv == 0); + + // Wait for removal to take effect + TEST_CHECK(wait_until([&tp](){ return tp.thread_count() == 3; })); + + // Should fail: 3 is min_threads + rv = tp.remove_thread(); + TEST_CHECK(rv == -EINVAL); + TEST_CHECK(tp.thread_count() == 3); +} + +void +test_tp_add_thread_refuses_above_scaling_max() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 2; + cfg.max_threads = 3; + + ThreadPool tp(3, 8, "test.addscmax", cfg); + + TEST_CHECK(tp.thread_count() == 3); + + // Should fail: already at max_threads + int rv = tp.add_thread(); + TEST_CHECK(rv == -EAGAIN); + TEST_CHECK(tp.thread_count() == 3); +} + +void +test_tp_autoscale_mixed_resize_and_autoscale_enqueue() +{ + ThreadPool::ScalingConfig cfg; + + cfg.min_threads = 2; + cfg.max_threads = 10; + cfg.idle_threshold_usecs = 1000000; + cfg.sample_interval_usecs = 50000; + cfg.cooldown_usecs = 10000; + + ThreadPool tp(3, 64, "test.easmix", cfg); + + const int PRODUCERS = 4; + const int PER_PRODUCER = 300; + const int TOTAL = PRODUCERS * PER_PRODUCER; + std::atomic counter{0}; + std::atomic producers_done{false}; + + std::thread resizer([&tp, &producers_done]() + { + int i = 0; + while(!producers_done.load()) + { + std::size_t target = 2 + (i % 8); + tp.set_threads(target); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + ++i; + } + }); + + std::vector producers; + for(int p = 0; p < PRODUCERS; ++p) + { + producers.emplace_back([&tp, &counter]() + { + auto ptok = tp.ptoken(); + for(int i = 0; i < PER_PRODUCER; ++i) + tp.enqueue_work_autoscale(ptok, [&counter]() + { + counter.fetch_add(1); + }); + }); + } + + for(auto &t : producers) + t.join(); + + producers_done.store(true); + resizer.join(); + + TEST_CHECK(wait_until([&counter, TOTAL](){ return counter.load() == TOTAL; }, 15000)); + TEST_CHECK(counter.load() == TOTAL); +} + void test_config_has_key() { @@ -3161,5 +3757,28 @@ TEST_LIST = {"tp_heavy_repeated_construct_destroy",test_tp_heavy_repeated_construct_destroy}, {"tp_heavy_enqueue_task_mixed_outcomes",test_tp_heavy_enqueue_task_mixed_outcomes}, {"tp_heavy_add_remove_churn_under_enqueue",test_tp_heavy_add_remove_churn_under_enqueue}, + {"tp_construct_zero_queue_depth_throws",test_tp_construct_zero_queue_depth_throws}, + {"tp_construct_scaling_min_gt_max_throws",test_tp_construct_scaling_min_gt_max_throws}, + {"tp_construct_scaling_negative_interval_throws",test_tp_construct_scaling_negative_interval_throws}, + {"tp_construct_scaling_negative_cooldown_throws",test_tp_construct_scaling_negative_cooldown_throws}, + {"tp_construct_scaling_negative_idle_throws",test_tp_construct_scaling_negative_idle_throws}, + {"tp_construct_scaling_clamps_initial_count",test_tp_construct_scaling_clamps_initial_count}, + {"tp_introspection_thread_count",test_tp_introspection_thread_count}, + {"tp_introspection_queue_capacity",test_tp_introspection_queue_capacity}, + {"tp_introspection_tasks_enqueued_completed",test_tp_introspection_tasks_enqueued_completed}, + {"tp_introspection_tasks_pending",test_tp_introspection_tasks_pending}, + {"tp_introspection_queue_depth",test_tp_introspection_queue_depth}, + {"tp_enqueue_work_autoscale",test_tp_enqueue_work_autoscale}, + {"tp_enqueue_work_autoscale_with_ptoken",test_tp_enqueue_work_autoscale_with_ptoken}, + {"tp_autoscale_grows_under_pressure",test_tp_autoscale_grows_under_pressure}, + {"tp_autoscale_idle_self_exit",test_tp_autoscale_idle_self_exit}, + {"tp_set_threads_clamps_to_scaling_bounds",test_tp_set_threads_clamps_to_scaling_bounds}, + {"tp_enqueue_task_autoscale_path",test_tp_enqueue_task_autoscale_path}, + {"tp_autoscale_stress_concurrent_enqueue",test_tp_autoscale_stress_concurrent_enqueue}, + {"tp_autoscale_with_slow_tasks_stress",test_tp_autoscale_with_slow_tasks_stress}, + {"tp_autoscale_repeated_construct_destroy",test_tp_autoscale_repeated_construct_destroy}, + {"tp_remove_thread_refuses_below_scaling_min",test_tp_remove_thread_refuses_below_scaling_min}, + {"tp_add_thread_refuses_above_scaling_max",test_tp_add_thread_refuses_above_scaling_max}, + {"tp_autoscale_mixed_resize_and_autoscale_enqueue",test_tp_autoscale_mixed_resize_and_autoscale_enqueue}, {NULL,NULL} }; diff --git a/vendored/libfuse/include/thread_pool.hpp b/vendored/libfuse/include/thread_pool.hpp index 69c15c6a..1f8a7302 100644 --- a/vendored/libfuse/include/thread_pool.hpp +++ b/vendored/libfuse/include/thread_pool.hpp @@ -96,6 +96,7 @@ private: void _remove_self(pthread_t t_); bool _try_remove_self_if_above(pthread_t t_, std::size_t min_count_); bool _try_claim_removal(); + bool _try_claim_and_remove_self(pthread_t t_); void _maybe_grow_on_pressure(); static std::uint64_t _now_usecs(); static void _set_thread_name(pthread_t t_, const std::string &name_); @@ -494,17 +495,13 @@ ThreadPool::_process_func(Func &func_) if(_stop.load(std::memory_order_acquire)) return BREAK; - // Try to claim a pending shrink request. Each call to - // _remove_thread_scaling_locked increments _remove_count and - // enqueues a wakeup pill. If the count is still positive, - // this pill is ours — detach and exit. If the count is zero, - // a worker already claimed the decrement via the task- - // completion path below; this pill is stale — keep working. - if(_try_claim_removal()) - { - _remove_self(pthread_self()); - return EXIT; - } + // Try to claim a pending shrink request and atomically remove + // self under both _scaling_mutex and _threads_mutex. This + // prevents set_threads from observing a stale effective count + // (decremented _remove_count but _threads not yet shrunk) and + // issuing excess removals that drain the pool to zero. + if(_try_claim_and_remove_self(pthread_self())) + return EXIT; return CONTINUE; } @@ -534,13 +531,12 @@ ThreadPool::_process_func(Func &func_) // _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. - if(_try_claim_removal()) - { - _remove_self(pthread_self()); - return EXIT; - } + // detaches itself and exits. Uses the combined claim+remove to + // prevent the TOCTOU race with set_threads. During shutdown + // (_stop set), _try_claim_and_remove_self returns false, so + // the destructor still joins us. + if(_try_claim_and_remove_self(pthread_self())) + return EXIT; return CONTINUE; } @@ -880,6 +876,61 @@ ThreadPool::_try_claim_removal() } +// Atomically claim a pending removal AND remove self from the thread +// list under both _scaling_mutex and _threads_mutex. This closes the +// TOCTOU race between the old _try_claim_removal + _remove_self +// two-step: without atomicity, set_threads can observe the decremented +// _remove_count before _threads is shrunk, miscalculate effective +// count, and issue excess removals that drain the pool to zero. +inline +bool +ThreadPool::_try_claim_and_remove_self(pthread_t t_) +{ + mutex_lockguard(_scaling_mutex); + mutex_lockguard(_threads_mutex); + + if(_stop.load(std::memory_order_acquire)) + return false; + + // Try to claim a pending removal. + 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_acq_rel)) + goto claimed; + } + return false; + +claimed: + // Safety floor: never remove the last thread. Use the same + // min_allowed logic as _remove_thread_scaling_locked. + { + std::size_t min_allowed = 1; + if(_dynamic_scaling_enabled && _scaling_config.min_threads > min_allowed) + min_allowed = _scaling_config.min_threads; + if(_threads.size() <= min_allowed) + { + // Restore the claim — a future set_threads or grow will + // either cancel it or the pool will be large enough. + _remove_count.fetch_add(1,std::memory_order_relaxed); + return false; + } + } + + auto it = std::find(_threads.begin(),_threads.end(),t_); + if(it == _threads.end()) + { + _remove_count.fetch_add(1,std::memory_order_relaxed); + return false; + } + + _threads.erase(it); + pthread_detach(t_); + return true; +} + + inline void ThreadPool::_remove_self(pthread_t t_) @@ -1156,6 +1207,7 @@ ThreadPool::set_threads(const std::size_t count_) std::size_t effective = (current > pending) ? (current - pending) : 0; int rv = 0; + bool changed = false; for(std::size_t i = effective; i < count; ++i) { @@ -1165,6 +1217,7 @@ ThreadPool::set_threads(const std::size_t count_) rv = err; break; } + changed = true; } for(std::size_t i = count; i < effective; ++i) @@ -1175,8 +1228,14 @@ ThreadPool::set_threads(const std::size_t count_) rv = err; break; } + changed = true; } + // Stamp the cooldown timer so the monitor doesn't immediately + // perturb the thread count after an explicit set_threads call. + if(changed && _dynamic_scaling_enabled) + _last_scale_time_usecs.store(_now_usecs(),std::memory_order_relaxed); + return rv; }