Merge pull request #147 from google/gbg/btbench

add benchmark tool and doc

bumble/device.py

@@ -50,6 +50,7 @@ from .hci import (
     HCI_LE_EXTENDED_CREATE_CONNECTION_COMMAND,
     HCI_LE_RAND_COMMAND,
     HCI_LE_READ_PHY_COMMAND,
+    HCI_LE_SET_PHY_COMMAND,
     HCI_MITM_NOT_REQUIRED_GENERAL_BONDING_AUTHENTICATION_REQUIREMENTS,
     HCI_MITM_NOT_REQUIRED_NO_BONDING_AUTHENTICATION_REQUIREMENTS,
     HCI_MITM_REQUIRED_GENERAL_BONDING_AUTHENTICATION_REQUIREMENTS,
@@ -1245,6 +1246,11 @@ class Device(CompositeEventEmitter):
         # Done
         self.powered_on = True
 
+    async def power_off(self) -> None:
+        if self.powered_on:
+            await self.host.flush()
+            self.powered_on = False
+
     def supports_le_feature(self, feature):
         return self.host.supports_le_feature(feature)
 
@@ -1669,7 +1675,7 @@ class Device(CompositeEventEmitter):
                         )
                     )
                     if not phys:
-                        raise ValueError('least one supported PHY needed')
+                        raise ValueError('at least one supported PHY needed')
 
                     phy_count = len(phys)
                     initiating_phys = phy_list_to_bits(phys)
@@ -1810,7 +1816,7 @@ class Device(CompositeEventEmitter):
 
                 try:
                     return await self.abort_on('flush', pending_connection)
-                except ConnectionError as error:
+                except core.ConnectionError as error:
                     raise core.TimeoutError() from error
         finally:
             self.remove_listener('connection', on_connection)
@@ -2044,21 +2050,31 @@ class Device(CompositeEventEmitter):
     async def set_connection_phy(
         self, connection, tx_phys=None, rx_phys=None, phy_options=None
     ):
+        if not self.host.supports_command(HCI_LE_SET_PHY_COMMAND):
+            logger.warning('ignoring request, command not supported')
+            return
+
         all_phys_bits = (1 if tx_phys is None else 0) | (
             (1 if rx_phys is None else 0) << 1
         )
 
-        return await self.send_command(
+        result = await self.send_command(
             HCI_LE_Set_PHY_Command(
                 connection_handle=connection.handle,
                 all_phys=all_phys_bits,
                 tx_phys=phy_list_to_bits(tx_phys),
                 rx_phys=phy_list_to_bits(rx_phys),
                 phy_options=0 if phy_options is None else int(phy_options),
-            ),
-            check_result=True,
+            )
         )
 
