compatibility with recent host ACL property changes

apps: Add a controller loopback throughput test app

.gitignore

@@ -10,3 +10,5 @@ __pycache__
 bumble/_version.py
 .vscode/launch.json
 /.idea
+venv/
+.venv/

.vscode/settings.json

@@ -22,6 +22,7 @@
         "cmac",
         "CONNECTIONLESS",
         "csip",
+        "csis",
         "csrcs",
         "CVSD",
         "datagram",

apps/controller_info.py

@@ -32,10 +32,14 @@ from bumble.hci import (
     HCI_Command,
     HCI_Command_Complete_Event,
     HCI_Command_Status_Event,
+    HCI_READ_BUFFER_SIZE_COMMAND,
+    HCI_Read_Buffer_Size_Command,
     HCI_READ_BD_ADDR_COMMAND,
     HCI_Read_BD_ADDR_Command,
     HCI_READ_LOCAL_NAME_COMMAND,
     HCI_Read_Local_Name_Command,
+    HCI_LE_READ_BUFFER_SIZE_COMMAND,
+    HCI_LE_Read_Buffer_Size_Command,
     HCI_LE_READ_MAXIMUM_DATA_LENGTH_COMMAND,
     HCI_LE_Read_Maximum_Data_Length_Command,
     HCI_LE_READ_NUMBER_OF_SUPPORTED_ADVERTISING_SETS_COMMAND,
@@ -59,7 +63,7 @@ def command_succeeded(response):
 
 
 # -----------------------------------------------------------------------------
-async def get_classic_info(host):
+async def get_classic_info(host: Host) -> None:
     if host.supports_command(HCI_READ_BD_ADDR_COMMAND):
         response = await host.send_command(HCI_Read_BD_ADDR_Command())
         if command_succeeded(response):
@@ -80,7 +84,7 @@ async def get_classic_info(host):
 
 
 # -----------------------------------------------------------------------------
-async def get_le_info(host):
+async def get_le_info(host: Host) -> None:
     print()
 
     if host.supports_command(HCI_LE_READ_NUMBER_OF_SUPPORTED_ADVERTISING_SETS_COMMAND):
@@ -136,6 +140,31 @@ async def get_le_info(host):
         print('  ', name_or_number(HCI_LE_SUPPORTED_FEATURES_NAMES, feature))
 
 
+# -----------------------------------------------------------------------------
+async def get_acl_flow_control_info(host: Host) -> None:
+    print()
+
+    if host.supports_command(HCI_READ_BUFFER_SIZE_COMMAND):
+        response = await host.send_command(
+            HCI_Read_Buffer_Size_Command(), check_result=True
+        )
+        print(
+            color('ACL Flow Control:', 'yellow'),
+            f'{response.return_parameters.hc_total_num_acl_data_packets} '
+            f'packets of size {response.return_parameters.hc_acl_data_packet_length}',
+        )
+
+    if host.supports_command(HCI_LE_READ_BUFFER_SIZE_COMMAND):
+        response = await host.send_command(
+            HCI_LE_Read_Buffer_Size_Command(), check_result=True
+        )
+        print(
+            color('LE ACL Flow Control:', 'yellow'),
+            f'{response.return_parameters.hc_total_num_le_acl_data_packets} '
+            f'packets of size {response.return_parameters.hc_le_acl_data_packet_length}',
+        )
+
+
 # -----------------------------------------------------------------------------
 async def async_main(transport):
     print('<<< connecting to HCI...')
@@ -168,6 +197,9 @@ async def async_main(transport):
         # Get the LE info
         await get_le_info(host)
 
+        # Print the ACL flow control info
+        await get_acl_flow_control_info(host)
+
         # Print the list of commands supported by the controller
         print()
         print(color('Supported Commands:', 'yellow'))

apps/controller_loopback.py

@@ -0,0 +1,200 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# -----------------------------------------------------------------------------
+# Imports
+# -----------------------------------------------------------------------------
+import asyncio
+import logging
+import os
+import time
+from typing import Optional
+from bumble.colors import color
+from bumble.hci import (
+    HCI_READ_LOOPBACK_MODE_COMMAND,
+    HCI_Read_Loopback_Mode_Command,
+    HCI_WRITE_LOOPBACK_MODE_COMMAND,
+    HCI_Write_Loopback_Mode_Command,
+    LoopbackMode,
+)
+from bumble.host import Host
+from bumble.transport import open_transport_or_link
+import click
+
+
+class Loopback:
+    """Send and receive ACL data packets in local loopback mode"""
+
+    def __init__(self, packet_size: int, packet_count: int, transport: str):
+        self.transport = transport
+        self.packet_size = packet_size
+        self.packet_count = packet_count
+        self.connection_handle: Optional[int] = None
+        self.connection_event = asyncio.Event()
+        self.done = asyncio.Event()
+        self.expected_cid = 0
+        self.bytes_received = 0
+        self.start_timestamp = 0.0
+        self.last_timestamp = 0.0
+
+    def on_connection(self, connection_handle: int, *args):
+        """Retrieve connection handle from new connection event"""
+        if not self.connection_event.is_set():
+            # save first connection handle for ACL
+            # subsequent connections are SCO
+            self.connection_handle = connection_handle
+            self.connection_event.set()
+
+    def on_l2cap_pdu(self, connection_handle: int, cid: int, pdu: bytes):
+        """Calculate packet receive speed"""
+        now = time.time()
+        print(f'<<< Received packet {cid}: {len(pdu)} bytes')
+        assert connection_handle == self.connection_handle
+        assert cid == self.expected_cid
+        self.expected_cid += 1
+        if cid == 0:
+            self.start_timestamp = now
+        else:
+            elapsed_since_start = now - self.start_timestamp
+            elapsed_since_last = now - self.last_timestamp
+            self.bytes_received += len(pdu)
+            instant_rx_speed = len(pdu) / elapsed_since_last
+            average_rx_speed = self.bytes_received / elapsed_since_start
+            print(
+                color(
+                    f'@@@ RX speed: instant={instant_rx_speed:.4f},'
+                    f' average={average_rx_speed:.4f}',
+                    'cyan',
+                )
+            )
+
+        self.last_timestamp = now
+
+        if self.expected_cid == self.packet_count:
+            print(color('@@@ Received last packet', 'green'))
+            self.done.set()
+
+    async def run(self):
+        """Run a loopback throughput test"""
+        print(color('>>> Connecting to HCI...', 'green'))
+        async with await open_transport_or_link(self.transport) as (
+            hci_source,
+            hci_sink,
+        ):
+            print(color('>>> Connected', 'green'))
+
+            host = Host(hci_source, hci_sink)
+            await host.reset()
+
+            # make sure data can fit in one l2cap pdu
+            l2cap_header_size = 4
+            max_packet_size = host.acl_packet_queue.max_packet_size - l2cap_header_size
+            if self.packet_size > max_packet_size:
+                print(
+                    color(
+                        f'!!! Packet size ({self.packet_size}) larger than max supported'
+                        f' size ({max_packet_size})',
+                        'red',
+                    )
+                )
+                return
+
+            if not host.supports_command(
+                HCI_WRITE_LOOPBACK_MODE_COMMAND
+            ) or not host.supports_command(HCI_READ_LOOPBACK_MODE_COMMAND):
+                print(color('!!! Loopback mode not supported', 'red'))
+                return
+
+            # set event callbacks
+            host.on('connection', self.on_connection)
+            host.on('l2cap_pdu', self.on_l2cap_pdu)
+
+            loopback_mode = LoopbackMode.LOCAL
+
+            print(color('### Setting loopback mode', 'blue'))
+            await host.send_command(
+                HCI_Write_Loopback_Mode_Command(loopback_mode=LoopbackMode.LOCAL),
+                check_result=True,
+            )
+
+            print(color('### Checking loopback mode', 'blue'))
+            response = await host.send_command(
+                HCI_Read_Loopback_Mode_Command(), check_result=True
+            )
+            if response.return_parameters.loopback_mode != loopback_mode:
+                print(color('!!! Loopback mode mismatch', 'red'))
+                return
+
+            await self.connection_event.wait()
+            print(color('### Connected', 'cyan'))
+
+            print(color('=== Start sending', 'magenta'))
+            start_time = time.time()
+            bytes_sent = 0
+            for cid in range(0, self.packet_count):
+                # using the cid as an incremental index
+                host.send_l2cap_pdu(
+                    self.connection_handle, cid, bytes(self.packet_size)
+                )
+                print(
+                    color(
+                        f'>>> Sending packet {cid}: {self.packet_size} bytes', 'yellow'
+                    )
+                )
+                bytes_sent += self.packet_size  # don't count L2CAP or HCI header sizes
+                await asyncio.sleep(0)  # yield to allow packet receive
+
+            await self.done.wait()
+            print(color('=== Done!', 'magenta'))
+
+            elapsed = time.time() - start_time
+            average_tx_speed = bytes_sent / elapsed
+            print(
+                color(
+                    f'@@@ TX speed: average={average_tx_speed:.4f} ({bytes_sent} bytes'
+                    f' in {elapsed:.2f} seconds)',
+                    'green',
+                )
+            )
+
+
+# -----------------------------------------------------------------------------
+@click.command()
+@click.option(
+    '--packet-size',
+    '-s',
+    metavar='SIZE',
+    type=click.IntRange(8, 4096),
+    default=500,
+    help='Packet size',
+)
+@click.option(
+    '--packet-count',
+    '-c',
+    metavar='COUNT',
+    type=int,
+    default=10,
+    help='Packet count',
+)
+@click.argument('transport')
+def main(packet_size, packet_count, transport):
+    logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'WARNING').upper())
+
+    loopback = Loopback(packet_size, packet_count, transport)
+    asyncio.run(loopback.run())
+
+
+# -----------------------------------------------------------------------------
+if __name__ == '__main__':
+    main()

