perf: batch NIO dispatch on project open (one HTTP per compute)

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.
This commit is contained in:
YueGuobin committed 2026-08-10 23:42:50 +08:00
1 parent 96d4d82716
commit 38c49a655c
7 files changed
+280 -29

No files matched your search

+5 -2
View File
@@ -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)
+131 -7
View File
@@ -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:
+38 -18
View File
@@ -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)