From ad7ffc48290daab11df259d6efc3486f477a2a8d Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Mon, 13 Jul 2026 20:01:35 +0300 Subject: [PATCH] EventEmitter: offAndWait()/removeAllListenersAndWait() for the destroy-vs-in-flight-handler race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An owner that removes a listener and immediately frees the state the listener captured (the ListeningRpcCommunicator::destroy() pattern from the phase-C6 crashes) could race a handler invocation still running on another thread — a use-after-free. The phase-C6 attempt to close this made off() itself wait for in-flight invocations and had to be reverted: it hung Layer1ScaleTest.MultipleLayer1Dht and the 64-node ContentDeliveryLayerNode test. Reproducing the hang with thread samples showed two deadlock shapes, so blocking cannot be the default off() semantics: 1. Handler-handler cycle: one thread ran a connection's Data handler (handshake success -> stopHandshaker -> off) while another ran the same connection's Disconnected handler (close tie-break -> off); each waited forever for the other. One of the two threads was a Simulator association-executor delivery task, which is why the hang surfaced as Simulator::stop()'s inFlightOperations drain never reaching zero. 2. Caller-held-lock ABBA: ConnectionManager::acceptNewConnection holds the Endpoint state mutex when detachFromPendingConnection removes the Connected listener, while the in-flight Connected handler blocks on that same mutex. The landed design keeps off()/removeAllListeners() non-blocking (their historical semantics) and adds explicit waiting variants: - offAndWait()/removeAllListenersAndWait() remove the listener(s), cancel not-yet-started invocations of an executing emit loop, and wait until no invocation is running on another thread. - The wait is skipped when the caller is itself inside any handler invocation, tracked by a thread-local invocation depth bumped around every handler call: every blocked waiter then runs no handler itself, so no waiter can be part of a cycle of handlers waiting for each other (shape 1). Callers must not hold locks the event's handlers take (shape 2) — documented on the methods. - ListeningRpcCommunicator::destroy() uses the waiting variants. Also hardened the Simulator's delivery lambda: folly::SerialExecutor swallows task exceptions, so an exception thrown by a delivery call-out would silently skip the inFlightOperations decrement and park stop()'s drain forever. The decrement now runs on all paths and the exception is logged. (No such exception was observed at runtime; this closes the hazard found while investigating.) Regression tests: offAndWait/removeAllListenersAndWait complete only after the in-flight handler finishes (fails on the non-waiting semantics), and the shape-1 deadlock replica (two handlers removing each other across threads) completes without deadlock. Verification: the reintroduced blocking-off() build deadlocked 8 seconds into Layer1ScaleTest.MultipleLayer1Dht; with this design the dht scale test and the 64-node test passed 8/8 runs (instrumented and clean) with zero blocked-wait log lines. Co-Authored-By: Claude Fable 5 --- .../connection/simulator/Simulator.cppm | 35 ++- .../transport/ListeningRpcCommunicator.cppm | 12 +- .../modules/EventEmitter.cppm | 244 +++++++++++++++++- .../test/unit/EventEmitterTest.cpp | 130 ++++++++++ 4 files changed, 401 insertions(+), 20 deletions(-) diff --git a/packages/streamr-dht/modules/connection/simulator/Simulator.cppm b/packages/streamr-dht/modules/connection/simulator/Simulator.cppm index 78e8eab6..4052d19e 100644 --- a/packages/streamr-dht/modules/connection/simulator/Simulator.cppm +++ b/packages/streamr-dht/modules/connection/simulator/Simulator.cppm @@ -30,6 +30,7 @@ module; #include #include #include +#include #include #include #include @@ -301,16 +302,30 @@ private: this->inFlightOperations++; lock.unlock(); executor->add([this, operation]() { - switch (operation.type) { - case OperationType::CONNECT: - this->executeConnectOperation(operation); - break; - case OperationType::CLOSE: - this->executeCloseOperation(operation); - break; - case OperationType::SEND: - this->executeSendOperation(operation); - break; + // The decrement below must run even if the call-out + // throws: folly's SerialExecutor swallows task exceptions + // (SerialExecutor::worker invokeCatchingExns), so a + // propagated exception would silently skip the decrement + // and park stop()'s drain-wait forever. + try { + switch (operation.type) { + case OperationType::CONNECT: + this->executeConnectOperation(operation); + break; + case OperationType::CLOSE: + this->executeCloseOperation(operation); + break; + case OperationType::SEND: + this->executeSendOperation(operation); + break; + } + } catch (const std::exception& e) { + SLogger::error( + "Simulator operation threw an exception: " + + std::string(e.what())); + } catch (...) { + SLogger::error( + "Simulator operation threw a non-std exception"); } // Notify under the lock: once stop()'s drain-wait sees the // count hit zero the Simulator may be destroyed, so this diff --git a/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm b/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm index cec92642..f5a72bca 100644 --- a/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm +++ b/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm @@ -78,13 +78,21 @@ public: } void destroy() { - transport.off(this->onMessageHandlerToken); + // The waiting variant: the owner destroys this communicator right + // after destroy() returns, and a Message/Disconnected handler + // mid-invocation on another thread would then dereference freed + // memory. offAndWait() returns only once no such invocation is + // running (unless destroy() itself is called from inside a + // handler invocation, where waiting could deadlock and the + // historical non-blocking semantics are kept). + transport.offAndWait( + this->onMessageHandlerToken); // Also detach the Disconnected listener: leaving it registered // dangles `this` on the transport after destruction, and a live // listener could still add client errors while a retired // communicator waits to be destroyed (see // RecursiveOperationManager::retiredSessionCommunicators). - transport.off( + transport.offAndWait( this->onDisconnectedHandlerToken); } using RpcCommunicator::registerRpcMethod; diff --git a/packages/streamr-eventemitter/modules/EventEmitter.cppm b/packages/streamr-eventemitter/modules/EventEmitter.cppm index 67c1d4f7..9651f03b 100644 --- a/packages/streamr-eventemitter/modules/EventEmitter.cppm +++ b/packages/streamr-eventemitter/modules/EventEmitter.cppm @@ -4,17 +4,50 @@ // this file is now the source of truth. module; +#include +#include #include #include #include #include #include #include +#include #include #include +#include export module streamr.eventemitter.EventEmitter; +namespace streamr::eventemitter::detail { + +// Number of event-handler invocations currently on THIS thread's call +// stack, across ALL emitter instances and event types. offAndWait() and +// removeAllListenersAndWait() block on in-flight invocations only when +// this is zero: a wait issued from inside a handler can deadlock, +// because handlers routinely remove each other's listeners from within +// handler context. The proven cycle (Layer1ScaleTest, 2026-07-13): +// thread A ran a connection's Data handler (handshake success -> +// stopHandshaker -> off) while thread B ran the same +// connection's Disconnected handler (close tie-break -> off) — +// each waited for the other's invocation to end, forever. Depth must be +// global (not per-emitter): the cycle spans emitter instances. +// With the depth guard, every blocked waiter runs no handler itself, so +// wait edges only point from non-handler threads to handler-running +// threads and no cycle can form among them. +inline thread_local size_t handlerInvocationDepth = 0; + +struct InvocationDepthGuard { + InvocationDepthGuard() { ++handlerInvocationDepth; } + ~InvocationDepthGuard() { --handlerInvocationDepth; } + InvocationDepthGuard(const InvocationDepthGuard&) = delete; + InvocationDepthGuard& operator=(const InvocationDepthGuard&) = delete; + InvocationDepthGuard(InvocationDepthGuard&&) = delete; + InvocationDepthGuard& operator=(InvocationDepthGuard&&) = delete; +}; + +} // namespace streamr::eventemitter::detail + export namespace streamr::eventemitter { // Event class that all events must inherit from. @@ -135,6 +168,65 @@ private: std::map mExecutingEmitLoopHandlers; + // Invocations currently RUNNING, per handler id (one entry per + // invoking thread; replays and the emit loop may overlap). Guarded + // by mExecutingEmitLoopHandlersMutex. offAndWait() blocks on + // mInvocationFinished until the removed handler has no in-flight + // invocation on another thread, so a listener's owner may free the + // captured state right after offAndWait() returns — without this, a + // handler mid-invocation on the emit thread dereferences freed + // memory (seen as ListeningRpcCommunicator crashes when a transport + // Message raced the communicator's destroy()). + std::map> mInFlightInvocations; + std::condition_variable mInvocationFinished; + + // Call under mExecutingEmitLoopHandlersMutex. + [[nodiscard]] bool hasInFlightInvocationOnOtherThread( + size_t handlerId) const { + const auto it = mInFlightInvocations.find(handlerId); + if (it == mInFlightInvocations.end()) { + return false; + } + return std::ranges::any_of( + it->second, [](const std::thread::id& threadId) { + return threadId != std::this_thread::get_id(); + }); + } + + // Call under mExecutingEmitLoopHandlersMutex. + [[nodiscard]] bool hasAnyInFlightInvocationOnOtherThread() const { + return std::ranges::any_of(mInFlightInvocations, [](const auto& entry) { + return std::ranges::any_of( + entry.second, [](const std::thread::id& threadId) { + return threadId != std::this_thread::get_id(); + }); + }); + } + + void beginInvocation(size_t handlerId) { + std::lock_guard guard{mExecutingEmitLoopHandlersMutex}; + mInFlightInvocations[handlerId].push_back(std::this_thread::get_id()); + } + + void endInvocation(size_t handlerId) { + { + std::lock_guard guard{mExecutingEmitLoopHandlersMutex}; + const auto it = mInFlightInvocations.find(handlerId); + if (it != mInFlightInvocations.end()) { + auto& threads = it->second; + const auto threadIt = + std::ranges::find(threads, std::this_thread::get_id()); + if (threadIt != threads.end()) { + threads.erase(threadIt); + } + if (threads.empty()) { + mInFlightInvocations.erase(it); + } + } + } + mInvocationFinished.notify_all(); + } + std::optional> mLatestEvent; std::mutex mLatestEventMutex; @@ -172,6 +264,12 @@ private: mExecutingEmitLoopHandlers.erase(handlerIterator); + // Marked in-flight atomically with the pop, so off() sees the + // handler either in the pending snapshot or in-flight — never + // in the gap between the two. + mInFlightInvocations[handler.getId()].push_back( + std::this_thread::get_id()); + return handler; } @@ -244,7 +342,16 @@ public: } } if (replayArguments.has_value()) { - invokeLatestEvent(handler, std::move(replayArguments.value())); + this->beginInvocation(handler.getId()); + try { + detail::InvocationDepthGuard depthGuard; + invokeLatestEvent( + handler, std::move(replayArguments.value())); + } catch (...) { + this->endInvocation(handler.getId()); + throw; + } + this->endInvocation(handler.getId()); if (once) { return HandlerToken{}; } @@ -284,14 +391,67 @@ public: template EventType> void off(HandlerToken handlerReference) { - std::lock_guard guard{mEventHandlersMutex}; + { + std::lock_guard guard{mEventHandlersMutex}; + mEventHandlers.remove_if( + [&handlerReference]( + const typename EmitterEventType::Handler& handler) { + return handler.getId() == handlerReference.getId(); + }); + } + // Also remove the pending copy an executing emit loop may still + // hold. NOTE: an invocation of this handler that already STARTED + // on another thread may still be running when off() returns; an + // owner that frees the handler's captures right after off() must + // use offAndWait() instead. + this->removeHandlerFromExecutingEmitLoops(handlerReference); + } - mEventHandlers.remove_if( - [&handlerReference]( - const typename EmitterEventType::Handler& handler) { - return handler.getId() == handlerReference.getId(); + /** + * @brief Remove an event listener and wait until it is no longer + * running, so the caller may free the listener's captured state + * immediately afterwards. + * + * @details Waits out an invocation already running on ANOTHER + * thread. The wait is skipped when called from inside any event + * handler invocation (on any emitter): handlers legitimately remove + * each other's listeners from handler context, and blocking there + * deadlocks — two threads, each inside a handler the other wants + * removed, would wait on each other forever (seen between a + * connection's Data and Disconnected handlers in the Layer1 scale + * tests). The caller must also not hold any lock that this event's + * handlers can take, or the wait deadlocks on that lock (seen + * between Endpoint's state mutex and a PendingConnection Connected + * handler). + * + * @tparam EventType The type of the event to remove the listener from. + * @param handlerReference The token of the listener to remove. + */ + + template EventType> + void offAndWait(HandlerToken handlerReference) { + { + std::lock_guard guard{mEventHandlersMutex}; + mEventHandlers.remove_if( + [&handlerReference]( + const typename EmitterEventType::Handler& handler) { + return handler.getId() == handlerReference.getId(); + }); + } + // Remove the pending snapshot copy and wait out an invocation + // already running on another thread. NB: the wait must not hold + // mEventHandlersMutex: the running handler may need it + // (once-removal, on()/off() reentry). + std::unique_lock lock{mExecutingEmitLoopHandlersMutex}; + mExecutingEmitLoopHandlers.erase(handlerReference.getId()); + // Wait only from non-handler context (see the method comment and + // detail::handlerInvocationDepth). + if (detail::handlerInvocationDepth == 0) { + mInvocationFinished.wait(lock, [this, &handlerReference]() { + return !this->hasInFlightInvocationOnOtherThread( + handlerReference.getId()); }); - removeHandlerFromExecutingEmitLoops(handlerReference); + } } /** @@ -319,6 +479,33 @@ public: mEventHandlers.clear(); } + /** + * @brief Remove all listeners for an event type and wait until none + * of them is still running — same guarantee and same caller + * constraints as offAndWait(), for every handler at once. + * + * @tparam EventType The type of the event to remove all listeners for. + */ + + template EventType> + void removeAllListenersAndWait() { + { + std::lock_guard guard{mEventHandlersMutex}; + mEventHandlers.clear(); + } + // Also cancel invocations an in-progress emit loop has not started + // yet — after this method returns, no handler may start. + std::unique_lock lock{mExecutingEmitLoopHandlersMutex}; + mExecutingEmitLoopHandlers.clear(); + // Wait only from non-handler context — same deadlock-avoidance + // rule as offAndWait(). + if (detail::handlerInvocationDepth == 0) { + mInvocationFinished.wait(lock, [this]() { + return !this->hasAnyInFlightInvocationOnOtherThread(); + }); + } + } + /** * @brief Emit an event to the event emitter. * @@ -362,7 +549,14 @@ public: // event, for instance). The per-handler copy is the intended // fan-out semantics; Handler::operator() then moves the copy // on into the stored std::function. - std::invoke(handler, args...); + try { + detail::InvocationDepthGuard depthGuard; + std::invoke(handler, args...); + } catch (...) { + this->endInvocation(handler.getId()); + throw; + } + this->endInvocation(handler.getId()); if (handler.isOnce()) { std::lock_guard guard{mEventHandlersMutex}; mEventHandlers.remove(handler); @@ -413,8 +607,10 @@ public: using EventEmitterImpl::on...; using EventEmitterImpl::once...; using EventEmitterImpl::off...; + using EventEmitterImpl::offAndWait...; using EventEmitterImpl::listenerCount...; using EventEmitterImpl::removeAllListeners...; + using EventEmitterImpl::removeAllListenersAndWait...; using EventEmitterImpl::emit...; template @@ -437,6 +633,21 @@ public: EventTypes>(), ...); } + + /** + * @brief Remove all listeners for all event types and wait until none + * of them is still running (see + * EventEmitterImpl::removeAllListenersAndWait for the caller + * constraints). + * + */ + + template + void removeAllListenersAndWait() { + (EventEmitterImpl::template removeAllListenersAndWait< + EventTypes>(), + ...); + } }; // ReplayEventEmitter is an EventEmitter that replays the latest event to new @@ -458,8 +669,10 @@ public: using EventEmitterImpl::on...; using EventEmitterImpl::once...; using EventEmitterImpl::off...; + using EventEmitterImpl::offAndWait...; using EventEmitterImpl::listenerCount...; using EventEmitterImpl::removeAllListeners...; + using EventEmitterImpl::removeAllListenersAndWait...; using EventEmitterImpl::emit...; template @@ -485,6 +698,21 @@ public: EventTypes>(), ...); } + + /** + * @brief Remove all listeners for all event types and wait until none + * of them is still running (see + * EventEmitterImpl::removeAllListenersAndWait for the caller + * constraints). + * + */ + + template + void removeAllListenersAndWait() { + (EventEmitterImpl::template removeAllListenersAndWait< + EventTypes>(), + ...); + } }; } // namespace streamr::eventemitter diff --git a/packages/streamr-eventemitter/test/unit/EventEmitterTest.cpp b/packages/streamr-eventemitter/test/unit/EventEmitterTest.cpp index efa2d787..47402922 100644 --- a/packages/streamr-eventemitter/test/unit/EventEmitterTest.cpp +++ b/packages/streamr-eventemitter/test/unit/EventEmitterTest.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -509,3 +510,132 @@ TEST_F(EventEmitterTest, TestEventMethod) { eventEmitter.emit("Hello, world!"); ASSERT_EQ(promise1.get_future().get(), "Hello, world!"); } + +// --- Regression tests for the destroy-vs-in-flight-handler race ------------- +// An owner that removes a listener and then frees the state its handler +// captured must be able to rely on the handler no longer running: +// offAndWait()/removeAllListenersAndWait() wait out an invocation in +// flight on another thread — but only when called from OUTSIDE any +// handler invocation: handlers legitimately remove each other's listeners +// from within handler context, and waiting there deadlocks (two threads, +// each inside a handler the other wants removed, waited on each other +// forever in the Layer1 scale tests). Plain off()/removeAllListeners() +// never wait. + +constexpr int inFlightHandlerDurationMs = 300; +constexpr int deadlockWatchdogSeconds = 20; + +TEST_F(EventEmitterTest, OffWaitsForInFlightHandlerBeforeOwnerFreesState) { + struct TestEvent : Event<> {}; + using Events = std::tuple; + + EventEmitter eventEmitter; + + std::promise handlerEntered; + // Stands in for owner state the handler dereferences right up to its + // last statement; pre-wait off() semantics let the owner free it while + // the handler was still running (use-after-free in real owners). + std::atomic handlerFinished{false}; + + auto token = eventEmitter.on( + [&handlerEntered, &handlerFinished]() -> void { + handlerEntered.set_value(); + std::this_thread::sleep_for( + std::chrono::milliseconds(inFlightHandlerDurationMs)); + handlerFinished.store(true); + }); + + std::thread emitterThread( + [&eventEmitter]() { eventEmitter.emit(); }); + + handlerEntered.get_future().wait(); + eventEmitter.offAndWait(token); + // offAndWait() returned from a non-handler thread: the in-flight + // invocation must be complete, so freeing the captured state is now + // safe. + EXPECT_EQ(handlerFinished.load(), true); + + emitterThread.join(); +} + +TEST_F(EventEmitterTest, RemoveAllListenersWaitsForInFlightHandler) { + struct TestEvent : Event<> {}; + using Events = std::tuple; + + EventEmitter eventEmitter; + + std::promise handlerEntered; + std::atomic handlerFinished{false}; + + eventEmitter.on([&handlerEntered, &handlerFinished]() -> void { + handlerEntered.set_value(); + std::this_thread::sleep_for( + std::chrono::milliseconds(inFlightHandlerDurationMs)); + handlerFinished.store(true); + }); + + std::thread emitterThread( + [&eventEmitter]() { eventEmitter.emit(); }); + + handlerEntered.get_future().wait(); + eventEmitter.removeAllListenersAndWait(); + EXPECT_EQ(handlerFinished.load(), true); + + emitterThread.join(); +} + +TEST_F(EventEmitterTest, HandlersRemovingEachOtherAcrossThreadsDoNotDeadlock) { + // The exact shape that deadlocked the Layer1 scale tests: thread A is + // inside the Data handler and off()s the Disconnected listener while + // thread B is inside the Disconnected handler and off()s the Data + // listener. off() from handler context must not block on the foreign + // in-flight invocation. + struct EventA : Event<> {}; + struct EventB : Event<> {}; + using Events = std::tuple; + + EventEmitter eventEmitter; + + std::atomic handlerAStarted{false}; + std::atomic handlerBStarted{false}; + HandlerToken tokenA; + HandlerToken tokenB; + + tokenA = eventEmitter.on( + [&eventEmitter, &handlerAStarted, &handlerBStarted, &tokenB]() -> void { + handlerAStarted.store(true); + while (!handlerBStarted.load()) { + std::this_thread::yield(); + } + eventEmitter.offAndWait(tokenB); + }); + tokenB = eventEmitter.on( + [&eventEmitter, &handlerAStarted, &handlerBStarted, &tokenA]() -> void { + handlerBStarted.store(true); + while (!handlerAStarted.load()) { + std::this_thread::yield(); + } + eventEmitter.offAndWait(tokenA); + }); + + auto futureA = std::async( + std::launch::async, [&eventEmitter]() { eventEmitter.emit(); }); + auto futureB = std::async( + std::launch::async, [&eventEmitter]() { eventEmitter.emit(); }); + + // Both handlers are guaranteed concurrently in flight (each spins for + // the other's start flag), so pre-fix both off() calls waited forever. + ASSERT_EQ( + futureA.wait_for(std::chrono::seconds(deadlockWatchdogSeconds)), + std::future_status::ready) + << "deadlock: off() called from inside a handler blocked on a" + " foreign in-flight invocation"; + ASSERT_EQ( + futureB.wait_for(std::chrono::seconds(deadlockWatchdogSeconds)), + std::future_status::ready) + << "deadlock: off() called from inside a handler blocked on a" + " foreign in-flight invocation"; + + ASSERT_EQ(eventEmitter.listenerCount(), 0); + ASSERT_EQ(eventEmitter.listenerCount(), 0); +}