Simplify IOController pool identity and address PR review feedback.

TrentHouliston · cursoragent · TrentHouliston · commit 4f82a76a6272 · 2026-06-17T13:25:34.000+10:00
Move pool constexprs onto IOController, raise poll to NORMAL priority, colocate inline bump reactions with their handlers, inline the poll loop, and bring Windows onto the same scheduler-driven model.

Co-authored-by: Cursor &lt;cursoragent@cursor.com&gt;
diff --git a/docs/explanation/scheduler.md b/docs/explanation/scheduler.md
@@ -277,7 +277,14 @@ Persistent pools (`ThreadPoolDescriptor::persistent`) continue accepting tasks d
 
 ### Blocking pools and priority ordering
 
-Some extensions use a single-consumer pool with a blocking wait in the worker thread (for example, IOController on POSIX blocks in `::poll()`). Priority does not preempt inside the blocking call; instead, cross-thread control work is enqueued at HIGH priority on the same pool, a notify pipe wakes the poll thread, the LOW poll task finishes its iteration and resubmits, and the worker dequeues the already-queued HIGH task before the resubmitted poll. Control handlers that mutate poll state from another thread pair with separate default-pool bump reactions registered after the HIGH handlers so the bump runs from a thread that is not blocked in `::poll()`.
+Some extensions use a single-consumer pool with a blocking wait in the worker thread.
+For example, IOController blocks in `::poll()` or `WSAWaitForMultipleEvents()`.
+
+Priority does not preempt inside the blocking call.
+Instead, cross-thread control work is enqueued at HIGH priority on the same pool.
+An inline bump reaction writes the notify pipe or event from the emitting thread.
+The poll task finishes its iteration and resubmits at NORMAL priority.
+The worker then dequeues the already-queued HIGH task before the resubmitted poll.
 
 ## Design tradeoffs
 
diff --git a/docs/reference/extensions/built-in-extensions.md b/docs/reference/extensions/built-in-extensions.md
@@ -95,12 +95,13 @@ Monitors file descriptors for IO readiness events and triggers corresponding rea
 
 **Implementation:**
 
-**Implementation:**
-
-Uses platform-native polling mechanisms:
+Uses platform-native polling mechanisms on a dedicated `IOController` pool (`concurrency = 1`, MPSC):
 
-- **POSIX (Linux/macOS):** `poll()` with `pollfd` arrays, driven by a dedicated `IOController` pool (`concurrency = 1`, MPSC). A LOW-priority poll task blocks in `::poll()` and resubmits itself after each iteration. Control handlers (`IOConfiguration`, `IOFinished`, `Unbind`, `Shutdown`) run at HIGH priority on the same pool; separate default-pool bump reactions wake the poll thread via a notify pipe once the HIGH task is queued.
-- **Windows:** `WSAWaitForMultipleEvents` with `WSAEVENT` handles, still using `on<Always>` (scheduler-driven parity deferred).
+- **POSIX (Linux/macOS):** `poll()` with `pollfd` arrays.
+  A NORMAL-priority poll task blocks in `::poll()` and resubmits itself after each iteration.
+  Control handlers (`IOConfiguration`, `IOFinished`, `Unbind`, `Shutdown`) run at HIGH priority on the same pool.
+  Each control handler is followed by an `Inline::ALWAYS` bump reaction that writes the notify pipe from the emitting thread.
+- **Windows:** `WSAWaitForMultipleEvents` with `WSAEVENT` handles, using the same scheduler-driven pool and priority model.
 
 When events are detected on registered file descriptors, the controller creates tasks for the corresponding reactions, passing the event flags through `ThreadStore` so the `get` method can report which specific events occurred.
 
