diff --git a/bumble/transport/usb.py b/bumble/transport/usb.py index 2185b9b5..48e02cd4 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,62 @@ def __init__(self, emit: Callable[[bytes], Any]) -> None: super().__init__(length_offset=2, length_size=2, emit=emit) +# Fallback USB interrupt polling sync window (4 ms) when the endpoint descriptor +# 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)() + 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__(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, + poll_sync: bool | None = None, + ): super().__init__() self.device = device self.metadata = metadata @@ -502,11 +557,24 @@ 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: + 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 = {} + 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 +628,80 @@ 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() + 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: + # 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: + 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 + ): + # 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 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() + 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 +769,14 @@ async def dequeue(self): except Exception: logger.exception(color('!!! Exception in sink.on_packet', 'red')) + def on_transport_lost(self): + self._clear_acl_queue() + super().on_transport_lost() + def close(self): with self.lock: self.closed = True + self._clear_acl_queue() async def terminate(self): self.close() @@ -766,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: @@ -782,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: @@ -801,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 @@ -817,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(':') @@ -969,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 6295282e..571ab757 100644 --- a/tests/transport/usb_test.py +++ b/tests/transport/usb_test.py @@ -205,3 +205,307 @@ 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.05, + ) + 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) + 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 + # 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) + 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' + + # 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 + 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() + + +@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, + {'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, + )