apps/l2cap_bridge.py

@@ -49,14 +49,16 @@ class ServerBridge:
         self.tcp_port = tcp_port
 
     async def start(self, device: Device) -> None:
-        # Listen for incoming L2CAP CoC connections
+        # Listen for incoming L2CAP channel connections
         device.create_l2cap_server(
             spec=l2cap.LeCreditBasedChannelSpec(
                 psm=self.psm, mtu=self.mtu, mps=self.mps, max_credits=self.max_credits
             ),
-            handler=self.on_coc,
+            handler=self.on_channel,
+        )
+        print(
+            color(f'### Listening for channel connection on PSM {self.psm}', 'yellow')
         )
-        print(color(f'### Listening for CoC connection on PSM {self.psm}', 'yellow'))
 
         def on_ble_connection(connection):
             def on_ble_disconnection(reason):
@@ -73,7 +75,7 @@ class ServerBridge:
         await device.start_advertising(auto_restart=True)
 
     # Called when a new L2CAP connection is established
-    def on_coc(self, l2cap_channel):
+    def on_channel(self, l2cap_channel):
         print(color('*** L2CAP channel:', 'cyan'), l2cap_channel)
 
         class Pipe:
@@ -83,7 +85,7 @@ class ServerBridge:
                 self.l2cap_channel = l2cap_channel
 
                 l2cap_channel.on('close', self.on_l2cap_close)
-                l2cap_channel.sink = self.on_coc_sdu
+                l2cap_channel.sink = self.on_channel_sdu
 
             async def connect_to_tcp(self):
                 # Connect to the TCP server
@@ -128,7 +130,7 @@ class ServerBridge:
                 if self.tcp_transport is not None:
                     self.tcp_transport.close()
 
-            def on_coc_sdu(self, sdu):
+            def on_channel_sdu(self, sdu):
                 print(color(f'<<< [L2CAP SDU]: {len(sdu)} bytes', 'cyan'))
                 if self.tcp_transport is None:
                     print(color('!!! TCP socket not open, dropping', 'red'))
@@ -183,7 +185,7 @@ class ClientBridge:
             peer_name = writer.get_extra_info('peer_name')
             print(color(f'<<< TCP connection from {peer_name}', 'magenta'))
 
-            def on_coc_sdu(sdu):
+            def on_channel_sdu(sdu):
                 print(color(f'<<< [L2CAP SDU]: {len(sdu)} bytes', 'cyan'))
                 l2cap_to_tcp_pipe.write(sdu)
 
@@ -209,7 +211,7 @@ class ClientBridge:
                 writer.close()
                 return
 
-            l2cap_channel.sink = on_coc_sdu
+            l2cap_channel.sink = on_channel_sdu
             l2cap_channel.on('close', on_l2cap_close)
 
             # Start a flow control pipe from L2CAP to TCP
@@ -274,23 +276,29 @@ async def run(device_config, hci_transport, bridge):
 @click.pass_context
 @click.option('--device-config', help='Device configuration file', required=True)
 @click.option('--hci-transport', help='HCI transport', required=True)
-@click.option('--psm', help='PSM for L2CAP CoC', type=int, default=1234)
+@click.option('--psm', help='PSM for L2CAP', type=int, default=1234)
 @click.option(
-    '--l2cap-coc-max-credits',
-    help='Maximum L2CAP CoC Credits',
+    '--l2cap-max-credits',
+    help='Maximum L2CAP Credits',
     type=click.IntRange(1, 65535),
     default=128,
 )
 @click.option(
-    '--l2cap-coc-mtu',
-    help='L2CAP CoC MTU',
-    type=click.IntRange(23, 65535),
-    default=1022,
+    '--l2cap-mtu',
+    help='L2CAP MTU',
+    type=click.IntRange(
+        l2cap.L2CAP_LE_CREDIT_BASED_CONNECTION_MIN_MTU,
+        l2cap.L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_MTU,
+    ),
+    default=1024,
 )
 @click.option(
-    '--l2cap-coc-mps',
-    help='L2CAP CoC MPS',
-    type=click.IntRange(23, 65533),
+    '--l2cap-mps',
+    help='L2CAP MPS',
+    type=click.IntRange(
+        l2cap.L2CAP_LE_CREDIT_BASED_CONNECTION_MIN_MPS,
+        l2cap.L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_MPS,
+    ),
     default=1024,
 )
 def cli(
@@ -298,17 +306,17 @@ def cli(
     device_config,
     hci_transport,
     psm,
-    l2cap_coc_max_credits,
-    l2cap_coc_mtu,
-    l2cap_coc_mps,
+    l2cap_max_credits,
+    l2cap_mtu,
+    l2cap_mps,
 ):
     context.ensure_object(dict)
     context.obj['device_config'] = device_config
     context.obj['hci_transport'] = hci_transport
     context.obj['psm'] = psm
-    context.obj['max_credits'] = l2cap_coc_max_credits
-    context.obj['mtu'] = l2cap_coc_mtu
-    context.obj['mps'] = l2cap_coc_mps
+    context.obj['max_credits'] = l2cap_max_credits
+    context.obj['mtu'] = l2cap_mtu
+    context.obj['mps'] = l2cap_mps
 
 
 # -----------------------------------------------------------------------------

bumble/controller.py

@@ -134,12 +134,14 @@ class Controller:
             '0000000060000000'
         )  # BR/EDR Not Supported, LE Supported (Controller)
         self.manufacturer_name = 0xFFFF
