Extract shared IOController scheduler wiring to reduce duplication.

TrentHouliston · cursoragent · TrentHouliston · commit 62617d2979ca · 2026-06-18T17:00:32.000+10:00
Move bump reactions, shutdown control, and poll resubmit logic into
IOController_Common.ipp so POSIX and Windows paths share one implementation
and satisfy SonarCloud duplicated-lines quality gate.

Co-authored-by: Cursor &lt;cursoragent@cursor.com&gt;
diff --git a/src/extension/IOController.cpp b/src/extension/IOController.cpp
@@ -25,3 +25,5 @@
 #else
     #include "IOController_Posix.ipp"
 #endif  // _WIN32
+
+#include "IOController_Common.ipp"
diff --git a/src/extension/IOController.hpp b/src/extension/IOController.hpp
@@ -28,6 +28,7 @@
 #include "../util/platform.hpp"
 
 #include <atomic>
+#include <functional>
 
 namespace NUClear {
 namespace extension {
@@ -126,6 +127,25 @@ namespace extension {
          */
         void bump() const;
 
+        /// Inline bump reactions that wake the poll task from the emitting thread.
+        void register_inline_bump_reactions();
+
+        /// HIGH-priority shutdown handler on the IO pool.
+        void register_shutdown_control();
+
+        /**
+         * Registers the self-resubmitting poll task.
+         *
+         * @param wait_and_process platform-specific blocking wait and event dispatch
+         */
+        void register_poll_loop(std::function<void()> wait_and_process);
+
+        /// Returns false when the poll loop should exit without blocking.
+        bool prepare_poll_iteration();
+
+        /// Resubmits the poll reaction after one blocking iteration.
+        void resubmit_poll_task();
+
     public:
         explicit IOController(std::unique_ptr<NUClear::Environment> environment);
 
diff --git a/src/extension/IOController_Common.ipp b/src/extension/IOController_Common.ipp
@@ -0,0 +1,71 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2026 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 "IOController.hpp"
+
+#include "../threading/ReactionTask.hpp"
+
+namespace NUClear {
+namespace extension {
+
+    void IOController::register_inline_bump_reactions() {
+        on<Trigger<dsl::word::IOConfiguration>, Inline::ALWAYS>().then("Configure IO bump", [this] { bump(); });
+        on<Trigger<dsl::word::IOFinished>, Inline::ALWAYS>().then("IO Finished bump", [this] { bump(); });
+        on<Trigger<dsl::operation::Unbind<IO>>, Inline::ALWAYS>().then("Unbind IO bump", [this] { bump(); });
+        on<Shutdown, Inline::ALWAYS>().then("Shutdown IO bump", [this] { bump(); });
+    }
+
+    void IOController::register_shutdown_control() {
+        on<Shutdown, Pool<IOPool>, Priority::HIGH, Inline::NEVER>().then("Shutdown IO Controller", [this] {
+            running.store(false, std::memory_order_release);
+        });
+    }
+
+    bool IOController::prepare_poll_iteration() {
+        if (!running.load(std::memory_order_acquire)) {
+            return false;
+        }
+
+        if (dirty.load(std::memory_order_acquire)) {
+            rebuild_list();
+        }
+
+        return true;
+    }
+
+    void IOController::resubmit_poll_task() {
+        powerplant.submit(threading::ReactionTask::get_current_task()->parent->get_task());
+    }
+
+    void IOController::register_poll_loop(std::function<void()> wait_and_process) {
+        on<Startup, Pool<IOPool>, Priority::NORMAL, Inline::NEVER>().then("IO Poll", [this, wait = std::move(wait_and_process)] {
+            if (!prepare_poll_iteration()) {
+                return;
+            }
+
+            wait();
+            resubmit_poll_task();
+        });
+    }
+
+}  // namespace extension
+}  // namespace NUClear
diff --git a/src/extension/IOController_Posix.ipp b/src/extension/IOController_Posix.ipp
@@ -22,8 +22,6 @@
 
 #include "IOController.hpp"
 
-#include "../threading/ReactionTask.hpp"
-
 namespace NUClear {
 namespace extension {
 
@@ -169,9 +167,6 @@ namespace extension {
         // Start by rebuilding the list
         rebuild_list();
 
-        // Control handlers run on the IO pool at HIGH priority with Inline::NEVER so they are
-        // always queued to the IO pool rather than running on the emitting thread. Inline wake
-        // reactions registered after them only write to the notify pipe so ::poll() unblocks.
         on<Trigger<dsl::word::IOConfiguration>, Pool<IOPool>, Priority::HIGH, Inline::NEVER>().then(
             "Configure IO Reaction",
             [this](const dsl::word::IOConfiguration& config) {
@@ -181,7 +176,6 @@ 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>, Pool<IOPool>, Priority::HIGH, Inline::NEVER>().then("IO Finished", [this](const dsl::word::IOFinished& event) {
             auto task = std::find_if(tasks.begin(), tasks.end(), [&event](const Task& t) {
@@ -207,7 +201,6 @@ namespace extension {
                 }
             }
         });
-        on<Trigger<dsl::word::IOFinished>, Inline::ALWAYS>().then("IO Finished bump", [this] { bump(); });
 
         on<Trigger<dsl::operation::Unbind<IO>>, Pool<IOPool>, Priority::HIGH, Inline::NEVER>().then(
             "Unbind IO Reaction",
@@ -222,22 +215,10 @@ 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, Pool<IOPool>, Priority::HIGH, Inline::NEVER>().then("Shutdown IO Controller", [this] {
-            running.store(false, std::memory_order_release);
-        });
-        on<Shutdown, Inline::ALWAYS>().then("Shutdown IO bump", [this] { bump(); });
-
-        on<Startup, Pool<IOPool>, Priority::NORMAL, Inline::NEVER>().then("IO Poll", [this] {
-            if (!running.load(std::memory_order_acquire)) {
-                return;
-            }
-
-            if (dirty.load(std::memory_order_acquire)) {
-                rebuild_list();
-            }
 
+        register_shutdown_control();
+        register_inline_bump_reactions();
+        register_poll_loop([this] {
             if (::poll(watches.data(), nfds_t(watches.size()), -1) < 0) {
                 throw std::system_error(network_errno,
                                         std::system_category(),
@@ -249,8 +230,6 @@ namespace extension {
                     process_event(fd);
                 }
             }
-
-            powerplant.submit(threading::ReactionTask::get_current_task()->parent->get_task());
         });
     }
 
diff --git a/src/extension/IOController_Windows.ipp b/src/extension/IOController_Windows.ipp
@@ -22,8 +22,6 @@
 
 #include "IOController.hpp"
 
-#include "../threading/ReactionTask.hpp"
-
 namespace NUClear {
 namespace extension {
 
@@ -163,7 +161,6 @@ namespace extension {
                 tasks.insert(std::make_pair(event, Task{config.fd, config.events, config.reaction}));
                 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>, Pool<IOPool>, Priority::HIGH, Inline::NEVER>().then("IO Finished", [this](const dsl::word::IOFinished& event) {
             auto it = std::find_if(tasks.begin(), tasks.end(), [&event](const std::pair<WSAEVENT, Task>& t) {
@@ -182,7 +179,6 @@ namespace extension {
                 }
             }
         });
-        on<Trigger<dsl::word::IOFinished>, Inline::ALWAYS>().then("IO Finished bump", [this] { bump(); });
 
         on<Trigger<dsl::operation::Unbind<IO>>, Pool<IOPool>, Priority::HIGH, Inline::NEVER>().then(
             "Unbind IO Reaction",
@@ -197,22 +193,10 @@ 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, Pool<IOPool>, Priority::HIGH, Inline::NEVER>().then("Shutdown IO Controller", [this] {
-            running.store(false, std::memory_order_release);
-        });
-        on<Shutdown, Inline::ALWAYS>().then("Shutdown IO bump", [this] { bump(); });
-
-        on<Startup, Pool<IOPool>, Priority::NORMAL, Inline::NEVER>().then("IO Poll", [this] {
-            if (!running.load(std::memory_order_acquire)) {
-                return;
-            }
-
-            if (dirty.load(std::memory_order_acquire)) {
-                rebuild_list();
-            }
 
+        register_shutdown_control();
+        register_inline_bump_reactions();
+        register_poll_loop([this] {
             const DWORD event_index = WSAWaitForMultipleEvents(static_cast<DWORD>(watches.size()),
                                                                watches.data(),
                                                                false,
@@ -223,8 +207,6 @@ namespace extension {
                 auto& event = watches[event_index - WSA_WAIT_EVENT_0];
                 process_event(event);
             }
-
-            powerplant.submit(threading::ReactionTask::get_current_task()->parent->get_task());
         });
     }
 
