Fix premature global idle when a parked Sync waiter becomes runnable

Pool::get_task honoured the pending_idle latch by firing an idle epoch
before attempting to dequeue. When a parked external waiter (e.g. a task
blocked on a Sync group's token) becomes runnable and is drained into the
pool's queue, firing idle up-front dropped the pool's active count and
released its active_pools slot while a runnable task was still queued,
letting a global idle epoch fire prematurely.

This reorders idle-driven reactions and, with shutdown-on-idle, can
quiesce the powerplant while real work is pending. It manifested in the
NUbots Director, which dispatches provider reactions onto the default
pool while still holding its Sync<Director> token; the re-entrant
provider parks, and on token release the premature idle reordered the
Director's idle-driven steps.

Consume the pending_idle latch without firing idle: its only job is to
wake the worker so it re-checks its queue. The existing dequeue-first /
!got path then decides correctly - a drained-runnable waiter is dequeued
and run (no idle), while a still-parked waiter leaves the queue empty so
the !got branch fires idle exactly as before (preserving cross-pool
idle-wake / deadlock-break behaviour).

Add the IdleDirectorPingPong regression test reproducing the Director
topology. It is fully deterministic and uses no sleeps: the provider and
the global idle reaction share a single-worker pool, and priority
ordering (REALTIME idle vs LOW provider) means that if the buggy
scheduler fires idle while the drained provider is still queued, the idle
reaction is dequeued first and observes the pending provider. It fails
deterministically before the fix and passes after.

Author: TrentHouliston

src/threading/scheduler/Pool.cpp

@@ -336,28 +336,50 @@ namespace threading {
                     continue;
                 }
 
-                // If a waiter was parked for this pool since the last time this worker looked,
-                // ensure we fire one idle epoch before dispatching the next task. This is the
-                // counterpart of the OLD scheduler behaviour where a parked task with a failing
-                // group lock sat in the pool queue and forced the worker to poll-fail-and-fall-
-                // through to get_idle_task; in the fast path the task is parked in the Group's
-                // wait_buckets instead, so without this latch the worker can be preempted long
-                // enough for the drained (lock-OK) task to arrive in the queue before the worker
-                // polls and end up running it directly, swallowing the idle fire.
+                // A waiter was parked for this pool since the last time this worker looked, so it
+                // set pending_idle to wake us. Consume the latch here, but do NOT fire an idle
+                // epoch up-front: the latch's only job is to WAKE this worker so it re-checks its
+                // queue. Whether this is actually an idle situation must be decided by the normal
+                // dequeue-first path below.
                 //
-                // get_idle_task() is a no-op when this thread is already idle (local_lock set),
-                // so a wasted consume here is harmless: the worker just falls through to the
-                // normal dequeue path below.
+                // This matters for the case where the parked waiter has since become runnable and
+                // been drained into this pool's queue (e.g. a Sync group released its token). If we
+                // fired get_idle_task() here, before try_dequeue_task(), we would drop this pool's
+                // "active" count to zero and release its active_pools slot while a runnable task is
+                // still sitting in the queue. That can let a GLOBAL idle epoch fire even though real
+                // work is pending (premature idle) - which reorders idle-driven reactions and, with
+                // shutdown-on-idle, can quiesce the powerplant early.
                 //
-                // The relaxed load short-circuits the (more expensive) read-modify-write on the
-                // common path where nothing has been latched, so a busy worker never pays for the
-                // exclusive cacheline acquire that exchange() would force every iteration.
-                if (pending_idle.load(std::memory_order_acquire)
-                    && pending_idle.exchange(false, std::memory_order_acq_rel)) {
-                    auto idle_task = get_idle_task();
-                    if (idle_task.task != nullptr) {
-                        return idle_task;
-                    }
+                // By only consuming the latch here and letting the dequeue-first / !got path below
+                // decide, a drained-runnable waiter is dequeued and run (no idle), while a still-
+                // parked waiter leaves the queue empty so the !got branch fires idle exactly as
+                // before (preserving the cross-pool idle-wake / deadlock-break behavior).
+                //
+                // The relaxed-ish load short-circuits the (more expensive) store on the common path
+                // where nothing has been latched, so a busy worker never pays for the exclusive
+                // cacheline acquire that any unconditional atomic RMW - store, exchange, or even a
+                // failed compare_exchange - would force every iteration. On real hardware only a
+                // plain load can be satisfied from a cache line held Shared; a store/exchange/CAS
+                // always requires exclusive ownership of the line, even when the value doesn't
+                // change (a "failed" CAS still takes the lock on x86, still faults the exclusive
+                // monitor on ARM). So gating the write behind a load is the only way to keep this
+                // hot per-dispatch check free of cross-core cache-line ping-pong on a busy pool.
+                //
+                // A plain store (rather than exchange) is safe here even though we don't hold the
+                // mutex: we never branch on the old value, so there is nothing for exchange to give
+                // us that store doesn't. The apparent "lost wakeup" if a new waiter's
+                // register_external_waiter() sees the latch already true (skips its notify) and we
+                // then clear it here is not actually a correctness issue, because neither
+                // notify_one() call is what makes a parked waiter's task eventually run: submit()
+                // (when the drained task is enqueued) and unregister_external_waiter() (when
+                // external_waiters returns to 0) both notify unconditionally, under the pool's
+                // mutex, on every transition that the wait predicate below actually depends on.
+                // pending_idle's own notify is purely a latency optimization to promptly wake a
+                // worker that is sleeping for no other reason than "nothing has happened yet"; if
+                // it is occasionally skipped, the worker is woken anyway by one of those other
+                // unconditional notifies once there is something to actually act on.
+                if (pending_idle.load(std::memory_order_acquire)) {
+                    pending_idle.store(false, std::memory_order_release);
                 }
 
                 bool got = false;

src/threading/scheduler/Pool.hpp

@@ -283,12 +283,14 @@ namespace threading {
             std::atomic<std::size_t> external_waiters{0};
             /// Latched "an external waiter was parked for this pool since you last polled".
             ///
-            /// Consumed (exchanged to false) at the top of every get_task iteration. If set and
-            /// this thread is not already idle, a single idle fire is dispatched before any task
-            /// from the queue is returned. This preserves the OLD scheduler's invariant that a
-            /// waiting-but-not-runnable task on the destination pool would always force one idle
-            /// fire per parking, even when the worker is preempted long enough for the drained
-            /// (RunningLock-OK) task to be sitting in the queue by the time the worker resumes.
+            /// Consumed (cleared to false) at the top of every get_task iteration purely to WAKE a
+            /// sleeping worker so it re-checks its queue; it does NOT by itself force an idle fire.
+            /// Whether this is actually an idle situation is decided by the normal dequeue-first
+            /// path: if the parked waiter has since become runnable and been drained into this
+            /// pool's queue (e.g. a Sync group released its token), the worker dequeues and runs it
+            /// with no idle epoch. Only if the queue is genuinely empty after dequeuing does the
+            /// !got path fire idle, preserving the cross-pool idle-wake / deadlock-break behavior
+            /// without prematurely firing idle while real work is still pending.
             ///
             /// This is only ever set when idle_relevant() is true (some idle reaction could fire
             /// on this pool), so on the hot contended path with no idle reactions the latch stays

tests/tests/dsl/IdleParkedSyncWaiter.cpp

@@ -0,0 +1,165 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2024 NUClear Contributors
+ *
+ * This file is part of the NUClear codebase.
+ * See https://github.com/Fastcode/NUClear for further info.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
+ * WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+ * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+#include <catch2/catch_message.hpp>
+#include <catch2/catch_test_macros.hpp>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "nuclear"
+#include "test_util/diff_string.hpp"
+
+// Deterministic regression test for a premature global-idle epoch, WITHOUT relying on any sleeps or
+// timing windows.
+//
+// Topology:
+//   * outer        - runs on a single-worker WorkerPool while holding the Sync group token, and
+//                    dispatches the re-entrant inner task onto that same pool while STILL holding the
+//                    token (so it cannot run and instead parks as an external waiter).
+//   * inner        - the re-entrant task: same Sync group, same WorkerPool, LOW priority.
+//   * on<Idle<>>   - the global idle reaction, also pinned to the single-worker WorkerPool, at
+//                    REALTIME priority.
+//
+// The bug: when the WorkerPool worker is woken by the parked inner task's pending_idle latch, the
+// buggy scheduler fires a global idle epoch BEFORE dequeuing the (now drained and runnable) inner
+// task, dropping the pool's active count and releasing its active_pools slot while real work is
+// queued.
+//
+// Why this is deterministic (no sleeps):
+//   * Submission of the inner task is synchronous inside the outer reaction, so while the outer
+//     reaction still holds the Sync token the inner task is parked (arming pending_idle) before the
+//     outer reaction returns. Returning releases the token and drains the inner task into the
+//     WorkerPool queue as a runnable task.
+//   * The inner task and the idle reaction share the SAME single-worker WorkerPool, so they can never
+//     run concurrently - the one worker serializes them.
+//   * The idle reaction is REALTIME and the inner task is LOW. In the buggy case the premature idle
+//     submits the idle reaction into the same queue that already holds the drained inner task; the
+//     single worker then dequeues the higher-priority idle reaction FIRST, so it observes the still
+//     pending inner task (awaiting_inner == true) and records the violation. In the fixed case the
+//     worker dequeues the inner task first (no idle epoch), so the idle reaction never runs while the
+//     inner task is pending.
+//
+// Result: without the fix the event sequence deterministically contains an extra
+// "idle-while-inner-pending" entry; with the fix it is exactly {"outer", "inner"}.
+
+namespace {
+
+struct WorkerPool {
+    static constexpr int concurrency = 1;
+};
+struct SyncGroup {};
+
+struct Outer {};
+struct Inner {};
+
+// NOLINTBEGIN(cppcoreguidelines-avoid-non-const-global-variables)
+std::mutex mtx;
+std::vector<std::string> events;
+bool awaiting_inner = false;
+bool inner_done     = false;
+bool shutting_down  = false;
+// NOLINTEND(cppcoreguidelines-avoid-non-const-global-variables)
+
+class TestReactor : public NUClear::Reactor {
+public:
+    explicit TestReactor(std::unique_ptr<NUClear::Environment> environment) : Reactor(std::move(environment)) {
+
+        // Global idle, pinned to the (single-worker) WorkerPool at REALTIME so that if the buggy
+        // scheduler submits it while the drained inner task is still queued on the same pool, it is
+        // dequeued before the lower-priority inner task - making the observation deterministic.
+        on<Idle<>, Pool<WorkerPool>, Priority::REALTIME>().then([this] {
+            bool do_shutdown = false;
+            {
+                const std::lock_guard<std::mutex> lock(mtx);
+                if (awaiting_inner) {
+                    // Real work (the parked-then-drained inner task) is still pending on this pool, yet
+                    // a global idle epoch has fired: this is the premature-idle bug.
+                    events.push_back("idle-while-inner-pending");
+                }
+                else if (inner_done && !shutting_down) {
+                    shutting_down = true;
+                    do_shutdown   = true;
+                }
+            }
+            if (do_shutdown) {
+                powerplant.shutdown();
+            }
+        });
+
+        // Kick off exactly one cycle.
+        on<Startup>().then([this] { emit(std::make_unique<Outer>()); });
+
+        // The outer reaction: holds the Sync token on the WorkerPool and dispatches the re-entrant
+        // inner task onto that same pool while still holding that token.
+        on<Trigger<Outer>, Sync<SyncGroup>, Pool<WorkerPool>, Priority::REALTIME>().then([this] {
+            {
+                const std::lock_guard<std::mutex> lock(mtx);
+                awaiting_inner = true;
+                events.push_back("outer");
+            }
+            // Submitted synchronously: because we still hold the Sync<SyncGroup> token, the inner task
+            // fails the group lock and PARKS as an external waiter (arming the WorkerPool's
+            // pending_idle latch) before this reaction returns. Returning then releases the token,
+            // draining the parked inner task into the WorkerPool queue as a runnable task.
+            emit(std::make_unique<Inner>());
+        });
+
+        // The re-entrant inner task: same Sync group, same WorkerPool, lower priority than idle.
+        on<Trigger<Inner>, Sync<SyncGroup>, Pool<WorkerPool>, Priority::LOW>().then([this] {
+            const std::lock_guard<std::mutex> lock(mtx);
+            events.push_back("inner");
+            awaiting_inner = false;
+            inner_done     = true;
+        });
+    }
+};
+
+}  // namespace
+
+TEST_CASE("Test global idle does not fire while a parked Sync waiter is pending", "[api][dsl][Idle][Pool][Sync]") {
+
+    {
+        const std::lock_guard<std::mutex> lock(mtx);
+        events.clear();
+        awaiting_inner = false;
+        inner_done     = false;
+        shutting_down  = false;
+    }
+
+    NUClear::Configuration config;
+    config.default_pool_concurrency = 1;
+    NUClear::PowerPlant powerplant(config);
+    powerplant.install<TestReactor>();
+    powerplant.start();
+
+    const std::vector<std::string> expected = {
+        "outer",
+        "inner",
+    };
+
+    INFO(test_util::diff_string(expected, events));
+
+    REQUIRE(events == expected);
+}