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(); }