diff --git a/src/extension/IOController.hpp b/src/extension/IOController.hpp
@@ -32,15 +32,6 @@
 namespace NUClear {
 namespace extension {
 
-    /// Dedicated single-consumer pool for IOController scheduler-driven polling.
-    namespace io_pool {
-        struct IO {
-            static constexpr const char* name           = "IOController";
-            static constexpr int concurrency            = 1;
-            static constexpr bool counts_for_idle       = false;
-        };
-    }  // namespace io_pool
-
     class IOController : public Reactor {
     public:
         struct Task;
@@ -104,6 +95,13 @@ namespace extension {
         };
 
     private:
+        /// Dedicated single-consumer pool for scheduler-driven IO polling.
+        struct IOPool {
+            static constexpr const char* name     = "IOController";
+            static constexpr int concurrency      = 1;
+            static constexpr bool counts_for_idle = false;
+        };
+
         /**
          * Rebuilds the list of file descriptors to poll.
          *
@@ -133,20 +131,6 @@ namespace extension {
         // NOLINTNEXTLINE(readability-make-member-function-const) this changes states
         void bump();
 
-        /**
-         * Runs one poll iteration on the IO pool and reschedules the poll task when running.
-         *
-         * The poll loop is a LOW-priority task on a dedicated single-consumer pool. After each
-         * iteration it resubmits at LOW; cross-thread control work is queued at HIGH on the same
-         * pool before a separate bump reaction wakes ::poll(), so the worker dequeues HIGH first.
-         */
-        void poll_iteration();
-
-        /**
-         * Resubmit the poll reaction to the scheduler (same pool, LOW priority).
-         */
-        void reschedule_poll();
-
     public:
         explicit IOController(std::unique_ptr<NUClear::Environment> environment);
 
diff --git a/src/extension/IOController_Posix.ipp b/src/extension/IOController_Posix.ipp
@@ -22,6 +22,7 @@
 
 #include "IOController.hpp"
 
+#include "../dsl/word/Inline.hpp"
 #include "../dsl/word/Pool.hpp"
 #include "../dsl/word/Priority.hpp"
 #include "../dsl/word/Startup.hpp"
@@ -30,8 +31,6 @@
 namespace NUClear {
 namespace extension {
 
-    using IOPool = dsl::word::Pool<io_pool::IO>;
-
     namespace {
 
         /**
@@ -97,52 +96,6 @@ namespace extension {
 
     }  // namespace
 
-    void IOController::reschedule_poll() {
-        if (!running.load(std::memory_order_acquire)) {
-            return;
-        }
-        if (const auto* current = threading::ReactionTask::get_current_task()) {
-            powerplant.submit(current->parent->get_task());
-        }
-    }
-
-    void IOController::poll_iteration() {
-        // Rebuild the list if something changed
-        if (dirty.load(std::memory_order_acquire)) {
-            rebuild_list();
-        }
-
-        // Wait for an event to happen on one of our file descriptors
-        bool polled = false;
-        /* mutex scope */ {
-            const NotifierPollScope poll(notifier);
-            if (!poll.wake_pending()) {
-                if (::poll(watches.data(), nfds_t(watches.size()), -1) < 0) {
-                    throw std::system_error(network_errno,
-                                            std::system_category(),
-                                            "There was an IO error while attempting to poll the file descriptors");
-                }
-                polled = true;
-            }
-        }
-
-        if (polled) {
-            // Get the lock so we don't concurrently modify the list
-            const std::lock_guard<std::mutex> lock(tasks_mutex);
-            for (auto& fd : watches) {
-
-                // Collect the events that happened into the tasks list
-                // Something happened
-                if (fd.revents != 0) {
-                    process_event(fd);
-                }
-            }
-        }
-
-        // Yield to the scheduler. Any HIGH control tasks queued before the bump wake are dequeued first.
-        reschedule_poll();
-    }
-
     void IOController::rebuild_list() {
         // Get the lock so we don't concurrently modify the list
         const std::lock_guard<std::mutex> lock(tasks_mutex);
@@ -286,11 +239,7 @@ namespace extension {
         // Start by rebuilding the list
         rebuild_list();
 
-        // Control handlers run on the IO pool at HIGH priority. Separate bump reactions (default pool,
-        // registered after the HIGH handlers) wake ::poll() once the HIGH task is already queued.
-        // TypeCallbackStore iteration follows registration order, so HIGH is enqueued before bump.
-
-        on<Trigger<dsl::word::IOConfiguration>, IOPool, Priority::HIGH>().then(
+        on<Trigger<dsl::word::IOConfiguration>, Pool<IOPool>, Priority::HIGH>().then(
             "Configure IO Reaction",
             [this](const dsl::word::IOConfiguration& config) {
                 const std::lock_guard<std::mutex> lock(tasks_mutex);
@@ -301,8 +250,9 @@ namespace extension {
                 std::sort(tasks.begin(), tasks.end());
                 dirty.store(true, std::memory_order_release);
             });
+        on<Trigger<dsl::word::IOConfiguration>, Inline::ALWAYS>().then("Configure IO bump", [this] { bump(); });
 
-        on<Trigger<dsl::word::IOFinished>, IOPool, Priority::HIGH>().then("IO Finished", [this](const dsl::word::IOFinished& event) {
+        on<Trigger<dsl::word::IOFinished>, Pool<IOPool>, Priority::HIGH>().then("IO Finished", [this](const dsl::word::IOFinished& event) {
             const std::lock_guard<std::mutex> lock(tasks_mutex);
 
             auto task = std::find_if(tasks.begin(), tasks.end(), [&event](const Task& t) {
@@ -315,7 +265,6 @@ namespace extension {
                     tasks.erase(task);
                 }
                 else {
-                    // Mutate watches[] only while poll is out of ::poll() — wake-then-lock on the IO thread.
                     NotifierWakeGuard wake(notifier);
                     wake.signal();
                     const auto notifier_lock = wake.lock();
@@ -333,8 +282,9 @@ namespace extension {
                 }
             }
         });
+        on<Trigger<dsl::word::IOFinished>, Inline::ALWAYS>().then("IO Finished bump", [this] { bump(); });
 
-        on<Trigger<dsl::operation::Unbind<IO>>, IOPool, Priority::HIGH>().then(
+        on<Trigger<dsl::operation::Unbind<IO>>, Pool<IOPool>, Priority::HIGH>().then(
             "Unbind IO Reaction",
             [this](const dsl::operation::Unbind<IO>& unbind) {
                 const std::lock_guard<std::mutex> lock(tasks_mutex);
@@ -349,18 +299,46 @@ namespace extension {
 
                 dirty.store(true, std::memory_order_release);
             });
+        on<Trigger<dsl::operation::Unbind<IO>>, Inline::ALWAYS>().then("Unbind IO bump", [this] { bump(); });
 
-        on<Shutdown, IOPool, Priority::HIGH>().then("Shutdown IO Controller", [this] {
+        on<Shutdown, Pool<IOPool>, Priority::HIGH>().then("Shutdown IO Controller", [this] {
             running.store(false, std::memory_order_release);
         });
+        on<Shutdown, Inline::ALWAYS>().then("Shutdown IO bump", [this] { bump(); });
 
-        on<Trigger<dsl::word::IOConfiguration>>().then("Configure IO bump", [this] { bump(); });
-        on<Trigger<dsl::word::IOFinished>>().then("IO Finished bump", [this] { bump(); });
-        on<Trigger<dsl::operation::Unbind<IO>>>().then("Unbind IO bump", [this] { bump(); });
-        on<Shutdown>().then("Shutdown IO bump", [this] { bump(); });
+        on<Startup, Pool<IOPool>, Priority::NORMAL>().then("IO Poll", [this] {
+            if (!running.load(std::memory_order_acquire)) {
+                return;
+            }
 
-        // Scheduler-driven poll: dedicated single-consumer IO pool, LOW priority, self-resubmitting.
-        on<Startup, IOPool, Priority::LOW>().then("IO Poll", [this] { poll_iteration(); });
+            if (dirty.load(std::memory_order_acquire)) {
+                rebuild_list();
+            }
+
+            bool polled = false;
+            {
+                const NotifierPollScope poll(notifier);
+                if (!poll.wake_pending()) {
+                    if (::poll(watches.data(), nfds_t(watches.size()), -1) < 0) {
+                        throw std::system_error(network_errno,
+                                                std::system_category(),
+                                                "There was an IO error while attempting to poll the file descriptors");
+                    }
+                    polled = true;
+                }
+            }
+
+            if (polled) {
+                const std::lock_guard<std::mutex> lock(tasks_mutex);
+                for (auto& fd : watches) {
+                    if (fd.revents != 0) {
+                        process_event(fd);
+                    }
+                }
+            }
+
+            powerplant.submit(threading::ReactionTask::get_current_task()->parent->get_task());
+        });
     }
 
 }  // namespace extension
diff --git a/src/extension/IOController_Windows.ipp b/src/extension/IOController_Windows.ipp
