Merge pull request #944 from zxzxwu/data-length-class

Refactor connection data length to use dataclass

apps/bench.py

@@ -133,8 +133,8 @@ def print_connection(connection: Connection) -> None:
     if connection.transport == PhysicalTransport.LE:
         params.append(
             'DL=('
-            f'TX:{connection.data_length[0]}/{connection.data_length[1]},'
-            f'RX:{connection.data_length[2]}/{connection.data_length[3]}'
+            f'TX:{connection.data_length.max_tx_octets}/{connection.data_length.max_tx_time},'
+            f'RX:{connection.data_length.max_rx_octets}/{connection.data_length.max_rx_time}'
             ')'
         )
 

bumble/device.py

@@ -1804,6 +1804,13 @@ class Connection(utils.CompositeEventEmitter):
         subrate_factor: int = 1
         continuation_number: int = 0
 
+    @dataclass(frozen=True)
+    class LeDataLength:
+        max_tx_octets: int
+        max_tx_time: int
+        max_rx_octets: int
+        max_rx_time: int
+
     def __init__(
         self,
         device: Device,
@@ -1832,7 +1839,7 @@ class Connection(utils.CompositeEventEmitter):
         self.authenticated = False
         self.sc = False
         self.att_mtu = att.ATT_DEFAULT_MTU
-        self.data_length: tuple[int, int, int, int] = DEVICE_DEFAULT_DATA_LENGTH
+        self.data_length = self.LeDataLength(*DEVICE_DEFAULT_DATA_LENGTH)
         self.gatt_client = gatt_client.Client(self)  # Per-connection client
         self.gatt_server = (
             device.gatt_server
@@ -6794,11 +6801,11 @@ class Device(utils.CompositeEventEmitter):
             f'*** Connection Data Length Change: [0x{connection.handle:04X}] '
             f'{connection.peer_address} as {connection.role_name}'
         )
-        connection.data_length = (
-            max_tx_octets,
-            max_tx_time,
-            max_rx_octets,
-            max_rx_time,
+        connection.data_length = Connection.LeDataLength(
+            max_tx_octets=max_tx_octets,
+            max_tx_time=max_tx_time,
+            max_rx_octets=max_rx_octets,
+            max_rx_time=max_rx_time,
         )
         connection.emit(connection.EVENT_CONNECTION_DATA_LENGTH_CHANGE)
 

bumble/host.py

@@ -418,6 +418,7 @@ class Host(utils.EventEmitter):
                         hci.HCI_HARDWARE_ERROR_EVENT,
                         hci.HCI_FLUSH_OCCURRED_EVENT,
                         hci.HCI_ROLE_CHANGE_EVENT,
+                        hci.HCI_NUMBER_OF_COMPLETED_PACKETS_EVENT,
                         hci.HCI_MODE_CHANGE_EVENT,
                         hci.HCI_RETURN_LINK_KEYS_EVENT,
                         hci.HCI_PIN_CODE_REQUEST_EVENT,

bumble/transport/usb.py

@@ -37,6 +37,20 @@ from bumble.transport.common import BaseSource, Transport, TransportInitError
 logger = logging.getLogger(__name__)
 
 
+# -----------------------------------------------------------------------------
+def _safe_call_soon(loop: asyncio.AbstractEventLoop, callback, *args) -> None:
+    """Schedule `callback` on `loop` from the libusb event thread, tolerating the
+    case where the loop has already been closed during process/transport
+    teardown. Without this guard, a libusb transfer callback that fires after the
+    asyncio loop is closed raises 'Event loop is closed' on the C callback
+    thread, which can escalate to a libusb mutex assertion and crash the process
+    (SIGABRT). This makes an unclean shutdown a no-op instead of a core dump."""
+    try:
+        loop.call_soon_threadsafe(callback, *args)
+    except RuntimeError:
+        pass  # loop already closed; nothing left to schedule
+
+
 # -----------------------------------------------------------------------------
 # Constants
 # -----------------------------------------------------------------------------
@@ -266,7 +280,7 @@ class UsbPacketSink:
         self.packets.put_nowait(packet)
 
     def transfer_callback(self, transfer):
-        self.loop.call_soon_threadsafe(self.out_transfer_ready.release)
+        _safe_call_soon(self.loop, self.out_transfer_ready.release)
         status = transfer.getStatus()
 
         logger.debug(f"OUT CALLBACK: {status}")
@@ -535,8 +549,8 @@ class UsbPacketSource(asyncio.Protocol, BaseSource):
         self.dequeue_task = self.loop.create_task(self.dequeue())
 
     def queue_packet(self, packet_type: int, packet_data: bytes) -> None:
-        self.loop.call_soon_threadsafe(
-            self.queue.put_nowait, bytes([packet_type]) + packet_data
+        _safe_call_soon(
+            self.loop, self.queue.put_nowait, bytes([packet_type]) + packet_data
         )
 
     def transfer_callback(self, transfer):
@@ -582,16 +596,16 @@ class UsbPacketSource(asyncio.Protocol, BaseSource):
                         transfer.submit()
                     except usb1.USBError as error:
                         logger.warning(f"Failed to re-submit transfer: {error}")
-                        self.loop.call_soon_threadsafe(self.on_transport_lost)
+                        _safe_call_soon(self.loop, self.on_transport_lost)
         elif status == usb1.TRANSFER_CANCELLED:
             logger.debug(f"IN[{packet_type}] transfer canceled")
-            self.loop.call_soon_threadsafe(self.done[transfer].set)
+            _safe_call_soon(self.loop, self.done[transfer].set)
         else:
             logger.warning(
                 color(f'!!! IN[{packet_type}] transfer not completed', 'red')
             )
-            self.loop.call_soon_threadsafe(self.done[transfer].set)
-            self.loop.call_soon_threadsafe(self.on_transport_lost)
+            _safe_call_soon(self.loop, self.done[transfer].set)
+            _safe_call_soon(self.loop, self.on_transport_lost)
 
     async def dequeue(self):
         while not self.closed:
@@ -704,7 +718,7 @@ class UsbTransport(Transport):
                 logger.warning(f'!!! Exception while handling events: {error}')
 
         logger.debug('ending USB event loop')
-        self.loop.call_soon_threadsafe(self.event_loop_done.set_result, None)
+        _safe_call_soon(self.loop, self.event_loop_done.set_result, None)
 
     async def close(self):
         self.source.close()

pyproject.toml

@@ -27,7 +27,7 @@ dependencies = [
     "grpcio >= 1.62.1; platform_system!='Emscripten'",
     "humanize >= 4.6.0; platform_system!='Emscripten'",
     "libusb1 >= 2.0.1; platform_system!='Emscripten'",
-    "libusb-package == 1.0.26.1; platform_system!='Emscripten' and platform_system!='Android'",
+    "libusb-package == 1.0.26.4; platform_system!='Emscripten' and platform_system!='Android'",
     "platformdirs >= 3.10.0; platform_system!='Emscripten'",
     "prompt_toolkit >= 3.0.16; platform_system!='Emscripten'",
     "prettytable >= 3.6.0; platform_system!='Emscripten'",
@@ -43,9 +43,9 @@ dependencies = [
 [project.optional-dependencies]
 build = ["build >= 0.7"]
 test = [
-    "pytest >= 8.2",
-    "pytest-asyncio >= 0.23.5",
-    "pytest-html >= 3.2.0",
+    "pytest >= 9.0",
+    "pytest-asyncio >= 1.4",
+    "pytest-html >= 4.2",
     "coverage >= 6.4",
 ]
 development = [

tests/host_test.py

@@ -74,6 +74,20 @@ async def test_reset(supported_commands: set[int], max_lmp_features_page_number:
     )
 
 
+@pytest.mark.asyncio
+async def test_reset_enables_number_of_completed_packets_event() -> None:
+    controller = Controller('C')
+    controller.total_num_le_acl_data_packets = 3
+    host = Host(controller, AsyncPipeSink(controller))
+
+    await host.reset()
+
+    completed_packets_event_bit = 1 << (hci.HCI_NUMBER_OF_COMPLETED_PACKETS_EVENT - 1)
+    assert controller.event_mask & completed_packets_event_bit
+    assert host.le_acl_packet_queue is not None
+    assert host.le_acl_packet_queue.max_in_flight == 3
+
+
 # -----------------------------------------------------------------------------
 def test_data_packet_queue():
     controller = unittest.mock.Mock()

tests/l2cap_test.py

@@ -49,19 +49,19 @@ def test_helpers():
     psm = l2cap.L2CAP_Connection_Request.serialize_psm(0x242311)
     assert psm == bytes([0x11, 0x23, 0x24])
 
-    (offset, psm) = l2cap.L2CAP_Connection_Request.parse_psm(
+    offset, psm = l2cap.L2CAP_Connection_Request.parse_psm(
         bytes([0x00, 0x01, 0x00, 0x44]), 1
     )
     assert offset == 3
     assert psm == 0x01
 
-    (offset, psm) = l2cap.L2CAP_Connection_Request.parse_psm(
+    offset, psm = l2cap.L2CAP_Connection_Request.parse_psm(
         bytes([0x00, 0x23, 0x10, 0x44]), 1
     )
     assert offset == 3
     assert psm == 0x1023
 
-    (offset, psm) = l2cap.L2CAP_Connection_Request.parse_psm(
+    offset, psm = l2cap.L2CAP_Connection_Request.parse_psm(
         bytes([0x00, 0x11, 0x23, 0x24, 0x44]), 1
     )
     assert offset == 4

tests/usb_test.py

@@ -21,6 +21,31 @@ from bumble import hci
 from bumble.transport import usb
 
 
+def test_safe_call_soon_ignores_closed_loop():
+    closed_loop = asyncio.new_event_loop()
+    closed_loop.close()
+    callback = mock.Mock()
+
+    usb._safe_call_soon(closed_loop, callback)
+
+    callback.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_safe_call_soon_schedules_callback_on_open_loop():
+    called = asyncio.Event()
+    args = []
+
+    def callback(value):
+        args.append(value)
+        called.set()
+
+    usb._safe_call_soon(asyncio.get_running_loop(), callback, 'scheduled')
+
+    await asyncio.wait_for(called.wait(), timeout=1.0)
+    assert args == ['scheduled']
+
+
 @pytest.mark.asyncio
 async def test_usb_packet_sink_iso_routing():
     # Mock usb1 device and endpoints