perf: batch marker-def fan-out to one PUT /nios/batch per compute

When a marker definition is created (or re-applied on data_link_type
change), the fan-out used to call inherit_marker -> update() on every
link, issuing one PUT /nio per link end (5000+ round-trips on a 2500-link
project). On started nodes each round-trip also reconfigured uBridge.

Two-phase fan-out:
- inherit_marker/start_marker gain memory_only: writes the marker into
  link._markers and refreshes _link_data without any HTTP/emit/dump.
- _apply_def_to_all_links applies memory-only to every link, then
  _batch_update_link_nios groups the updated NIO specs by compute and
  sends a single PUT /projects/{id}/nios/batch per compute.

compute: new PUT /projects/{id}/nios/batch endpoint with _get_existing_nio
+ _update_nio_binding dispatch (mirrors create_batch_nios), re-applies
filters+markers to uBridge on started nodes.

Precise per-marker operations (update_marker bpf change, stop_marker on
def delete) are untouched — they deliberately avoid a full reapply to
preserve sibling marker pcaps.

gns3server/api/routes/compute/projects.py

@@ -167,6 +167,65 @@ async def _add_nio_binding(node, adapter_number, port_number, nio):
         )
 
 
+def _get_existing_nio(node, adapter_number, port_number):
+    """
+    Fetch the already-bound NIO for a port, preserving its UDP endpoints
+    (lport/rhost/rport) so a marker/filter update only changes markers/filters.
+    Dispatch keys off the manager class name, mirroring _add_nio_binding.
+    """
+
+    manager_name = type(node.manager).__name__
+    if manager_name in ("Docker", "Qemu", "VMware", "VirtualBox"):
+        return node.get_nio(adapter_number)
+    elif manager_name == "IOU":
+        return node.get_nio(adapter_number, port_number)
+    elif manager_name in ("VPCS", "Builtin"):
+        return node.get_nio(port_number)
+    elif manager_name == "Dynamips":
+        # Dynamips routers expose NIOs via the slot/adapter; switches/hubs
+        # via get_nio(port).
+        if hasattr(node, "get_nio"):
+            import inspect as _inspect
+            if len(_inspect.signature(node.get_nio).parameters) >= 2:
+                return node.get_nio(adapter_number, port_number)
+            return node.get_nio(port_number)
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Dynamips node '{node.name}' has no get_nio for batch update",
+        )
+    raise HTTPException(
+        status_code=status.HTTP_400_BAD_REQUEST,
+        detail=f"Batch NIO update not supported for node type '{manager_name}'",
+    )
+
+
+async def _update_nio_binding(node, adapter_number, port_number, nio):
+    """
+    Re-apply a NIO binding (filters + markers) to a started node's uBridge.
+    Dispatch keys off the manager class name, mirroring _add_nio_binding.
+    """
+
+    manager_name = type(node.manager).__name__
+    if manager_name in ("Docker", "Qemu", "VMware", "VirtualBox"):
+        await node.adapter_update_nio_binding(adapter_number, nio)
+    elif manager_name == "IOU":
+        await node.adapter_update_nio_binding(adapter_number, port_number, nio)
+    elif manager_name == "VPCS":
+        await node.port_update_nio_binding(port_number, nio)
+    elif manager_name == "Dynamips":
+        if hasattr(node, "slot_update_nio_binding"):
+            await node.slot_update_nio_binding(adapter_number, port_number, nio)
+        else:
+            await node.update_nio(port_number, nio)
+    elif manager_name == "Builtin":
+        await node.update_nio(port_number, nio)
+    else:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Batch NIO update not supported for node type '{manager_name}'",
+        )
+
+
 @router.post(
     "/projects/{project_id}/nios/batch",
     status_code=status.HTTP_201_CREATED,
@@ -202,6 +261,37 @@ async def create_batch_nios(
     return {"added": added}
 
 
+@router.put(
+    "/projects/{project_id}/nios/batch",
+    status_code=status.HTTP_200_OK,
+)
+async def update_batch_nios(
+        project_id: UUID,
+        batch: schemas.BatchNIOCreate,
+        project: Project = Depends(dep_project),
+) -> dict:
+    """
+    Update many NIO bindings (filters + markers) across nodes in a single
+    request, re-applying them to uBridge on started nodes.
+
+    Used by the controller when a project-level marker definition changes and
+    must fan out to every affected link — replacing one PUT /nio round-trip per
+    link end with one round-trip per compute. Each entry fetches the already-
+    bound NIO (preserving its UDP endpoints), overlays the new markers/filters,
+    and re-binds it.
+    """
+
+    updated = 0
+    for entry in batch.nios:
+        node = project.get_node(entry.node_id)
+        nio = _get_existing_nio(node, entry.adapter_number, entry.port_number)
+        nio.filters = entry.nio.filters or {}
+        nio.markers = entry.nio.markers or {}
+        await _update_nio_binding(node, entry.adapter_number, entry.port_number, nio)
+        updated += 1
+    return {"updated": updated}
+
+
 @router.get("/projects/{project_id}/files", response_model=List[schemas.ProjectFile])
 async def get_compute_project_files(project: Project = Depends(dep_project)) -> List[schemas.ProjectFile]:
     """

gns3server/controller/link.py

@@ -114,7 +114,7 @@ class Link:
         """
         return self._markers
 
-    async def inherit_marker(self, def_name, marker_def, dump=True):
+    async def inherit_marker(self, def_name, marker_def, dump=True, memory_only=False):
         """
         Apply a project-level marker definition to this link.
 
@@ -148,6 +148,7 @@ class Link:
             enabled=not marker_def.get("paused", False),
             inherited_from=def_name,
             dump=dump,
+            memory_only=memory_only,
         )
 
     def _persist_markers(self):

gns3server/controller/project.py

@@ -1210,17 +1210,57 @@ class Project:
         Fan out a single marker definition to every existing link in the project.
         Links that have no capable node (``_MARKER_CAPABLE_TYPES``) are silently
         skipped — the marker can only live on a uBridge bridge.
+
+        Two-phase to avoid one HTTP round-trip per link end: (1) write the
+        inherited marker into each link's memory (``memory_only`` refreshes
+        ``_link_data`` without pushing), then (2) batch-update every affected
+        NIO via a single ``PUT /projects/{id}/nios/batch`` per compute.
         """
 
         d = self._marker_definitions[def_name]
-        # dump=False: per-link topology writes are the dominant cost on large
-        # projects — the callers (create/update_marker_definition) dump once
-        # after the fan-out.
-        await self._marker_apply_concurrently(
-            list(self._links.values()),
-            lambda link: link.inherit_marker(def_name, d, dump=False),
-            lambda link, e: f"Marker definition '{def_name}' could not be applied to link {link.id}: {e}",
-        )
+        affected = []
+        for link in self._links.values():
+            try:
+                await link.inherit_marker(def_name, d, dump=False, memory_only=True)
+                affected.append(link)
+            except ControllerError as e:
+                log.warning("Marker definition '%s' could not be applied to link %s: %s", def_name, link.id, e)
+        await self._batch_update_link_nios(affected)
+
+    async def _batch_update_link_nios(self, links):
+        """
+        Push the current ``_link_data`` (markers/filters) of *links* to their
+        computes in one ``PUT /projects/{id}/nios/batch`` per compute — replacing
+        one PUT /nio round-trip per link end. Started nodes re-apply uBridge;
+        stopped nodes update in memory.
+        """
+
+        per_compute = {}
+        for link in links:
+            if len(link._link_data) < 2:
+                continue
+            for i, side in enumerate(link._nodes):
+                node = side["node"]
+                per_compute.setdefault(node.compute, []).append(
+                    {
+                        "node_id": node.id,
+                        "adapter_number": side["adapter_number"],
+                        "port_number": side["port_number"],
+                        "nio": link._link_data[i],
+                    }
+                )
+
+        async def _dispatch(compute, entries):
+            await compute.put(
+                f"/projects/{self._id}/nios/batch",
+                data={"nios": entries},
+                timeout=300,
+            )
+
+        if per_compute:
+            await asyncio.gather(
+                *[_dispatch(c, n) for c, n in per_compute.items()]
+            )
 
     async def apply_defs_to_new_link(self, link):
         """

gns3server/controller/udp_link.py

@@ -407,7 +407,7 @@ class UDPLink(Link):
         # explicitly deletes a marker via the REST API, and a marker is torn
         # down automatically only when its link is deleted.
 
-    async def start_marker(self, name, bpf, tag=None, direction=None, data_link_type="DLT_EN10MB", capture_node_id=None, color=None, highlight_duration=None, enabled=True, inherited_from=None, dump=True):
+    async def start_marker(self, name, bpf, tag=None, direction=None, data_link_type="DLT_EN10MB", capture_node_id=None, color=None, highlight_duration=None, enabled=True, inherited_from=None, dump=True, memory_only=False):
         """
         Attach a traffic-insight marker to this link.
 
@@ -463,6 +463,12 @@ class UDPLink(Link):
         if inherited_from:
             marker_entry["inherited_from"] = inherited_from
         self._markers[name] = marker_entry
+        if memory_only:
+            # Project-open prepare / marker-def fan-out: only refresh the
+            # in-memory NIO specs so a later batch dispatch carries the new
+            # markers — no per-link update HTTP, notification or dump.
+            self._refresh_link_data()
+            return
         if self._created:
             await self.update()
         self._project.emit_notification("link.updated", self.asdict())
@@ -471,6 +477,26 @@ class UDPLink(Link):
         if dump:
             self._project.dump()
 
+    def _refresh_link_data(self):
+        """
+        Recompute the filters / markers / suspend fields of ``_link_data`` from
+        the current link state without pushing to computes. Used by the
+        memory-only marker path so a batch dispatch picks up the new markers.
+        """
+
+        if len(self._link_data) < 2:
+            return
+        node1 = self._nodes[0]["node"]
+        node2 = self._nodes[1]["node"]
+        node1_filters, node2_filters = self._get_node_filters(node1, node2)
+        node1_markers, node2_markers = self._get_node_markers(node1, node2)
+        self._link_data[0]["filters"] = node1_filters
+        self._link_data[0]["markers"] = node1_markers
+        self._link_data[0]["suspend"] = self._suspended
+        self._link_data[1]["filters"] = node2_filters
+        self._link_data[1]["markers"] = node2_markers
+        self._link_data[1]["suspend"] = self._suspended
+
     async def stop_marker(self, name, inherited=False, dump=True):
         """
         Remove a traffic-insight marker from this link.