diff --git a/include/kpn/diagnostics.hpp b/include/kpn/diagnostics.hpp index b94c8f2..884d668 100644 --- a/include/kpn/diagnostics.hpp +++ b/include/kpn/diagnostics.hpp @@ -220,6 +220,11 @@ struct ResourceSnapshot { struct IResourceProbe { virtual ~IResourceProbe() = default; virtual ResourceSnapshot snapshot(const std::string& name) const = 0; + + /// Release every thread waiting for the resource, so teardown is not held + /// up by one. A network calls this on the resources registered with it when + /// it halts; default no-op for probes with nothing to wake. + virtual void close() {} }; } // namespace kpn diff --git a/include/kpn/shared_resource.hpp b/include/kpn/shared_resource.hpp index 79924c3..9919508 100644 --- a/include/kpn/shared_resource.hpp +++ b/include/kpn/shared_resource.hpp @@ -14,6 +14,18 @@ namespace kpn { template class Channel; // forward declaration for acquire_balanced +/// Thrown by a pending acquire() when the resource is closed underneath it. +/// +/// acquire() blocks on a condition variable with no timeout and no stop +/// condition, so a node parked there ignored teardown entirely: the worker +/// never returned, the pool's join never completed, and shutdown hung on a +/// resource nobody was going to release. Closing the resource turns that into +/// an exception the node's normal error path already handles. +class ResourceClosedError : public std::runtime_error { +public: + ResourceClosedError() : std::runtime_error("shared resource closed") {} +}; + // ── SharedResource ──────────────────────────────────────────────────────────── // // Wraps an exclusive resource (e.g. an ONNX session, a CUDA stream) and @@ -72,6 +84,7 @@ public: template Guard acquire(PriorityFn&& fn) { std::unique_lock lock(mutex_); + if (closed_) throw ResourceClosedError{}; if (!held_) { held_ = true; acq_.fetch_add(1, std::memory_order_relaxed); @@ -83,18 +96,46 @@ public: current_waiters_.store(waiters_.size(), std::memory_order_relaxed); auto t0 = w.wait_start; - w.cv.wait(lock, [&w] { return w.ready; }); + // Woken either by release() handing over ownership, or by close() + // giving up on the wait entirely. + w.cv.wait(lock, [&w] { return w.ready || w.closed; }); int64_t wait_us = std::chrono::duration_cast( clock_t::now() - t0).count(); waiters_.erase(std::find(waiters_.begin(), waiters_.end(), &w)); current_waiters_.store(waiters_.size(), std::memory_order_relaxed); - acq_.fetch_add(1, std::memory_order_relaxed); total_wait_us_.fetch_add(static_cast(wait_us > 0 ? wait_us : 0), std::memory_order_relaxed); + + // Closed without being handed ownership: no Guard, so nothing to + // release, and held_ is left exactly as close() found it. + if (!w.ready) throw ResourceClosedError{}; + + acq_.fetch_add(1, std::memory_order_relaxed); return Guard(this); } + /// Wake every waiter and refuse further acquisitions. + /// + /// Teardown is the whole point: a node parked in acquire() is not + /// observing stop flags, so without this the only way out is for whoever + /// holds the resource to release it — which, if that node is also being + /// stopped, may never happen. Idempotent, and safe to call from any thread. + void close() override { + std::lock_guard lock(mutex_); + closed_ = true; + for (Waiter* w : waiters_) { + w->closed = true; + w->cv.notify_one(); + } + } + + /// Reopen after a close(). For reuse across runs; not needed for teardown. + void reopen() { + std::lock_guard lock(mutex_); + closed_ = false; + } + // Acquire with no priority (all waiters treated equally, order is fair-ish). Guard acquire() { return acquire([] { return 0.5f; }); @@ -132,7 +173,10 @@ public: private: void release() { std::unique_lock lock(mutex_); - if (waiters_.empty()) { + // Hand over only to a waiter that is still waiting. A closed one is on + // its way out and will not take ownership, so treating it as the next + // holder would leave held_ true with nobody holding it. + if (closed_ || waiters_.empty()) { held_ = false; return; } @@ -162,7 +206,8 @@ private: std::function priority_fn; clock_t::time_point wait_start; std::condition_variable cv; - bool ready{false}; + bool ready{false}; // handed ownership by release() + bool closed{false}; // woken by close() instead Waiter(std::function fn, clock_t::time_point t) : priority_fn(std::move(fn)), wait_start(t) {} @@ -172,6 +217,7 @@ private: T resource_; bool held_{false}; + bool closed_{false}; mutable std::mutex mutex_; std::vector waiters_; std::atomic acq_{0}; diff --git a/include/kpn/static_network.hpp b/include/kpn/static_network.hpp index 75ad6e6..4eda22e 100644 --- a/include/kpn/static_network.hpp +++ b/include/kpn/static_network.hpp @@ -160,6 +160,11 @@ public: #ifdef KPN_WEB_DEBUG if (web_server_) web_server_->stop(); #endif + // Release anything parked on a shared resource first. A node blocked in + // acquire() is not watching stop flags, so stopping it would wait on a + // handover that may never come — its holder is being stopped too. + for (auto& [rname, probe] : resource_probes_) { (void)rname; probe->close(); } + for (auto it = fanout_nodes_ptr_.rbegin(); it != fanout_nodes_ptr_.rend(); ++it) (*it)->stop(); for (auto it = user_nodes_topo_.rbegin(); it != user_nodes_topo_.rend(); ++it) @@ -173,6 +178,7 @@ public: #ifdef KPN_WEB_DEBUG if (web_server_) web_server_->stop(); #endif + for (auto& [rname, probe] : resource_probes_) { (void)rname; probe->close(); } // user_nodes_topo_ is already in sources-first order. // Stop each node and drain its output channels before moving on. for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) { diff --git a/tests/test_shared_resource.cpp b/tests/test_shared_resource.cpp index 7468f11..9e8abf6 100644 --- a/tests/test_shared_resource.cpp +++ b/tests/test_shared_resource.cpp @@ -237,3 +237,51 @@ TEST_CASE("make_shared_resource constructs with forwarded args", "[shared_resour auto g = res.acquire(); REQUIRE(*g == "hello"); } + +// Regression: a waiter must be releasable, or teardown waits on it forever. +// +// acquire() blocks on a condition variable whose predicate only becomes true +// when release() hands over ownership. There was no timeout and no stop +// condition, so a node parked there ignored teardown entirely: its worker never +// returned, the pool's join never completed, and shutdown hung waiting for a +// resource nobody was going to release — which is exactly the case when the +// holder is being stopped too. +// +// close() turns that into an exception the node's existing error path already +// handles, and networks now call it on registered resources before stopping any +// node, for the same reason. +TEST_CASE("closing a shared resource releases its waiters", "[shared_resource]") { + SharedResource res(42); + + auto holder = res.acquire(); // resource is now held + + std::atomic threw{false}, returned{false}; + std::thread waiter([&] { + try { + auto g = res.acquire(); // blocks: someone else holds it + (void)g; + } catch (const ResourceClosedError&) { + threw.store(true, std::memory_order_release); + } + returned.store(true, std::memory_order_release); + }); + + // Let it park, then tear down without ever releasing the holder. + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + REQUIRE_FALSE(returned.load(std::memory_order_acquire)); + + res.close(); + waiter.join(); + + CHECK(threw.load(std::memory_order_acquire)); +} + +TEST_CASE("acquiring a closed resource fails immediately", "[shared_resource]") { + SharedResource res(7); + res.close(); + CHECK_THROWS_AS(res.acquire(), ResourceClosedError); + + // Reusable across runs once reopened. + res.reopen(); + CHECK_NOTHROW(res.acquire()); +}