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()