From 38c49a655cb0b3d1b31a66ee594d25df25837484 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Mon, 10 Aug 2026 23:42:50 +0800 Subject: [PATCH] perf: batch NIO dispatch on project open (one HTTP per compute) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Project open used to create each link by issuing two NIO POSTs from the controller to the compute — ~5000 HTTP round-trips for a 2500-link topology, all funnelling through the single shared controller/compute event loop and capping throughput near 12 links/s. Replace it with a bulk path: - UDPLink split into _prepare() (local: ports, peer addrs, link_data) and _commit_nios() (dispatch). create() = prepare + commit (interactive). - Link.add_node gains batch=True: attach both nodes without triggering per-link NIO HTTP. - compute: new POST /projects/{id}/nios/batch endpoint with a unified _add_nio_binding dispatch across node types (docker/qemu/iou/vpcs/ builtin differ in signature). - project.open: prepare all links locally, group NIO entries by compute, send each compute a single /nios/batch, then finalise (wire node/port refs, mark created, notify, apply marker defs) in parallel. Cuts controller->compute HTTP from O(links) to O(computes). Test added for the batch endpoint. --- gns3server/api/routes/compute/projects.py | 55 +++++++ gns3server/controller/link.py | 7 +- gns3server/controller/project.py | 138 +++++++++++++++++- gns3server/controller/udp_link.py | 56 ++++--- gns3server/schemas/__init__.py | 2 +- gns3server/schemas/compute/nios.py | 23 ++- tests/api/routes/compute/test_docker_nodes.py | 28 ++++ 7 files changed, 280 insertions(+), 29 deletions(-) diff --git a/gns3server/api/routes/compute/projects.py b/gns3server/api/routes/compute/projects.py index 71245dfe7..5a4cb63fd 100644 --- a/gns3server/api/routes/compute/projects.py +++ b/gns3server/api/routes/compute/projects.py @@ -34,6 +34,7 @@ from uuid import UUID from gns3server.compute.project_manager import ProjectManager from gns3server.compute.project import Project +from gns3server.compute.base_manager import BaseManager from gns3server.utils.path import is_safe_path from gns3server import schemas @@ -131,6 +132,60 @@ async def delete_compute_project(project: Project = Depends(dep_project)) -> Non ProjectManager.instance().remove_project(project.id) +async def _add_nio_binding(node, adapter_number, port_number, nio): + """ + Unified NIO-binding dispatch across node types. Each node type exposes a + different method signature, so centralise the fan-out here for the batch + endpoint. Dispatch keys off the manager class name (only dynamips/iou/qemu + carry a ``_NODE_TYPE`` attribute, so it can't be used universally). + """ + + manager_name = type(node.manager).__name__ + # Adapter-based nodes: docker / qemu / vmware / virtualbox take + # (adapter_number, nio); iou additionally takes port_number. + if manager_name in ("Docker", "Qemu", "VMware", "VirtualBox"): + await node.adapter_add_nio_binding(adapter_number, nio) + elif manager_name == "IOU": + await node.adapter_add_nio_binding(adapter_number, port_number, nio) + elif manager_name == "VPCS": + await node.port_add_nio_binding(port_number, nio) + elif manager_name == "Builtin": + # ethernet_switch / ethernet_hub / cloud / nat: add_nio(nio, port_number) + await node.add_nio(nio, port_number) + else: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Batch NIO creation not supported for node type '{manager_name}'", + ) + + +@router.post( + "/projects/{project_id}/nios/batch", + status_code=status.HTTP_201_CREATED, +) +async def create_batch_nios( + project_id: UUID, + batch: schemas.BatchNIOCreate, + project: Project = Depends(dep_project), +) -> dict: + """ + Create many NIO bindings across nodes in a single request. + + Used by the controller during project open to avoid one HTTP round-trip per + NIO. Each entry resolves its node via the project, builds the NIO through + the node's manager, and binds it. Nodes that are not started perform the + binding in memory; started nodes additionally wire uBridge. + """ + + added = 0 + for entry in batch.nios: + node = project.get_node(entry.node_id) + nio = node.manager.create_nio(jsonable_encoder(entry.nio, exclude_unset=True)) + await _add_nio_binding(node, entry.adapter_number, entry.port_number, nio) + added += 1 + return {"added": added} + + @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]: """ diff --git a/gns3server/controller/link.py b/gns3server/controller/link.py index e716bfd03..0cc00a140 100644 --- a/gns3server/controller/link.py +++ b/gns3server/controller/link.py @@ -257,11 +257,14 @@ class Link: """ return self._created - async def add_node(self, node, adapter_number, port_number, label=None, dump=True): + async def add_node(self, node, adapter_number, port_number, label=None, dump=True, batch=False): """ Add a node to the link :param dump: Dump project on disk + :param batch: When True, do not create the link on the computes once + both nodes are attached — the caller drives creation via the + project-open bulk path. Used to avoid one HTTP round-trip per link. """ port = node.get_port(adapter_number, port_number) @@ -305,7 +308,7 @@ class Link: {"node": node, "adapter_number": adapter_number, "port_number": port_number, "port": port, "label": label} ) - if len(self._nodes) == 2: + if len(self._nodes) == 2 and not batch: await self.create() for n in self._nodes: n["node"].add_link(self) diff --git a/gns3server/controller/project.py b/gns3server/controller/project.py index a6bcc159c..a23c1095f 100644 --- a/gns3server/controller/project.py +++ b/gns3server/controller/project.py @@ -827,6 +827,81 @@ class Project: # a link should have 2 attached nodes, this can happen with corrupted projects await self.delete_link(link.id, force_delete=True) + async def _prepare_link_from_topology(self, link_data): + """ + Build a link locally from topology data WITHOUT dispatching NIOs to the + computes. Returns ``(link, entries)`` where ``entries`` is the list of + ``(node, adapter_number, port_number, nio_data)`` tuples produced by + ``UDPLink._prepare()``, or ``None`` if the link is invalid/incomplete. + + Used by the project-open bulk path so all NIOs can be sent in a single + batch HTTP call per compute instead of one round-trip per link. + """ + + link = await self.add_link(link_id=link_data["link_id"]) + if "filters" in link_data: + try: + await link.update_filters(link_data["filters"]) + except ControllerError as e: + log.warning("Dropping invalid filters on link %s: %s", link_data.get("link_id"), e) + for name, marker in (link_data.get("markers") or {}).items(): + bpf = marker.get("bpf") + if not bpf: + log.warning("Dropping marker %s on link %s: missing bpf", name, link_data.get("link_id")) + continue + result = validate_bpf_syntax(bpf) + if not result.get("valid"): + log.warning( + "Dropping marker %s on link %s: invalid BPF (%s)", + name, link_data.get("link_id"), result.get("error") + ) + continue + link._markers[name] = { + "bpf": bpf, + "tag": marker.get("tag"), + "enabled": marker.get("enabled", True), + "color": marker.get("color"), + "highlight_duration": marker.get("highlight_duration"), + "capture_node_id": marker.get("capture_node_id"), + "direction": marker.get("direction"), + } + if "link_style" in link_data: + await link.update_link_style(link_data["link_style"]) + if "show_filters_icon" in link_data: + await link.update_show_filters_icon(link_data["show_filters_icon"]) + for node_link in link_data.get("nodes", []): + node = self.get_node(node_link["node_id"]) + port = node.get_port(node_link["adapter_number"], node_link["port_number"]) + if port is None: + log.warning( + "Port {}/{} for {} not found".format( + node_link["adapter_number"], node_link["port_number"], node.name + ) + ) + continue + if port.link is not None: + log.warning( + "Port {}/{} is already connected to link ID {}".format( + node_link["adapter_number"], node_link["port_number"], port.link.id + ) + ) + continue + # batch=True: attach the node without triggering per-link NIO HTTP + await link.add_node( + node, + node_link["adapter_number"], + node_link["port_number"], + label=node_link.get("label"), + dump=False, + batch=True, + ) + if len(link.nodes) != 2: + # a link should have 2 attached nodes, this can happen with corrupted projects + await self.delete_link(link.id, force_delete=True) + return None + entries = await link._prepare() + return (link, entries) + @open_required async def add_link(self, link_id=None, dump=True): """ @@ -1648,13 +1723,62 @@ class Project: count = ports_per_compute.get(compute.id, 0) if count > 0: await self.preallocate_udp_ports_for_compute(compute, count) - # Create links in parallel for improved performance - pool = Pool(concurrency=100) - for link_data in topology.get("links", []): - if "link_id" not in link_data.keys(): - continue - pool.append(self._create_link_from_topology_data, link_data) - await pool.join() + # Create links via the bulk path: build every link locally (no NIO + # HTTP), then dispatch all NIOs to each compute in a single batch + # request. This replaces one HTTP round-trip per link (~5000 for a + # 2500-link topology) with one round-trip per compute. + link_data_list = [d for d in topology.get("links", []) if "link_id" in d.keys()] + sem = asyncio.Semaphore(100) + + async def _prepare_one(data): + async with sem: + try: + return await self._prepare_link_from_topology(data) + except Exception as e: + log.warning("Could not load link %s: %s", data.get("link_id"), e) + return None + + prepared = await asyncio.gather(*[_prepare_one(d) for d in link_data_list]) + valid = [p for p in prepared if p is not None] + + # Group the prepared NIO entries by destination compute and send + # each compute a single /nios/batch request. + per_compute = {} # compute -> list of {node_id, adapter_number, port_number, nio} + for link, entries in valid: + for node, adapter_number, port_number, nio_data in entries: + per_compute.setdefault(node.compute, []).append( + { + "node_id": node.id, + "adapter_number": adapter_number, + "port_number": port_number, + "nio": nio_data, + } + ) + + async def _dispatch_batch(compute, nio_entries): + await compute.post( + f"/projects/{self._id}/nios/batch", + data={"nios": nio_entries}, + timeout=300, + ) + + if per_compute: + await asyncio.gather( + *[_dispatch_batch(c, n) for c, n in per_compute.items()] + ) + + # Finalise every link: wire node/port back-references, mark created, + # notify clients, and apply project-level marker definitions. + for link, _entries in valid: + for n in link._nodes: + n["node"].add_link(link) + n["port"].link = link + link._created = True + self.emit_notification("link.created", link.asdict()) + if valid: + await asyncio.gather( + *[self.apply_defs_to_new_link(link) for link, _ in valid] + ) # Release any pre-allocated UDP ports that were not consumed by links for compute_id, ports in self._preallocated_udp_ports.items(): if ports: diff --git a/gns3server/controller/udp_link.py b/gns3server/controller/udp_link.py index 680fcba51..7e9517989 100644 --- a/gns3server/controller/udp_link.py +++ b/gns3server/controller/udp_link.py @@ -17,7 +17,6 @@ import asyncio -import time import logging from .controller_error import ControllerError, ControllerNotFoundError @@ -88,9 +87,15 @@ class UDPLink(Link): """ return self._markers_for_node(node1), self._markers_for_node(node2) - async def create(self): + async def _prepare(self): """ - Create the link on the nodes + Local-only link setup: resolve peer addresses, reserve UDP ports and + build the two NIO tunnel specs (``self._link_data``). No NIO is sent to + the computes — the caller decides how to dispatch them (one-by-one via + :meth:`create`, or batched via the project-open bulk path). + + :returns: list of two ``(node, adapter_number, port_number, nio_data)`` + tuples, ready to be POSTed to each node's compute. """ node1 = self._nodes[0]["node"] @@ -101,12 +106,10 @@ class UDPLink(Link): port_number2 = self._nodes[1]["port_number"] # Get an IP allowing communication between both host - _t0 = time.perf_counter() try: (node1_host, node2_host) = await node1.compute.get_ip_on_same_subnet(node2.compute) except ValueError as e: raise ControllerError(f"Cannot get an IP address on same subnet: {e}") - _t1 = time.perf_counter() # Reserve a UDP port on both sides in parallel. Pre-allocated ports # (used during batch project loading) are popped from memory; otherwise @@ -121,7 +124,6 @@ class UDPLink(Link): self._node1_port, self._node2_port = await asyncio.gather( _allocate_port(node1.compute), _allocate_port(node2.compute) ) - _t2 = time.perf_counter() node1_filters, node2_filters = self._get_node_filters(node1, node2) node1_markers, node2_markers = self._get_node_markers(node1, node2) @@ -151,17 +153,32 @@ class UDPLink(Link): } ) - # Create the NIO tunnel on both sides in parallel. The two ends are - # independent once the ports and peer addresses are known -- each node - # talks to its own compute/uBridge with no shared lock between them -- - # so the two POSTs overlap. If either fails, roll back whichever side - # succeeded before re-raising the first error. + return [ + (node1, adapter_number1, port_number1, self._link_data[0]), + (node2, adapter_number2, port_number2, self._link_data[1]), + ] + + async def _commit_nios(self, entries): + """ + Send the two NIO tunnel POSTs in parallel and roll back on failure. + + :param entries: the two ``(node, adapter_number, port_number, nio_data)`` + tuples returned by :meth:`_prepare`. + """ + + (node1, adapter_number1, port_number1, nio_data1), \ + (node2, adapter_number2, port_number2, nio_data2) = entries + + # The two ends are independent once the ports and peer addresses are + # known — each node talks to its own compute/uBridge with no shared + # lock between them — so the two POSTs overlap. If either fails, roll + # back whichever side succeeded before re-raising the first error. results = await asyncio.gather( node1.post( - f"/adapters/{adapter_number1}/ports/{port_number1}/nio", data=self._link_data[0], timeout=120 + f"/adapters/{adapter_number1}/ports/{port_number1}/nio", data=nio_data1, timeout=120 ), node2.post( - f"/adapters/{adapter_number2}/ports/{port_number2}/nio", data=self._link_data[1], timeout=120 + f"/adapters/{adapter_number2}/ports/{port_number2}/nio", data=nio_data2, timeout=120 ), return_exceptions=True, ) @@ -179,12 +196,15 @@ class UDPLink(Link): if cleanup: await asyncio.gather(*cleanup, return_exceptions=True) raise errors[0] - _t3 = time.perf_counter() - log.info( - "UDPLink.create timing get_ip=%.3fms ports=%.3fms nio=%.3fms total=%.3fms", - 1000 * (_t1 - _t0), 1000 * (_t2 - _t1), 1000 * (_t3 - _t2), 1000 * (_t3 - _t0) - ) self._created = True + + async def create(self): + """ + Create the link on the nodes (interactive path: prepare + commit). + """ + + entries = await self._prepare() + await self._commit_nios(entries) # New links automatically inherit every active project-level marker # definition so the user doesn't have to reconfigure. await self._project.apply_defs_to_new_link(self) diff --git a/gns3server/schemas/__init__.py b/gns3server/schemas/__init__.py index af026c241..7d4ba6c22 100644 --- a/gns3server/schemas/__init__.py +++ b/gns3server/schemas/__init__.py @@ -91,7 +91,7 @@ from .controller.templates.dynamips_templates import ( ) # Compute schemas -from .compute.nios import UDPNIO, TAPNIO, EthernetNIO, MarkerToggle, MarkerRebuild +from .compute.nios import UDPNIO, TAPNIO, EthernetNIO, MarkerToggle, MarkerRebuild, BatchNIOEntry, BatchNIOCreate from .compute.atm_switch_nodes import ATMSwitchCreate, ATMSwitchUpdate, ATMSwitch from .compute.cloud_nodes import CloudCreate, CloudUpdate, Cloud from .compute.docker_nodes import DockerCreate, DockerUpdate, Docker diff --git a/gns3server/schemas/compute/nios.py b/gns3server/schemas/compute/nios.py index 5830847c5..a1d8641b3 100644 --- a/gns3server/schemas/compute/nios.py +++ b/gns3server/schemas/compute/nios.py @@ -16,7 +16,7 @@ from pydantic import BaseModel, Field -from typing import Optional +from typing import Optional, List from enum import Enum @@ -91,3 +91,24 @@ class MarkerRebuild(BaseModel): direction: Optional[str] = None enabled: bool = True link_id: str = "" + + +class BatchNIOEntry(BaseModel): + """ + A single NIO binding to create as part of a project-wide batch. + """ + + node_id: str = Field(..., description="Node the NIO is attached to") + adapter_number: int = Field(0, ge=0, description="Adapter number") + port_number: int = Field(0, ge=0, description="Port number") + nio: UDPNIO = Field(..., description="NIO settings") + + +class BatchNIOCreate(BaseModel): + """ + Body for the project-wide batch NIO endpoint: create many NIO bindings in a + single request (used during project open) to avoid one HTTP round-trip per + NIO between controller and compute. + """ + + nios: List[BatchNIOEntry] = Field(..., description="NIO bindings to create") diff --git a/tests/api/routes/compute/test_docker_nodes.py b/tests/api/routes/compute/test_docker_nodes.py index ebbba442f..13e4609b7 100644 --- a/tests/api/routes/compute/test_docker_nodes.py +++ b/tests/api/routes/compute/test_docker_nodes.py @@ -211,6 +211,34 @@ class TestDockerNodesRoutes: assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" + async def test_docker_nio_batch_create(self, app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: + """ + Exercise the project-wide batch NIO endpoint: bind two NIOs on the same + docker node in a single request (the path used during project open). + """ + + params = { + "nios": [ + { + "node_id": vm["node_id"], + "adapter_number": 0, + "port_number": 0, + "nio": {"type": "nio_udp", "lport": 4242, "rport": 4343, "rhost": "127.0.0.1"}, + }, + { + "node_id": vm["node_id"], + "adapter_number": 1, + "port_number": 0, + "nio": {"type": "nio_udp", "lport": 4243, "rport": 4344, "rhost": "127.0.0.1"}, + }, + ] + } + + url = app.url_path_for("compute:create_batch_nios", project_id=vm["project_id"]) + response = await compute_client.post(url, json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["added"] == 2 + async def test_docker_update_nio(self, app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: