diff --git a/packages/streamr-dht/modules/connection/webrtc/WebrtcConnection.cppm b/packages/streamr-dht/modules/connection/webrtc/WebrtcConnection.cppm index 468132d2..f4e093f9 100644 --- a/packages/streamr-dht/modules/connection/webrtc/WebrtcConnection.cppm +++ b/packages/streamr-dht/modules/connection/webrtc/WebrtcConnection.cppm @@ -83,6 +83,13 @@ private: streamr::eventemitter::EventEmitter signallingEvents; std::recursive_mutex mMutex; + // Per-connection FIFO dispatch of the datachannel open/message emit + // chains: the rtc worker pool is fixed-size and also runs ICE/DTLS + // processing for every connection in the process, so the emit chains + // must not run inline there (the WebsocketConnection idiom; see its + // mDispatchExecutor). + streamr::utils::SharedSerialExecutor mDispatchExecutor{ + streamr::utils::SharedExecutors::worker()}; explicit WebrtcConnection(WebrtcConnectionParams&& params) : Connection(ConnectionType::WEBRTC), params(std::move(params)) {} @@ -368,7 +375,14 @@ private: defaultBufferThresholdLow)); this->dataChannel->onOpen([self]() { SLogger::trace("dc.onOpened"); - self->onDataChannelOpen(); + // The Connected emit chain (handshaker, connection manager) + // must not run on the rtc worker thread — it starves the + // fixed-size pool that also processes ICE/DTLS handshakes + // (the same starvation the websocket poll thread had). The + // per-connection serial executor keeps Connected ordered + // before the Data emits below. + self->mDispatchExecutor.add( + [self]() { self->onDataChannelOpen(); }); }); this->dataChannel->onClosed([self]() { SLogger::trace("dc.closed"); @@ -382,7 +396,15 @@ private: this->dataChannel->onMessage([self](rtc::message_variant message) { SLogger::trace("dc.onMessage"); if (std::holds_alternative(message)) { - self->emit(std::get(std::move(message))); + // Thin enqueue (see onOpen above): the rtc worker thread + // only copies the frame; the emit chain runs on the + // per-connection serial executor in arrival order. + self->mDispatchExecutor.add( + [self, + data = + std::get(std::move(message))]() mutable { + self->emit(std::move(data)); + }); } }); } diff --git a/packages/streamr-dht/modules/connection/webrtc/WebrtcConnector.cppm b/packages/streamr-dht/modules/connection/webrtc/WebrtcConnector.cppm index 344200ac..d0c54983 100644 --- a/packages/streamr-dht/modules/connection/webrtc/WebrtcConnector.cppm +++ b/packages/streamr-dht/modules/connection/webrtc/WebrtcConnector.cppm @@ -261,26 +261,78 @@ public: }); // TS delFunc: the attempt is dropped once the connection settles - // either way (erasing an absent key is a no-op). + // either way. Every handler below erases the map entry ONLY if it + // still belongs to the connection/pending pair that fired — the map + // is keyed by nodeId and repeated connects to the same peer insert + // new pairs, so an unconditional erase(nodeId) from a stale handler + // (an older connection's Disconnected arriving late) deleted a + // freshly inserted attempt and orphaned its WebrtcConnection + // (observed: insert and stale erase within the same millisecond). + // An orphan is immortal — its rtc callbacks capture a strong self, + // broken only by doClose — and its dispatch executor's worker-pool + // KeepAlive then blocks process exit in the shared pool destructor + // (joinKeepAliveOnce). The captured raw pointers are compared for + // identity only, never dereferenced. connection->on( - [this, nodeId]( + [this, nodeId, connPtr = connection.get()]( bool /*gracefulLeave*/, uint64_t /*code*/, const std::string& /*reason*/) { std::scoped_lock lock(this->mMutex); - this->ongoingConnectAttempts.erase(nodeId); + const auto it = this->ongoingConnectAttempts.find(nodeId); + if (it == this->ongoingConnectAttempts.end()) { + return; + } + if (it->second.connection.get() != connPtr) { + return; + } + this->ongoingConnectAttempts.erase(it); }); pendingConnection->on( - [this, nodeId](bool /*gracefulLeave*/) { - std::scoped_lock lock(this->mMutex); - this->ongoingConnectAttempts.erase(nodeId); + [this, nodeId, pendingPtr = pendingConnection.get()]( + bool /*gracefulLeave*/) { + // The attempt's WebrtcConnection must be closed HERE, not + // just dropped from the map: for an answerer that never + // received a handshake request, no handshaker links the + // pending connection's Disconnected to connection->close() + // (IncomingHandshaker wires that link only inside + // onHandshakeRequest), so a bare erase would remove the last + // owner stop() could still destroy and leave the connection + // orphaned (see the immortality note above — this was the + // 1-in-10 post-PASSED exit hang of the 22-node webrtc test). + // close() is idempotent, so the offerer path (whose + // OutgoingHandshaker closes the connection from this same + // event first) is unaffected. Identity-checked like the + // handler above. + std::shared_ptr connectionToClose; + { + std::scoped_lock lock(this->mMutex); + const auto it = this->ongoingConnectAttempts.find(nodeId); + if (it == this->ongoingConnectAttempts.end() || + it->second.managedConnection.get() != pendingPtr) { + return; + } + connectionToClose = it->second.connection; + this->ongoingConnectAttempts.erase(it); + } + // Outside mMutex (locking policy: no call-outs under the + // connector mutex — the close emits Disconnected and defers + // the rtc teardown to the worker pool). + if (connectionToClose) { + connectionToClose->close(false); + } }); pendingConnection->on( - [this, nodeId]( + [this, nodeId, pendingPtr = pendingConnection.get()]( const PeerDescriptor& /*peerDescriptor*/, const std::shared_ptr& /*connection*/) { std::scoped_lock lock(this->mMutex); - this->ongoingConnectAttempts.erase(nodeId); + const auto it = this->ongoingConnectAttempts.find(nodeId); + if (it == this->ongoingConnectAttempts.end() || + it->second.managedConnection.get() != pendingPtr) { + return; + } + this->ongoingConnectAttempts.erase(it); }); connection->webrtcEvents().on( diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketConnection.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketConnection.cppm index 7fe34830..ce630dd0 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketConnection.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketConnection.cppm @@ -7,12 +7,15 @@ module; #include #include +#include + #include export module streamr.dht.WebsocketConnection; import streamr.logger.SLogger; import streamr.utils.EnableSharedFromThis; +import streamr.utils.SharedExecutors; import streamr.dht.Connection; // Hoisted from the former header (file scope, NOT exported); @@ -43,6 +46,33 @@ protected: std::atomic mConnectedEmitted{false}; // NOLINT std::recursive_mutex mMutex; // NOLINT +private: + // All libdatachannel callbacks are dispatched through this per-connection + // serial view of the shared worker pool instead of running inline. Every + // websocket in the process shares ONE libdatachannel poll thread; running + // the emit chains (handshake, RPC parse/dispatch) inline there starved + // concurrent websocket upgrades so badly that a 12-node start storm + // pushed connect latency past the 1 s connectivity-check timeout + // (measured: the poll thread was >95% busy in Data emit chains while + // accepts sat waiting). The serial executor preserves the per-connection + // event order the emitter contract relies on: Connected before any Data, + // Data in arrival order, Disconnected last (the same per-association + // ordering idiom as the Simulator's delivery executors). + streamr::utils::SharedSerialExecutor mDispatchExecutor{ + streamr::utils::SharedExecutors::worker()}; + + // Enqueue a callback body; the task keeps the connection alive until it + // has run. Callers are live rtc callbacks (rtc's callback mutex + // guarantees the object cannot be mid-destruction there — the + // destructor's resetCallbacks() blocks on in-flight callbacks) or + // methods invoked through an owning shared_ptr (setSocket's + // already-open path), so sharedFromThis is safe. + void dispatch(std::function task) { + auto self = this->sharedFromThis(); + mDispatchExecutor.add([self, task = std::move(task)]() { task(); }); + } + +protected: // Only allow subclasses to be created explicit WebsocketConnection(ConnectionType type) : Connection(type), @@ -157,23 +187,32 @@ protected: SLogger::trace("setSocket() after move " + getConnectionTypeString()); - // Set socket callbacks. The message callback is deliberately NOT - // attached here — see startReceiving(). + // Set socket callbacks. Each is a thin wrapper that only enqueues + // the real body onto the per-connection dispatch executor — the rtc + // callback threads (the shared poll thread, the per-server accept + // thread) must never run the emit chains (see mDispatchExecutor). + // The message callback is deliberately NOT attached here — see + // startReceiving(). - mSocket->onError(this->onError); - mSocket->onClosed(this->onClosed); - mSocket->onOpen(this->onOpen); + mSocket->onError([this](std::string error) { + this->dispatch( + [this, error = std::move(error)]() { this->onError(error); }); + }); + mSocket->onClosed( + [this]() { this->dispatch([this]() { this->onClosed(); }); }); + mSocket->onOpen( + [this]() { this->dispatch([this]() { this->onOpen(); }); }); // rtc does not retro-fire the open callback: if the websocket // handshake completed on the processor thread before the callback // above was attached, onOpen would never run and Connected would - // never be emitted. Emit it here in that case (mConnectedEmitted + // never be emitted. Enqueue it here in that case (mConnectedEmitted // keeps the two paths idempotent). if (mSocket->readyState() == rtc::WebSocket::State::Open) { SLogger::trace( "setSocket() socket already open, emitting Connected " + getConnectionTypeString()); - this->onOpen(); + this->dispatch([this]() { this->onOpen(); }); } SLogger::trace("setSocket() end " + getConnectionTypeString()); @@ -203,7 +242,17 @@ public: SLogger::trace("startReceiving() " + getConnectionTypeString()); std::scoped_lock lock(mMutex); if (!mDestroyed && mSocket) { - mSocket->onMessage(this->onMessage, this->onStringMessage); + // Thin wrapper (see setSocket): the poll thread only copies the + // frame into a task; the emit chain runs on the dispatch + // executor. The attach-time flush of queued frames enqueues + // them here in order, ahead of any later-arriving frame. + mSocket->onMessage( + [this](rtc::binary data) { + this->dispatch([this, data = std::move(data)]() mutable { + this->onMessage(std::move(data)); + }); + }, + this->onStringMessage); } } @@ -287,6 +336,19 @@ public: socket->close(); } } + + // Barrier: returns once every dispatch task enqueued so far has run. + // resetCallbacks() only stops NEW rtc callbacks; already-enqueued + // bodies still run afterwards, so an owner whose listeners capture it + // by raw pointer must drain before destroying itself (WebsocketServer + // does, for half-ready connections). MUST NOT be called from a task + // running on this connection's own dispatch executor — that would + // deadlock the serial queue. + void drainDispatchQueue() { + folly::Baton<> baton; + mDispatchExecutor.add([&baton]() { baton.post(); }); + baton.wait(); + } }; } // namespace streamr::dht::connection::websocket diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServer.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServer.cppm index 101de48a..c0e58840 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServer.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketServer.cppm @@ -112,19 +112,37 @@ public: void stop() { SLogger::info("stop() start"); - if (mStopped) { + if (mStopped.exchange(true)) { SLogger::info("stop() already stopped, returning"); return; } - mStopped = true; - removeAllListeners(); - SLogger::info( - "stop() removeAllListeners() end, calling mServer->stop()"); + // Quiesce the accept thread first: mServer->stop() joins it, and + // handleIncomingClient is gated on mStopped, so no new half-ready + // connection can be inserted after this. if (mServer) { mServer->stop(); mServer = nullptr; } - SLogger::info("stop() mServer->stop() end"); + // Tear down the half-ready connections. Their listeners capture + // this server by raw pointer and run on the connections' dispatch + // executors, so each queue must be drained before the caller may + // destroy this object (removeAllListeners() first makes queued + // not-yet-run emits no-ops). + std::unordered_map< + std::string, + std::shared_ptr> + halfReady; + { + std::scoped_lock lock(mHalfReadyConnectionsMutex); + halfReady.swap(mHalfReadyConnections); + } + for (auto& [id, connection] : halfReady) { + connection->removeAllListeners(); + connection->destroy(); + connection->drainDispatchQueue(); + } + removeAllListeners(); + SLogger::info("stop() end"); } void updateCertificate(const std::string& cert, const std::string& key) { @@ -142,6 +160,10 @@ private: void handleIncomingClient(std::shared_ptr ws) { std::scoped_lock lock(mHalfReadyConnectionsMutex); + if (mStopped) { + // Racing stop(): dropping ws here closes the socket. + return; + } auto websocketServerConnection = WebsocketServerConnection::newInstance(); auto id = Uuid::v4(); @@ -151,7 +173,12 @@ private: SLogger::info( "Half-ready WebSocketServerConnection emitted Connected, emitting it as Connected"); std::scoped_lock lock(mHalfReadyConnectionsMutex); - auto readyConnection = mHalfReadyConnections.at(id); + const auto entry = mHalfReadyConnections.find(id); + if (entry == mHalfReadyConnections.end()) { + // stop() swapped the map out while this task was queued. + return; + } + auto readyConnection = entry->second; readyConnection->removeAllListeners(); diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnection.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnection.cppm index ad26455f..44c341d8 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnection.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnection.cppm @@ -37,6 +37,15 @@ private: "WebsocketServerConnection() The half-ready socket is fully connected"); mResourceURL = Url{mSocket->path().value()}; mRemoteAddress = mSocket->remoteAddress().value(); + // libdatachannel formats this as ":"; the + // consumers (TS-parity remoteAddress, the connectivity + // probe URL, ipv4ToNumber) need the bare host. The service + // is always the suffix after the LAST colon, so this also + // handles unbracketed IPv6 hosts. + const auto portSep = mRemoteAddress.rfind(':'); + if (portSep != std::string::npos) { + mRemoteAddress = mRemoteAddress.substr(0, portSep); + } this->off( this->mHalfReadyConnectedHandlerToken); }); diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm index 2fc3acf1..9f15e1df 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm @@ -175,6 +175,18 @@ public: const std::shared_ptr< WebsocketServerConnection>& serverSocket) { + if (this->abortController.getSignal().aborted) { + // destroy() has run (or is running): drop the late + // connection instead of touching the cleared maps. + try { + serverSocket->close(false); + } catch (...) { + SLogger::debug( + "Error closing incoming Websocket connection" + " after abort"); + } + return; + } const auto resourceUrl = serverSocket->getResourceURL(); const auto action = getActionFromUrl(resourceUrl); SLogger::trace( diff --git a/packages/streamr-logger/modules/Logger.cppm b/packages/streamr-logger/modules/Logger.cppm index ee6ae622..f673d5a1 100644 --- a/packages/streamr-logger/modules/Logger.cppm +++ b/packages/streamr-logger/modules/Logger.cppm @@ -4,6 +4,7 @@ // this file is now the source of truth. module; +#include #include #include #include @@ -11,6 +12,10 @@ module; #include #include +// The C symbol for the process environment: must be declared in the +// global module fragment (see FollyLoggerImpl). +extern "C" char** environ; // NOLINT + export module streamr.logger.Logger; import streamr.json.toJson; @@ -31,6 +36,15 @@ private: std::shared_ptr mLoggerImpl; StreamrLogLevel mLoggerLogLevel; nlohmann::json mContextBindings; + // Cheapest enabled level across this logger AND every + // LOG_LEVEL_ env override (those can only raise verbosity + // for a file category). log() early-outs below this bound BEFORE + // marshalling the metadata to JSON: the marshalling (plus the + // per-call environment scan in FollyLoggerImpl) ran on every + // filtered-out trace/debug call and dominated the per-message cost + // of the connection delivery chains (measured 20-90 ms per network + // message in Debug builds under the full-node tests). + int mMinEnabledLogLevelValue{0}; public: /** @@ -67,6 +81,25 @@ public: mContextBindings = ensureJsonObject(toJson(contextBindings), "contextBindings"); + // Snapshot the minimum enabled level once (see the member's + // comment). Env vars cannot meaningfully change mid-process, and + // FollyLoggerImpl re-reads them per message anyway for the + // messages that pass this bound. + int minEnabled = getStreamrLogLevelValue(mLoggerLogLevel); + for (char** env = environ; *env != nullptr; ++env) { // NOLINT + const std::string_view envVar{*env}; + if (envVar.starts_with(detail::envCategoryLogLevelName)) { + const auto eq = envVar.find('='); + if (eq != std::string_view::npos) { + const auto categoryLevel = getStreamrLogLevelByName( + envVar.substr(eq + 1), mLoggerLogLevel); + minEnabled = std::min( + minEnabled, getStreamrLogLevelValue(categoryLevel)); + } + } + } + mMinEnabledLogLevelValue = minEnabled; + if (!loggerImpl) { mLoggerImpl = std::make_shared(); } else { @@ -209,6 +242,13 @@ private: std::string_view msg, MetadataType metadata, const std::source_location& location) { + // Early-out below the cheapest enabled level (see + // mMinEnabledLogLevelValue): no category could emit this message, + // so skip the metadata marshalling and the LoggerImpl call. + if (getStreamrLogLevelValue(messageLogLevel) < + mMinEnabledLogLevelValue) { + return; + } // Merge the possible metadata with the context bindings auto metadataJson = ensureJsonObject(toJson(metadata), "metadata"); diff --git a/packages/streamr-trackerless-network/test/unit/WebrtcFullNodeNetworkTest.cpp b/packages/streamr-trackerless-network/test/unit/WebrtcFullNodeNetworkTest.cpp index 17d84e56..55390e13 100644 --- a/packages/streamr-trackerless-network/test/unit/WebrtcFullNodeNetworkTest.cpp +++ b/packages/streamr-trackerless-network/test/unit/WebrtcFullNodeNetworkTest.cpp @@ -43,12 +43,8 @@ using streamr::utils::waitForCondition; namespace { -// TS runs this with NUM_OF_NODES = 22; the C++ entry point's -// websocket accept path degrades under a concurrent connectivity-check -// storm (measured: ~9 accepts in the first second, then ~1/s), so the -// late checks exceed the 1 s connect timeout beyond ~8 nodes. Scaled -// down until the accept path is offloaded — see the follow-up task. -constexpr size_t numOfNodes = 8; +// TS parity: 22 nodes start concurrently against one entry point. +constexpr size_t numOfNodes = 22; constexpr uint16_t entryPointPort = 14444; // Generous for the 3-core CI runners: the TS test tolerates up to its // 220 s jest budget for convergence. diff --git a/packages/streamr-trackerless-network/test/unit/WebsocketFullNodeNetworkTest.cpp b/packages/streamr-trackerless-network/test/unit/WebsocketFullNodeNetworkTest.cpp index eefe23ea..b1fa6201 100644 --- a/packages/streamr-trackerless-network/test/unit/WebsocketFullNodeNetworkTest.cpp +++ b/packages/streamr-trackerless-network/test/unit/WebsocketFullNodeNetworkTest.cpp @@ -44,12 +44,8 @@ using streamr::utils::waitForCondition; namespace { -// TS runs this with NUM_OF_NODES = 12; the C++ entry point's -// websocket accept path degrades under a concurrent connectivity-check -// storm (measured: ~9 accepts in the first second, then ~1/s), so the -// late checks exceed the 1 s connect timeout beyond ~8 nodes. Scaled -// down until the accept path is offloaded — see the follow-up task. -constexpr size_t numOfNodes = 8; +// TS parity: 12 nodes start concurrently against one entry point. +constexpr size_t numOfNodes = 12; constexpr uint16_t entryPointPort = 15555; // Generous for the 3-core CI runners: the TS test tolerates up to its // 220 s jest budget for convergence.