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); +}