Performance: accelerate project opening with parallel link creation and batch UDP port allocation

- Narrow IOU lock scope to only cover application_id allocation,
  allowing concurrent IOU node creation via Pool(concurrency=5)
- Parallelize link creation during project.open() using Pool(concurrency=5)
  instead of sequential processing
- Add batch UDP port allocation endpoint on compute to allocate N ports
  in a single HTTP call
- Pre-allocate UDP ports per compute before link creation during project
  loading, reducing HTTP round-trips
- UDPLink.create() falls back to individual port allocation if no
  pre-allocated port is available
This commit is contained in:
YueGuobin committed 2026-06-15 12:42:18 +08:00
1 parent b605d71a8a
commit 5d0284e7be
3 files changed
+131 -48

No files matched your search

+101 -44
View File
@@ -158,6 +158,7 @@ class Project:
self.dump()
self._iou_id_lock = asyncio.Lock()
self._preallocated_udp_ports = {} # compute_id -> list of pre-allocated UDP ports
log.debug(f'Project "{self.name}" [{self._id}] loaded')
self.emit_controller_notification("project.created", self.asdict())
@@ -216,6 +217,7 @@ class Project:
self._load_snapshot_config()
# Create the project on demand on the compute node
self._project_created_on_compute = set()
self._preallocated_udp_ports = {}
@property
def scene_height(self):
@@ -618,9 +620,9 @@ class Project:
if node_type == "iou":
async with self._iou_id_lock:
# wait for an IOU node to be completely created before adding a new one
# this is important otherwise we allocate the same application ID (used
# to generate MAC addresses) when creating multiple IOU node at the same time
# Only hold the lock while allocating a unique application ID to avoid
# serializing the HTTP call to the compute across all IOU nodes.
# Multiple IOU nodes can be created concurrently once IDs are assigned.
if "properties" in kwargs.keys():
# allocate a new application id for nodes loaded from the project
kwargs.get("properties")["application_id"] = get_next_application_id(
@@ -629,7 +631,7 @@ class Project:
elif "application_id" not in kwargs.keys() and not kwargs.get("properties"):
# allocate a new application id for nodes added to the project
kwargs["application_id"] = get_next_application_id(self._controller.projects, self._computes)
node = await self._create_node(compute, name, node_id, node_type, **kwargs)
node = await self._create_node(compute, name, node_id, node_type, **kwargs)
else:
node = await self._create_node(compute, name, node_id, node_type, **kwargs)
self.emit_notification("node.created", node.asdict())
@@ -751,6 +753,56 @@ class Project:
self.dump()
self.emit_notification("drawing.deleted", drawing.asdict())
async def _create_link_from_topology_data(self, link_data):
"""
Create a link from topology data (used during project loading).
Extracted into a separate method so links can be created in parallel
via Pool() during project.open().
:param link_data: Link data from the topology file
"""
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
)
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
await link.add_node(
node,
node_link["adapter_number"],
node_link["port_number"],
label=node_link.get("label"),
dump=False,
)
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)
@open_required
async def add_link(self, link_id=None, dump=True):
"""
@@ -766,6 +818,35 @@ class Project:
self.dump()
return link
async def preallocate_udp_ports_for_compute(self, compute, count):
"""
Pre-allocate UDP ports from a compute in a single batch call.
Used during project loading to reduce HTTP round-trips when
creating many links.
:param compute: Compute instance
:param count: Number of UDP ports to pre-allocate
"""
if count <= 0:
return
response = await compute.post(f"/projects/{self._id}/ports/udp/batch", data={"count": count})
ports = response.json["udp_ports"]
self._preallocated_udp_ports.setdefault(compute.id, [])
self._preallocated_udp_ports[compute.id].extend(ports)
def pop_preallocated_udp_port(self, compute_id):
"""
Pop a pre-allocated UDP port for a compute.
:param compute_id: Compute ID
:returns: UDP port number or None if no pre-allocated port is available
"""
ports = self._preallocated_udp_ports.get(compute_id, [])
if ports:
return ports.pop(0)
return None
@open_required
async def delete_link(self, link_id, force_delete=False):
link = self.get_link(link_id)
@@ -1225,50 +1306,26 @@ class Project:
for compute, name, node_id, node_data in nodes_to_create:
pool.append(self.add_node, compute, name, node_id, dump=False, **node_data)
await pool.join()
# Pre-allocate UDP ports for all links in batch to reduce HTTP round-trips
ports_per_compute = {}
for link_data in topology.get("links", []):
if "link_id" not in link_data.keys():
# skip the link
continue
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
)
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
await link.add_node(
node,
node_link["adapter_number"],
node_link["port_number"],
label=node_link.get("label"),
dump=False,
)
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)
node = self._nodes.get(node_link["node_id"])
if node:
ports_per_compute[node.compute.id] = ports_per_compute.get(node.compute.id, 0) + 1
for compute in self.computes:
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=5)
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()
for drawing_data in topology.get("drawings", []):
await self.add_drawing(dump=False, **drawing_data)