+        self.hc_data_packet_length = 27
+        self.hc_total_num_data_packets = 64
         self.hc_le_data_packet_length = 27
         self.hc_total_num_le_data_packets = 64
         self.event_mask = 0
         self.event_mask_page_2 = 0
         self.supported_commands = bytes.fromhex(
-            '2000800000c000000000e40000002822000000000000040000f7ffff7f000000'
+            '2000800000c000000000e4000000a822000000000000040000f7ffff7f000000'
             '30f0f9ff01008004000000000000000000000000000000000000000000000000'
         )
         self.le_event_mask = 0
@@ -914,6 +916,19 @@ class Controller:
         '''
         return bytes([HCI_SUCCESS]) + self.lmp_features
 
+    def on_hci_read_buffer_size_command(self, _command):
+        '''
+        See Bluetooth spec Vol 4, Part E - 7.4.5 Read Buffer Size Command
+        '''
+        return struct.pack(
+            '<BHBHH',
+            HCI_SUCCESS,
+            self.hc_data_packet_length,
+            0,
+            self.hc_total_num_data_packets,
+            0,
+        )
+
     def on_hci_read_bd_addr_command(self, _command):
         '''
         See Bluetooth spec Vol 4, Part E - 7.4.6 Read BD_ADDR Command

bumble/hci.py

@@ -2026,6 +2026,17 @@ class OwnAddressType(enum.IntEnum):
         return {'size': 1, 'mapper': lambda x: OwnAddressType(x).name}
 
 
+# -----------------------------------------------------------------------------
+class LoopbackMode(enum.IntEnum):
+    DISABLED = 0
+    LOCAL = 1
+    REMOTE = 2
+
+    @classmethod
+    def type_spec(cls):
+        return {'size': 1, 'mapper': lambda x: LoopbackMode(x).name}
+
+
 # -----------------------------------------------------------------------------
 class HCI_Packet:
     '''
@@ -3352,6 +3363,27 @@ class HCI_Read_Encryption_Key_Size_Command(HCI_Command):
     '''
 
 
+# -----------------------------------------------------------------------------
+@HCI_Command.command(
+    return_parameters_fields=[
+        ('status', STATUS_SPEC),
+        ('loopback_mode', LoopbackMode.type_spec()),
+    ],
+)
+class HCI_Read_Loopback_Mode_Command(HCI_Command):
+    '''
+    See Bluetooth spec @ 7.6.1 Read Loopback Mode Command
+    '''
+
+
+# -----------------------------------------------------------------------------
+@HCI_Command.command([('loopback_mode', 1)])
+class HCI_Write_Loopback_Mode_Command(HCI_Command):
+    '''
+    See Bluetooth spec @ 7.6.2 Write Loopback Mode Command
+    '''
+
+
 # -----------------------------------------------------------------------------
 @HCI_Command.command([('le_event_mask', 8)])
 class HCI_LE_Set_Event_Mask_Command(HCI_Command):

bumble/hid.py

@@ -402,7 +402,7 @@ class Device(HID):
         report_type = pdu[0] & 0x03
         buffer_flag = (pdu[0] & 0x08) >> 3
         report_id = pdu[1]
-        logger.debug("buffer_flag: " + str(buffer_flag))
+        logger.debug(f"buffer_flag: {buffer_flag}")
         if buffer_flag == 1:
             buffer_size = (pdu[3] << 8) | pdu[2]
         else:

bumble/host.py

@@ -21,7 +21,7 @@ import collections
 import logging
 import struct
 
-from typing import Any, Awaitable, Callable, Dict, Optional, Union, cast, TYPE_CHECKING
+from typing import Any, Awaitable, Callable, Deque, Dict, Optional, cast, TYPE_CHECKING
 
 from bumble.colors import color
 from bumble.l2cap import L2CAP_PDU
@@ -91,16 +91,49 @@ logger = logging.getLogger(__name__)
 
 
 # -----------------------------------------------------------------------------
-# Constants
-# -----------------------------------------------------------------------------
-# fmt: off
+class AclPacketQueue:
+    max_packet_size: int
 
-HOST_DEFAULT_HC_LE_ACL_DATA_PACKET_LENGTH = 27
-HOST_HC_TOTAL_NUM_LE_ACL_DATA_PACKETS     = 1
-HOST_DEFAULT_HC_ACL_DATA_PACKET_LENGTH    = 27
-HOST_HC_TOTAL_NUM_ACL_DATA_PACKETS        = 1
+    def __init__(
+        self,
+        max_packet_size: int,
+        max_in_flight: int,
+        send: Callable[[HCI_Packet], None],
+    ) -> None:
+        self.max_packet_size = max_packet_size
+        self.max_in_flight = max_in_flight
+        self.in_flight = 0
+        self.send = send
+        self.packets: Deque[HCI_AclDataPacket] = collections.deque()
 
-# fmt: on
+    def enqueue(self, packet: HCI_AclDataPacket) -> None:
+        self.packets.appendleft(packet)
+        self.check_queue()
+
+        if self.packets:
+            logger.debug(
+                f'{self.in_flight} ACL packets in flight, '
+                f'{len(self.packets)} in queue'
+            )
+
+    def check_queue(self) -> None:
+        while self.packets and self.in_flight < self.max_in_flight:
+            packet = self.packets.pop()
+            self.send(packet)
+            self.in_flight += 1
+
+    def on_packets_completed(self, packet_count: int) -> None:
+        if packet_count > self.in_flight:
+            logger.warning(
+                color(
+                    '!!! {packet_count} completed but only '
+                    f'{self.in_flight} in flight'
+                )
+            )
+            packet_count = self.in_flight
+
+        self.in_flight -= packet_count
+        self.check_queue()
 
 
 # -----------------------------------------------------------------------------
@@ -111,6 +144,13 @@ class Connection:
         self.peer_address = peer_address
         self.assembler = HCI_AclDataPacketAssembler(self.on_acl_pdu)
         self.transport = transport
+        acl_packet_queue: Optional[AclPacketQueue] = (
+            host.le_acl_packet_queue
+            if transport == BT_LE_TRANSPORT
+            else host.acl_packet_queue
+        )
+        assert acl_packet_queue
+        self.acl_packet_queue = acl_packet_queue
 
     def on_hci_acl_data_packet(self, packet: HCI_AclDataPacket) -> None:
         self.assembler.feed_packet(packet)
@@ -123,7 +163,8 @@ class Connection:
 # -----------------------------------------------------------------------------
 class Host(AbortableEventEmitter):
     connections: Dict[int, Connection]
