From 05934fa8e85300125877c4c97c1b38fba1fe02b9 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Mon, 10 Aug 2026 23:06:54 +0800 Subject: [PATCH] perf: offload per-NIO ubridge commands to thread-pool executor Replace the 3-5 sequential await _ubridge_send calls in _connect_nio with a single run_in_executor batch. The batch holds the node-level asyncio Lock to prevent interleaving with async sends, then uses the hypervisor's new send_batch_sync method which does blocking socket sendall/recv inside the thread pool. Different nodes' batches now run in true OS-thread parallelism rather than serialising through the asyncio event loop between every command. - ubridge_hypervisor.send_batch_sync: blocking batch send using the underlying socket from the asyncio transport, protected by threading.Lock. - _connect_nio: builds command list (add_nio_udp, start_capture, bridge start, reset_packet_filters, add_packet_filter) and dispatches to the default executor. --- gns3server/compute/docker/docker_vm.py | 54 ++++++++------- .../compute/ubridge/ubridge_hypervisor.py | 66 +++++++++++++++++++ tests/compute/docker/test_docker_vm.py | 9 ++- 3 files changed, 103 insertions(+), 26 deletions(-) diff --git a/gns3server/compute/docker/docker_vm.py b/gns3server/compute/docker/docker_vm.py index fb7bf66c5..6f2c425ba 100644 --- a/gns3server/compute/docker/docker_vm.py +++ b/gns3server/compute/docker/docker_vm.py @@ -27,7 +27,6 @@ import aiohttp import subprocess import os import re -import time from gns3server.utils.asyncio.ssh_server import AsyncioSSHServer from gns3server.utils.asyncio.telnet_server import AsyncioTelnetServer @@ -1216,34 +1215,43 @@ class DockerVM(BaseNode): return int(result["State"]["Pid"]) async def _connect_nio(self, adapter_number, nio): - _t0 = time.perf_counter() bridge_name = f"bridge{adapter_number}" - await self._ubridge_send( - "bridge add_nio_udp {bridge_name} {lport} {rhost} {rport}".format( - bridge_name=bridge_name, lport=nio.lport, rhost=nio.rhost, rport=nio.rport - ) - ) - _t1 = time.perf_counter() + # Build the command batch for this NIO. We send everything in one + # executor call so that a single async-lock acquisition covers the + # whole batch, and different nodes' batches can overlap in the + # thread pool via blocking socket I/O. + commands = [ + f"bridge add_nio_udp {bridge_name} {nio.lport} {nio.rhost} {nio.rport}", + ] if nio.capturing: - await self._ubridge_send( - 'bridge start_capture {bridge_name} "{pcap_file}"'.format( - bridge_name=bridge_name, pcap_file=nio.pcap_output_file - ) + commands.append(f'bridge start_capture {bridge_name} "{nio.pcap_output_file}"') + commands.append(f"bridge start {bridge_name}") + commands.append(f"bridge reset_packet_filters {bridge_name}") + for packet_filter in self._build_filter_list(nio.filters): + commands.append(f"bridge add_packet_filter {bridge_name} {packet_filter}") + + # Hold the per-node ubridge lock across the entire executor batch so + # that no async _ubridge_send for this node can interleave with the + # sync socket writes. The lock is created lazily (mirrors the + # @locking decorator on _ubridge_send). + lock_name = "___ubridge_send_lock" + if not hasattr(self, lock_name): + setattr(self, lock_name, asyncio.Lock()) + async with getattr(self, lock_name): + loop = asyncio.get_running_loop() + await loop.run_in_executor( + None, # default thread-pool executor + self._ubridge_hypervisor.send_batch_sync, + commands, ) - await self._ubridge_send(f"bridge start {bridge_name}") - _t2 = time.perf_counter() - await self._ubridge_apply_filters(bridge_name, nio.filters) - _t3 = time.perf_counter() + + # Traffic-insight markers are rare — keep them on the async path so + # they benefit from the existing marker-management logic. The + # per-node lock is already released at this point, but markers + # serialise themselves via _ubridge_send's own @locking. await self._ubridge_apply_markers(bridge_name, nio) - _t4 = time.perf_counter() - log.info( - "NIO timing [adapter=%d] add_nio_udp=%.3fms start=%.3fms filters=%.3fms markers=%.3fms total=%.3fms", - adapter_number, - 1000 * (_t1 - _t0), 1000 * (_t2 - _t1), - 1000 * (_t3 - _t2), 1000 * (_t4 - _t3), - 1000 * (_t4 - _t0)) async def adapter_add_nio_binding(self, adapter_number, nio): """ diff --git a/gns3server/compute/ubridge/ubridge_hypervisor.py b/gns3server/compute/ubridge/ubridge_hypervisor.py index 83f765d16..740c22378 100644 --- a/gns3server/compute/ubridge/ubridge_hypervisor.py +++ b/gns3server/compute/ubridge/ubridge_hypervisor.py @@ -18,6 +18,7 @@ import re import time import logging import asyncio +import threading from gns3server.utils.asyncio import locking from .ubridge_error import UbridgeError @@ -57,6 +58,8 @@ class UBridgeHypervisor: self._timeout = timeout self._reader = None self._writer = None + self._recv_buf = b"" # leftover bytes from last sync recv + self._send_lock = threading.Lock() async def connect(self, timeout=10): """ @@ -266,3 +269,66 @@ class UBridgeHypervisor: log.debug(f"returned result {data}") return data + + def send_batch_sync(self, commands): + """ + Send multiple commands to uBridge using blocking socket I/O. Designed + to run inside ``loop.run_in_executor`` so that a single per-node batch + doesn't bounce through the event loop between every command, and + batches for *different* nodes run in parallel across the thread pool. + + :param commands: iterable of command strings + :raises UbridgeError: if any command fails + """ + if self._writer is None: + raise UbridgeError("Not connected") + transport = self._writer.transport + if transport is None or transport.is_closing(): + raise UbridgeError("Transport closed") + sock = transport.get_extra_info("socket") + if sock is None: + raise UbridgeError("No underlying socket for sync send_batch") + + # Serialise access to this hypervisor's socket — only one batch (sync + # or async) talks to uBridge at a time. The node-level async lock + # (:func:`_ubridge_send`) is held for the entire executor call, so no + # async ``send()`` can interleave. + with self._send_lock: + sock.setblocking(True) + try: + for command in commands: + cmd = (command.strip() + "\n").encode() + sock.sendall(cmd) + + # Read until the terminating line (100-… or 2xx-…) + buf = self._recv_buf + while True: + try: + chunk = sock.recv(4096) + except BlockingIOError: + continue + if not chunk: + raise UbridgeError( + f"uBridge closed connection during '{command}'" + ) + buf += chunk + decoded = buf.decode("utf-8", errors="replace") + # Last complete line determines termination + tail = decoded.rsplit("\r\n", 1)[-1] + if tail and tail[0] in "12" and tail[1:3].isdigit() and len(tail) >= 4 and tail[3] == "-": + break + + # Check for error codes (2xx-…) + last_line = decoded.strip().split("\r\n")[-1] + if self.error_re.match(last_line): + raise UbridgeError(last_line[4:]) + + # Keep any leftover bytes (after the trailing \r\n) for the + # next read in the batch + trailer_start = decoded.rfind("\r\n") + if trailer_start >= 0: + self._recv_buf = buf[trailer_start + 2:] + else: + self._recv_buf = b"" + finally: + sock.setblocking(False) diff --git a/tests/compute/docker/test_docker_vm.py b/tests/compute/docker/test_docker_vm.py index 9d8bee2b8..ae40d91d9 100644 --- a/tests/compute/docker/test_docker_vm.py +++ b/tests/compute/docker/test_docker_vm.py @@ -1440,9 +1440,12 @@ async def test_add_ubridge_connection(vm): call.send('bridge create bridge0'), call.send("bridge add_nio_tap bridge0 tap-gns3-e0"), call.send('docker move_to_ns tap-gns3-e0 42 eth0'), - call.send('bridge add_nio_udp bridge0 4242 127.0.0.1 4343'), - call.send('bridge start_capture bridge0 "/tmp/capture.pcap"'), - call.send('bridge start bridge0') + call.send_batch_sync([ + 'bridge add_nio_udp bridge0 4242 127.0.0.1 4343', + 'bridge start_capture bridge0 "/tmp/capture.pcap"', + 'bridge start bridge0', + 'bridge reset_packet_filters bridge0', + ]), ] assert 'bridge0' in vm._bridges # We need to check any_order otherwise mock is confused by asyncio