Lock-free scheduler with per-priority queues

.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/threading.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:

mkdocs.yml

@@ -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

sonar-project.properties

@@ -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

src/Reactor.hpp

@@ -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/dsl/word/Watchdog.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();
                 }

src/dsl/word/emit/Watchdog.hpp

@@ -23,11 +23,8 @@
 #ifndef NUCLEAR_DSL_WORD_EMIT_WATCHDOG_HPP
 #define NUCLEAR_DSL_WORD_EMIT_WATCHDOG_HPP
 
-#include <stdexcept>
-
 #include "../../../PowerPlant.hpp"
-#include "../../../util/TypeMap.hpp"
-#include "../../../util/demangle.hpp"
+#include "../Watchdog.hpp"
 
 namespace NUClear {
 namespace dsl {
@@ -47,8 +44,6 @@ namespace dsl {
             template <typename WatchdogGroup, typename RuntimeType = void>
             struct WatchdogServicer {
                 using MapType = std::remove_cv_t<RuntimeType>;
-                using WatchdogStore =
-                    util::TypeMap<WatchdogGroup, MapType, std::map<MapType, NUClear::clock::time_point>>;
 
                 /**
                  * Construct a new Watchdog Servicer object
@@ -63,18 +58,14 @@ namespace dsl {
                 explicit WatchdogServicer(const RuntimeType& data) : data(data) {}
 
                 /**
-                 * Services the watchdog
+                 * Services the watchdog.
                  *
-                 * The watchdog timer that is specified by the WatchdogGroup/RuntimeType/data combination will have its
-                 * service time updated to whatever is stored in when.
+                 * Delegates to @ref word::WatchdogDataStore::service so the write happens under the
+                 * same mutex that guards reads in the chrono controller; otherwise the time_point
+                 * would be torn-read / torn-written across threads.
                  */
                 void service() {
-                    if (WatchdogStore::get() == nullptr || WatchdogStore::get()->count(data) == 0) {
-                        throw std::domain_error("Store for <" + util::demangle(typeid(WatchdogGroup).name()) + ", "
-                                                + util::demangle(typeid(RuntimeType).name())
-                                                + "> has not been created yet or no watchdog has been set up");
-                    }
-                    WatchdogStore::get()->at(data) = when;
+                    word::WatchdogDataStore<WatchdogGroup, RuntimeType>::service(data, when);
                 }
 
             private:
@@ -94,19 +85,15 @@ namespace dsl {
              */
             template <typename WatchdogGroup>
             struct WatchdogServicer<WatchdogGroup, void> {
-                using WatchdogStore = util::TypeMap<WatchdogGroup, void, NUClear::clock::time_point>;
 
                 /**
-                 * Services the watchdog
+                 * Services the watchdog.
                  *
-                 * The watchdog timer for WatchdogGroup will have its service time updated to whatever is stored in when
+                 * Delegates to @ref word::WatchdogDataStore::service so the write happens under the
+                 * same mutex that guards reads in the chrono controller.
                  */
                 void service() {
-                    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::set(std::make_shared<NUClear::clock::time_point>(when));
+                    word::WatchdogDataStore<WatchdogGroup, void>::service(when);
                 }
 
             private:

src/extension/IOController.hpp

@@ -27,6 +27,8 @@
 #include "../dsl/word/IO.hpp"
 #include "../util/platform.hpp"
 
+#include <atomic>
+
 namespace NUClear {
 namespace extension {
 
@@ -51,6 +53,8 @@ namespace extension {
             fd_t recv{-1};     ///< This is the file descriptor that is waited on by poll
             fd_t send{-1};     ///< This is the file descriptor that is written to to wake up the poll command
             std::mutex mutex;  ///< This mutex is used to ensure that a write to poll has worked
+            /// Armed by NotifierWakeGuard during the wake-then-lock handoff; checked under mutex before ::poll().
+            std::atomic<bool> wake_requested{false};
         };
 #endif