fix(runtime): drain autoscaler example phases

.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

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<Scheduler, Triggers...>` now
   constructs and dispatches each configured trigger correctly instead of
   failing template instantiation because the trigger argument was omitted.

examples/autoscaler_example.cpp

@@ -3,22 +3,129 @@
 #include <iostream>
 #include <atomic>
 #include <chrono>
+#include <functional>
 #include <random>
+#include <string_view>
+#include <thread>
+#include <vector>
 
 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<void> workload_task(std::atomic<int>& counter) {
-    // Simulate variable work duration (10-100ms)
+coro::task<void> workload_task(
+        std::atomic<int>& counter,
+        std::chrono::milliseconds min_work,
+        std::chrono::milliseconds max_work) {
     static thread_local std::mt19937 rng(std::hash<std::thread::id>{}(std::this_thread::get_id()));
-    std::uniform_int_distribution<int> dist(10, 100);
+    std::uniform_int_distribution<int> dist(
+        static_cast<int>(min_work.count()),
+        static_cast<int>(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<int> completed{0};
+    std::vector<coro::join_handle<void>> tasks;
+    tasks.reserve(static_cast<size_t>(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<int> 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<int> 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();

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