Lock-free scheduler with per-priority queues (#193)

## Summary

Rework `Scheduler` / `Pool` / `Group` to remove the global mutex
contention that showed up on the existing implementation under load:

- **Per-priority lock-free task queues.** Each `Pool` now owns five
independent MPMC `TaskQueue`s (one per priority bucket) instead of a
single mutex-protected priority queue. Submitting and dequeuing on
different priorities no longer contend at all, and same-priority submits
use a block-based lock-free FIFO with a graveyard-style block
reclamation scheme to avoid ABA.
- **MPSC specialisation for single-consumer pools.** Pools with
`concurrency == 1` (e.g. `MainThread`, `TraceController`) now construct
`MPSCQueue` instead of `TaskQueue`, giving a non-atomic consumer side
and removing the dequeue CAS entirely for those pools. Both queues sit
behind a `Queue<T>` polymorphic interface so `Pool` can hold an array of
buckets and dispatch the right type at construction.
- **Counting-semaphore Group fast path.** Single-`Sync` groups now
acquire via a signed `std::atomic<int> tokens` plus a lock-free
`TaskQueue` per priority for waiters, falling back to the mutex-backed
`GroupLock` only for multi-group submissions. Sync ordering inside a
group is strict; equal-priority tasks across pools see relaxed global
FIFO.
- **Idle bookkeeping off the pool mutex.** Pool idle tracking moved off
the main pool mutex so wake-ups no longer block submitters.

New lock-free primitives (`TaskQueue`, `MPSCQueue`, `Semaphore`) and the
Group counting fast path are covered by Catch2 BDD-style tests, plus a
scheduler microbenchmark (`tests/tests/Benchmark.cpp`, hidden behind the
`[.benchmark]` tag).

## TSAN race fixes (pre-existing)

While validating the changes under TSAN on macOS clang and Linux gcc 13
three pre-existing data races surfaced and are fixed in this PR:

- **`IOController_Posix.ipp` — `pollfd::events`.** The `IOFinished`
handler mutated `watches[].events` while only holding `tasks_mutex`, and
the poll thread read the same field from inside `::poll()` while only
holding `notifier.mutex`. `bump()` woke poll but released
`notifier.mutex` before the mutation, leaving the race window open. The
handler now writes the wake byte inline and holds `notifier.mutex`
across the `watches` update and the follow-up `fire_event` call (which
can also touch `watches[].events`).
- **`Scheduler::submit` — `Reaction::scheduler_data`.** The cached pool
was read/written from any submitting thread without synchronisation. It
is now a non-owning raw `Pool*` cached in a `std::atomic<void*>`
(release store / acquire load) rather than a `std::shared_ptr<void>`
accessed via `std::atomic_load`/`atomic_store` — the shared_ptr atomics
fall back to a small global mutex pool on libstdc++ and become a
contention point on hot submission paths. Pools outlive all reactions
(PowerPlant tears reactors down before the scheduler), so a raw pointer
is safe, and all racers resolve the same pool, so the benign store is
last-writer-wins.
- **`Watchdog` data store.** The service `time_point` was read by the
chrono controller while being mutated by user threads emitting a service
event, and the void specialisation returned a reference through a
temporary `shared_ptr` (latent dangling reference). Centralised
reads/writes through a per-`(WatchdogGroup, RuntimeType)` `std::mutex`,
made `get` return by value, and routed `WatchdogServicer::service`
through `WatchdogDataStore::service` so writes share the read mutex.

## Other

- `.gitignore` now matches `build-*/` so out-of-tree TSAN/ASAN/Release
build directories don't appear in `git status`.
- SonarCloud suppressions for the deliberate lock-free / placement-new
idioms live in `sonar-project.properties`.

## Test plan

- [x] macOS clang TSAN: `dsl/IO`, `dsl/Inline`, `dsl/Watchdog` 30/30
clean each; full suite ✓
- [x] macOS clang Release: full suite ✓
- [x] Linux gcc 13 TSAN (Docker): same three tests 30/30 clean; hot-path
tests under repeated runs with no TSAN warnings
- [x] Linux gcc 9 Release (Docker): full suite ✓
- [x] Lock-free primitives have BDD-style unit tests (including
queue-destructor leak coverage)
- [x] Scheduler microbenchmark exercises submit / dequeue / group
acquire under contention


Made with [Cursor](https://cursor.com)

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: TrentHouliston

.gitignore

@@ -14,6 +14,7 @@
 
 # Build & CMake files
 build/
+build-*/
 CMakeCache.txt
 CMakeFiles
 Makefile

CMakeLists.txt

@@ -81,11 +81,13 @@ if(CI_BUILD)
   endif()
 endif(CI_BUILD)
 
+# Tests must be declared before src/ so NUClear can expose test-only APIs when enabled.
+option(BUILD_TESTS "Builds all of the NUClear unit tests." ON)
+
 # Add the src directory
 add_subdirectory(src)
 
 # Add the tests directory
-option(BUILD_TESTS "Builds all of the NUClear unit tests." ON)
 if(BUILD_TESTS)
   enable_testing()
   add_subdirectory(tests)

cmake/TestRunner.cmake

@@ -54,7 +54,8 @@ foreach(target ${all_targets})
   list(APPEND report_outputs ${junit_report_file})
   add_custom_command(
     OUTPUT ${junit_report_file} ${raw_coverage}
-    COMMAND ${command_prefix} $<TARGET_FILE:${target}> --reporter console --reporter JUnit::out=${junit_report_file}
+    COMMAND ${command_prefix} $<TARGET_FILE:${target}> --allow-running-no-tests --reporter console
+            --reporter JUnit::out=${junit_report_file}
     WORKING_DIRECTORY ${PROJECT_BINARY_DIR}
     DEPENDS ${target}
     USES_TERMINAL

docs/explanation/index.md

@@ -11,6 +11,7 @@ If you've already followed the tutorials and know how to use NUClear, this is wh
 | --------------------------------- | --------------------------------------------------------------------------------------------- |
 | [Architecture](architecture.md)   | Why NUClear exists, the problems it solves, and the event-driven reactive pattern at its core |
 | [Threading Model](threading.md)   | How tasks are scheduled across thread pools, priority queues, and group constraints           |
+| [Scheduler](scheduler.md)         | Internal design of the lock-free scheduler: pools, queues, groups, idle tasks, and shutdown   |
 | [Lifecycle](lifecycle.md)         | The three phases of a NUClear system: initialisation, execution, and shutdown                 |
 | [The DSL System](dsl-system.md)   | How `on<>().then()` works from top to bottom — template metaprogramming in action             |
 | [Message Flow](message-flow.md)   | What happens when you emit data, from call site to reaction execution                         |

docs/explanation/scheduler.md

@@ -3,6 +3,8 @@
 NUClear's threading model is designed around a simple goal: **you should never have to write a mutex**.
 The framework handles concurrency for you through immutable messages, thread pools, and a priority-based scheduler.
 
+For the internal design of the scheduler (lock-free queues, group tokens, idle detection, shutdown), see [Scheduler](scheduler.md).
+
 ## Thread Pool Architecture
 
 NUClear uses multiple thread pools, each serving a different purpose:

docs/explanation/threading.md

@@ -181,6 +181,7 @@ nav:
       - explanation/index.md
       - Architecture: explanation/architecture.md
       - Threading Model: explanation/threading.md
+      - Scheduler: explanation/scheduler.md
       - Lifecycle: explanation/lifecycle.md
       - The DSL System: explanation/dsl-system.md
       - Message Flow: explanation/message-flow.md

mkdocs.yml

@@ -0,0 +1,26 @@
+# SonarCloud issue suppressions for deliberate lock-free / placement-new code.
+# projectKey, organization, sources, tests and coverage settings are passed on
+# the scanner CLI in .github/workflows/sonarcloud.yaml; only the ignore rules
+# below are configured here.
+
+sonar.issue.ignore.multicriteria=e1,e2,e3,e4,e5
+
+# S8417: explicit memory_order arguments are intentional in this concurrency
+# framework; the carefully chosen relaxed/acquire/release/acq_rel orderings are
+# required for performance and must not be forced to seq_cst.
+sonar.issue.ignore.multicriteria.e1.ruleKey=cpp:S8417
+sonar.issue.ignore.multicriteria.e1.resourceKey=src/threading/**
+sonar.issue.ignore.multicriteria.e2.ruleKey=cpp:S8417
+sonar.issue.ignore.multicriteria.e2.resourceKey=src/extension/**
+
+# S5025 (manual new/delete), S3630 (reinterpret_cast) and S3432 (explicit
+# destructor call) are unavoidable in the lock-free queues: manual Block
+# lifetime is dictated by the graveyard reclamation scheme and the
+# reinterpret_cast + explicit ~T() are the placement-new idiom for the aligned
+# slot storage. Scope these to the queue files only.
+sonar.issue.ignore.multicriteria.e3.ruleKey=cpp:S5025
+sonar.issue.ignore.multicriteria.e3.resourceKey=**/scheduler/queue/*.hpp
+sonar.issue.ignore.multicriteria.e4.ruleKey=cpp:S3630
+sonar.issue.ignore.multicriteria.e4.resourceKey=**/scheduler/queue/*.hpp
+sonar.issue.ignore.multicriteria.e5.ruleKey=cpp:S3432
+sonar.issue.ignore.multicriteria.e5.resourceKey=**/scheduler/queue/*.hpp

sonar-project.properties

@@ -390,7 +390,7 @@ class Reactor {
 
     public:
         template <typename... Args>
-        Binder(Reactor& r, Args&&... args) : reactor(r), args(std::forward<Args>(args)...) {}
+        Binder(Reactor& r, Args&&... args_) : reactor(r), args(std::forward<Args>(args_)...) {}
 
         template <typename Label, typename Function>
         auto then(Label&& label, Function&& callback) {

src/Reactor.hpp

@@ -23,6 +23,7 @@
 #ifndef NUCLEAR_DSL_WORD_WATCHDOG_HPP
 #define NUCLEAR_DSL_WORD_WATCHDOG_HPP
 
+#include <mutex>
 #include <stdexcept>
 
 #include "../../threading/Reaction.hpp"
@@ -52,12 +53,25 @@ namespace dsl {
             using MapType       = std::remove_cv_t<RuntimeType>;
             using WatchdogStore = util::TypeMap<WatchdogGroup, MapType, std::map<MapType, NUClear::clock::time_point>>;
 
+            /**
+             * Mutex protecting structural and value updates to the underlying map for this
+             * (WatchdogGroup, RuntimeType) pair. Watchdog timers are read by the chrono controller
+             * thread (via @ref get) while being written by user threads that emit a service event
+             * (via @ref service), and the underlying std::map is also mutated by init/unbind, so a
+             * single shared mutex serialises all of those operations.
+             */
+            static std::mutex& mutex() {
+                static std::mutex m;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+                return m;
+            }
+
             /**
              * Ensures the data store is initialised correctly.
              *
              * @param data The runtime argument for the current watchdog in the WatchdogGroup/RuntimeType group
              */
             static void init(const RuntimeType& data) {
+                const std::lock_guard<std::mutex> lock(mutex());
                 if (WatchdogStore::get() == nullptr) {
                     WatchdogStore::set(std::make_shared<std::map<MapType, NUClear::clock::time_point>>());
                 }
@@ -67,11 +81,15 @@ namespace dsl {
             }
 
             /**
-             * Gets the current service time for the WatchdogGroup/RuntimeType/data watchdog
+             * Gets the current service time for the WatchdogGroup/RuntimeType/data watchdog.
+             *
+             * Returned by value so the caller never holds a reference into the (mutex-protected)
+             * map. The time_point is small and trivially copyable so the copy is essentially free.
              *
              * @param data The runtime argument for the current watchdog in the WatchdogGroup/RuntimeType group
              */
-            static const NUClear::clock::time_point& get(const RuntimeType& data) {
+            static NUClear::clock::time_point get(const RuntimeType& data) {
+                const std::lock_guard<std::mutex> lock(mutex());
                 if (WatchdogStore::get() == nullptr || WatchdogStore::get()->count(data) == 0) {
                     throw std::domain_error("Store for <" + util::demangle(typeid(WatchdogGroup).name()) + ", "
                                             + util::demangle(typeid(MapType).name())
@@ -80,12 +98,29 @@ namespace dsl {
                 return WatchdogStore::get()->at(data);
             }
 
+            /**
+             * Atomically updates the service time for the WatchdogGroup/RuntimeType/data watchdog.
+             *
+             * Called by @ref emit::WatchdogServicer::service to keep the write under the same
+             * mutex that @ref get uses for reads.
+             */
+            static void service(const RuntimeType& data, const NUClear::clock::time_point& when) {
+                const std::lock_guard<std::mutex> lock(mutex());
+                if (WatchdogStore::get() == nullptr || WatchdogStore::get()->count(data) == 0) {
+                    throw std::domain_error("Store for <" + util::demangle(typeid(WatchdogGroup).name()) + ", "
+                                            + util::demangle(typeid(MapType).name())
+                                            + "> has not been created yet or no watchdog has been set up");
+                }
+                WatchdogStore::get()->at(data) = when;
+            }
+
             /**
              * Cleans up any allocated storage for the WatchdogGroup/RuntimeType/data watchdog
              *
              * @param data The runtime argument for the current watchdog in the WatchdogGroup/RuntimeType group
              */
             static void unbind(const RuntimeType& data) {
+                const std::lock_guard<std::mutex> lock(mutex());
                 if (WatchdogStore::get() != nullptr) {
                     WatchdogStore::get()->erase(data);
                 }
@@ -105,30 +140,54 @@ namespace dsl {
         struct WatchdogDataStore<WatchdogGroup, void> {
             using WatchdogStore = util::TypeMap<WatchdogGroup, void, NUClear::clock::time_point>;
 
+            /// See the documentation on the runtime-arg specialisation.
+            static std::mutex& mutex() {
+                static std::mutex m;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+                return m;
+            }
+
             /**
              * Ensures the data store is initialised correctly.
              */
             static void init() {
+                const std::lock_guard<std::mutex> lock(mutex());
                 if (WatchdogStore::get() == nullptr) {
                     WatchdogStore::set(std::make_shared<NUClear::clock::time_point>(NUClear::clock::now()));
                 }
             }
 
             /**
              * Gets the current service time for the WatchdogGroup watchdog.
+             *
+             * Returned by value so the caller never reads from the time_point while it is being
+             * mutated by @ref service on another thread.
              */
-            static const NUClear::clock::time_point& get() {
+            static NUClear::clock::time_point get() {
+                const std::lock_guard<std::mutex> lock(mutex());
                 if (WatchdogStore::get() == nullptr) {
                     throw std::domain_error("Store for <" + util::demangle(typeid(WatchdogGroup).name())
                                             + "> is trying to field a service call for an unknown data type");
                 }
                 return *WatchdogStore::get();
             }
 
+            /**
+             * Atomically updates the service time for the WatchdogGroup watchdog.
+             */
+            static void service(const NUClear::clock::time_point& when) {
+                const std::lock_guard<std::mutex> lock(mutex());
+                if (WatchdogStore::get() == nullptr) {
+                    throw std::domain_error("Store for <" + util::demangle(typeid(WatchdogGroup).name())
+                                            + "> has not been created yet or no watchdog has been set up");
+                }
+                *WatchdogStore::get() = when;
+            }
+
             /**
              * Cleans up any allocated storage for the WatchdogGroup watchdog.
              */
             static void unbind() {
+                const std::lock_guard<std::mutex> lock(mutex());
                 if (WatchdogStore::get() != nullptr) {
                     WatchdogStore::get().reset();
                 }