-    acl_packet_queue: collections.deque[HCI_AclDataPacket]
+    acl_packet_queue: Optional[AclPacketQueue] = None
+    le_acl_packet_queue: Optional[AclPacketQueue] = None
     hci_sink: Optional[TransportSink] = None
     hci_metadata: Dict[str, Any]
     long_term_key_provider: Optional[
@@ -143,12 +184,6 @@ class Host(AbortableEventEmitter):
         self.connections = {}  # Connections, by connection handle
         self.pending_command = None
         self.pending_response = None
-        self.hc_le_acl_data_packet_length = HOST_DEFAULT_HC_LE_ACL_DATA_PACKET_LENGTH
-        self.hc_total_num_le_acl_data_packets = HOST_HC_TOTAL_NUM_LE_ACL_DATA_PACKETS
-        self.hc_acl_data_packet_length = HOST_DEFAULT_HC_ACL_DATA_PACKET_LENGTH
-        self.hc_total_num_acl_data_packets = HOST_HC_TOTAL_NUM_ACL_DATA_PACKETS
-        self.acl_packet_queue = collections.deque()
-        self.acl_packets_in_flight = 0
         self.local_version = None
         self.local_supported_commands = bytes(64)
         self.local_le_features = 0
@@ -254,46 +289,54 @@ class Host(AbortableEventEmitter):
             response = await self.send_command(
                 HCI_Read_Buffer_Size_Command(), check_result=True
             )
-            self.hc_acl_data_packet_length = (
+            hc_acl_data_packet_length = (
                 response.return_parameters.hc_acl_data_packet_length
             )
-            self.hc_total_num_acl_data_packets = (
+            hc_total_num_acl_data_packets = (
                 response.return_parameters.hc_total_num_acl_data_packets
             )
 
             logger.debug(
                 'HCI ACL flow control: '
-                f'hc_acl_data_packet_length={self.hc_acl_data_packet_length},'
-                f'hc_total_num_acl_data_packets={self.hc_total_num_acl_data_packets}'
+                f'hc_acl_data_packet_length={hc_acl_data_packet_length},'
+                f'hc_total_num_acl_data_packets={hc_total_num_acl_data_packets}'
             )
 
+            self.acl_packet_queue = AclPacketQueue(
+                max_packet_size=hc_acl_data_packet_length,
+                max_in_flight=hc_total_num_acl_data_packets,
+                send=self.send_hci_packet,
+            )
+
+        hc_le_acl_data_packet_length = 0
+        hc_total_num_le_acl_data_packets = 0
         if self.supports_command(HCI_LE_READ_BUFFER_SIZE_COMMAND):
             response = await self.send_command(
                 HCI_LE_Read_Buffer_Size_Command(), check_result=True
             )
-            self.hc_le_acl_data_packet_length = (
+            hc_le_acl_data_packet_length = (
                 response.return_parameters.hc_le_acl_data_packet_length
             )
-            self.hc_total_num_le_acl_data_packets = (
+            hc_total_num_le_acl_data_packets = (
                 response.return_parameters.hc_total_num_le_acl_data_packets
             )
 
             logger.debug(
                 'HCI LE ACL flow control: '
-                f'hc_le_acl_data_packet_length={self.hc_le_acl_data_packet_length},'
-                'hc_total_num_le_acl_data_packets='
-                f'{self.hc_total_num_le_acl_data_packets}'
+                f'hc_le_acl_data_packet_length={hc_le_acl_data_packet_length},'
+                f'hc_total_num_le_acl_data_packets={hc_total_num_le_acl_data_packets}'
             )
 
-            if (
-                response.return_parameters.hc_le_acl_data_packet_length == 0
-                or response.return_parameters.hc_total_num_le_acl_data_packets == 0
-            ):
-                # LE and Classic share the same values
-                self.hc_le_acl_data_packet_length = self.hc_acl_data_packet_length
-                self.hc_total_num_le_acl_data_packets = (
-                    self.hc_total_num_acl_data_packets
-                )
+        if hc_le_acl_data_packet_length == 0 or hc_total_num_le_acl_data_packets == 0:
+            # LE and Classic share the same queue
+            self.le_acl_packet_queue = self.acl_packet_queue
+        else:
+            # Create a separate queue for LE
+            self.le_acl_packet_queue = AclPacketQueue(
+                max_packet_size=hc_le_acl_data_packet_length,
+                max_in_flight=hc_total_num_le_acl_data_packets,
+                send=self.send_hci_packet,
+            )
 
         if self.supports_command(
             HCI_LE_READ_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND
@@ -332,14 +375,13 @@ class Host(AbortableEventEmitter):
         self.hci_metadata = getattr(source, 'metadata', self.hci_metadata)
 
     def send_hci_packet(self, packet: HCI_Packet) -> None:
+        logger.debug(f'{color("### HOST -> CONTROLLER", "blue")}: {packet}')
         if self.snooper:
             self.snooper.snoop(bytes(packet), Snooper.Direction.HOST_TO_CONTROLLER)
         if self.hci_sink:
             self.hci_sink.on_packet(bytes(packet))
 
     async def send_command(self, command, check_result=False):
-        logger.debug(f'{color("### HOST -> CONTROLLER", "blue")}: {command}')
-
         # Wait until we can send (only one pending command at a time)
         async with self.command_semaphore:
             assert self.pending_command is None
@@ -387,6 +429,17 @@ class Host(AbortableEventEmitter):
         asyncio.create_task(send_command(command))
 
     def send_l2cap_pdu(self, connection_handle: int, cid: int, pdu: bytes) -> None:
+        if not (connection := self.connections.get(connection_handle)):
+            logger.warning(f'connection 0x{connection_handle:04X} not found')
+            return
+        packet_queue = connection.acl_packet_queue
+        if packet_queue is None:
+            logger.warning(
+                f'no ACL packet queue for connection 0x{connection_handle:04X}'
+            )
+            return
+
+        # Create a PDU
         l2cap_pdu = bytes(L2CAP_PDU(cid, pdu))
 
         # Send the data to the controller via ACL packets
@@ -394,8 +447,7 @@ class Host(AbortableEventEmitter):
         offset = 0
         pb_flag = 0
         while bytes_remaining:
-            # TODO: support different LE/Classic lengths
-            data_total_length = min(bytes_remaining, self.hc_le_acl_data_packet_length)
+            data_total_length = min(bytes_remaining, packet_queue.max_packet_size)
             acl_packet = HCI_AclDataPacket(
                 connection_handle=connection_handle,
                 pb_flag=pb_flag,
@@ -403,34 +455,12 @@ class Host(AbortableEventEmitter):
                 data_total_length=data_total_length,
                 data=l2cap_pdu[offset : offset + data_total_length],
             )
-            logger.debug(
-                f'{color("### HOST -> CONTROLLER", "blue")}: (CID={cid}) {acl_packet}'
-            )
-            self.queue_acl_packet(acl_packet)
+            logger.debug(f'>>> ACL packet enqueue: (CID={cid}) {acl_packet}')
+            packet_queue.enqueue(acl_packet)
             pb_flag = 1
             offset += data_total_length
             bytes_remaining -= data_total_length
 
-    def queue_acl_packet(self, acl_packet: HCI_AclDataPacket) -> None:
-        self.acl_packet_queue.appendleft(acl_packet)
-        self.check_acl_packet_queue()
-
-        if len(self.acl_packet_queue):
-            logger.debug(
-                f'{self.acl_packets_in_flight} ACL packets in flight, '
-                f'{len(self.acl_packet_queue)} in queue'
-            )
-
-    def check_acl_packet_queue(self) -> None:
-        # Send all we can (TODO: support different LE/Classic limits)
-        while (
-            len(self.acl_packet_queue) > 0
-            and self.acl_packets_in_flight < self.hc_total_num_le_acl_data_packets
-        ):
-            packet = self.acl_packet_queue.pop()
-            self.send_hci_packet(packet)
-            self.acl_packets_in_flight += 1
-
     def supports_command(self, command):
         # Find the support flag position for this command
         for octet, flags in enumerate(HCI_SUPPORTED_COMMANDS_FLAGS):
@@ -553,7 +583,7 @@ class Host(AbortableEventEmitter):
             # This is used just for the Num_HCI_Command_Packets field, not related to
             # an actual command
             logger.debug('no-command event')
-            return None
+            return
 
         return self.on_command_processed(event)
 
@@ -561,18 +591,17 @@ class Host(AbortableEventEmitter):
         return self.on_command_processed(event)
 
     def on_hci_number_of_completed_packets_event(self, event):
-        total_packets = sum(event.num_completed_packets)
-        if total_packets <= self.acl_packets_in_flight:
-            self.acl_packets_in_flight -= total_packets
-            self.check_acl_packet_queue()
-        else:
-            logger.warning(
-                color(
-                    '!!! {total_packets} completed but only '
-                    f'{self.acl_packets_in_flight} in flight'
+        for connection_handle, num_completed_packets in zip(
+            event.connection_handles, event.num_completed_packets
+        ):
+            if not (connection := self.connections.get(connection_handle)):
+                logger.warning(
+                    'received packet completion event for unknown handle '
+                    f'0x{connection_handle:04X}'
                 )
-            )
-            self.acl_packets_in_flight = 0
+                continue
+
+            connection.acl_packet_queue.on_packets_completed(num_completed_packets)
 
     # Classic only
     def on_hci_connection_request_event(self, event):

bumble/l2cap.py

@@ -149,9 +149,10 @@ L2CAP_INVALID_CID_IN_REQUEST_REASON = 0x0002
 
 L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_CREDITS             = 65535
 L2CAP_LE_CREDIT_BASED_CONNECTION_MIN_MTU                 = 23
+L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_MTU                 = 65535
 L2CAP_LE_CREDIT_BASED_CONNECTION_MIN_MPS                 = 23
 L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_MPS                 = 65533
-L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MTU             = 2046
+L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MTU             = 2048
 L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MPS             = 2048
 L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_INITIAL_CREDITS = 256
 
@@ -188,8 +189,11 @@ class LeCreditBasedChannelSpec:
             or self.max_credits > L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_CREDITS
         ):
             raise ValueError('max credits out of range')
-        if self.mtu < L2CAP_LE_CREDIT_BASED_CONNECTION_MIN_MTU:
-            raise ValueError('MTU too small')
+        if (
+            self.mtu < L2CAP_LE_CREDIT_BASED_CONNECTION_MIN_MTU
+            or self.mtu > L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_MTU
+        ):
+            raise ValueError('MTU out of range')
         if (
             self.mps < L2CAP_LE_CREDIT_BASED_CONNECTION_MIN_MPS
             or self.mps > L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_MPS
@@ -1644,12 +1648,13 @@ class ChannelManager:
 
     def send_pdu(self, connection, cid: int, pdu: Union[SupportsBytes, bytes]) -> None:
         pdu_str = pdu.hex() if isinstance(pdu, bytes) else str(pdu)
+        pdu_bytes = bytes(pdu)
         logger.debug(
             f'{color(">>> Sending L2CAP PDU", "blue")} '
             f'on connection [0x{connection.handle:04X}] (CID={cid}) '
-            f'{connection.peer_address}: {pdu_str}'
+            f'{connection.peer_address}: {len(pdu_bytes)} bytes, {pdu_str}'
         )
-        self.host.send_l2cap_pdu(connection.handle, cid, bytes(pdu))
+        self.host.send_l2cap_pdu(connection.handle, cid, pdu_bytes)
 
     def on_pdu(self, connection: Connection, cid: int, pdu: bytes) -> None:
         if cid in (L2CAP_SIGNALING_CID, L2CAP_LE_SIGNALING_CID):

bumble/profiles/csip.py

@@ -19,7 +19,7 @@
 from __future__ import annotations
 import enum
 import struct
-from typing import Optional
+from typing import Optional, Tuple
 
 from bumble import core
 from bumble import crypto
@@ -31,6 +31,9 @@ from bumble import gatt_client
 # -----------------------------------------------------------------------------
 # Constants
 # -----------------------------------------------------------------------------
+SET_IDENTITY_RESOLVING_KEY_LENGTH = 16
+
+
 class SirkType(enum.IntEnum):
     '''Coordinated Set Identification Service - 5.1 Set Identity Resolving Key.'''
 
@@ -66,6 +69,10 @@ def k1(n: bytes, salt: bytes, p: bytes) -> bytes:
 def sef(k: bytes, r: bytes) -> bytes:
     '''
     Coordinated Set Identification Service - 4.5 SIRK encryption function sef.
+
+    SIRK decryption function sdf shares the same algorithm. The only difference is that argument r is:
+      * Plaintext in encryption
+      * Cipher in decryption
     '''
     return crypto.xor(k1(k, s1(b'SIRKenc'[::-1]), b'csis'[::-1]), r)
 
@@ -105,6 +112,11 @@ class CoordinatedSetIdentificationService(gatt.TemplateService):
         set_member_lock: Optional[MemberLock] = None,
         set_member_rank: Optional[int] = None,
     ) -> None:
+        if len(set_identity_resolving_key) != SET_IDENTITY_RESOLVING_KEY_LENGTH:
+            raise ValueError(
+                f'Invalid SIRK length {len(set_identity_resolving_key)}, expected {SET_IDENTITY_RESOLVING_KEY_LENGTH}'
+            )
+
         characteristics = []
 
         self.set_identity_resolving_key = set_identity_resolving_key
@@ -113,7 +125,7 @@ class CoordinatedSetIdentificationService(gatt.TemplateService):
             uuid=gatt.GATT_SET_IDENTITY_RESOLVING_KEY_CHARACTERISTIC,
             properties=gatt.Characteristic.Properties.READ
             | gatt.Characteristic.Properties.NOTIFY,
-            permissions=gatt.Characteristic.Permissions.READABLE,
+            permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
             value=gatt.CharacteristicValue(read=self.on_sirk_read),
         )
         characteristics.append(self.set_identity_resolving_key_characteristic)
@@ -123,7 +135,7 @@ class CoordinatedSetIdentificationService(gatt.TemplateService):
                 uuid=gatt.GATT_COORDINATED_SET_SIZE_CHARACTERISTIC,
                 properties=gatt.Characteristic.Properties.READ
                 | gatt.Characteristic.Properties.NOTIFY,
-                permissions=gatt.Characteristic.Permissions.READABLE,
+                permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
                 value=struct.pack('B', coordinated_set_size),
             )
             characteristics.append(self.coordinated_set_size_characteristic)
@@ -134,7 +146,7 @@ class CoordinatedSetIdentificationService(gatt.TemplateService):
                 properties=gatt.Characteristic.Properties.READ
                 | gatt.Characteristic.Properties.NOTIFY
                 | gatt.Characteristic.Properties.WRITE,
-                permissions=gatt.Characteristic.Permissions.READABLE
+                permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION
                 | gatt.Characteristic.Permissions.WRITEABLE,
                 value=struct.pack('B', set_member_lock),
             )
@@ -145,18 +157,32 @@ class CoordinatedSetIdentificationService(gatt.TemplateService):
                 uuid=gatt.GATT_SET_MEMBER_RANK_CHARACTERISTIC,
                 properties=gatt.Characteristic.Properties.READ
                 | gatt.Characteristic.Properties.NOTIFY,
-                permissions=gatt.Characteristic.Permissions.READABLE,
+                permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
                 value=struct.pack('B', set_member_rank),
             )
             characteristics.append(self.set_member_rank_characteristic)
 
         super().__init__(characteristics)
 
-    def on_sirk_read(self, _connection: Optional[device.Connection]) -> bytes:
+    async def on_sirk_read(self, connection: Optional[device.Connection]) -> bytes:
         if self.set_identity_resolving_key_type == SirkType.PLAINTEXT:
-            return bytes([SirkType.PLAINTEXT]) + self.set_identity_resolving_key
+            sirk_bytes = self.set_identity_resolving_key
         else:
-            raise NotImplementedError('TODO: Pending async Characteristic read.')
+            assert connection
+
+            if connection.transport == core.BT_LE_TRANSPORT:
+                key = await connection.device.get_long_term_key(
+                    connection_handle=connection.handle, rand=b'', ediv=0
+                )
+            else:
+                key = await connection.device.get_link_key(connection.peer_address)
+
+            if not key:
+                raise RuntimeError('LTK or LinkKey is not present')
+
+            sirk_bytes = sef(key, self.set_identity_resolving_key)
+
+        return bytes([self.set_identity_resolving_key_type]) + sirk_bytes
 
     def get_advertising_data(self) -> bytes:
         return bytes(
@@ -203,3 +229,29 @@ class CoordinatedSetIdentificationProxy(gatt_client.ProfileServiceProxy):
             gatt.GATT_SET_MEMBER_RANK_CHARACTERISTIC
         ):
             self.set_member_rank = characteristics[0]
+
+    async def read_set_identity_resolving_key(self) -> Tuple[SirkType, bytes]:
+        '''Reads SIRK and decrypts if encrypted.'''
+        response = await self.set_identity_resolving_key.read_value()
+        if len(response) != SET_IDENTITY_RESOLVING_KEY_LENGTH + 1:
+            raise RuntimeError('Invalid SIRK value')
+
+        sirk_type = SirkType(response[0])
+        if sirk_type == SirkType.PLAINTEXT:
+            sirk = response[1:]
+        else:
+            connection = self.service_proxy.client.connection
+            device = connection.device
+            if connection.transport == core.BT_LE_TRANSPORT:
+                key = await device.get_long_term_key(
+                    connection_handle=connection.handle, rand=b'', ediv=0
+                )
+            else:
+                key = await device.get_link_key(connection.peer_address)
+
+            if not key:
+                raise RuntimeError('LTK or LinkKey is not present')
+
+            sirk = sef(key, response[1:])
+
+        return (sirk_type, sirk)

bumble/rfcomm.py

@@ -118,8 +118,8 @@ CRC_TABLE = bytes([
     0XBA, 0X2B, 0X59, 0XC8, 0XBD, 0X2C, 0X5E, 0XCF
 ])
 
-RFCOMM_DEFAULT_INITIAL_RX_CREDITS = 7
-RFCOMM_DEFAULT_PREFERRED_MTU      = 1280
+RFCOMM_DEFAULT_WINDOW_SIZE    = 16
+RFCOMM_DEFAULT_MAX_FRAME_SIZE = 2000
 
 RFCOMM_DYNAMIC_CHANNEL_NUMBER_START = 1
 RFCOMM_DYNAMIC_CHANNEL_NUMBER_END   = 30
@@ -438,20 +438,24 @@ class DLC(EventEmitter):
         multiplexer: Multiplexer,
         dlci: int,
         max_frame_size: int,
-        initial_tx_credits: int,
+        window_size: int,
     ) -> None:
         super().__init__()
         self.multiplexer = multiplexer
         self.dlci = dlci
-        self.rx_credits = RFCOMM_DEFAULT_INITIAL_RX_CREDITS
-        self.rx_threshold = self.rx_credits // 2
-        self.tx_credits = initial_tx_credits
+        self.max_frame_size = max_frame_size
+        self.window_size = window_size
+        self.rx_credits = window_size
+        self.rx_threshold = window_size // 2
+        self.tx_credits = window_size
         self.tx_buffer = b''
         self.state = DLC.State.INIT
         self.role = multiplexer.role
         self.c_r = 1 if self.role == Multiplexer.Role.INITIATOR else 0
         self.sink = None
         self.connection_result = None
+        self.drained = asyncio.Event()
+        self.drained.set()
 
         # Compute the MTU
         max_overhead = 4 + 1  # header with 2-byte length + fcs
@@ -537,11 +541,11 @@ class DLC(EventEmitter):
         if len(data) and self.sink:
             self.sink(data)  # pylint: disable=not-callable
 
-        # Update the credits
-        if self.rx_credits > 0:
-            self.rx_credits -= 1
-        else:
-            logger.warning(color('!!! received frame with no rx credits', 'red'))
+            # Update the credits
+            if self.rx_credits > 0:
+                self.rx_credits -= 1
+            else:
+                logger.warning(color('!!! received frame with no rx credits', 'red'))
 
         # Check if there's anything to send (including credits)
         self.process_tx()
@@ -580,9 +584,9 @@ class DLC(EventEmitter):
             cl=0xE0,
             priority=7,
             ack_timer=0,
-            max_frame_size=RFCOMM_DEFAULT_PREFERRED_MTU,
+            max_frame_size=self.max_frame_size,
             max_retransmissions=0,
-            window_size=RFCOMM_DEFAULT_INITIAL_RX_CREDITS,
+            window_size=self.window_size,
         )
         mcc = RFCOMM_Frame.make_mcc(mcc_type=RFCOMM_MCC_PN_TYPE, c_r=0, data=bytes(pn))
         logger.debug(f'>>> PN Response: {pn}')
