From 5481049ab1450cb33a2c017502056f8f44084074 Mon Sep 17 00:00:00 2001 From: uier Date: Tue, 13 Oct 2026 16:44:45 +0000 Subject: [PATCH] transport/usb: Preserve HCI event and ACL packet ordering in UsbPacketSource Defer incoming USB Bulk IN ACL packets by one USB interrupt polling interval (default 4ms, matching Linux btusb.c force_poll_sync) and flush deferred ACL packets immediately after queueing incoming USB Interrupt IN HCI events. This prevents Bulk IN ACL packets (such as SMP Identity Information) from racing ahead of concurrent Interrupt IN HCI events (such as HCI_Encryption_Change) on controllers like Intel BE200. Special handling: - For HCI_Disconnection_Complete (0x05), flush deferred ACL packets before queueing the disconnection event so ACL payloads transmitted by the peer prior to disconnect are delivered while the connection state is still active. - Cancel any pending ACL flush timer and clear deferred ACL packets in close() and on_transport_lost(). --- bumble/transport/usb.py | 115 +++++++++++++++++++++++++++++++++--- tests/transport/usb_test.py | 37 ++++++++++++ 2 files changed, 145 insertions(+), 7 deletions(-) diff --git a/bumble/transport/usb.py b/bumble/transport/usb.py index 2185b9b5..445dda6e 100644 --- a/bumble/transport/usb.py +++ b/bumble/transport/usb.py @@ -22,6 +22,7 @@ import logging import platform import threading +from collections import deque from collections.abc import Callable from typing import Any @@ -492,8 +493,22 @@ def __init__(self, emit: Callable[[bytes], Any]) -> None: super().__init__(length_offset=2, length_size=2, emit=emit) +# Default USB interrupt polling sync window (4 ms, matching 1 kernel jiffy at HZ=250 +# in Linux btusb.c force_poll_sync). +DEFAULT_INTR_INTERVAL = 0.004 + + class UsbPacketSource(asyncio.Protocol, BaseSource): - def __init__(self, device, metadata, interrupt_in, bulk_in, isochronous_in): + + def __init__( + self, + device, + metadata, + interrupt_in, + bulk_in, + isochronous_in, + intr_interval: float | None = None, + ): super().__init__() self.device = device self.metadata = metadata @@ -502,11 +517,23 @@ def __init__(self, device, metadata, interrupt_in, bulk_in, isochronous_in): self.bulk_in = bulk_in self.bulk_in_transfer = None self.isochronous_in = isochronous_in - self.isochronous_in_transfers = [] + self.isochronous_in_transfers: list[Any] = [] self.loop = asyncio.get_running_loop() - self.queue = asyncio.Queue() + self.queue: asyncio.Queue[bytes] = asyncio.Queue() + self.acl_queue: deque[tuple[float, bytes]] = deque() + self.acl_flush_handle: asyncio.TimerHandle | None = None + if intr_interval is not None: + self.intr_interval = float(intr_interval) + else: + raw_interval = getattr(interrupt_in, 'getInterval', lambda: None)() + if isinstance(raw_interval, (int, float)) and raw_interval > 0: + self.intr_interval = max( + float(raw_interval) / 1000.0, DEFAULT_INTR_INTERVAL + ) + else: + self.intr_interval = DEFAULT_INTR_INTERVAL self.dequeue_task = None - self.done = {} + self.done: dict[Any, asyncio.Event] = {} self.splitters = { hci.HCI_EVENT_PACKET: EventPacketSplitter( lambda packet: self.queue_packet(hci.HCI_EVENT_PACKET, packet) @@ -560,10 +587,73 @@ def start(self): self.dequeue_task = self.loop.create_task(self.dequeue()) + def _flush_acl_queue(self) -> None: + if self.acl_flush_handle is not None: + self.acl_flush_handle.cancel() + self.acl_flush_handle = None + while self.acl_queue: + _, packet = self.acl_queue.popleft() + self.queue.put_nowait(packet) + + def _on_acl_timer(self) -> None: + self.acl_flush_handle = None + now = self.loop.time() + # Pop the head packet unconditionally because asyncio.call_at may fire + # slightly before the nominal deadline due to OS timer resolution. + if self.acl_queue: + _, packet = self.acl_queue.popleft() + self.queue.put_nowait(packet) + while self.acl_queue and self.acl_queue[0][0] <= now: + _, packet = self.acl_queue.popleft() + self.queue.put_nowait(packet) + if self.acl_queue: + self.acl_flush_handle = self.loop.call_at( + self.acl_queue[0][0], self._on_acl_timer + ) + + def _enqueue_packet(self, packet_type: int, packet_data: bytes) -> None: + if self.closed: + return + + packet = bytes([packet_type]) + packet_data + if packet_type == hci.HCI_ACL_DATA_PACKET and self.intr_interval > 0: + # Defer each ACL packet by its own intr_interval deadline so that + # any HCI event generated by the controller at the same time (which + # waits for the periodic Interrupt IN poll) is delivered first, + # without allowing an earlier ACL packet's timer expiry to + # prematurely flush a newer ACL packet. Mirrors Linux btusb.c + # force_poll_sync / rx_work behavior. + deadline = self.loop.time() + self.intr_interval + self.acl_queue.append((deadline, packet)) + if self.acl_flush_handle is None: + self.acl_flush_handle = self.loop.call_at(deadline, self._on_acl_timer) + return + + if ( + packet_type == hci.HCI_EVENT_PACKET + and self.acl_queue + and len(packet_data) > 0 + and packet_data[0] == hci.HCI_DISCONNECTION_COMPLETE_EVENT + ): + # A connection cannot receive valid ACL data after Disconnection + # Complete. Flush all pending ACL packets (including any on other + # active handles, since flushing early is a safe no-op for ordering + # once an event arrives) BEFORE delivering Disconnection Complete + # so the Host does not tear down the connection handle before + # processing the final ACL packet. + self._flush_acl_queue() + self.queue.put_nowait(packet) + return + + # Enqueue HCI Event (or SCO/unbuffered packet) BEFORE flushing deferred + # ACL packets, matching upstream btusb patch: + # "Bluetooth: btusb: Preserve event and ACL packet ordering". + self.queue.put_nowait(packet) + if packet_type == hci.HCI_EVENT_PACKET and self.acl_queue: + self._flush_acl_queue() + def queue_packet(self, packet_type: int, packet_data: bytes) -> None: - _safe_call_soon( - self.loop, self.queue.put_nowait, bytes([packet_type]) + packet_data - ) + _safe_call_soon(self.loop, self._enqueue_packet, packet_type, packet_data) def transfer_callback(self, transfer): packet_type = transfer.getUserData() @@ -631,9 +721,20 @@ async def dequeue(self): except Exception: logger.exception(color('!!! Exception in sink.on_packet', 'red')) + def on_transport_lost(self): + if self.acl_flush_handle is not None: + self.acl_flush_handle.cancel() + self.acl_flush_handle = None + self.acl_queue.clear() + super().on_transport_lost() + def close(self): with self.lock: self.closed = True + if self.acl_flush_handle is not None: + self.acl_flush_handle.cancel() + self.acl_flush_handle = None + self.acl_queue.clear() async def terminate(self): self.close() diff --git a/tests/transport/usb_test.py b/tests/transport/usb_test.py index 6295282e..350f9742 100644 --- a/tests/transport/usb_test.py +++ b/tests/transport/usb_test.py @@ -205,3 +205,40 @@ async def test_open_usb_transport_by_index_checks_interfaces(device_class): pytest.raises(usb.TransportInitError, match='no compatible interface'), ): await usb.open_usb_transport('0') + + +@pytest.mark.asyncio +async def test_usb_packet_source_acl_ordering(): + source = usb.UsbPacketSource( + mock.Mock(), + {}, + mock.Mock(), + mock.Mock(), + None, + intr_interval=0.01, + ) + acl = b'\x00\x28\x02\x00\xaa\xbb' + enc_event = b'\x08\x04\x00\x00\x08\x01' + disconn_event = bytes( + [hci.HCI_DISCONNECTION_COMPLETE_EVENT, 0x04, 0x00, 0x00, 0x08, 0x13] + ) + + # 1. Event arrives while ACL is deferred -> Event is queued before ACL. + source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl) + source.splitters[hci.HCI_EVENT_PACKET].feed(enc_event) + assert await source.queue.get() == bytes([hci.HCI_EVENT_PACKET]) + enc_event + assert await source.queue.get() == bytes([hci.HCI_ACL_DATA_PACKET]) + acl + + # 2. Disconnection Complete arrives while ACL is deferred -> ACL is flushed first. + source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl) + source.splitters[hci.HCI_EVENT_PACKET].feed(disconn_event) + assert await source.queue.get() == bytes([hci.HCI_ACL_DATA_PACKET]) + acl + assert await source.queue.get() == bytes([hci.HCI_EVENT_PACKET]) + disconn_event + + # 3. Standalone ACL packet is flushed after intr_interval expires. + source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl) + assert source.queue.empty() + assert await asyncio.wait_for(source.queue.get(), timeout=1.0) == ( + bytes([hci.HCI_ACL_DATA_PACKET]) + acl + ) + source.close()