From a5c016833ddf4e045700bbd8d4bb7eba893e71fe Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Wed, 5 Aug 2026 13:39:26 +0200 Subject: [PATCH] fix: install channel callbacks before any node runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ThreadSanitizer reported ten data races on a plain multi-node network, all the same one: Read in Channel::try_push -> push_callback_() (worker thread) Write in Channel::set_push_callback -> push_callback_ = ... (main thread) A node's push and space callbacks are std::function members living on channels it shares with its neighbours. register_callbacks() wrote them from inside start(), and a network starts its nodes one at a time — so by the time node N is being started, nodes 1..N-1 are already running and pushing into N's input channel, reading the very std::function that start() is assigning. Concurrent read and write of a std::function is a data race on its vtable pointer and buffer, not a benign one. This is the cause of the symptom a8cfe73 patched. That commit found nodes missing their startup wake because "enable_inputs() opens the channel several statements before register_callbacks() installs the push callback", and fixed it by re-asking the question with on_input_ready(). The gap it described is this race: the callback is not merely late, it is being written while another thread reads it. INode gains prepare(), which installs callbacks and starts nothing. Networks call it on every node before starting any of them, so every write happens while the pipeline is idle and the callbacks are read-only once it is live. start() calls prepare() itself when a node is used standalone, and prepare() is idempotent so both paths are safe. The flag is never cleared: the callbacks capture `this` and stay valid across a restart, so re-registering them would only add a pointless write to a live channel. a8cfe73's on_input_ready() stays, and is still needed — a network starts nodes one at a time, so an upstream node can still push into this one between its prepare() and its start(), where on_input_ready() returns early on stop_flag_ and the empty->non-empty edge is spent. It is now a level-triggered check against a benign ordering rather than cover for a race. Verified with -DKPN_SANITIZER=thread: ten races before, none of these after, across the unit suite and the contended channel stress suite. One unrelated race remains, on overlapping fire_once invocations; it is pre-existing and is fixed separately. --- include/kpn/inode.hpp | 17 ++++++++++++ include/kpn/network.hpp | 3 +++ include/kpn/pool_node.hpp | 49 +++++++++++++++++++++++++--------- include/kpn/static_network.hpp | 9 +++++++ include/kpn/variant_node.hpp | 1 + 5 files changed, 66 insertions(+), 13 deletions(-) diff --git a/include/kpn/inode.hpp b/include/kpn/inode.hpp index a7454e2..1ad9747 100644 --- a/include/kpn/inode.hpp +++ b/include/kpn/inode.hpp @@ -22,6 +22,23 @@ enum class NodeEvent { Overflow, Closed }; struct INode { virtual ~INode() = default; + + // Install channel callbacks, without starting anything. + // + // A node's push/space callbacks live in std::function members on channels + // it shares with its neighbours, and a neighbour that is already running + // reads them on its own thread. Writing one while the pipeline runs is a + // data race on the std::function — ThreadSanitizer reports it, and the + // consequence in the field was the missed startup wake a8cfe73 had to + // patch around. + // + // So a network calls prepare() on every node before it calls start() on + // any of them: all the writes happen while nothing is running, and once a + // node is live the callbacks are read-only. start() calls prepare() itself + // if it has not been called, so standalone nodes still work; it is + // idempotent, and the network relies on that. + virtual void prepare() {} + virtual void start() = 0; virtual void stop() = 0; virtual bool running() const = 0; diff --git a/include/kpn/network.hpp b/include/kpn/network.hpp index 239ad88..67686f3 100644 --- a/include/kpn/network.hpp +++ b/include/kpn/network.hpp @@ -134,6 +134,9 @@ public: void start() override { start_time_ = clock_t::now(); + // Callbacks first, everywhere, before anything runs — see INode::prepare. + for (auto& name : topo_) + nodes_.at(name)->prepare(); for (auto& name : topo_) nodes_.at(name)->start(); start_watchdog(); diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index bff47b3..cf52f2b 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -81,29 +81,36 @@ public: // ── INode ───────────────────────────────────────────────────────────────── + void prepare() override { + if (prepared_) return; // idempotent: the network calls this, + prepared_ = true; // and start() calls it again if not. + register_callbacks(std::make_index_sequence{}); + } + void start() override { + prepare(); enable_inputs(std::make_index_sequence{}); stop_flag_.store(false, std::memory_order_relaxed); gate_.force_idle(); - register_callbacks(std::make_index_sequence{}); if constexpr (input_count == 0) try_submit(0.5f); else // Never start with a wake already outstanding — the startup case of // the invariant 9c5ce5f established for the running pipeline. // - // enable_inputs() opens the channel several statements before - // register_callbacks() installs the push callback, and the network - // starts nodes sources-first, so an upstream node is already firing - // into this one during that gap. A push landing there is accepted by - // the ring but wakes nobody: Channel::push only invokes the callback - // on the empty→non-empty transition, and at that instant the - // callback is still null. Every later push sees a non-empty ring and - // stays silent, so the node is never submitted — the pipeline reads - // as wedged from the first frame, with no item ever delivered. + // The callback is installed by prepare(), before any node runs, but + // a network still starts its nodes one at a time: an upstream node + // that is already firing can push into this one between the two + // calls. The push is accepted by the ring and does invoke the + // callback, but on_input_ready() sees stop_flag_ still set and + // returns. Every later push sees a non-empty ring and stays silent + // — Channel invokes push_callback_ only on the empty->non-empty + // transition — so without this the node is never submitted and the + // pipeline reads as wedged from the first frame. // - // on_input_ready() is the level-triggered form of the same question, - // so asking it once here converts the missed edge into a state check. + // on_input_ready() is the level-triggered form of the same + // question, so asking it once here converts the missed edge into a + // state check. on_input_ready(); } @@ -584,6 +591,11 @@ private: /// Serialises firings and records wakes that arrive during one. See /// submit_gate.hpp for why this cannot be two separate flags. SubmitGate gate_; + /// Whether prepare() has installed the channel callbacks. Only ever touched + /// from the thread driving start()/stop(), never from a worker, and never + /// cleared: the callbacks capture `this` and stay valid across a restart, so + /// re-registering them would be a pointless write to a live channel. + bool prepared_{false}; /// The hidden one-slot output buffer (see push_outputs). Holding the value /// here is what lets a node stop running without dropping it or occupying a @@ -646,11 +658,17 @@ public: ~PoolObjectNode() override { stop(); } + void prepare() override { + if (prepared_) return; + prepared_ = true; + register_callbacks(std::make_index_sequence{}); + } + void start() override { + prepare(); enable_inputs(std::make_index_sequence{}); stop_flag_.store(false, std::memory_order_relaxed); gate_.force_idle(); - register_callbacks(std::make_index_sequence{}); if constexpr (input_count == 0) try_submit(0.5f); else @@ -1050,6 +1068,11 @@ private: /// Serialises firings and records wakes that arrive during one. See /// submit_gate.hpp for why this cannot be two separate flags. SubmitGate gate_; + /// Whether prepare() has installed the channel callbacks. Only ever touched + /// from the thread driving start()/stop(), never from a worker, and never + /// cleared: the callbacks capture `this` and stay valid across a restart, so + /// re-registering them would be a pointless write to a live channel. + bool prepared_{false}; /// The hidden one-slot output buffer (see push_outputs). Holding the value /// here is what lets a node stop running without dropping it or occupying a diff --git a/include/kpn/static_network.hpp b/include/kpn/static_network.hpp index bb34a2c..9c2dc32 100644 --- a/include/kpn/static_network.hpp +++ b/include/kpn/static_network.hpp @@ -126,6 +126,15 @@ public: for (auto* node : user_nodes_topo_) node->set_network_error_callback(error_handler_); } + // Install every node's channel callbacks before starting any of them. + // Those callbacks are std::function members on channels shared with + // neighbours; a neighbour that is already running reads them from its + // own thread, so writing one after the pipeline is live is a data race + // (ThreadSanitizer reports it on any multi-node network). Doing all the + // writes here, while nothing runs, makes them read-only thereafter. + for (auto* n : user_nodes_topo_) n->prepare(); + for (auto* n : fanout_nodes_ptr_) n->prepare(); + for (auto* n : user_nodes_topo_) n->start(); for (auto* n : fanout_nodes_ptr_) n->start(); #ifdef KPN_WEB_DEBUG diff --git a/include/kpn/variant_node.hpp b/include/kpn/variant_node.hpp index 9a98599..26320ca 100644 --- a/include/kpn/variant_node.hpp +++ b/include/kpn/variant_node.hpp @@ -162,6 +162,7 @@ public: // ── INode ───────────────────────────────────────────────────────────────── + void prepare() override { node_.prepare(); } void start() override { node_.start(); } void stop() override { node_.stop(); } bool running() const override { return node_.running(); }