From b46e43e83d92a076b04edf6eacac3e0da9f92796 Mon Sep 17 00:00:00 2001 From: kkkkk1k1 Date: Thu, 9 Jul 2026 19:22:35 +0800 Subject: [PATCH 1/2] Fix serial disconnects: concurrent, exactly-once URB handling The URB traffic loop processed submissions sequentially and returned ETIMEDOUT (-110) whenever an IN endpoint had no data ready. For serial devices (CDC-ACM, CH34x, ...) this broke the whole chain: - Interrupt/bulk IN endpoints that legitimately have no data pending were completed with -110 instead of staying pending like real hardware, which made the CH340 modem-status read fail and ttyUSB/ttyACM open() return EIO. - A single blocking IN read head-of-line blocked control and OUT transfers, so open() timed out. - During flashing the client frequently unlinks slow URBs; the worker still sent a RET_SUBMIT for an already-unlinked seqnum, which makes the Linux vhci_hcd abort the connection ("cannot find a urb of seqnum ...") and drop the serial port. Rework the loop so the receive thread only does socket I/O and dispatches each URB to a worker thread: - IN endpoints stay pending (polled in coarse slices) and complete only when data arrives or the URB is unlinked - no more spurious -110. - Per-endpoint locks keep same-endpoint transfers ordered while different endpoints run concurrently, so a pending IN read never stalls control/OUT. - Exactly-once completion (per-URB lock + flag) guarantees a URB is finished by either RET_SUBMIT or RET_UNLINK, never both, fixing the fatal RET_UNLINK-before-RET_SUBMIT ordering. - Socket writes are serialised; USBIP_CMD_UNLINK now actually cancels the pending URB. Add mocked pytest regression tests (no hardware) covering concurrency, pending-IN unlink, and the exactly-once unlink race. Co-Authored-By: Claude Opus 4.8 --- pyproject.toml | 5 + tests/test_urb_handling.py | 201 +++++++++++++++++++++ usbip_server.py | 351 ++++++++++++++++++++++++++++--------- 3 files changed, 476 insertions(+), 81 deletions(-) create mode 100644 tests/test_urb_handling.py diff --git a/pyproject.toml b/pyproject.toml index ec0ff9f..cf0c781 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,7 @@ dev = [ "ruff>=0.14.0", "mypy>=1.0.0", "reuse>=6.2.0", + "pytest>=7.0.0", ] [project.urls] @@ -51,6 +52,10 @@ usbipd = "usbipd:main" [tool.setuptools] py-modules = ["usbipd", "usb_device", "usbip_server", "binding_configuration", "libusb_backend"] +[tool.pytest.ini_options] +pythonpath = ["."] +testpaths = ["tests"] + [tool.ruff] target-version = "py311" line-length = 100 diff --git a/tests/test_urb_handling.py b/tests/test_urb_handling.py new file mode 100644 index 0000000..639f02d --- /dev/null +++ b/tests/test_urb_handling.py @@ -0,0 +1,201 @@ +# SPDX-FileCopyrightText: 2026 usbipd-python contributors +# SPDX-License-Identifier: GPL-3.0-or-later +"""Regression tests for concurrent URB handling in ``usbip_server``. + +These exercise the URB traffic loop with a mocked USB device and a scripted +in-memory socket (no libusb / no hardware, per CONTRIBUTING): + +* a pending IN read must not head-of-line block OUT / other endpoints; +* an unlinked pending IN read must not emit a spurious ``RET_SUBMIT``; +* every URB is completed exactly once, so an ``UNLINK`` racing with the + transfer never produces a ``RET_UNLINK`` before ``RET_SUBMIT`` for the same + seqnum (which makes the Linux ``vhci_hcd`` abort the whole connection). +""" + +import struct +import threading +import time + +import usb.core + +import usbip_server as srv + +CMD_SUBMIT = 1 +CMD_UNLINK = 2 +RET_SUBMIT = 3 +RET_UNLINK = 4 +DIR_OUT = 0 +DIR_IN = 1 + + +class FakeDevice: + """Minimal stand-in for a ``usb.core.Device``. + + Bulk OUT writes are echoed back on the matching IN endpoint so a simple + loopback can be asserted. IN reads with no buffered data raise + ``USBTimeoutError`` like libusb would. + """ + + def __init__(self, write_delay: float = 0.0) -> None: + self._echo: dict[int, bytearray] = {} + self._lock = threading.Lock() + self._write_delay = write_delay + + def write(self, endpoint: int, data: bytes, timeout: int = 0) -> int: + if self._write_delay: + time.sleep(self._write_delay) + in_addr = (endpoint & 0x0F) | 0x80 + with self._lock: + self._echo.setdefault(in_addr, bytearray()).extend(bytes(data)) + return len(data) + + def read(self, endpoint: int, length: int, timeout: int = 0) -> bytes: + with self._lock: + buf = self._echo.get(endpoint) + if buf: + out = bytes(buf[:length]) + del buf[:length] + return out + time.sleep(min(timeout, 20) / 1000.0) + raise usb.core.USBTimeoutError("timeout") + + def ctrl_transfer(self, *args: object, **kwargs: object) -> bytes: + return b"" + + +class FakeExportedDevice: + """Stand-in for the exported-device wrapper used by the server.""" + + def __init__(self, device: FakeDevice) -> None: + self.device = device + + def claim(self) -> bool: + return True + + def release(self) -> None: + pass + + +class FakeSocket: + """Feeds a scripted client byte stream and captures server responses.""" + + def __init__(self, stream: bytes) -> None: + self._stream = stream + self._pos = 0 + self.sent: list[bytes] = [] + self._lock = threading.Lock() + + def recv(self, length: int) -> bytes: + if self._pos >= len(self._stream): + # Give worker threads a moment to finish, then signal EOF. + time.sleep(0.4) + return b"" + chunk = self._stream[self._pos : self._pos + length] + self._pos += len(chunk) + return chunk + + def sendall(self, data: bytes) -> None: + with self._lock: + self.sent.append(bytes(data)) + + def close(self) -> None: + pass + + +def _submit( + seqnum: int, + direction: int, + endpoint: int, + buflen: int = 0, + payload: bytes = b"", + setup: bytes = b"\x00" * 8, +) -> bytes: + header = struct.pack(">IIIII", CMD_SUBMIT, seqnum, 0, direction, endpoint) + header += struct.pack(">IIIII", 0, buflen, 0, 0, 0) + header += setup + return header + (payload if direction == DIR_OUT else b"") + + +def _unlink(seqnum: int, target: int) -> bytes: + header = struct.pack(">IIIII", CMD_UNLINK, seqnum, 0, 0, 0) + header += struct.pack(">I", target) + b"\x00" * 24 + return header[:48] + + +def _run(stream: bytes, device: FakeDevice) -> list[tuple[str, int, int, bytes]]: + server = srv.USBIPServer() + server._running = True + server._exported_devices = {"1-1": FakeExportedDevice(device)} # type: ignore[dict-item] + sock = FakeSocket(stream) + thread = threading.Thread(target=server._handle_urb_traffic, args=(sock, "1-1"), daemon=True) + thread.start() + thread.join(timeout=10) + assert not thread.is_alive(), "URB traffic handler did not terminate" + + parsed: list[tuple[str, int, int, bytes]] = [] + for msg in sock.sent: + command, seqnum = struct.unpack(">II", msg[:8]) + if command == RET_SUBMIT: + status, actual = struct.unpack(">iI", msg[20:28]) + parsed.append(("submit", seqnum, status, msg[48 : 48 + actual])) + elif command == RET_UNLINK: + parsed.append(("unlink", seqnum, 0, b"")) + return parsed + + +def test_pending_in_does_not_block_out_or_echo() -> None: + """A pending interrupt IN read must not stall bulk OUT / bulk IN.""" + device = FakeDevice() + stream = b"".join( + [ + _submit(101, DIR_IN, 1, buflen=8), # interrupt IN -> stays pending + _submit(102, DIR_OUT, 2, buflen=5, payload=b"HELLO"), # bulk OUT + _submit(103, DIR_IN, 2, buflen=64), # bulk IN -> echoes HELLO + _unlink(104, 101), # release the pending interrupt IN + ] + ) + resp = _run(stream, device) + submits = {seqnum: (status, data) for kind, seqnum, status, data in resp if kind == "submit"} + unlinks = {seqnum for kind, seqnum, _, _ in resp if kind == "unlink"} + + assert submits.get(102, (None, b""))[0] == 0 # OUT completed (not HOL blocked) + assert submits.get(103, (None, b""))[1] == b"HELLO" # IN echoed loopback data + assert 104 in unlinks # unlink acknowledged + assert 101 not in submits # pending IN that was unlinked never RET_SUBMITs + + +def test_exactly_once_completion_under_unlink_race() -> None: + """UNLINK racing with completion must never yield RET_UNLINK before + RET_SUBMIT for the same URB, and never a duplicate completion.""" + device = FakeDevice(write_delay=0.002) # keep workers busy to force the race + count = 200 + parts: list[bytes] = [] + unlink_to_orig: dict[int, int] = {} + for i in range(count): + orig = 1000 + i + unlink_cmd = 500000 + i + unlink_to_orig[unlink_cmd] = orig + parts.append(_submit(orig, DIR_OUT, 2, buflen=4, payload=b"XXXX")) + parts.append(_unlink(unlink_cmd, orig)) + + resp = _run(b"".join(parts), device) + + submit_pos: dict[int, int] = {} + unlink_pos: dict[int, int] = {} + submit_count: dict[int, int] = {} + for pos, (kind, seqnum, _status, _data) in enumerate(resp): + if kind == "submit": + submit_pos.setdefault(seqnum, pos) + submit_count[seqnum] = submit_count.get(seqnum, 0) + 1 + else: + unlink_pos.setdefault(seqnum, pos) + + for unlink_cmd, orig in unlink_to_orig.items(): + sp = submit_pos.get(orig) + up = unlink_pos.get(unlink_cmd) + # The fatal ordering is RET_UNLINK(orig) before RET_SUBMIT(orig). + assert not (sp is not None and up is not None and up < sp), ( + f"fatal RET_UNLINK-before-RET_SUBMIT for seqnum {orig}" + ) + assert submit_count.get(orig, 0) <= 1, f"duplicate RET_SUBMIT for seqnum {orig}" + assert sp is not None or up is not None, f"seqnum {orig} got no terminal response" diff --git a/usbip_server.py b/usbip_server.py index 9a73ff1..f0a48b7 100644 --- a/usbip_server.py +++ b/usbip_server.py @@ -47,9 +47,62 @@ # Default server port DEFAULT_PORT = 3240 +# Poll slice (ms) for pending IN endpoints. Because each URB runs in its own +# thread this only bounds how quickly a pending IN read notices an unlink / +# shutdown; it does NOT affect throughput or cause head-of-line blocking. +_IN_POLL_MS = 200 + logger = logging.getLogger(__name__) +class _PendingUrb: + """Tracks an in-flight URB so USBIP_CMD_UNLINK can cancel it. + + ``lock`` + ``completed`` enforce exactly-once completion: whichever of the + worker (RET_SUBMIT) or the unlink handler (RET_UNLINK) gets there first + claims the URB; the other must not also complete it. A late RET_SUBMIT for + an already-unlinked seqnum makes the Linux vhci_hcd abort the whole + connection, so this invariant is what keeps the port from dropping. + """ + + __slots__ = ("cancel", "endpoint", "lock", "completed") + + def __init__(self, endpoint: int) -> None: + self.cancel = threading.Event() + self.endpoint = endpoint + self.lock = threading.Lock() + self.completed = False + + +class _UrbConnection: + """Per-client URB processing context. + + Serialises writes on the shared socket, tracks pending URBs so they can be + unlinked, and hands out per-endpoint locks so transfers on *different* + endpoints run concurrently while transfers on the *same* endpoint stay + ordered (important so serial byte streams are never reordered). + """ + + def __init__(self, sock: socket.socket, device: "usb.core.Device") -> None: + self.sock = sock + self.device = device + self.send_lock = threading.Lock() + self.pending_lock = threading.Lock() + self.pending: dict[int, _PendingUrb] = {} + self._ep_locks: dict[int, threading.Lock] = {} + self._ep_guard = threading.Lock() + self.stop = threading.Event() + self.workers: list[threading.Thread] = [] + + def endpoint_lock(self, key: int) -> threading.Lock: + with self._ep_guard: + lock = self._ep_locks.get(key) + if lock is None: + lock = threading.Lock() + self._ep_locks[key] = lock + return lock + + class USBIPServer: """A USB/IP server that exports USB devices over the network.""" @@ -448,6 +501,12 @@ def _build_import_device_info(self, bus_id: str, device: usb.core.Device) -> byt def _handle_urb_traffic(self, client_socket: socket.socket, bus_id: str) -> None: """Handle URB traffic for an imported device. + The receive loop only touches the socket (parse the header and read any + OUT payload) and then hands each URB to a worker thread. This keeps a + blocking / pending IN read from head-of-line blocking control and OUT + transfers, and lets IN URBs stay pending like real hardware until data + arrives or the client unlinks them. + Args: client_socket: The client socket. bus_id: The bus ID of the imported device. @@ -464,6 +523,7 @@ def _handle_urb_traffic(self, client_socket: socket.socket, bus_id: str) -> None return logger.info(f"Starting URB traffic handling for {bus_id}") + conn = _UrbConnection(client_socket, usb_device.device) try: while self._running: @@ -478,16 +538,9 @@ def _handle_urb_traffic(self, client_socket: socket.socket, bus_id: str) -> None ) if command == USBIP_CMD_SUBMIT: - self._handle_urb_submit( - client_socket, - usb_device.device, - header, - seqnum, - direction, - endpoint, - ) + self._dispatch_urb_submit(conn, header, seqnum, direction, endpoint) elif command == USBIP_CMD_UNLINK: - self._handle_urb_unlink(client_socket, header, seqnum) + self._handle_urb_unlink(conn, header, seqnum) else: logger.warning(f"Unknown URB command: 0x{command:08x}") break @@ -500,30 +553,31 @@ def _handle_urb_traffic(self, client_socket: socket.socket, bus_id: str) -> None logger.error(f"Error handling URB: {error}") break finally: - # Release the device + # Tear down: stop workers and cancel any pending IN URBs so their + # threads wake and exit, then release the device. + conn.stop.set() + with conn.pending_lock: + for urb in conn.pending.values(): + urb.cancel.set() + for worker in list(conn.workers): + worker.join(timeout=1.0) usb_device.release() logger.info(f"URB traffic handling ended for {bus_id}") - def _handle_urb_submit( + def _dispatch_urb_submit( self, - client_socket: socket.socket, - device: usb.core.Device, + conn: "_UrbConnection", header: bytes, seqnum: int, direction: int, endpoint: int, ) -> None: - """Handle USBIP_CMD_SUBMIT command. + """Parse a SUBMIT, read any OUT payload from the socket (in stream + order), then hand the URB to a worker thread for the actual transfer. - Args: - client_socket: The client socket. - device: The USB device. - header: The full header bytes. - seqnum: The sequence number. - direction: The transfer direction. - endpoint: The endpoint number. + Runs in the single receive thread so socket reads stay ordered; the + (possibly blocking / pending) USB transfer happens off-thread. """ - # Parse the rest of the submit header ( transfer_flags, transfer_buffer_length, @@ -534,57 +588,149 @@ def _handle_urb_submit( setup = header[40:48] - # For control transfers, check the setup packet to determine if we need to read data - # For bulk/interrupt OUT transfers, read the data buffer + # Read the outbound data buffer here, while we own the socket stream. transfer_buffer = b"" if transfer_buffer_length > 0: if endpoint == 0: - # Control transfer - check bmRequestType for direction bmRequestType = setup[0] is_device_to_host = (bmRequestType & 0x80) != 0 if not is_device_to_host: - # Host to Device (OUT) - read the data - recv_result = self._recv_exact(client_socket, transfer_buffer_length) + recv_result = self._recv_exact(conn.sock, transfer_buffer_length) if recv_result is None: return transfer_buffer = recv_result elif direction == USBIP_DIR_OUT: - # Bulk/Interrupt OUT - read the data - recv_result = self._recv_exact(client_socket, transfer_buffer_length) + recv_result = self._recv_exact(conn.sock, transfer_buffer_length) if recv_result is None: return transfer_buffer = recv_result - # Execute the USB transfer + urb = _PendingUrb(endpoint) + with conn.pending_lock: + conn.pending[seqnum] = urb + + worker = threading.Thread( + target=self._process_urb, + args=( + conn, + urb, + seqnum, + direction, + endpoint, + setup, + transfer_buffer, + transfer_buffer_length, + start_frame, + number_of_packets, + ), + daemon=True, + ) + conn.workers.append(worker) + worker.start() + + # Keep the worker handle list from growing without bound. + if len(conn.workers) > 64: + conn.workers[:] = [w for w in conn.workers if w.is_alive()] + + def _process_urb( + self, + conn: "_UrbConnection", + urb: "_PendingUrb", + seqnum: int, + direction: int, + endpoint: int, + setup: bytes, + transfer_buffer: bytes, + transfer_buffer_length: int, + start_frame: int, + number_of_packets: int, + ) -> None: + """Execute one URB in its own thread and send RET_SUBMIT. + + Transfers are serialised per endpoint (so same-endpoint data keeps its + order) but run concurrently across different endpoints, so a pending IN + read never blocks control/OUT traffic. + """ + # Per-endpoint serialisation key: control shares EP0; bulk/interrupt is + # keyed by address + direction. + if endpoint == 0: + ep_key = 0 + else: + ep_key = (endpoint & 0x0F) | (0x80 if direction == USBIP_DIR_IN else 0x00) + actual_length = 0 status = 0 response_data = b"" try: - if endpoint == 0: - # Control transfer - response_data, actual_length = self._do_control_transfer( - device, setup, transfer_buffer, transfer_buffer_length, direction - ) - else: - # Bulk/Interrupt transfer - response_data, actual_length = self._do_bulk_interrupt_transfer( - device, endpoint, transfer_buffer, transfer_buffer_length, direction + with conn.endpoint_lock(ep_key): + if urb.cancel.is_set() or conn.stop.is_set(): + self._finish_urb(conn, seqnum) + return + try: + if endpoint == 0: + response_data, actual_length = self._do_control_transfer( + conn.device, + setup, + transfer_buffer, + transfer_buffer_length, + direction, + ) + else: + result = self._do_bulk_interrupt_transfer( + conn.device, + endpoint, + transfer_buffer, + transfer_buffer_length, + direction, + cancel=urb.cancel, + ) + if result is None: + # Unlinked while pending: the client already got a + # RET_UNLINK, so do not send a RET_SUBMIT. + self._finish_urb(conn, seqnum) + return + response_data, actual_length = result + except usb.core.USBTimeoutError: + logger.debug(f"USB timeout on endpoint {endpoint}") + status = -110 + except usb.core.USBError as error: + logger.debug(f"USB error on endpoint {endpoint}: {error}") + status = -error.errno if error.errno is not None else -5 + except Exception as error: # worker must never die silently + logger.error(f"URB worker error seqnum={seqnum}: {error}") + status = -5 + + # Exactly-once completion: if an UNLINK already claimed this URB, do + # NOT send a RET_SUBMIT. A late RET_SUBMIT for an unlinked seqnum makes + # vhci_hcd abort the connection ("cannot find a urb of seqnum ..."), + # which is what dropped the serial port during flashing. Hold urb.lock + # across the check + send so it is atomic vs. the unlink handler. + with urb.lock: + if not urb.completed: + urb.completed = True + self._send_ret_submit( + conn, + seqnum, + status, + actual_length, + response_data, + start_frame, + number_of_packets, ) - except usb.core.USBTimeoutError: - # Timeout - return ETIMEDOUT (-110 on Linux) - logger.debug(f"USB timeout on endpoint {endpoint}") - status = -110 - except usb.core.USBError as error: - logger.debug(f"USB error on endpoint {endpoint}: {error}") - # Map common USB errors to Linux errno values - if error.errno is not None: - status = -error.errno - else: - # Generic I/O error - status = -5 # EIO + self._finish_urb(conn, seqnum) - # Build response + def _send_ret_submit( + self, + conn: "_UrbConnection", + seqnum: int, + status: int, + actual_length: int, + response_data: bytes, + start_frame: int, + number_of_packets: int, + ) -> None: + """Serialise the RET_SUBMIT write on the shared socket.""" response = struct.pack( ">IIIII", USBIP_RET_SUBMIT, @@ -593,7 +739,6 @@ def _handle_urb_submit( 0, # direction 0, # endpoint ) - response += struct.pack( ">iI I I i", status, @@ -602,19 +747,27 @@ def _handle_urb_submit( number_of_packets, 0, # error_count ) - # Padding (8 bytes) response += b"\x00" * 8 - # Add response data for IN transfers (including control IN transfers) if actual_length > 0 and len(response_data) > 0: response += response_data[:actual_length] - client_socket.sendall(response) + with conn.send_lock: + try: + conn.sock.sendall(response) + except OSError: + conn.stop.set() + return logger.debug( f"Sent URB response: seqnum={seqnum}, status={status}, actual_length={actual_length}" ) + def _finish_urb(self, conn: "_UrbConnection", seqnum: int) -> None: + """Remove a completed/cancelled URB from the pending registry.""" + with conn.pending_lock: + conn.pending.pop(seqnum, None) + def _do_control_transfer( self, device: usb.core.Device, @@ -683,19 +836,28 @@ def _do_bulk_interrupt_transfer( data: bytes, length: int, direction: int, - ) -> tuple[bytes, int]: + cancel: "threading.Event | None" = None, + ) -> "tuple[bytes, int] | None": """ Execute a bulk or interrupt transfer. + For IN transfers the URB is kept *pending* (polled in coarse slices) and + only completes when data arrives or ``cancel`` is set (the client sent + USBIP_CMD_UNLINK or the connection is closing). This mirrors real + hardware, which just NAKs an IN endpoint that has no data rather than + completing the URB with an error. + Args: device: The USB device. endpoint: The endpoint number (without direction bit). data: The data buffer for OUT transfers. length: The expected transfer length. direction: The transfer direction (USBIP_DIR_IN or USBIP_DIR_OUT). + cancel: Event set when the URB is unlinked / the connection closes. Returns: - A tuple of (response_data, actual_length). + A tuple of (response_data, actual_length), or ``None`` if the URB + was cancelled while still pending (caller must not send RET_SUBMIT). """ # Construct the full endpoint address with direction bit if direction == USBIP_DIR_IN: @@ -703,14 +865,18 @@ def _do_bulk_interrupt_transfer( logger.debug( f"Bulk/Interrupt IN transfer: endpoint=0x{endpoint_addr:02x}, length={length}" ) - try: - # Use a shorter timeout for interrupt endpoints to avoid blocking - result = device.read(endpoint_addr, length, timeout=1000) - return bytes(result), len(result) - except usb.core.USBTimeoutError: - # Timeout is normal for interrupt endpoints with no data - logger.debug(f"Read timeout on endpoint 0x{endpoint_addr:02x}") - raise + # Stay pending like real hardware. Because each URB runs in its own + # thread, blocking here does not stall other endpoints or control + # transfers, so we poll in coarse slices only to notice cancellation. + while cancel is None or not cancel.is_set(): + if not self._running: + return None + try: + result = device.read(endpoint_addr, length, timeout=_IN_POLL_MS) + return bytes(result), len(result) + except usb.core.USBTimeoutError: + continue # no data yet -> keep the URB pending + return None # unlinked while pending: do not complete else: endpoint_addr = endpoint & 0x0F # Clear direction bit for OUT logger.debug( @@ -719,18 +885,13 @@ def _do_bulk_interrupt_transfer( result = device.write(endpoint_addr, data, timeout=5000) return b"", result - def _handle_urb_unlink(self, client_socket: socket.socket, header: bytes, seqnum: int) -> None: - """ - Handle USBIP_CMD_UNLINK command. + def _send_ret_unlink(self, conn: "_UrbConnection", seqnum: int, status: int = -104) -> None: + """Serialise a RET_UNLINK write on the shared socket. - Args: - client_socket: The client socket. - header: The full header bytes. - seqnum: The sequence number. + ``seqnum`` is the UNLINK command's own seqnum; status defaults to + -ECONNRESET (-104), the value the Linux stub uses for a successful + unlink. """ - unlink_seqnum = struct.unpack(">I", header[20:24])[0] - - # Build response (we don't actually track pending URBs in this simple implementation) response = struct.pack( ">IIIII", USBIP_RET_UNLINK, @@ -739,12 +900,40 @@ def _handle_urb_unlink(self, client_socket: socket.socket, header: bytes, seqnum 0, # direction 0, # endpoint ) - - # Status: -ECONNRESET (-104) for successful unlink - response += struct.pack(">i", -104) - + response += struct.pack(">i", status) # Padding (24 bytes) response += b"\x00" * 24 + with conn.send_lock: + try: + conn.sock.sendall(response) + except OSError: + conn.stop.set() + + def _handle_urb_unlink(self, conn: "_UrbConnection", header: bytes, seqnum: int) -> None: + """ + Handle USBIP_CMD_UNLINK command: claim the target URB's completion so + its worker cannot send a (now-fatal) late RET_SUBMIT, wake any pending + IN read, and acknowledge with RET_UNLINK. + + Args: + conn: The per-client URB connection context. + header: The full header bytes. + seqnum: The sequence number of this UNLINK command. + """ + unlink_seqnum = struct.unpack(">I", header[20:24])[0] + + with conn.pending_lock: + target = conn.pending.get(unlink_seqnum) + + if target is not None: + # Claim completion and ack under the URB lock so we serialise + # against the worker's RET_SUBMIT on the same URB (exactly-once). + with target.lock: + target.completed = True + target.cancel.set() + self._send_ret_unlink(conn, seqnum) + else: + # URB already completed and retired; still acknowledge the unlink. + self._send_ret_unlink(conn, seqnum) - client_socket.sendall(response) logger.debug(f"Unlinked URB seqnum={unlink_seqnum}") From df2ad5a4c38584a2439f6f66edea67e1676f9f81 Mon Sep 17 00:00:00 2001 From: kkkkk1k1 Date: Tue, 21 Jul 2026 15:46:09 +0800 Subject: [PATCH 2/2] Enable TCP_NODELAY on accepted client sockets USB/IP is a request/response protocol made up of many small PDUs. Leaving Nagle's algorithm enabled on the accepted client socket lets small URB responses be held back waiting for an ACK, which interacts badly with the peer's delayed-ACK timer. The Linux usbip userspace sets TCP_NODELAY on its sockets for this reason (usbip_net_set_nodelay); do the same here. Note: this is protocol hygiene, not a fix for any specific measured regression. On one tested setup (client reaching the server through an SSH tunnel) the dominant per-URB latency was ~48 ms and lived in the tunnel hops, not in this socket - enabling TCP_NODELAY did not change it there. Factored into a small helper so it can be unit-tested without a running server. Co-Authored-By: Claude Opus 4.8 --- tests/test_urb_handling.py | 24 ++++++++++++++++++++++++ usbip_server.py | 18 ++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/tests/test_urb_handling.py b/tests/test_urb_handling.py index 639f02d..1ef646c 100644 --- a/tests/test_urb_handling.py +++ b/tests/test_urb_handling.py @@ -199,3 +199,27 @@ def test_exactly_once_completion_under_unlink_race() -> None: ) assert submit_count.get(orig, 0) <= 1, f"duplicate RET_SUBMIT for seqnum {orig}" assert sp is not None or up is not None, f"seqnum {orig} got no terminal response" + + +def test_client_socket_has_nagle_disabled() -> None: + """Accepted client sockets must have TCP_NODELAY set. + + USB/IP is a small-PDU request/response protocol; Nagle plus the peer's + delayed-ACK timer adds tens to hundreds of ms per URB, which breaks the + DTR/RTS auto-reset timing serial bootloaders depend on. + """ + import socket as _socket + + listener = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) + listener.bind(("127.0.0.1", 0)) + listener.listen(1) + client = _socket.create_connection(listener.getsockname()) + accepted, _ = listener.accept() + try: + assert accepted.getsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY) == 0 + srv.USBIPServer._configure_client_socket(accepted) + assert accepted.getsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY) != 0 + finally: + accepted.close() + client.close() + listener.close() diff --git a/usbip_server.py b/usbip_server.py index f0a48b7..d18994a 100644 --- a/usbip_server.py +++ b/usbip_server.py @@ -187,6 +187,7 @@ def start(self) -> None: self._server_socket.settimeout(1.0) try: client_socket, client_address = self._server_socket.accept() + self._configure_client_socket(client_socket) logger.info(f"Connection from {client_address}") client_thread = threading.Thread( target=self._handle_client, @@ -200,6 +201,23 @@ def start(self) -> None: except OSError: break + @staticmethod + def _configure_client_socket(client_socket: socket.socket) -> None: + """Disable Nagle's algorithm on an accepted client connection. + + USB/IP is a latency-sensitive request/response protocol made up of many + small PDUs. With Nagle enabled, small URB responses are held back waiting + for an ACK; combined with the peer's delayed-ACK timer this adds tens to + hundreds of milliseconds per URB. That is enough to break timing-critical + sequences such as the DTR/RTS auto-reset serial bootloaders rely on (the + device then never enters download mode). The Linux usbip userspace sets + TCP_NODELAY for the same reason. + """ + try: + client_socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + except OSError: + logger.debug("Could not enable TCP_NODELAY on client socket") + def stop(self) -> None: """Stop the USB/IP server.""" self._running = False