From 5481049ab1450cb33a2c017502056f8f44084074 Mon Sep 17 00:00:00 2001 From: uier Date: Tue, 13 Oct 2026 16:44:45 +0000 Subject: [PATCH 1/6] 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() From b3f7ab456862f64b74e092e3a92d53a72f4995bb Mon Sep 17 00:00:00 2001 From: uier Date: Mon, 19 Oct 2026 08:35:26 +0000 Subject: [PATCH 2/6] tests/transport/usb: Expand UsbPacketSource ACL deferral and timer tests --- tests/transport/usb_test.py | 107 +++++++++++++++++++++++++++++++++--- 1 file changed, 99 insertions(+), 8 deletions(-) diff --git a/tests/transport/usb_test.py b/tests/transport/usb_test.py index 350f9742..72b41a09 100644 --- a/tests/transport/usb_test.py +++ b/tests/transport/usb_test.py @@ -215,7 +215,7 @@ async def test_usb_packet_source_acl_ordering(): mock.Mock(), mock.Mock(), None, - intr_interval=0.01, + intr_interval=0.05, ) acl = b'\x00\x28\x02\x00\xaa\xbb' enc_event = b'\x08\x04\x00\x00\x08\x01' @@ -226,19 +226,110 @@ async def test_usb_packet_source_acl_ordering(): # 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 + await asyncio.sleep(0) + assert source.queue.get_nowait() == bytes([hci.HCI_EVENT_PACKET]) + enc_event + assert source.queue.get_nowait() == bytes([hci.HCI_ACL_DATA_PACKET]) + acl - # 2. Disconnection Complete arrives while ACL is deferred -> ACL is flushed first. + # 2. Disconnection Complete arrives while ACL is deferred -> ACL is flushed + # immediately before the timer expires, ahead of Disconnection Complete. 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. + await asyncio.sleep(0) + assert source.acl_flush_handle is None + assert len(source.acl_queue) == 0 + assert source.queue.get_nowait() == bytes([hci.HCI_ACL_DATA_PACKET]) + acl + assert source.queue.get_nowait() == bytes([hci.HCI_EVENT_PACKET]) + disconn_event + + # 3. Standalone ACL packet remains deferred after _enqueue_packet runs, and is + # flushed only after intr_interval expires. source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl) + await asyncio.sleep(0) assert source.queue.empty() + assert len(source.acl_queue) == 1 assert await asyncio.wait_for(source.queue.get(), timeout=1.0) == ( bytes([hci.HCI_ACL_DATA_PACKET]) + acl ) source.close() + + +@pytest.mark.asyncio +async def test_usb_packet_source_staggered_acl_deadlines(): + loop = asyncio.get_running_loop() + now = 100.0 + with mock.patch.object(loop, 'time', side_effect=lambda: now): + source = usb.UsbPacketSource( + mock.Mock(), + {}, + mock.Mock(), + mock.Mock(), + None, + intr_interval=0.04, + ) + acl_1 = b'\x00\x28\x02\x00\x01\x01' + acl_2 = b'\x00\x28\x02\x00\x02\x02' + + source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl_1) + await asyncio.sleep(0) + now = 100.02 + source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl_2) + await asyncio.sleep(0) + assert len(source.acl_queue) == 2 + + # When acl_1's deadline (100.04) expires, acl_2 (100.06) remains deferred. + now = 100.04 + assert await asyncio.wait_for(source.queue.get(), timeout=1.0) == ( + bytes([hci.HCI_ACL_DATA_PACKET]) + acl_1 + ) + assert source.queue.empty() + assert len(source.acl_queue) == 1 + + now = 100.06 + assert await asyncio.wait_for(source.queue.get(), timeout=1.0) == ( + bytes([hci.HCI_ACL_DATA_PACKET]) + acl_2 + ) + source.close() + + +@pytest.mark.asyncio +async def test_usb_packet_source_close_cancels_pending_timer(): + source = usb.UsbPacketSource( + mock.Mock(), + {}, + mock.Mock(), + mock.Mock(), + None, + intr_interval=0.03, + ) + acl = b'\x00\x28\x02\x00\xaa\xbb' + + source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl) + await asyncio.sleep(0) + assert source.acl_flush_handle is not None + assert len(source.acl_queue) == 1 + + source.close() + assert source.acl_flush_handle is None + assert len(source.acl_queue) == 0 + + await asyncio.sleep(0.04) + assert source.queue.empty() + + +@pytest.mark.asyncio +async def test_usb_packet_source_zero_intr_interval_bypass(): + source = usb.UsbPacketSource( + mock.Mock(), + {}, + mock.Mock(), + mock.Mock(), + None, + intr_interval=0, + ) + acl = b'\x00\x28\x02\x00\xaa\xbb' + + source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl) + await asyncio.sleep(0) + assert source.acl_flush_handle is None + assert len(source.acl_queue) == 0 + assert source.queue.get_nowait() == bytes([hci.HCI_ACL_DATA_PACKET]) + acl + source.close() From aadac14e379731bfff8346928ca73712a88db201 Mon Sep 17 00:00:00 2001 From: uier Date: Mon, 19 Oct 2026 08:36:22 +0000 Subject: [PATCH 3/6] transport/usb: Factor ACL timer cancellation into _clear_acl_queue --- bumble/transport/usb.py | 16 ++++++++-------- tests/transport/usb_test.py | 19 +++++++++++++++++++ 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/bumble/transport/usb.py b/bumble/transport/usb.py index 445dda6e..be45a615 100644 --- a/bumble/transport/usb.py +++ b/bumble/transport/usb.py @@ -587,6 +587,12 @@ def start(self): self.dequeue_task = self.loop.create_task(self.dequeue()) + def _clear_acl_queue(self) -> None: + if self.acl_flush_handle is not None: + self.acl_flush_handle.cancel() + self.acl_flush_handle = None + self.acl_queue.clear() + def _flush_acl_queue(self) -> None: if self.acl_flush_handle is not None: self.acl_flush_handle.cancel() @@ -722,19 +728,13 @@ async def dequeue(self): 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() + self._clear_acl_queue() 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() + self._clear_acl_queue() async def terminate(self): self.close() diff --git a/tests/transport/usb_test.py b/tests/transport/usb_test.py index 72b41a09..31f14fdf 100644 --- a/tests/transport/usb_test.py +++ b/tests/transport/usb_test.py @@ -302,6 +302,25 @@ async def test_usb_packet_source_close_cancels_pending_timer(): ) acl = b'\x00\x28\x02\x00\xaa\xbb' + # 1. _clear_acl_queue() directly cancels pending timer and clears acl_queue. + source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl) + await asyncio.sleep(0) + assert source.acl_flush_handle is not None + assert len(source.acl_queue) == 1 + source._clear_acl_queue() + assert source.acl_flush_handle is None + assert len(source.acl_queue) == 0 + + # 2. on_transport_lost() cancels pending timer and clears acl_queue. + source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl) + await asyncio.sleep(0) + assert source.acl_flush_handle is not None + assert len(source.acl_queue) == 1 + source.on_transport_lost() + assert source.acl_flush_handle is None + assert len(source.acl_queue) == 0 + + # 3. close() cancels pending timer and clears acl_queue. source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl) await asyncio.sleep(0) assert source.acl_flush_handle is not None From c2ebe7f527cf0ffc09bd14ee80ea42e752976cf9 Mon Sep 17 00:00:00 2001 From: uier Date: Mon, 19 Oct 2026 08:37:06 +0000 Subject: [PATCH 4/6] transport/usb: Clarify UsbPacketSource ACL deferral comments vs btusb --- bumble/transport/usb.py | 35 ++++++++++++++++++----------------- 1 file changed, 18 insertions(+), 17 deletions(-) diff --git a/bumble/transport/usb.py b/bumble/transport/usb.py index be45a615..86f5a4ae 100644 --- a/bumble/transport/usb.py +++ b/bumble/transport/usb.py @@ -493,8 +493,8 @@ 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). +# Fallback USB interrupt polling sync window (4 ms) when the endpoint descriptor +# does not report a valid bInterval. DEFAULT_INTR_INTERVAL = 0.004 @@ -623,12 +623,14 @@ def _enqueue_packet(self, packet_type: int, packet_data: bytes) -> None: 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. + # Inspired by Linux btusb poll_sync: defer ACL packets over the + # Interrupt IN polling window so an HCI event generated by the + # controller at the same time (which waits for the periodic + # Interrupt IN poll) is delivered first. Unlike btusb's single + # rx_work delayed work item (which drains the entire queue when the + # first packet's timer expires), each ACL packet gets its own + # intr_interval deadline so an earlier timer expiry does not + # prematurely flush a newer ACL packet. deadline = self.loop.time() + self.intr_interval self.acl_queue.append((deadline, packet)) if self.acl_flush_handle is None: @@ -641,19 +643,18 @@ def _enqueue_packet(self, packet_type: int, packet_data: bytes) -> None: 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. + # Unlike btusb, flush any deferred ACL packets BEFORE delivering + # HCI_Disconnection_Complete_Event. A connection cannot receive + # valid ACL data after Disconnection Complete, so flushing first + # ensures the Host processes the final ACL frames before tearing + # down connection state. 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". + # Enqueue the HCI Event (or SCO/unbuffered packet) BEFORE flushing + # deferred ACL packets so state-transition events (such as + # HCI_Encryption_Change) are processed ahead of subsequent ACL data. self.queue.put_nowait(packet) if packet_type == hci.HCI_EVENT_PACKET and self.acl_queue: self._flush_acl_queue() From 17777896e46bfeac8250436af81a66df33047b77 Mon Sep 17 00:00:00 2001 From: uier Date: Mon, 19 Oct 2026 08:38:17 +0000 Subject: [PATCH 5/6] transport/usb: Derive interrupt interval from bInterval and device speed --- bumble/transport/usb.py | 42 ++++++++++++++++++++++++++++++------- tests/transport/usb_test.py | 38 +++++++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 7 deletions(-) diff --git a/bumble/transport/usb.py b/bumble/transport/usb.py index 86f5a4ae..3b612749 100644 --- a/bumble/transport/usb.py +++ b/bumble/transport/usb.py @@ -498,6 +498,40 @@ def __init__(self, emit: Callable[[bytes], Any]) -> None: DEFAULT_INTR_INTERVAL = 0.004 +def _get_intr_interval(device, interrupt_in) -> float: + raw_interval = getattr(interrupt_in, 'getInterval', lambda: None)() + if not isinstance(raw_interval, (int, float)) or raw_interval <= 0: + return DEFAULT_INTR_INTERVAL + + speed = None + get_speed = getattr(device, 'getDeviceSpeed', None) + if callable(get_speed): + candidate = get_speed() + if isinstance(candidate, int): + speed = candidate + if speed is None: + get_device = getattr(device, 'getDevice', None) + if callable(get_device): + usb_device = get_device() + get_usb_speed = getattr(usb_device, 'getDeviceSpeed', None) + if callable(get_usb_speed): + candidate = get_usb_speed() + if isinstance(candidate, int): + speed = candidate + + # pylint: disable=no-member + if speed in ( + usb1.SPEED_HIGH, + usb1.SPEED_SUPER, + usb1.SPEED_SUPER_PLUS, + usb1.SPEED_SUPER_PLUS_X2, + ): + exponent = min(max(int(raw_interval), 1), 16) + return (1 << (exponent - 1)) * 0.000125 + + return float(raw_interval) / 1000.0 + + class UsbPacketSource(asyncio.Protocol, BaseSource): def __init__( @@ -525,13 +559,7 @@ def __init__( 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.intr_interval = _get_intr_interval(device, interrupt_in) self.dequeue_task = None self.done: dict[Any, asyncio.Event] = {} self.splitters = { diff --git a/tests/transport/usb_test.py b/tests/transport/usb_test.py index 31f14fdf..61dbc0d4 100644 --- a/tests/transport/usb_test.py +++ b/tests/transport/usb_test.py @@ -352,3 +352,41 @@ async def test_usb_packet_source_zero_intr_interval_bypass(): assert len(source.acl_queue) == 0 assert source.queue.get_nowait() == bytes([hci.HCI_ACL_DATA_PACKET]) + acl source.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ('speed', 'via_handle', 'b_interval', 'expected'), + [ + (usb.usb1.SPEED_FULL, False, 1, 0.001), + (usb.usb1.SPEED_LOW, False, 10, 0.010), + (usb.usb1.SPEED_HIGH, False, 4, 0.001), + (usb.usb1.SPEED_HIGH, True, 1, 0.000125), + (usb.usb1.SPEED_SUPER, True, 4, 0.001), + (usb.usb1.SPEED_FULL, False, 0, usb.DEFAULT_INTR_INTERVAL), + (usb.usb1.SPEED_FULL, False, None, usb.DEFAULT_INTR_INTERVAL), + ], +) +async def test_usb_packet_source_endpoint_interval_derivation( + speed, via_handle, b_interval, expected +): + mock_device = mock.Mock(spec=['getDevice'] if via_handle else ['getDeviceSpeed']) + if via_handle: + usb_device = mock.Mock(spec=['getDeviceSpeed']) + usb_device.getDeviceSpeed.return_value = speed + mock_device.getDevice.return_value = usb_device + else: + mock_device.getDeviceSpeed.return_value = speed + + mock_interrupt_in = mock.Mock(spec=['getInterval']) + mock_interrupt_in.getInterval.return_value = b_interval + + source = usb.UsbPacketSource( + mock_device, + {}, + mock_interrupt_in, + mock.Mock(), + None, + ) + assert source.intr_interval == pytest.approx(expected) + source.close() From 829fc2c50e25f8cd36f260f4e848c1b324492824 Mon Sep 17 00:00:00 2001 From: uier Date: Mon, 19 Oct 2026 08:40:07 +0000 Subject: [PATCH 6/6] transport/usb: Scope ACL poll-sync to Intel controllers or explicit opt-in --- bumble/transport/usb.py | 55 +++++++++++++--- tests/transport/usb_test.py | 121 +++++++++++++++++++++++++++++++++++- 2 files changed, 166 insertions(+), 10 deletions(-) diff --git a/bumble/transport/usb.py b/bumble/transport/usb.py index 3b612749..48e02cd4 100644 --- a/bumble/transport/usb.py +++ b/bumble/transport/usb.py @@ -497,6 +497,11 @@ def __init__(self, emit: Callable[[bytes], Any]) -> None: # does not report a valid bInterval. DEFAULT_INTR_INTERVAL = 0.004 +# USB Vendor IDs that require ACL-vs-Event poll synchronization by default due +# to Bulk IN completing ahead of periodic Interrupt IN polling. +USB_VENDOR_ID_INTEL = 0x8087 +POLL_SYNC_VENDOR_IDS = {USB_VENDOR_ID_INTEL} + def _get_intr_interval(device, interrupt_in) -> float: raw_interval = getattr(interrupt_in, 'getInterval', lambda: None)() @@ -542,6 +547,7 @@ def __init__( bulk_in, isochronous_in, intr_interval: float | None = None, + poll_sync: bool | None = None, ): super().__init__() self.device = device @@ -559,7 +565,14 @@ def __init__( if intr_interval is not None: self.intr_interval = float(intr_interval) else: - self.intr_interval = _get_intr_interval(device, interrupt_in) + if poll_sync is None: + poll_sync = ( + isinstance(metadata, dict) + and metadata.get('vendor_id') in POLL_SYNC_VENDOR_IDS + ) + self.intr_interval = ( + _get_intr_interval(device, interrupt_in) if poll_sync else 0.0 + ) self.dequeue_task = None self.done: dict[Any, asyncio.Event] = {} self.splitters = { @@ -896,7 +909,11 @@ async def close(self): self.context.close() -async def open_usb_transport(spec: str) -> Transport: +async def open_usb_transport( + spec: str, + intr_interval: float | None = None, + poll_sync: bool | None = None, +) -> Transport: ''' Open a USB transport. The moniker string has this syntax: @@ -912,7 +929,8 @@ async def open_usb_transport(spec: str) -> Transport: Opotionally, the moniker may include a +sco= suffix to enable SCO/eSCO and specify the alternate setting to use for SCO/eSCO transfers, with 0 meaning an - automatic selection. + automatic selection, and/or a +poll_sync (or +poll_sync=0 / +poll_sync=1) suffix to + explicitly enable or disable Interrupt IN / Bulk IN packet poll synchronization. In addition, if the moniker ends with the symbol "!", the device will be used in "forced" mode: @@ -931,6 +949,7 @@ async def open_usb_transport(spec: str) -> Transport: mode. 0+sco=0 --> the first BT USB dongle, with SCO enabled using auto-selection. 0+sco=5 --> the first BT USB dongle, with SCO enabled using alternate setting 5. + 0+poll_sync --> the first BT USB dongle, with ACL poll-sync explicitly enabled. ''' # Find the device according to the spec moniker @@ -947,11 +966,23 @@ async def open_usb_transport(spec: str) -> Transport: else: forced_mode = False - if '+sco=' in spec: - spec, sco_alternate_str = spec.split('+sco=') - sco_alternate = int(sco_alternate_str) - else: - sco_alternate = None + spec_parts = spec.split('+') + spec = spec_parts[0] + sco_alternate = None + for option in spec_parts[1:]: + if option.startswith('sco='): + sco_alternate = int(option.split('=', 1)[1]) + elif option == 'poll_sync': + if poll_sync is None: + poll_sync = True + elif option.startswith('poll_sync='): + if poll_sync is None: + poll_sync = option.split('=', 1)[1].lower() not in ( + '0', + 'false', + 'no', + 'off', + ) if ':' in spec: vendor_id, product_id = spec.split(':') @@ -1099,7 +1130,13 @@ def device_is_bluetooth_hci(device): logger.warning('failed to set configuration') source = UsbPacketSource( - device, device_metadata, interrupt_in, bulk_in, isochronous_in + device, + device_metadata, + interrupt_in, + bulk_in, + isochronous_in, + intr_interval=intr_interval, + poll_sync=poll_sync, ) sink = UsbPacketSink( device, bulk_out, isochronous_out, acl_interface.getNumber() diff --git a/tests/transport/usb_test.py b/tests/transport/usb_test.py index 61dbc0d4..571ab757 100644 --- a/tests/transport/usb_test.py +++ b/tests/transport/usb_test.py @@ -383,10 +383,129 @@ async def test_usb_packet_source_endpoint_interval_derivation( source = usb.UsbPacketSource( mock_device, - {}, + {'vendor_id': usb.USB_VENDOR_ID_INTEL}, + mock_interrupt_in, + mock.Mock(), + None, + ) + assert source.intr_interval == pytest.approx(expected) + source.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ('metadata', 'intr_interval', 'poll_sync', 'expected'), + [ + # Non-Intel controller defaults to 0.0 (no ACL latency penalty) + ({'vendor_id': 0x0A12, 'product_id': 0x0001}, None, None, 0.0), + ({}, None, None, 0.0), + # Intel controller enables poll-sync automatically by default + ({'vendor_id': 0x8087, 'product_id': 0x0036}, None, None, 0.001), + # Explicit poll_sync=True enables deferral on non-Intel controller + ({'vendor_id': 0x0A12, 'product_id': 0x0001}, None, True, 0.001), + # Explicit poll_sync=False disables deferral on Intel controller + ({'vendor_id': 0x8087, 'product_id': 0x0036}, None, False, 0.0), + # Explicit intr_interval overrides everything + ({'vendor_id': 0x0A12, 'product_id': 0x0001}, 0.005, None, 0.005), + ], +) +async def test_usb_packet_source_poll_sync_vendor_and_opt_in( + metadata, intr_interval, poll_sync, expected +): + mock_device = mock.Mock(spec=['getDeviceSpeed']) + mock_device.getDeviceSpeed.return_value = usb.usb1.SPEED_HIGH + mock_interrupt_in = mock.Mock(spec=['getInterval']) + mock_interrupt_in.getInterval.return_value = 4 # 1 ms at High-Speed + + source = usb.UsbPacketSource( + mock_device, + metadata, mock_interrupt_in, mock.Mock(), None, + intr_interval=intr_interval, + poll_sync=poll_sync, ) assert source.intr_interval == pytest.approx(expected) source.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ('spec', 'kwargs', 'expected_poll_sync', 'expected_intr_interval'), + [ + ('0', {}, None, None), + ('0+poll_sync', {}, True, None), + ('0+poll_sync=0', {}, False, None), + ('0+sco=0+poll_sync=1', {}, True, None), + ('0', {'poll_sync': True, 'intr_interval': 0.002}, True, 0.002), + ], +) +async def test_open_usb_transport_poll_sync_plumbing( + spec, kwargs, expected_poll_sync, expected_intr_interval +): + device = mock.Mock() + device.getDeviceClass.return_value = usb.USB_DEVICE_CLASS_WIRELESS_CONTROLLER + device.getDeviceSubClass.return_value = usb.USB_DEVICE_SUBCLASS_RF_CONTROLLER + device.getDeviceProtocol.return_value = ( + usb.USB_DEVICE_PROTOCOL_BLUETOOTH_PRIMARY_CONTROLLER + ) + device.getVendorID.return_value = 0x0A12 + device.getProductID.return_value = 0x0001 + handle = mock.Mock() + handle.getConfiguration.return_value = 1 + device.open.return_value = handle + + context = mock.Mock() + context.getDeviceIterator.return_value = [device] + + acl_interface = mock.Mock() + acl_interface.getNumber.return_value = 0 + acl_interface.getAlternateSetting.return_value = 0 + interrupt_in = mock.Mock() + interrupt_in.getAddress.return_value = 0x81 + bulk_in = mock.Mock() + bulk_in.getAddress.return_value = 0x82 + bulk_out = mock.Mock() + bulk_out.getAddress.return_value = 0x02 + + mock_source = mock.Mock() + mock_sink = mock.Mock() + mock_transport = mock.Mock() + + with ( + mock.patch.object(usb, 'load_libusb'), + mock.patch.object(usb.usb1, 'USBContext', return_value=context), + mock.patch.object(usb.usb1, 'hasCapability', return_value=False), + mock.patch.object( + usb, + 'find_endpoints', + return_value=( + 1, + acl_interface, + None, + interrupt_in, + bulk_in, + None, + bulk_out, + None, + ), + ), + mock.patch.object( + usb, 'UsbPacketSource', return_value=mock_source + ) as source_cls, + mock.patch.object(usb, 'UsbPacketSink', return_value=mock_sink), + mock.patch.object(usb, 'UsbTransport', return_value=mock_transport), + ): + result = await usb.open_usb_transport(spec, **kwargs) + + assert result is mock_transport + source_cls.assert_called_once_with( + handle, + {'vendor_id': 0x0A12, 'product_id': 0x0001}, + interrupt_in, + bulk_in, + None, + intr_interval=expected_intr_interval, + poll_sync=expected_poll_sync, + )