+        if result.status != HCI_COMMAND_STATUS_PENDING:
+            logger.warning(
+                'HCI_LE_Set_PHY_Command failed: '
+                f'{HCI_Constant.error_name(result.status)}'
+            )
+            raise HCI_StatusError(result)
+
     async def set_default_phy(self, tx_phys=None, rx_phys=None):
         all_phys_bits = (1 if tx_phys is None else 0) | (
             (1 if rx_phys is None else 0) << 1
@@ -2497,7 +2513,7 @@ class Device(CompositeEventEmitter):
             self.advertising = False
 
         # Notify listeners
-        error = ConnectionError(
+        error = core.ConnectionError(
             error_code,
             transport,
             peer_address,
@@ -2570,7 +2586,7 @@ class Device(CompositeEventEmitter):
     @with_connection_from_handle
     def on_disconnection_failure(self, connection, error_code):
         logger.debug(f'*** Disconnection failed: {error_code}')
-        error = ConnectionError(
+        error = core.ConnectionError(
             error_code,
             connection.transport,
             connection.peer_address,

bumble/gatt_server.py

@@ -691,7 +691,7 @@ class Server(EventEmitter):
                 length=entry_size, attribute_data_list=b''.join(attribute_data_list)
             )
         else:
-            logging.warning(f"not found {request}")
+            logging.debug(f"not found {request}")
 
         self.send_response(connection, response)
 

bumble/l2cap.py

@@ -796,6 +796,11 @@ class Channel(EventEmitter):
         self.disconnection_result = asyncio.get_running_loop().create_future()
         return await self.disconnection_result
 
+    def abort(self):
+        if self.state == self.OPEN:
+            self.change_state(self.CLOSED)
+            self.emit('close')
+
     def send_configure_request(self):
         options = L2CAP_Control_Frame.encode_configuration_options(
             [
@@ -1105,6 +1110,10 @@ class LeConnectionOrientedChannel(EventEmitter):
         self.disconnection_result = asyncio.get_running_loop().create_future()
         return await self.disconnection_result
 
+    def abort(self):
+        if self.state == self.CONNECTED:
+            self.change_state(self.DISCONNECTED)
+
     def on_pdu(self, pdu):
         if self.sink is None:
             logger.warning('received pdu without a sink')
@@ -1492,8 +1501,12 @@ class ChannelManager:
     def on_disconnection(self, connection_handle, _reason):
         logger.debug(f'disconnection from {connection_handle}, cleaning up channels')
         if connection_handle in self.channels:
+            for _, channel in self.channels[connection_handle].items():
+                channel.abort()
             del self.channels[connection_handle]
         if connection_handle in self.le_coc_channels:
+            for _, channel in self.le_coc_channels[connection_handle].items():
+                channel.abort()
             del self.le_coc_channels[connection_handle]
         if connection_handle in self.identifiers:
             del self.identifiers[connection_handle]

bumble/rfcomm.py

@@ -852,17 +852,27 @@ class Server(EventEmitter):
         # Register ourselves with the L2CAP channel manager
         device.register_l2cap_server(RFCOMM_PSM, self.on_connection)
 
-    def listen(self, acceptor):
-        # Find a free channel number
-        for channel in range(
-            RFCOMM_DYNAMIC_CHANNEL_NUMBER_START, RFCOMM_DYNAMIC_CHANNEL_NUMBER_END + 1
-        ):
-            if channel not in self.acceptors:
-                self.acceptors[channel] = acceptor
-                return channel
+    def listen(self, acceptor, channel=0):
+        if channel:
+            if channel in self.acceptors:
+                # Busy
+                return 0
+        else:
+            # Find a free channel number
+            for candidate in range(
+                RFCOMM_DYNAMIC_CHANNEL_NUMBER_START,
+                RFCOMM_DYNAMIC_CHANNEL_NUMBER_END + 1,
+            ):
+                if candidate not in self.acceptors:
+                    channel = candidate
+                    break
 
-        # All channels used...
-        return 0
+            if channel == 0:
+                # All channels used...
+                return 0
+
+        self.acceptors[channel] = acceptor
+        return channel
 
     def on_connection(self, l2cap_channel):
         logger.debug(f'+++ new L2CAP connection: {l2cap_channel}')

bumble/utils.py

@@ -20,7 +20,7 @@ import logging
 import traceback
 import collections
 import sys
-from typing import Awaitable, TypeVar
+from typing import Awaitable, Set, TypeVar
 from functools import wraps
 from pyee import EventEmitter
 
@@ -157,6 +157,9 @@ class AsyncRunner:
     # Shared default queue
     default_queue = WorkQueue()
 
+    # Shared set of running tasks
+    running_tasks: Set[Awaitable] = set()
+
     @staticmethod
     def run_in_task(queue=None):
         """
@@ -187,6 +190,19 @@ class AsyncRunner:
 
         return decorator
 
+    @staticmethod
+    def spawn(coroutine):
+        """
+        Spawn a task to run a coroutine in a "fire and forget" mode.
+
+        Using this method instead of just calling `asyncio.create_task(coroutine)`
+        is necessary when you don't keep a reference to the task, because `asyncio`
+        only keeps weak references to alive tasks.
+        """
+        task = asyncio.create_task(coroutine)
+        AsyncRunner.running_tasks.add(task)
+        task.add_done_callback(AsyncRunner.running_tasks.remove)
+
 
 # -----------------------------------------------------------------------------
 class FlowControlAsyncPipe: