From a08106c307b62ae2bef22c03ab48de20c60fceb9 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sat, 4 Jul 2026 09:29:47 +0300 Subject: [PATCH] Make every package's header graph a verified DAG MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consolidation precondition 3 (MODERNIZATION.md Phase 2.6 memo) asked for the dht include cycles to become a partition DAG. A monorepo-wide analysis — include edges plus forward-declaration edges, which are what break cycles textually but still force cyclic imports once headers map onto module partitions — found exactly one cycle across all seven packages: the 6-header dht endpoint state-machine cluster (EndpointStateInterface held Endpoint& with its member definitions at the bottom of Endpoint.hpp). - EndpointStateInterface is now a pure abstract interface; Endpoint implements it (privately) and passes itself to the state objects. The state classes are unchanged. No behavioral change: the interface methods forwarded 1:1 to the Endpoint methods that are now their overrides (the extra recursive-mutex lock the old handleDisconnect forwarder took is subsumed by the lock the target method takes). - check-include-dag.py verifies the DAG property (both edge kinds) for every package; lint.sh runs it, so a new cycle fails CI with the offending headers listed. - MODERNIZATION.md: precondition 3 marked RESOLVED — consolidation is now gated only by clangd purview-lint coverage. Verified: streamr-dht suite 81/81, trackerless-network suites green (Release); checker reports all 8 packages acyclic. Co-Authored-By: Claude Fable 5 --- MODERNIZATION.md | 21 ++- check-include-dag.py | 132 ++++++++++++++++++ lint.sh | 8 ++ .../connection/endpoint/Endpoint.hpp | 88 +++--------- .../endpoint/EndpointStateInterface.hpp | 32 +++-- 5 files changed, 188 insertions(+), 93 deletions(-) create mode 100755 check-include-dag.py diff --git a/MODERNIZATION.md b/MODERNIZATION.md index 6cc9e18b..1fed16ad 100644 --- a/MODERNIZATION.md +++ b/MODERNIZATION.md @@ -743,9 +743,17 @@ verified NDK clang (18 = r27), with the textual fallback below it. mis-unifies preamble/BMI types; with code in partitions there is no header fallback for the linter — coverage would regress from "full" to "none" for consolidated code). -3. The dht intra-package include cycles (connection/endpoint cluster) - must become a partition DAG — untangle or use coarse per-cluster - partitions (deferred from 2.4 by design). +3. ~~The dht intra-package include cycles (connection/endpoint cluster) + must become a partition DAG~~ **RESOLVED (post-2.6)** — a monorepo-wide + analysis (include edges + forward-declaration edges, which are what + break cycles textually but still force cyclic imports between + partitions) found exactly ONE cycle in all seven packages: the + 6-header dht endpoint state-machine cluster. Untangled by making + `EndpointStateInterface` a pure abstract interface that `Endpoint` + implements (previously a concrete forwarder holding `Endpoint&` with + its member definitions at the bottom of Endpoint.hpp). Every + package's header graph is now a verified DAG, enforced continuously + by `check-include-dag.py` in `lint.sh`. ### What consolidation buys (quantified at the 2.4/2.5 checkpoints) - The −40% incremental-rebuild targets (currently: SLogger touch −6…−12%, @@ -762,9 +770,10 @@ per-package with a grep-enforced "nothing includes them" gate. The façade stage is COMPLETE and delivers: uniform `import streamr.` consumption, −24% clean builds, 250+ tests through import, and module infrastructure exercised on macOS/Linux/iOS/Android. Headers remain the -linted source of truth. With the Android blocker resolved, consolidation -is gated only by preconditions 2 (clangd lint coverage of purview code) -and 3 (dht partition DAG). This is a stable resting point. +linted source of truth. With the Android blocker resolved (2.6) and the +header graph a verified DAG (post-2.6), consolidation is gated only by +precondition 2: clangd lint coverage of purview code. This is a stable +resting point. ## Success metrics (measured macOS + Linux, end of Phase 2.5) - ≥25% clean-build wall-clock reduction (dev build with tests). diff --git a/check-include-dag.py b/check-include-dag.py new file mode 100755 index 00000000..a7954142 --- /dev/null +++ b/check-include-dag.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +"""Verify that each package's internal header graph is a DAG. + +The planned C++ modules consolidation (MODERNIZATION.md Phase 2.6) maps +headers onto module partitions, and module imports must be acyclic — so +the header dependency graph has to be a DAG. Two edge kinds are checked: + + * include edges: #include "streamr-/..." between headers of the + same package + * semantic edges: a forward declaration (`class X;` / `struct X;`) of + a type DEFINED in another header of the package. + Forward declarations break cycles textually, but a + forward-declared entity owned by another partition + still forces a cyclic import after consolidation. + +Run from the repository root: ./check-include-dag.py +Exits non-zero listing every cycle found. +""" + +import os +import re +import sys +from collections import defaultdict + +INC_RE = re.compile(r'#include\s+"((?:streamr|packages)[^"]*)"') +FWD_RE = re.compile(r"^\s*(?:class|struct)\s+([A-Za-z_]\w*)\s*;") +DEF_RE = re.compile(r"(?:class|struct)\s+([A-Za-z_]\w*)[^;{]*\{") + + +def check_package(include_root: str) -> list[list[str]]: + nodes: set[str] = set() + texts: dict[str, str] = {} + for dirpath, _, files in os.walk(include_root): + for f in files: + if not f.endswith(".hpp"): + continue + rel = os.path.relpath(os.path.join(dirpath, f), include_root) + nodes.add(rel) + with open( + os.path.join(dirpath, f), encoding="utf-8", errors="replace" + ) as fh: + texts[rel] = fh.read() + + owner: dict[str, str] = {} + for rel, txt in texts.items(): + for m in DEF_RE.finditer(txt): + owner.setdefault(m.group(1), rel) + + edges: dict[str, set[str]] = defaultdict(set) + for rel, txt in texts.items(): + for line in txt.splitlines(): + m = INC_RE.search(line) + if m and m.group(1) in nodes: + edges[rel].add(m.group(1)) + fm = FWD_RE.match(line) + if fm: + own = owner.get(fm.group(1)) + if own and own != rel: + edges[rel].add(own) + + # Tarjan SCC + sys.setrecursionlimit(10000) + counter = [0] + stack: list[str] = [] + index: dict[str, int] = {} + lowlink: dict[str, int] = {} + on_stack: dict[str, bool] = {} + sccs: list[list[str]] = [] + + def strongconnect(v: str) -> None: + index[v] = lowlink[v] = counter[0] + counter[0] += 1 + stack.append(v) + on_stack[v] = True + for w in edges.get(v, ()): + if w not in index: + strongconnect(w) + lowlink[v] = min(lowlink[v], lowlink[w]) + elif on_stack.get(w): + lowlink[v] = min(lowlink[v], index[w]) + if lowlink[v] == index[v]: + scc = [] + while True: + w = stack.pop() + on_stack[w] = False + scc.append(w) + if w == v: + break + sccs.append(scc) + + for v in sorted(nodes): + if v not in index: + strongconnect(v) + + return [s for s in sccs if len(s) > 1] + + +def main() -> int: + packages_dir = "packages" + if not os.path.isdir(packages_dir): + print("check-include-dag.py: run from the repository root", file=sys.stderr) + return 2 + + failed = False + for pkg in sorted(os.listdir(packages_dir)): + include_root = os.path.join(packages_dir, pkg, "include") + if not os.path.isdir(include_root): + continue + cycles = check_package(include_root) + if cycles: + failed = True + print(f"{pkg}: header dependency cycles found:") + for scc in sorted(cycles, key=len, reverse=True): + print(f" cycle of {len(scc)} headers:") + for h in sorted(scc): + print(f" {h}") + else: + print(f"{pkg}: OK") + + if failed: + print( + "\nHeader cycles block the module consolidation " + "(see MODERNIZATION.md Phase 2.6). Break the cycle with an " + "abstract interface or by merging/splitting headers.", + file=sys.stderr, + ) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/lint.sh b/lint.sh index 287ea25b..acce6a57 100755 --- a/lint.sh +++ b/lint.sh @@ -9,6 +9,14 @@ if ! ./sync-cmake-files.sh --check; then exit 1 fi +# Verify that every package's internal header graph is a DAG (including +# forward-declaration edges) — the module consolidation maps headers onto +# partitions and module imports must be acyclic. See check-include-dag.py. +if ! python3 ./check-include-dag.py; then + echo "::error title=header dependency cycle::A package's internal headers form a cycle; see the check-include-dag.py output above." + exit 1 +fi + # Parse MonorepoPackages.cmake and loop through them for package in $(cat MonorepoPackages.cmake | grep -v "set(MonorepoPackages" | grep -v ")"); do echo "" diff --git a/packages/streamr-dht/include/streamr-dht/connection/endpoint/Endpoint.hpp b/packages/streamr-dht/include/streamr-dht/connection/endpoint/Endpoint.hpp index 37d276d1..8599e3b9 100644 --- a/packages/streamr-dht/include/streamr-dht/connection/endpoint/Endpoint.hpp +++ b/packages/streamr-dht/include/streamr-dht/connection/endpoint/Endpoint.hpp @@ -39,14 +39,19 @@ using EndpointEvents = std::tuple< endpointevents::Connected, endpointevents::Disconnected>; +// Endpoint implements EndpointStateInterface (privately) so the state +// classes can drive the machine through the abstract interface — see +// EndpointStateInterface.hpp for why this replaces the former concrete +// forwarder member. class Endpoint : public EventEmitter, - public EnableSharedFromThis { + public EnableSharedFromThis, + private EndpointStateInterface { private: std::function removeSelfFromContainer; EndpointState* state; std::recursive_mutex mutex; - void emitData(const std::vector& data) { + void emitData(const std::vector& data) override { SLogger::debug("Endpoint::emitData start"); auto self = sharedFromThis(); std::scoped_lock lock(this->mutex); @@ -54,7 +59,7 @@ class Endpoint : public EventEmitter, SLogger::debug("Endpoint::emitData end"); } - void handleDisconnect(bool /*gracefulLeave*/) { + void handleDisconnect(bool /*gracefulLeave*/) override { SLogger::debug("Endpoint::handleDisconnect start"); auto self = sharedFromThis(); std::scoped_lock lock(this->mutex); @@ -65,7 +70,8 @@ class Endpoint : public EventEmitter, SLogger::debug("Endpoint::handleDisconnect end"); } - void changeToConnectedState(const std::shared_ptr& connection) { + void changeToConnectedState( + const std::shared_ptr& connection) override { SLogger::debug("Endpoint::changeToConnectedState start"); auto self = sharedFromThis(); std::scoped_lock lock(this->mutex); @@ -74,7 +80,7 @@ class Endpoint : public EventEmitter, SLogger::debug("Endpoint::changeToConnectedState end"); } - void handleConnected() { + void handleConnected() override { SLogger::debug("Endpoint::handleConnected start"); auto self = sharedFromThis(); std::scoped_lock lock(this->mutex); @@ -82,9 +88,6 @@ class Endpoint : public EventEmitter, SLogger::debug("Endpoint::handleConnected end"); } - friend class EndpointStateInterface; - - EndpointStateInterface stateInterface; std::shared_ptr initialState; std::shared_ptr connectedState; std::shared_ptr connectingState; @@ -94,14 +97,10 @@ class Endpoint : public EventEmitter, explicit Endpoint( PeerDescriptor peerDescriptor, std::function&& removeSelfFromContainer) - : stateInterface(*this), - connectedState( - ConnectedEndpointState::newInstance(this->stateInterface)), - connectingState( - ConnectingEndpointState::newInstance(this->stateInterface)), - disconnectedState( - DisconnectedEndpointState::newInstance(this->stateInterface)), - initialState(InitialEndpointState::newInstance(this->stateInterface)), + : connectedState(ConnectedEndpointState::newInstance(*this)), + connectingState(ConnectingEndpointState::newInstance(*this)), + disconnectedState(DisconnectedEndpointState::newInstance(*this)), + initialState(InitialEndpointState::newInstance(*this)), peerDescriptor(std::move(peerDescriptor)), removeSelfFromContainer(removeSelfFromContainer) { SLogger::debug("Endpoint constructor start"); @@ -128,7 +127,7 @@ class Endpoint : public EventEmitter, ~Endpoint() override { SLogger::debug("Endpoint destructor"); } void changeToConnectingState( - const std::shared_ptr& pendingConnection) { + const std::shared_ptr& pendingConnection) override { SLogger::debug("Endpoint::changeToConnectingState start"); auto self = sharedFromThis(); std::scoped_lock lock(this->mutex); @@ -190,61 +189,6 @@ class Endpoint : public EventEmitter, } }; -/* -inline void ConnectedEndpointState::changeToConnectingState( - const std::shared_ptr& pendingConnection) { - SLogger::debug("ConnectedEndpointState::changeToConnectingState start"); - std::scoped_lock lock(this->connectedEndpointStateMutex); - this->connection->close(true); - this->removeEventHandlers(); - this->stateInterface.changeToConnectingState(pendingConnection); - SLogger::debug("ConnectedEndpointState::changeToConnectingState end"); -} -*/ - -inline EndpointStateInterface::EndpointStateInterface(Endpoint& ep) - : endpoint(ep) { - SLogger::debug("EndpointStateInterface constructor"); -} - -inline void EndpointStateInterface::changeToConnectingState( - const std::shared_ptr& pendingConnection) { - SLogger::debug("EndpointStateInterface::changeToConnectingState start"); - this->endpoint.changeToConnectingState(pendingConnection); - SLogger::debug("EndpointStateInterface::changeToConnectingState end"); -} - -inline void EndpointStateInterface::changeToConnectedState( - const std::shared_ptr& connection) { - SLogger::debug("EndpointStateInterface::changeToConnectedState start"); - this->endpoint.changeToConnectedState(connection); - SLogger::debug("EndpointStateInterface::changeToConnectedState end"); -} - -inline void EndpointStateInterface::emitData( - const std::vector& data) { - SLogger::debug("EndpointStateInterface::emitData start"); - this->endpoint.emitData(data); - SLogger::debug("EndpointStateInterface::emitData end"); -} - -inline void EndpointStateInterface::handleDisconnect(bool gracefulLeave) { - SLogger::debug("EndpointStateInterface::handleDisconnect start"); - SLogger::debug( - "EndpointStateInterface::handleDisconnect acquiring scoped lock"); - std::scoped_lock lock(this->endpoint.mutex); - SLogger::debug("EndpointStateInterface::handleDisconnect lock acquired"); - this->endpoint.handleDisconnect(gracefulLeave); - SLogger::debug("EndpointStateInterface::handleDisconnect end"); -} - -inline void EndpointStateInterface::handleConnected() { - SLogger::debug("EndpointStateInterface::handleConnected start"); - - this->endpoint.handleConnected(); - SLogger::debug("EndpointStateInterface::handleConnected end"); -} - } // namespace streamr::dht::connection::endpoint #endif diff --git a/packages/streamr-dht/include/streamr-dht/connection/endpoint/EndpointStateInterface.hpp b/packages/streamr-dht/include/streamr-dht/connection/endpoint/EndpointStateInterface.hpp index 1b351c17..b6fa773e 100644 --- a/packages/streamr-dht/include/streamr-dht/connection/endpoint/EndpointStateInterface.hpp +++ b/packages/streamr-dht/include/streamr-dht/connection/endpoint/EndpointStateInterface.hpp @@ -6,30 +6,32 @@ #include #include "streamr-dht/connection/Connection.hpp" #include "streamr-dht/connection/IPendingConnection.hpp" -#include "streamr-dht/connection/endpoint/EndpointState.hpp" namespace streamr::dht::connection::endpoint { using streamr::dht::connection::Connection; using streamr::dht::connection::IPendingConnection; -class Endpoint; -class EndpointState; - +// Pure abstract callback interface through which the endpoint state +// classes drive the state machine. Endpoint implements it. Keeping this +// abstract — instead of the former concrete forwarder that held an +// Endpoint& and had its member definitions at the bottom of +// Endpoint.hpp — makes the endpoint header cluster acyclic: Endpoint +// depends on the states and the states depend only on this interface. +// (This was the only header cycle in the monorepo; the planned module +// consolidation needs the header graph to be a DAG.) class EndpointStateInterface { -private: - Endpoint& endpoint; - public: - explicit EndpointStateInterface(Endpoint& ep); - void changeToConnectingState( - const std::shared_ptr& pendingConnection); - void changeToConnectedState(const std::shared_ptr& connection); - void emitData(const std::vector& data); - void handleDisconnect(bool gracefulLeave); - void handleConnected(); + virtual ~EndpointStateInterface() = default; + virtual void changeToConnectingState( + const std::shared_ptr& pendingConnection) = 0; + virtual void changeToConnectedState( + const std::shared_ptr& connection) = 0; + virtual void emitData(const std::vector& data) = 0; + virtual void handleDisconnect(bool gracefulLeave) = 0; + virtual void handleConnected() = 0; }; } // namespace streamr::dht::connection::endpoint -#endif // STREAMR_DHT_CONNECTION_ENDPOINT_ENDPOINTSTATEINTERFACE_HPP \ No newline at end of file +#endif // STREAMR_DHT_CONNECTION_ENDPOINT_ENDPOINTSTATEINTERFACE_HPP