From abbb2d47700b38b7147b7c98e022bf951add7284 Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Wed, 5 Aug 2026 15:14:06 +0200 Subject: [PATCH] fix: submitting to a stopped pool must be refused, not fatal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ThreadPool::stop() ends with queues_.clear(), and submit() went straight to queues_[target] with no check. A submission arriving after stop indexed an empty vector and segfaulted. This is not a contrived teardown ordering. A node's space callback fires from whichever thread drained the channel, and that thread belongs to the *consumer*; the callback it runs belongs to the *producer*. 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: ThreadPool::submit <- source node's space_callback <- Channel::try_pop_now (relay draining its input) <- relay fire_once The static-network shutdown case in the next commit crashed about 12 runs in 20 on this. It survived until now because halt() stops in reverse topological order — consumers first — so the producer whose callback might fire is always still alive. shutdown() stops sources first and does not have that protection. Reading stopped_ without a lock would not fix it: the window between the read and the indexing is exactly where clear() runs. submit() takes a shared lock and stop() an exclusive one, so submissions still proceed in parallel with each other while being serialised against teardown. stop() sets the flag under the lock, releases it to join — a worker's task may itself call submit, and holding the lock across the join would deadlock against that — then retakes it to destroy the queues. Refusals are counted rather than silent. A teardown race is expected, but a node repeatedly trying to run after its pool is gone is worth being able to see. try_submit also checks stop_flag_ first, so a stopped node cannot claim the submit gate and leave it held. Not a smart-pointer problem, for anyone reading the crash: nothing here is owned by a raw pointer. It is std::vector::operator[] on a vector that was emptied by another thread. Verified in both directions: with the guard removed the new scheduler cases segfault; with it they pass. --- include/kpn/pool_node.hpp | 12 +++++++++ include/kpn/scheduler.hpp | 42 ++++++++++++++++++++++++++++- tests/test_scheduler.cpp | 57 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 110 insertions(+), 1 deletion(-) 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); + } +}