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,27 +336,65 @@ 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 the GLOBAL
+                // idle epoch up-front: that decision's only job is to WAKE this worker so it
+                // re-checks its queue. Whether a GLOBAL idle epoch is actually appropriate 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 the global idle check 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 the GLOBAL case, 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 LOCAL (per-pool, on<Idle<ThisPool>>) check is different: it is still fired
+                // eagerly here, right away. `active` is edge-triggered (CountingLock only succeeds on
+                // the exact transition to zero), so if we deferred it behind try_dequeue_task() too,
+                // a fleeting active-count-reaches-zero window could be missed forever whenever this
+                // pool happens to have unrelated work land in its queue in the same instant (the
+                // dequeue would then succeed and skip the idle check entirely for this iteration,
+                // with no guarantee `active` will ever read exactly zero again for this waiter's
+                // epoch). Firing the local check early is safe with respect to the premature-idle
+                // bug above because it only ever fires THIS pool's own Idle<ThisPool> reactions and
+                // never touches `scheduler.active_pools` - it cannot release a global idle slot.
+                //
+                // 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);
+
+                    auto local_idle_task = get_local_idle_task();
+                    if (local_idle_task.task != nullptr) {
+                        return local_idle_task;
                     }
                 }
 
@@ -400,13 +438,7 @@ namespace threading {
             throw ShutdownThreadException();
         }
 
-        Pool::Task Pool::get_idle_task() {
-            if (!running || !descriptor->counts_for_idle) {
-                return Task{};
-            }
-
-            std::vector<std::shared_ptr<Reaction>> tasks;
-
+        void Pool::collect_local_idle_reactions(std::vector<std::shared_ptr<Reaction>>& tasks) {
             auto& local_lock = thread_idle[std::this_thread::get_id()];
 
             if (local_lock == nullptr) {
@@ -415,16 +447,9 @@ namespace threading {
                     tasks.insert(tasks.end(), idle_tasks.begin(), idle_tasks.end());
                 }
             }
+        }
 
-            if (pool_idle == nullptr && active.load(std::memory_order_relaxed) == 0) {
-                pool_idle = std::make_unique<CountingLock>(scheduler.active_pools);
-
-                if (pool_idle->lock()) {
-                    const std::lock_guard<std::mutex> lock(scheduler.idle_mutex);
-                    tasks.insert(tasks.end(), scheduler.idle_tasks.begin(), scheduler.idle_tasks.end());
-                }
-            }
-
+        Pool::Task Pool::make_idle_dispatch_task(std::vector<std::shared_ptr<Reaction>>&& tasks) {
             if (tasks.empty()) {
                 return Task{};
             }
@@ -445,6 +470,36 @@ namespace threading {
             return Task{std::move(task)};
         }
 
+        Pool::Task Pool::get_local_idle_task() {
+            if (!running || !descriptor->counts_for_idle) {
+                return Task{};
+            }
+
+            std::vector<std::shared_ptr<Reaction>> tasks;
+            collect_local_idle_reactions(tasks);
+            return make_idle_dispatch_task(std::move(tasks));
+        }
+
+        Pool::Task Pool::get_idle_task() {
+            if (!running || !descriptor->counts_for_idle) {
+                return Task{};
+            }
+
+            std::vector<std::shared_ptr<Reaction>> tasks;
+            collect_local_idle_reactions(tasks);
+
+            if (pool_idle == nullptr && active.load(std::memory_order_relaxed) == 0) {
+                pool_idle = std::make_unique<CountingLock>(scheduler.active_pools);
+
+                if (pool_idle->lock()) {
+                    const std::lock_guard<std::mutex> lock(scheduler.idle_mutex);
+                    tasks.insert(tasks.end(), scheduler.idle_tasks.begin(), scheduler.idle_tasks.end());
+                }
+            }
+
+            return make_idle_dispatch_task(std::move(tasks));
+        }
+
         // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
         thread_local Pool* Pool::current_pool = nullptr;
 

src/threading/scheduler/Pool.hpp

@@ -257,6 +257,43 @@ namespace threading {
              */
             Task get_idle_task();
 
+            /**
+             * Get only this pool's own local idle task (on<Idle<ThisPool>> reactions), without considering the
+             * global (all-pools) idle epoch.
+             *
+             * This exists so the local, per-pool `active` transition can be checked eagerly - as soon as a woken
+             * worker notices `pending_idle` - without risking the premature-global-idle bug that firing the
+             * global check early can cause (see the comment in get_task() for details). The local `active`
+             * counter only ever gates this pool's OWN Idle<ThisPool> reactions, never `scheduler.active_pools`,
+             * so firing it eagerly cannot release the global active_pools slot early - it is safe to check as
+             * soon as possible, and doing so avoids missing a fleeting active-count-reaches-zero edge that a
+             * concurrent task submission could otherwise paper over before the deferred dequeue-first path gets
+             * around to checking it.
+             *
+             * @return the local idle task to execute if it is lockable, or hold if it is not
+             */
+            Task get_local_idle_task();
+
+            /**
+             * Collect this pool's own local idle reactions (on<Idle<ThisPool>>) if this worker is the one that
+             * takes the pool's `active` count to zero.
+             *
+             * Appends the reactions to fire to @p tasks; leaves it untouched if this worker did not win the
+             * local idle lock. Shared by both get_local_idle_task() and get_idle_task().
+             *
+             * @param tasks the accumulator to append any local idle reactions to
+             */
+            void collect_local_idle_reactions(std::vector<std::shared_ptr<Reaction>>& tasks);
+
+            /**
+             * Wrap a collected set of idle reactions in a dispatch task that submits them when run.
+             *
+             * @param tasks the idle reactions to dispatch (moved from)
+             *
+             * @return the dispatch task, or an empty Task if @p tasks is empty
+             */
+            Task make_idle_dispatch_task(std::vector<std::shared_ptr<Reaction>>&& tasks);
+
             friend class ExternalWaiterRegistration;
             void unregister_external_waiter();
 
@@ -283,12 +320,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