From b46e43e83d92a076b04edf6eacac3e0da9f92796 Mon Sep 17 00:00:00 2001 From: kkkkk1k1 Date: Thu, 9 Jul 2026 19:22:35 +0800 Subject: [PATCH] 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}")