@@ -591,7 +595,7 @@ class DLC(EventEmitter):
 
     def rx_credits_needed(self) -> int:
         if self.rx_credits <= self.rx_threshold:
-            return RFCOMM_DEFAULT_INITIAL_RX_CREDITS - self.rx_credits
+            return self.window_size - self.rx_credits
 
         return 0
 
@@ -631,6 +635,8 @@ class DLC(EventEmitter):
             )
 
             rx_credits_needed = 0
+            if not self.tx_buffer:
+                self.drained.set()
 
     # Stream protocol
     def write(self, data: Union[bytes, str]) -> None:
@@ -643,11 +649,11 @@ class DLC(EventEmitter):
                 raise ValueError('write only accept bytes or strings')
 
         self.tx_buffer += data
+        self.drained.clear()
         self.process_tx()
 
-    def drain(self) -> None:
-        # TODO
-        pass
+    async def drain(self) -> None:
+        await self.drained.wait()
 
     def __str__(self) -> str:
         return f'DLC(dlci={self.dlci},state={self.state.name})'
@@ -843,7 +849,12 @@ class Multiplexer(EventEmitter):
         )
         await self.disconnection_result
 
-    async def open_dlc(self, channel: int) -> DLC:
+    async def open_dlc(
+        self,
+        channel: int,
+        max_frame_size: int = RFCOMM_DEFAULT_MAX_FRAME_SIZE,
+        window_size: int = RFCOMM_DEFAULT_WINDOW_SIZE,
+    ) -> DLC:
         if self.state != Multiplexer.State.CONNECTED:
             if self.state == Multiplexer.State.OPENING:
                 raise InvalidStateError('open already in progress')
