From bda223aa8a24e649c8bb37269c62f8e98b962ba3 Mon Sep 17 00:00:00 2001 From: Coldwings Date: Thu, 13 Aug 2026 12:00:31 +0800 Subject: [PATCH] fix(runtime): drain autoscaler example phases --- .github/workflows/ci.yml | 4 + CHANGELOG.md | 4 + examples/autoscaler_example.cpp | 186 ++++++++++++++++++++++---------- wiki/Examples.md | 5 + 4 files changed, 144 insertions(+), 55 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1dbdd284..c8161cec 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -186,6 +186,10 @@ jobs: if: needs.changes.outputs.code == 'true' && matrix.build_type == 'Release' run: timeout --kill-after=5s 30s ./build/examples/scheduler_service_benchmark --smoke + - name: Run autoscaler example smoke test + if: needs.changes.outputs.code == 'true' && matrix.build_type == 'Release' + run: timeout --kill-after=5s 15s ./build/examples/autoscaler_example --smoke + package-consumer-check: needs: changes runs-on: ubuntu-24.04 diff --git a/CHANGELOG.md b/CHANGELOG.md index 90876651..6c965f07 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **Autoscaler example task lifetime**: Each load phase now retains join handles + and drains every submitted task before releasing its phase-local completion + counter or beginning the low-load phase. The shorter workload and CI smoke + mode keep termination coverage practical (#1022). - **Custom autoscaler trigger packs**: `autoscaler` now constructs and dispatches each configured trigger correctly instead of failing template instantiation because the trigger argument was omitted. diff --git a/examples/autoscaler_example.cpp b/examples/autoscaler_example.cpp index 0e3e0a9a..597d741d 100644 --- a/examples/autoscaler_example.cpp +++ b/examples/autoscaler_example.cpp @@ -3,22 +3,129 @@ #include #include #include +#include #include +#include +#include +#include using namespace elio; +using namespace std::chrono_literals; + +namespace { + +struct example_options { + int phase1_tasks = 400; + int phase2_tasks = 600; + int load_samples = 10; + int idle_samples = 20; + std::chrono::milliseconds tick_interval{100}; + std::chrono::seconds idle_delay{1}; + std::chrono::milliseconds min_work{5}; + std::chrono::milliseconds max_work{20}; + size_t max_workers = 8; +}; + +void print_usage(std::string_view program) { + std::cout << "Usage: " << program << " [--smoke]\n\n" + << " --smoke Run a short termination check for CI\n"; +} + +bool parse_options(int argc, char** argv, example_options& options) { + for (int i = 1; i < argc; ++i) { + const std::string_view argument = argv[i]; + if (argument == "--help") { + print_usage(argv[0]); + return false; + } + if (argument == "--smoke") { + options.phase1_tasks = 40; + options.phase2_tasks = 60; + options.load_samples = 4; + options.idle_samples = 10; + options.tick_interval = 20ms; + options.idle_delay = 0s; + options.min_work = 1ms; + options.max_work = 3ms; + options.max_workers = 4; + continue; + } + + std::cerr << "Unknown option: " << argument << '\n'; + print_usage(argv[0]); + return false; + } + return true; +} // Task that simulates work with random duration -coro::task workload_task(std::atomic& counter) { - // Simulate variable work duration (10-100ms) +coro::task workload_task( + std::atomic& counter, + std::chrono::milliseconds min_work, + std::chrono::milliseconds max_work) { static thread_local std::mt19937 rng(std::hash{}(std::this_thread::get_id())); - std::uniform_int_distribution dist(10, 100); + std::uniform_int_distribution dist( + static_cast(min_work.count()), + static_cast(max_work.count())); std::this_thread::sleep_for(std::chrono::milliseconds(dist(rng))); counter.fetch_add(1, std::memory_order_relaxed); co_return; } -int main() { +bool run_load_phase(runtime::scheduler& sched, + const example_options& options, + int task_count, + std::string_view title, + std::string_view separator) { + std::atomic completed{0}; + std::vector> tasks; + tasks.reserve(static_cast(task_count)); + + try { + for (int i = 0; i < task_count; ++i) { + tasks.push_back(sched.go_joinable(workload_task( + completed, options.min_work, options.max_work))); + } + + std::cout << title << std::endl; + std::cout << separator << std::endl; + + for (int i = 0; i < options.load_samples; ++i) { + std::this_thread::sleep_for(options.tick_interval); + + if (i % 2 == 0) { + std::cout << " Workers: " << sched.num_threads() + << ", Pending: " << sched.pending_tasks() + << ", Completed: " << completed.load() << std::endl; + } + } + } catch (...) { + for (const auto& task : tasks) { + task.wait_destroyed(); + } + throw; + } + + for (const auto& task : tasks) { + task.wait_destroyed(); + } + + const int final_completed = completed.load(std::memory_order_acquire); + const size_t final_pending = sched.pending_tasks(); + std::cout << " Drained: Pending: " << final_pending + << ", Completed: " << final_completed << std::endl; + return final_completed == task_count && final_pending == 0; +} + +} // namespace + +int main(int argc, char** argv) { + example_options options; + if (!parse_options(argc, argv, options)) { + return argc == 2 && std::string_view(argv[1]) == "--help" ? 0 : 2; + } + log::logger::instance().set_level(log::level::warning); std::cout << "=== Elio Autoscaler Example ===" << std::endl; @@ -27,12 +134,12 @@ int main() { // Configure autoscaler elio::runtime::autoscaler_config config; - config.tick_interval = std::chrono::milliseconds(200); + config.tick_interval = options.tick_interval; config.overload_threshold = 20; config.idle_threshold = 5; - config.idle_delay = std::chrono::seconds(3); + config.idle_delay = options.idle_delay; config.min_workers = 2; - config.max_workers = 8; + config.max_workers = options.max_workers; // Create scheduler with minimum workers runtime::scheduler sched(config.min_workers); @@ -45,57 +152,26 @@ int main() { std::cout << "Initial workers: " << sched.num_threads() << std::endl; std::cout << std::endl; - // Phase 1: High load - demonstrate scale-up - { - std::atomic completed{0}; - - // Submit heavy workload - for (int i = 0; i < 2000; ++i) { - sched.go([&completed]() { return workload_task(completed); }); - } - - std::cout << "Phase 1: High load - expecting scale-up..." << std::endl; - std::cout << "----------------------------------------" << std::endl; - - // Monitor autoscaler for 5 seconds - for (int i = 0; i < 25; ++i) { - std::this_thread::sleep_for(config.tick_interval); - - size_t workers = sched.num_threads(); - size_t pending = sched.pending_tasks(); - - if (i % 2 == 0) { - std::cout << " Workers: " << workers - << ", Pending: " << pending - << ", Completed: " << completed.load() << std::endl; - } - } + if (!run_load_phase( + sched, options, options.phase1_tasks, + "Phase 1: High load - expecting scale-up...", + "----------------------------------------")) { + std::cerr << "Phase 1 failed to drain its submitted tasks" << std::endl; + autoscaler.stop(); + sched.shutdown(); + return 1; } std::cout << std::endl; - // Phase 2: Even higher load - { - std::atomic completed2{0}; - - // Submit even heavier workload - for (int i = 0; i < 3000; ++i) { - sched.go([&completed2]() { return workload_task(completed2); }); - } - - std::cout << "Phase 2: Higher load - expecting more scale-up..." << std::endl; - std::cout << "-------------------------------------------" << std::endl; - - for (int i = 0; i < 25; ++i) { - std::this_thread::sleep_for(config.tick_interval); - - size_t pending = sched.pending_tasks(); - - if (i % 2 == 0) { - std::cout << " Workers: " << sched.num_threads() - << ", Pending: " << pending << std::endl; - } - } + if (!run_load_phase( + sched, options, options.phase2_tasks, + "Phase 2: Higher load - expecting more scale-up...", + "-------------------------------------------")) { + std::cerr << "Phase 2 failed to drain its submitted tasks" << std::endl; + autoscaler.stop(); + sched.shutdown(); + return 1; } std::cout << std::endl; @@ -106,7 +182,7 @@ int main() { std::cout << "------------------------------------------" << std::endl; // Wait longer for idle_delay to trigger scale-down - for (int i = 0; i < 30; ++i) { + for (int i = 0; i < options.idle_samples; ++i) { std::this_thread::sleep_for(config.tick_interval); size_t workers = sched.num_threads(); diff --git a/wiki/Examples.md b/wiki/Examples.md index 1a375a2d..7a31ee57 100644 --- a/wiki/Examples.md +++ b/wiki/Examples.md @@ -889,6 +889,11 @@ The autoscaler supports: - **Actions**: `scale_up`, `scale_down`, `log`, `null` - **Combinators**: `on_success`, `on_failure` +The complete `autoscaler_example` retains join handles for every submitted load +phase and waits for all task frames to be destroyed before releasing captured +phase state or beginning its low-load observation. Use +`./build/examples/autoscaler_example --smoke` for a short termination check. + `on_block` is a best-effort sampled diagnostic. A non-idle worker is considered blocked when its completed-resume counter does not advance for longer than `block_threshold`; detection can trail the threshold by approximately one