dtourolle
a5c016833d
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.
71 lines
3.1 KiB
C++
71 lines
3.1 KiB
C++
#pragma once
|
|
#include "diagnostics.hpp"
|
|
#include <chrono>
|
|
#include <functional>
|
|
#include <string>
|
|
#include <string_view>
|
|
|
|
namespace kpn {
|
|
|
|
// Called when a node's function throws. Return true to skip the failed
|
|
// invocation and keep running, false to stop the node.
|
|
using NodeErrorHandler = std::function<bool(std::string_view node_name, std::exception_ptr)>;
|
|
|
|
// Lightweight timestamp-only callback fired on per-node events.
|
|
// The node name is known at registration time so it is not included here.
|
|
using NodeEventCallback = std::function<void(std::chrono::steady_clock::time_point)>;
|
|
|
|
// Event types reported to the network-level aggregate callback.
|
|
enum class NodeEvent { Overflow, Closed };
|
|
|
|
// ── INode — type-erased interface for Network / watchdog ─────────────────────
|
|
|
|
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;
|
|
virtual const NodeStats& stats() const = 0;
|
|
virtual NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const = 0;
|
|
virtual void set_name(std::string name) = 0;
|
|
|
|
// Network-injected callbacks (slot 1 of each node's callback array).
|
|
// Default no-ops; overridden by PoolNode, PoolObjectNode, InterruptNode.
|
|
virtual void set_network_overflow_callback(NodeEventCallback) {}
|
|
virtual void set_network_closed_callback(NodeEventCallback) {}
|
|
|
|
// Network-level error listener. Consulted when a node's function throws
|
|
// and no per-node handler resolved it. Without this the exception is
|
|
// discarded and the failure is only visible as a Closed event, which says
|
|
// a node stopped but not why — the difference between a diagnosis and a
|
|
// guess. Same contract as NodeErrorHandler: true to continue, false to
|
|
// stop the node.
|
|
virtual void set_network_error_callback(NodeErrorHandler) {}
|
|
|
|
// halt(): alias for stop() — immediate, discards in-flight work.
|
|
virtual void halt() { stop(); }
|
|
|
|
// shutdown(): graceful drain before stopping. Base implementation falls
|
|
// back to stop(). Network and StaticNetwork override with topo-ordered drain.
|
|
virtual void shutdown() { stop(); }
|
|
};
|
|
|
|
} // namespace kpn
|