@@ -855,9 +866,9 @@ class Multiplexer(EventEmitter):
             cl=0xF0,
             priority=7,
             ack_timer=0,
-            max_frame_size=RFCOMM_DEFAULT_PREFERRED_MTU,
+            max_frame_size=max_frame_size,
             max_retransmissions=0,
-            window_size=RFCOMM_DEFAULT_INITIAL_RX_CREDITS,
+            window_size=window_size,
         )
         mcc = RFCOMM_Frame.make_mcc(mcc_type=RFCOMM_MCC_PN_TYPE, c_r=1, data=bytes(pn))
         logger.debug(f'>>> Sending MCC: {pn}')

bumble/transport/usb.py

@@ -108,7 +108,7 @@ async def open_usb_transport(spec: str) -> Transport:
         USB_DEVICE_PROTOCOL_BLUETOOTH_PRIMARY_CONTROLLER,
     )
 
-    READ_SIZE = 1024
+    READ_SIZE = 4096
 
     class UsbPacketSink:
         def __init__(self, device, acl_out):

docs/mkdocs/src/apps_and_tools/bench.md

@@ -7,16 +7,36 @@ throughput and/or latency between two devices.
 # General Usage
 
 ```
-Usage: bench.py [OPTIONS] COMMAND [ARGS]...
+Usage: bumble-bench [OPTIONS] COMMAND [ARGS]...
 
 Options:
   --device-config FILENAME        Device configuration file
   --role [sender|receiver|ping|pong]
   --mode [gatt-client|gatt-server|l2cap-client|l2cap-server|rfcomm-client|rfcomm-server]
   --att-mtu MTU                   GATT MTU (gatt-client mode)  [23<=x<=517]
-  -s, --packet-size SIZE          Packet size (server role)  [8<=x<=4096]
-  -c, --packet-count COUNT        Packet count (server role)
-  -sd, --start-delay SECONDS      Start delay (server role)
+  --extended-data-length TEXT     Request a data length upon connection,
+                                  specified as tx_octets/tx_time
+  --rfcomm-channel INTEGER        RFComm channel to use
+  --rfcomm-uuid TEXT              RFComm service UUID to use (ignored if
+                                  --rfcomm-channel is not 0)
+  --l2cap-psm INTEGER             L2CAP PSM to use
+  --l2cap-mtu INTEGER             L2CAP MTU to use
+  --l2cap-mps INTEGER             L2CAP MPS to use
+  --l2cap-max-credits INTEGER     L2CAP maximum number of credits allowed for
+                                  the peer
+  -s, --packet-size SIZE          Packet size (client or ping role)
+                                  [8<=x<=4096]
+  -c, --packet-count COUNT        Packet count (client or ping role)
+  -sd, --start-delay SECONDS      Start delay (client or ping role)
+  --repeat N                      Repeat the run N times (client and ping
+                                  roles)(0, which is the fault, to run just
+                                  once)
+  --repeat-delay SECONDS          Delay, in seconds, between repeats
+  --pace MILLISECONDS             Wait N milliseconds between packets (0,
+                                  which is the fault, to send as fast as
+                                  possible)
+  --linger                        Don't exit at the end of a run (server and
+                                  pong roles)
   --help                          Show this message and exit.
 
 Commands:
@@ -35,17 +55,18 @@ Options:
   --connection-interval, --ci CONNECTION_INTERVAL
                                   Connection interval (in ms)
   --phy [1m|2m|coded]             PHY to use
+  --authenticate                  Authenticate (RFComm only)
+  --encrypt                       Encrypt the connection (RFComm only)
   --help                          Show this message and exit.
 ```
 
