diff --git a/src/threading/scheduler/Pool.cpp b/src/threading/scheduler/Pool.cpp index 27ad4771..697fc644 100644 --- a/src/threading/scheduler/Pool.cpp +++ b/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>) 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 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> tasks; - + void Pool::collect_local_idle_reactions(std::vector>& 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(scheduler.active_pools); - - if (pool_idle->lock()) { - const std::lock_guard 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>&& 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> 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> tasks; + collect_local_idle_reactions(tasks); + + if (pool_idle == nullptr && active.load(std::memory_order_relaxed) == 0) { + pool_idle = std::make_unique(scheduler.active_pools); + + if (pool_idle->lock()) { + const std::lock_guard 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; diff --git a/src/threading/scheduler/Pool.hpp b/src/threading/scheduler/Pool.hpp index 0b32d394..b2004fb3 100644 --- a/src/threading/scheduler/Pool.hpp +++ b/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> 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 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>) 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>& 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>&& tasks); + friend class ExternalWaiterRegistration; void unregister_external_waiter(); @@ -283,12 +320,14 @@ namespace threading { std::atomic 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 diff --git a/tests/tests/dsl/IdleParkedSyncWaiter.cpp b/tests/tests/dsl/IdleParkedSyncWaiter.cpp new file mode 100644 index 00000000..9e34820d --- /dev/null +++ b/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 +#include +#include +#include +#include +#include +#include + +#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> - 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 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 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, Pool, Priority::REALTIME>().then([this] { + bool do_shutdown = false; + { + const std::lock_guard 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().then([this] { emit(std::make_unique()); }); + + // 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, Sync, Pool, Priority::REALTIME>().then([this] { + { + const std::lock_guard lock(mtx); + awaiting_inner = true; + events.push_back("outer"); + } + // Submitted synchronously: because we still hold the Sync 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()); + }); + + // The re-entrant inner task: same Sync group, same WorkerPool, lower priority than idle. + on, Sync, Pool, Priority::LOW>().then([this] { + const std::lock_guard 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 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(); + powerplant.start(); + + const std::vector expected = { + "outer", + "inner", + }; + + INFO(test_util::diff_string(expected, events)); + + REQUIRE(events == expected); +}