From 078cf92aef5b960e1067407ea8c0d0726e22194e Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 7 Aug 2026 23:50:41 +0800 Subject: [PATCH] marker: concurrent (bounded) definition fan-out The per-definition fan-out applied markers to links in a serial loop -- one compute round-trip per link. On a 1000-link project that serializes N HTTP round-trips (minutes on remote computes). Fan out with asyncio.gather + Semaphore(32): links are independent (own _markers/_link_data), per-link ControllerError stays isolated, and Project.dump is synchronous + atomic (tmp + rename) so concurrent dumps cannot corrupt the topology file. Converts the definition-create fan-out, the definition-update sync and re-fan-out loops, and the definition-delete cleanup to the shared _marker_apply_concurrently helper. apply_defs_to_new_link stays serial deliberately: all definitions share one link and each push carries the link's full marker set, so concurrent pushes would race and lose markers. --- gns3server/controller/project.py | 106 ++++++++++++++++++++----------- tests/controller/test_marker.py | 15 +++++ 2 files changed, 83 insertions(+), 38 deletions(-) diff --git a/gns3server/controller/project.py b/gns3server/controller/project.py index 785ca2eae..177a2cc9a 100644 --- a/gns3server/controller/project.py +++ b/gns3server/controller/project.py @@ -1045,30 +1045,33 @@ class Project: if data_link_type is not _UNSET: d["data_link_type"] = data_link_type + # Links that currently carry an inherited copy of this definition. + affected = [ + link for link in self._links.values() + if f"global-{name}" in link.markers + and link.markers[f"global-{name}"].get("inherited_from") == name + ] + if data_link_type is not _UNSET: # data_link_type decides which links host an inherited copy (serial # links are skipped unless a WAN encapsulation is chosen), so a change # needs a full re-fan-out: drop every copy, then re-apply. - for link in list(self._links.values()): - marker_name = f"global-{name}" - if marker_name in link.markers and link.markers[marker_name].get("inherited_from") == name: - try: - await link.stop_marker(marker_name, inherited=True) - except ControllerError: - log.warning( - "Failed to remove inherited marker %s from link %s", - marker_name, link.id - ) + await self._marker_apply_concurrently( + affected, + lambda link: link.stop_marker(f"global-{name}", inherited=True), + lambda link, e: f"Failed to remove inherited marker global-{name} from link {link.id}: {e}", + ) await self._apply_def_to_all_links(name) else: # Sync: update every inherited copy across all links. - for link in list(self._links.values()): - marker_name = f"global-{name}" - if marker_name in link.markers and link.markers[marker_name].get("inherited_from") == name: - await link.update_marker( - marker_name, bpf=d["bpf"], tag=d.get("tag"), direction=d.get("direction"), color=d.get("color"), - highlight_duration=d.get("highlight_duration"), inherited=True - ) + await self._marker_apply_concurrently( + affected, + lambda link: link.update_marker( + f"global-{name}", bpf=d["bpf"], tag=d.get("tag"), direction=d.get("direction"), + color=d.get("color"), highlight_duration=d.get("highlight_duration"), inherited=True + ), + lambda link, e: f"Failed to sync marker global-{name} on link {link.id}: {e}", + ) self.dump() self.emit_notification("project.updated", self.asdict()) @@ -1084,17 +1087,17 @@ class Project: del self._marker_definitions[name] - for link in list(self._links.values()): - marker_name = f"global-{name}" - if marker_name in link.markers and link.markers[marker_name].get("inherited_from") == name: - try: - await link.stop_marker(marker_name, inherited=True) - except ControllerError: - # A missing compute or broken link shouldn't block the delete. - log.warning( - "Failed to remove inherited marker %s from link %s", - marker_name, link.id - ) + affected = [ + link for link in self._links.values() + if f"global-{name}" in link.markers + and link.markers[f"global-{name}"].get("inherited_from") == name + ] + await self._marker_apply_concurrently( + affected, + lambda link: link.stop_marker(f"global-{name}", inherited=True), + # A missing compute or broken link shouldn't block the delete. + lambda link, e: f"Failed to remove inherited marker global-{name} from link {link.id}: {e}", + ) self.dump() self.emit_notification("project.updated", self.asdict()) @@ -1107,21 +1110,21 @@ class Project: """ d = self._marker_definitions[def_name] - for link in list(self._links.values()): - try: - await link.inherit_marker(def_name, d) - except ControllerError as e: - # Per-link failures (e.g. no capable node) shouldn't block the - # definition from serving the rest. - log.warning( - "Marker definition '%s' could not be applied to link %s: %s", - def_name, link.id, e - ) + await self._marker_apply_concurrently( + list(self._links.values()), + lambda link: link.inherit_marker(def_name, d), + lambda link, e: f"Marker definition '{def_name}' could not be applied to link {link.id}: {e}", + ) async def apply_defs_to_new_link(self, link): """ Apply every active marker definition to a newly created link so it inherits project-level rules automatically. + + Deliberately serial: all definitions share the same link, and each + ``inherit_marker`` pushes the link's full marker set — concurrent + pushes would race (a later push overwriting an earlier one's spec and + losing markers). """ for def_name, d in self._marker_definitions.items(): @@ -1133,6 +1136,33 @@ class Project: def_name, link.id, e ) + async def _marker_apply_concurrently(self, links, operation, fail_msg): + """ + Run an async per-link marker operation across *links* with bounded + concurrency. A serial loop takes N sequential compute round-trips — a + definition over 1000 links would take minutes on remote computes — so + fan out in parallel batches. Links are independent (own ``_markers`` / + ``_link_data``), so this is race-free; per-link ``ControllerError`` is + logged and skipped, preserving the serial loop's isolation semantics. + ``Project.dump`` is synchronous and writes atomically (tmp + rename), + so concurrent dumps from the fan-out cannot corrupt the topology file. + + :param links: iterable of links to operate on + :param operation: async callable ``(link) -> coroutine`` + :param fail_msg: callable ``(link, error) -> log message`` + """ + + sem = asyncio.Semaphore(32) + + async def guarded(link): + async with sem: + try: + await operation(link) + except ControllerError as e: + log.warning(fail_msg(link, e)) + + await asyncio.gather(*(guarded(link) for link in links)) + @property def snapshots(self): """ diff --git a/tests/controller/test_marker.py b/tests/controller/test_marker.py index 1c547f1ad..c5297aeff 100644 --- a/tests/controller/test_marker.py +++ b/tests/controller/test_marker.py @@ -433,6 +433,21 @@ async def test_create_marker_definition_fans_out(project): assert project.marker_definitions["arp"]["highlight_duration"] == 1200 +@pytest.mark.asyncio +async def test_definition_fans_out_over_many_links(project): + # The fan-out is concurrent (bounded) — a large topology must not serialize + # N compute round-trips — but behaviorally every link still receives the + # marker and per-link failures stay isolated. + with _valid_bpf(): + links = [await _make_link(project) for _ in range(20)] + await project.create_marker_definition("arp", "arp", highlight_duration=700) + + for link in links: + assert link.markers["global-arp"]["highlight_duration"] == 700 + assert link.markers["global-arp"]["inherited_from"] == "arp" + assert project.marker_definitions["arp"]["highlight_duration"] == 700 + + @pytest.mark.asyncio async def test_update_marker_definition_syncs(project):