-
-To test once device against another, one of the two devices must be running 
+To test once device against another, one of the two devices must be running
 the ``peripheral`` command and the other the ``central`` command. The device
 running the ``peripheral`` command will accept connections from the device
 running the ``central`` command.
 When using Bluetooth LE (all modes except for ``rfcomm-server`` and ``rfcomm-client``utils),
-the default addresses configured in the tool should be sufficient. But when using 
-Bluetooth Classic, the address of the Peripheral must be specified on the Central 
+the default addresses configured in the tool should be sufficient. But when using
+Bluetooth Classic, the address of the Peripheral must be specified on the Central
 using the ``--peripheral`` option. The address will be printed by the Peripheral when
 it starts.
 
@@ -83,7 +104,7 @@ the other on `usb:1`, and two consoles/terminals. We will run a command in each.
     $ bumble-bench central usb:1
     ```
 
-    In this default configuration, the Central runs a Sender, as a GATT client, 
+    In this default configuration, the Central runs a Sender, as a GATT client,
     connecting to the Peripheral running a Receiver, as a GATT server.
 
 !!! example "L2CAP Throughput"

examples/run_hid_device.py

@@ -590,12 +590,12 @@ async def main():
     def on_set_protocol_cb(protocol: int):
         retValue = hid_device.GetSetStatus()
         # We do not support SET_PROTOCOL.
-        print("SET_PROTOCOL report_id: " + str(protocol))
+        print(f"SET_PROTOCOL report_id: {protocol}")
         retValue.status = hid_device.GetSetReturn.ERR_UNSUPPORTED_REQUEST
         return retValue
 
     def on_virtual_cable_unplug_cb():
-        print(f'Received Virtual Cable Unplug')
+        print('Received Virtual Cable Unplug')
         asyncio.create_task(handle_virtual_cable_unplug())
 
     print('<<< connecting to HCI...')

extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/L2capClient.kt

@@ -28,8 +28,8 @@ private val Log = Logger.getLogger("btbench.l2cap-client")
 
 class L2capClient(
     private val viewModel: AppViewModel,
-    val bluetoothAdapter: BluetoothAdapter,
-    val context: Context
+    private val bluetoothAdapter: BluetoothAdapter,
+    private val context: Context
 ) {
     @SuppressLint("MissingPermission")
     fun run() {
@@ -74,12 +74,18 @@ class L2capClient(
                     gatt: BluetoothGatt?, status: Int, newState: Int
                 ) {
                     if (gatt != null && newState == BluetoothProfile.STATE_CONNECTED) {
-                        gatt.setPreferredPhy(
-                            BluetoothDevice.PHY_LE_2M_MASK,
-                            BluetoothDevice.PHY_LE_2M_MASK,
-                            BluetoothDevice.PHY_OPTION_NO_PREFERRED
-                        )
+                        if (viewModel.use2mPhy) {
+                            gatt.setPreferredPhy(
+                                BluetoothDevice.PHY_LE_2M_MASK,
+                                BluetoothDevice.PHY_LE_2M_MASK,
+                                BluetoothDevice.PHY_OPTION_NO_PREFERRED
+                            )
+                        }
                         gatt.readPhy()
+
+                        // Request an MTU update, even though we don't use GATT, because Android
+                        // won't request a larger link layer maximum data length otherwise.
+                        gatt.requestMtu(517)
                     }
                 }
             },

extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Model.kt

@@ -23,19 +23,20 @@ import androidx.compose.runtime.setValue
 import androidx.lifecycle.ViewModel
 import java.util.UUID
 
-val DEFAULT_RFCOMM_UUID = UUID.fromString("E6D55659-C8B4-4B85-96BB-B1143AF6D3AE")
+val DEFAULT_RFCOMM_UUID: UUID = UUID.fromString("E6D55659-C8B4-4B85-96BB-B1143AF6D3AE")
 const val DEFAULT_PEER_BLUETOOTH_ADDRESS = "AA:BB:CC:DD:EE:FF"
 const val DEFAULT_SENDER_PACKET_COUNT = 100
 const val DEFAULT_SENDER_PACKET_SIZE = 1024
+const val DEFAULT_PSM = 128
 
 class AppViewModel : ViewModel() {
     private var preferences: SharedPreferences? = null
     var peerBluetoothAddress by mutableStateOf(DEFAULT_PEER_BLUETOOTH_ADDRESS)
-    var l2capPsm by mutableStateOf(0)
+    var l2capPsm by mutableIntStateOf(DEFAULT_PSM)
     var use2mPhy by mutableStateOf(true)
-    var mtu by mutableStateOf(0)
-    var rxPhy by mutableStateOf(0)
-    var txPhy by mutableStateOf(0)
+    var mtu by mutableIntStateOf(0)
+    var rxPhy by mutableIntStateOf(0)
+    var txPhy by mutableIntStateOf(0)
     var senderPacketCountSlider by mutableFloatStateOf(0.0F)
     var senderPacketSizeSlider by mutableFloatStateOf(0.0F)
     var senderPacketCount by mutableIntStateOf(DEFAULT_SENDER_PACKET_COUNT)
@@ -79,18 +80,18 @@ class AppViewModel : ViewModel() {
     }
 
     fun updateSenderPacketCountSlider() {
-        if (senderPacketCount <= 10) {
-            senderPacketCountSlider = 0.0F
+        senderPacketCountSlider = if (senderPacketCount <= 10) {
+            0.0F
         } else if (senderPacketCount <= 50) {
-            senderPacketCountSlider = 0.2F
+            0.2F
         } else if (senderPacketCount <= 100) {
-            senderPacketCountSlider = 0.4F
+            0.4F
         } else if (senderPacketCount <= 500) {
-            senderPacketCountSlider = 0.6F
+            0.6F
         } else if (senderPacketCount <= 1000) {
-            senderPacketCountSlider = 0.8F
+            0.8F
         } else {
-            senderPacketCountSlider = 1.0F
+            1.0F
         }
 
         with(preferences!!.edit()) {
@@ -100,18 +101,18 @@ class AppViewModel : ViewModel() {
     }
 
     fun updateSenderPacketCount() {
-        if (senderPacketCountSlider < 0.1F) {
-            senderPacketCount = 10
+        senderPacketCount = if (senderPacketCountSlider < 0.1F) {
+            10
         } else if (senderPacketCountSlider < 0.3F) {
-            senderPacketCount = 50
+            50
         } else if (senderPacketCountSlider < 0.5F) {
-            senderPacketCount = 100
+            100
         } else if (senderPacketCountSlider < 0.7F) {
-            senderPacketCount = 500
+            500
         } else if (senderPacketCountSlider < 0.9F) {
-            senderPacketCount = 1000
+            1000
         } else {
-            senderPacketCount = 10000
+            10000
         }
 
         with(preferences!!.edit()) {
@@ -121,18 +122,18 @@ class AppViewModel : ViewModel() {
     }
 
     fun updateSenderPacketSizeSlider() {
-        if (senderPacketSize <= 16) {
-            senderPacketSizeSlider = 0.0F
+        senderPacketSizeSlider = if (senderPacketSize <= 16) {
+            0.0F
         } else if (senderPacketSize <= 256) {
-            senderPacketSizeSlider = 0.02F
+            0.02F
         } else if (senderPacketSize <= 512) {
-            senderPacketSizeSlider = 0.4F
+            0.4F
         } else if (senderPacketSize <= 1024) {
-            senderPacketSizeSlider = 0.6F
+            0.6F
         } else if (senderPacketSize <= 2048) {
-            senderPacketSizeSlider = 0.8F
+            0.8F
         } else {
-            senderPacketSizeSlider = 1.0F
+            1.0F
         }
 
         with(preferences!!.edit()) {
@@ -142,18 +143,18 @@ class AppViewModel : ViewModel() {
     }
 
     fun updateSenderPacketSize() {
-        if (senderPacketSizeSlider < 0.1F) {
-            senderPacketSize = 16
+        senderPacketSize = if (senderPacketSizeSlider < 0.1F) {
+            16
         } else if (senderPacketSizeSlider < 0.3F) {
-            senderPacketSize = 256
+            256
         } else if (senderPacketSizeSlider < 0.5F) {
-            senderPacketSize = 512
+            512
         } else if (senderPacketSizeSlider < 0.7F) {
-            senderPacketSize = 1024
+            1024
         } else if (senderPacketSizeSlider < 0.9F) {
-            senderPacketSize = 2048
+            2048
         } else {
-            senderPacketSize = 4096
+            4096
         }
 
         with(preferences!!.edit()) {

extras/android/RemoteHCI/app/src/main/java/com/github/google/bumble/remotehci/HciServer.java

@@ -42,6 +42,7 @@ public class HciServer {
         try (ServerSocket serverSocket = new ServerSocket(mPort)) {
             mListener.onMessage("Waiting for connection on port " + serverSocket.getLocalPort());
             try (Socket clientSocket = serverSocket.accept()) {
+                clientSocket.setTcpNoDelay(true);
                 mListener.onHostConnectionState(true);
                 mListener.onMessage("Connected");
                 HciParser parser = new HciParser(mListener);

tests/csip_test.py

@@ -20,6 +20,7 @@ import os
 import pytest
 import struct
 import logging
+from unittest import mock
 
 from bumble import device
 from bumble.profiles import csip
@@ -68,14 +69,18 @@ def test_sef():
 
 # -----------------------------------------------------------------------------
 @pytest.mark.asyncio
-async def test_csis():
+@pytest.mark.parametrize(
+    'sirk_type,', [(csip.SirkType.ENCRYPTED), (csip.SirkType.PLAINTEXT)]
+)
+async def test_csis(sirk_type):
     SIRK = bytes.fromhex('2f62c8ae41867d1bb619e788a2605faa')
+    LTK = bytes.fromhex('2f62c8ae41867d1bb619e788a2605faa')
 
     devices = TwoDevices()
     devices[0].add_service(
         csip.CoordinatedSetIdentificationService(
             set_identity_resolving_key=SIRK,
-            set_identity_resolving_key_type=csip.SirkType.PLAINTEXT,
+            set_identity_resolving_key_type=sirk_type,
             coordinated_set_size=2,
             set_member_lock=csip.MemberLock.UNLOCKED,
             set_member_rank=0,
@@ -83,15 +88,19 @@ async def test_csis():
     )
 
     await devices.setup_connection()
+
+    # Mock encryption.
+    devices.connections[0].encryption = 1
+    devices.connections[1].encryption = 1
+    devices[0].get_long_term_key = mock.AsyncMock(return_value=LTK)
+    devices[1].get_long_term_key = mock.AsyncMock(return_value=LTK)
+
     peer = device.Peer(devices.connections[1])
     csis_client = await peer.discover_service_and_create_proxy(
         csip.CoordinatedSetIdentificationProxy
     )
 
-    assert (
-        await csis_client.set_identity_resolving_key.read_value()
-        == bytes([csip.SirkType.PLAINTEXT]) + SIRK
-    )
+    assert await csis_client.read_set_identity_resolving_key() == (sirk_type, SIRK)
     assert await csis_client.coordinated_set_size.read_value() == struct.pack('B', 2)
     assert await csis_client.set_member_lock.read_value() == struct.pack(
         'B', csip.MemberLock.UNLOCKED

tests/device_test.py

@@ -29,7 +29,7 @@ from bumble.core import (
     ConnectionParameters,
 )
 from bumble.device import Connection, Device
-from bumble.host import Host
+from bumble.host import AclPacketQueue, Host
 from bumble.hci import (
     HCI_ACCEPT_CONNECTION_REQUEST_COMMAND,
     HCI_COMMAND_STATUS_PENDING,
@@ -73,6 +73,13 @@ async def test_device_connect_parallel():
     d1 = Device(host=Host(None, None))
     d2 = Device(host=Host(None, None))
 
+    def _send(packet):
+        pass
+
+    d0.host.acl_packet_queue = AclPacketQueue(0, 0, _send)
+    d1.host.acl_packet_queue = AclPacketQueue(0, 0, _send)
+    d2.host.acl_packet_queue = AclPacketQueue(0, 0, _send)
+
     # enable classic
     d0.classic_enabled = True
     d1.classic_enabled = True