Merge pull request #570 from google/gbg/bench-mobly-snippets

bench mobly snippets
2024-10-28 10:25:28 -07:00
parent 32642c5d7c fcd6bd7136
commit 4b25eed869
30 changed files with 1164 additions and 227 deletions
--- a/apps/bench.py
+++ b/apps/bench.py
@@ -19,6 +19,7 @@ import asyncio
 import enum
 import logging
 import os
+import statistics
 import struct
 import time

@@ -194,17 +195,19 @@ def make_sdp_records(channel):
    }


-def log_stats(title, stats):
+def log_stats(title, stats, precision=2):
    stats_min = min(stats)
    stats_max = max(stats)
-    stats_avg = sum(stats) / len(stats)
+    stats_avg = statistics.mean(stats)
+    stats_stdev = statistics.stdev(stats)
    logging.info(
        color(
            (
                f'### {title} stats: '
-                f'min={stats_min:.2f}, '
-                f'max={stats_max:.2f}, '
-                f'average={stats_avg:.2f}'
+                f'min={stats_min:.{precision}f}, '
+                f'max={stats_max:.{precision}f}, '
+                f'average={stats_avg:.{precision}f}, '
+                f'stdev={stats_stdev:.{precision}f}'
            ),
            'cyan',
        )
@@ -448,9 +451,9 @@ class Ping:
        self.repeat_delay = repeat_delay
        self.pace = pace
        self.done = asyncio.Event()
-        self.current_packet_index = 0
-        self.ping_sent_time = 0.0
-        self.latencies = []
+        self.ping_times = []
+        self.rtts = []
+        self.next_expected_packet_index = 0
        self.min_stats = []
        self.max_stats = []
        self.avg_stats = []
@@ -477,60 +480,57 @@ class Ping:
            logging.info(color('=== Sending RESET', 'magenta'))
            await self.packet_io.send_packet(bytes([PacketType.RESET]))

-            self.current_packet_index = 0
-            self.latencies = []
-            await self.send_next_ping()
+            packet_interval = self.pace / 1000
+            start_time = time.time()
+            self.next_expected_packet_index = 0
+            for i in range(self.tx_packet_count):
+                target_time = start_time + (i * packet_interval)
+                now = time.time()
+                if now < target_time:
+                    await asyncio.sleep(target_time - now)
+
+                packet = struct.pack(
+                    '>bbI',
+                    PacketType.SEQUENCE,
+                    (PACKET_FLAG_LAST if i == self.tx_packet_count - 1 else 0),
+                    i,
+                ) + bytes(self.tx_packet_size - 6)
+                logging.info(color(f'Sending packet {i}', 'yellow'))
+                self.ping_times.append(time.time())
+                await self.packet_io.send_packet(packet)

            await self.done.wait()

-            min_latency = min(self.latencies)
-            max_latency = max(self.latencies)
-            avg_latency = sum(self.latencies) / len(self.latencies)
+            min_rtt = min(self.rtts)
+            max_rtt = max(self.rtts)
+            avg_rtt = statistics.mean(self.rtts)
+            stdev_rtt = statistics.stdev(self.rtts)
            logging.info(
                color(
-                    '@@@ Latencies: '
-                    f'min={min_latency:.2f}, '
-                    f'max={max_latency:.2f}, '
-                    f'average={avg_latency:.2f}'
+                    '@@@ RTTs: '
+                    f'min={min_rtt:.2f}, '
+                    f'max={max_rtt:.2f}, '
+                    f'average={avg_rtt:.2f}, '
+                    f'stdev={stdev_rtt:.2f}'
                )
            )

-            self.min_stats.append(min_latency)
-            self.max_stats.append(max_latency)
-            self.avg_stats.append(avg_latency)
+            self.min_stats.append(min_rtt)
+            self.max_stats.append(max_rtt)
+            self.avg_stats.append(avg_rtt)

            run_counter = f'[{run + 1} of {self.repeat + 1}]' if self.repeat else ''
            logging.info(color(f'=== {run_counter} Done!', 'magenta'))

            if self.repeat:
-                log_stats('Min Latency', self.min_stats)
-                log_stats('Max Latency', self.max_stats)
-                log_stats('Average Latency', self.avg_stats)
+                log_stats('Min RTT', self.min_stats)
+                log_stats('Max RTT', self.max_stats)
+                log_stats('Average RTT', self.avg_stats)

        if self.repeat:
            logging.info(color('--- End of runs', 'blue'))

-    async def send_next_ping(self):
-        if self.pace:
-            await asyncio.sleep(self.pace / 1000)
-
-        packet = struct.pack(
-            '>bbI',
-            PacketType.SEQUENCE,
-            (
-                PACKET_FLAG_LAST
-                if self.current_packet_index == self.tx_packet_count - 1
-                else 0
-            ),
-            self.current_packet_index,
-        ) + bytes(self.tx_packet_size - 6)
-        logging.info(color(f'Sending packet {self.current_packet_index}', 'yellow'))
-        self.ping_sent_time = time.time()
-        await self.packet_io.send_packet(packet)
-
    def on_packet_received(self, packet):
-        elapsed = time.time() - self.ping_sent_time
-
        try:
            packet_type, packet_data = parse_packet(packet)
        except ValueError:
@@ -542,21 +542,23 @@ class Ping:
            return

        if packet_type == PacketType.ACK:
-            latency = elapsed * 1000
-            self.latencies.append(latency)
+            elapsed = time.time() - self.ping_times[packet_index]
+            rtt = elapsed * 1000
+            self.rtts.append(rtt)
            logging.info(
                color(
-                    f'<<< Received ACK [{packet_index}], latency={latency:.2f}ms',
+                    f'<<< Received ACK [{packet_index}], RTT={rtt:.2f}ms',
                    'green',
                )
            )

-            if packet_index == self.current_packet_index:
-                self.current_packet_index += 1
+            if packet_index == self.next_expected_packet_index:
+                self.next_expected_packet_index += 1
            else:
                logging.info(
                    color(
-                        f'!!! Unexpected packet, expected {self.current_packet_index} '
+                        f'!!! Unexpected packet, '
+                        f'expected {self.next_expected_packet_index} '
                        f'but received {packet_index}'
                    )
                )
@@ -565,8 +567,6 @@ class Ping:
            self.done.set()
            return

-        AsyncRunner.spawn(self.send_next_ping())
-

 # -----------------------------------------------------------------------------
 # Pong
@@ -583,8 +583,11 @@ class Pong:

    def reset(self):
        self.expected_packet_index = 0
+        self.receive_times = []

    def on_packet_received(self, packet):
+        self.receive_times.append(time.time())
+
        try:
            packet_type, packet_data = parse_packet(packet)
        except ValueError:
@@ -599,10 +602,16 @@ class Pong:
            packet_flags, packet_index = parse_packet_sequence(packet_data)
        except ValueError:
            return
+        interval = (
+            self.receive_times[-1] - self.receive_times[-2]
+            if len(self.receive_times) >= 2
+            else 0
+        )
        logging.info(
            color(
                f'<<< Received packet {packet_index}: '
-                f'flags=0x{packet_flags:02X}, {len(packet)} bytes',
+                f'flags=0x{packet_flags:02X}, {len(packet)} bytes, '
+                f'interval={interval:.4f}',
                'green',
            )
        )
@@ -623,8 +632,35 @@ class Pong:
            )
        )

-        if packet_flags & PACKET_FLAG_LAST and not self.linger:
-            self.done.set()
+        if packet_flags & PACKET_FLAG_LAST:
+            if len(self.receive_times) >= 3:
+                # Show basic stats
+                intervals = [
+                    self.receive_times[i + 1] - self.receive_times[i]
+                    for i in range(len(self.receive_times) - 1)
+                ]
+                log_stats('Packet intervals', intervals, 3)
+
+                # Show a histogram
+                bin_count = 20
+                bins = [0] * bin_count
+                interval_min = min(intervals)
+                interval_max = max(intervals)
+                interval_range = interval_max - interval_min
+                bin_thresholds = [
+                    interval_min + i * (interval_range / bin_count)
+                    for i in range(bin_count)
+                ]
+                for interval in intervals:
+                    for i in reversed(range(bin_count)):
+                        if interval >= bin_thresholds[i]:
+                            bins[i] += 1
+                            break
+                for i in range(bin_count):
+                    logging.info(f'@@@ >= {bin_thresholds[i]:.4f}: {bins[i]}')
+
+            if not self.linger:
+                self.done.set()

    async def run(self):
        await self.done.wait()
@@ -942,9 +978,12 @@ class RfcommClient(StreamedPacketIO):
            channel = await bumble.rfcomm.find_rfcomm_channel_with_uuid(
                connection, self.uuid
            )
-            logging.info(color(f'@@@ Channel number = {channel}', 'cyan'))
-            if channel == 0:
-                logging.info(color('!!! No RFComm service with this UUID found', 'red'))
+            if channel:
+                logging.info(color(f'@@@ Channel number = {channel}', 'cyan'))
+            else:
+                logging.warning(
+                    color('!!! No RFComm service with this UUID found', 'red')
+                )
                await connection.disconnect()
                return

@@ -1054,6 +1093,8 @@ class RfcommServer(StreamedPacketIO):
        if self.credits_threshold is not None:
            dlc.rx_credits_threshold = self.credits_threshold

+        self.ready.set()
+
    async def drain(self):
        assert self.dlc
        await self.dlc.drain()
@@ -1068,7 +1109,7 @@ class Central(Connection.Listener):
        transport,
        peripheral_address,
        classic,
-        role_factory,
+        scenario_factory,
        mode_factory,
        connection_interval,
        phy,
@@ -1081,7 +1122,7 @@ class Central(Connection.Listener):
        self.transport = transport
        self.peripheral_address = peripheral_address
        self.classic = classic
-        self.role_factory = role_factory
+        self.scenario_factory = scenario_factory
        self.mode_factory = mode_factory
        self.authenticate = authenticate
        self.encrypt = encrypt or authenticate
@@ -1134,7 +1175,7 @@ class Central(Connection.Listener):
                DEFAULT_CENTRAL_NAME, central_address, hci_source, hci_sink
            )
            mode = self.mode_factory(self.device)
-            role = self.role_factory(mode)
+            scenario = self.scenario_factory(mode)
            self.device.classic_enabled = self.classic

            # Set up a pairing config factory with minimal requirements.
@@ -1215,7 +1256,7 @@ class Central(Connection.Listener):

            await mode.on_connection(self.connection)

-            await role.run()
+            await scenario.run()
            await asyncio.sleep(DEFAULT_LINGER_TIME)
            await self.connection.disconnect()

@@ -1246,7 +1287,7 @@ class Peripheral(Device.Listener, Connection.Listener):
    def __init__(
        self,
        transport,
-        role_factory,
+        scenario_factory,
        mode_factory,
        classic,
        extended_data_length,
@@ -1254,11 +1295,11 @@ class Peripheral(Device.Listener, Connection.Listener):
    ):
        self.transport = transport
        self.classic = classic
-        self.role_factory = role_factory
+        self.scenario_factory = scenario_factory
        self.mode_factory = mode_factory
        self.extended_data_length = extended_data_length
        self.role_switch = role_switch
-        self.role = None
+        self.scenario = None
        self.mode = None
        self.device = None
        self.connection = None
@@ -1278,7 +1319,7 @@ class Peripheral(Device.Listener, Connection.Listener):
            )
            self.device.listener = self
            self.mode = self.mode_factory(self.device)
-            self.role = self.role_factory(self.mode)
+            self.scenario = self.scenario_factory(self.mode)
            self.device.classic_enabled = self.classic

            # Set up a pairing config factory with minimal requirements.
@@ -1315,7 +1356,7 @@ class Peripheral(Device.Listener, Connection.Listener):
            print_connection(self.connection)

            await self.mode.on_connection(self.connection)
-            await self.role.run()
+            await self.scenario.run()
            await asyncio.sleep(DEFAULT_LINGER_TIME)

    def on_connection(self, connection):
@@ -1344,7 +1385,7 @@ class Peripheral(Device.Listener, Connection.Listener):
    def on_disconnection(self, reason):
        logging.info(color(f'!!! Disconnection: reason={reason}', 'red'))
        self.connection = None
-        self.role.reset()
+        self.scenario.reset()

        if self.classic:
            AsyncRunner.spawn(self.device.set_discoverable(True))
@@ -1426,13 +1467,13 @@ def create_mode_factory(ctx, default_mode):


 # -----------------------------------------------------------------------------
-def create_role_factory(ctx, default_role):
-    role = ctx.obj['role']
-    if role is None:
-        role = default_role
+def create_scenario_factory(ctx, default_scenario):
+    scenario = ctx.obj['scenario']
+    if scenario is None:
+        scenarion = default_scenario

-    def create_role(packet_io):
-        if role == 'sender':
+    def create_scenario(packet_io):
+        if scenario == 'send':
            return Sender(
                packet_io,
                start_delay=ctx.obj['start_delay'],
@@ -1443,10 +1484,10 @@ def create_role_factory(ctx, default_role):
                packet_count=ctx.obj['packet_count'],
            )

-        if role == 'receiver':
+        if scenario == 'receive':
            return Receiver(packet_io, ctx.obj['linger'])

-        if role == 'ping':
+        if scenario == 'ping':
            return Ping(
                packet_io,
                start_delay=ctx.obj['start_delay'],
@@ -1457,12 +1498,12 @@ def create_role_factory(ctx, default_role):
                packet_count=ctx.obj['packet_count'],
            )

-        if role == 'pong':
+        if scenario == 'pong':
            return Pong(packet_io, ctx.obj['linger'])

-        raise ValueError('invalid role')
+        raise ValueError('invalid scenario')

-    return create_role
+    return create_scenario


 # -----------------------------------------------------------------------------
@@ -1470,7 +1511,7 @@ def create_role_factory(ctx, default_role):
 # -----------------------------------------------------------------------------
@click.group()
@click.option('--device-config', metavar='FILENAME', help='Device configuration file')
-@click.option('--role', type=click.Choice(['sender', 'receiver', 'ping', 'pong']))
+@click.option('--scenario', type=click.Choice(['send', 'receive', 'ping', 'pong']))
@click.option(
    '--mode',
    type=click.Choice(
@@ -1503,7 +1544,7 @@ def create_role_factory(ctx, default_role):
    '--rfcomm-channel',
    type=int,
    default=DEFAULT_RFCOMM_CHANNEL,
-    help='RFComm channel to use',
+    help='RFComm channel to use (specify 0 for channel discovery via SDP)',
 )
@click.option(
    '--rfcomm-uuid',
@@ -1565,7 +1606,7 @@ def create_role_factory(ctx, default_role):
    metavar='SIZE',
    type=click.IntRange(8, 8192),
    default=500,
-    help='Packet size (client or ping role)',
+    help='Packet size (send or ping scenario)',
 )
@click.option(
    '--packet-count',
@@ -1573,7 +1614,7 @@ def create_role_factory(ctx, default_role):
    metavar='COUNT',
    type=int,
    default=10,
-    help='Packet count (client or ping role)',
+    help='Packet count (send or ping scenario)',
 )
@click.option(
    '--start-delay',
@@ -1581,7 +1622,7 @@ def create_role_factory(ctx, default_role):
    metavar='SECONDS',
    type=int,
    default=1,
-    help='Start delay (client or ping role)',
+    help='Start delay (send or ping scenario)',
 )
@click.option(
    '--repeat',
@@ -1589,7 +1630,7 @@ def create_role_factory(ctx, default_role):
    type=int,
    default=0,
    help=(
-        'Repeat the run N times (client and ping roles)'
+        'Repeat the run N times (send and ping scenario)'
        '(0, which is the fault, to run just once) '
    ),
 )
@@ -1613,13 +1654,13 @@ def create_role_factory(ctx, default_role):
@click.option(
    '--linger',
    is_flag=True,
-    help="Don't exit at the end of a run (server and pong roles)",
+    help="Don't exit at the end of a run (receive and pong scenarios)",
 )
@click.pass_context
 def bench(
    ctx,
    device_config,
-    role,
+    scenario,
    mode,
    att_mtu,
    extended_data_length,
@@ -1645,7 +1686,7 @@ def bench(
 ):
    ctx.ensure_object(dict)
    ctx.obj['device_config'] = device_config
-    ctx.obj['role'] = role
+    ctx.obj['scenario'] = scenario
    ctx.obj['mode'] = mode
    ctx.obj['att_mtu'] = att_mtu
    ctx.obj['rfcomm_channel'] = rfcomm_channel
@@ -1699,7 +1740,7 @@ def central(
    ctx, transport, peripheral_address, connection_interval, phy, authenticate, encrypt
 ):
    """Run as a central (initiates the connection)"""
-    role_factory = create_role_factory(ctx, 'sender')
+    scenario_factory = create_scenario_factory(ctx, 'send')
    mode_factory = create_mode_factory(ctx, 'gatt-client')
    classic = ctx.obj['classic']

@@ -1708,7 +1749,7 @@ def central(
            transport,
            peripheral_address,
            classic,
-            role_factory,
+            scenario_factory,
            mode_factory,
            connection_interval,
            phy,
@@ -1726,13 +1767,13 @@ def central(
@click.pass_context
 def peripheral(ctx, transport):
    """Run as a peripheral (waits for a connection)"""
-    role_factory = create_role_factory(ctx, 'receiver')
+    scenario_factory = create_scenario_factory(ctx, 'receive')
    mode_factory = create_mode_factory(ctx, 'gatt-server')

    async def run_peripheral():
        await Peripheral(
            transport,
-            role_factory,
+            scenario_factory,
            mode_factory,
            ctx.obj['classic'],
            ctx.obj['extended_data_length'],
@@ -1743,7 +1784,11 @@ def peripheral(ctx, transport):


 def main():
-    logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper())
+    logging.basicConfig(
+        level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper(),
+        format="[%(asctime)s.%(msecs)03d] %(levelname)s:%(name)s:%(message)s",
+        datefmt="%H:%M:%S",
+    )
    bench()


--- a/apps/rfcomm_bridge.py
+++ b/apps/rfcomm_bridge.py
@@ -237,6 +237,7 @@ class ClientBridge:
        address: str,
        tcp_host: str,
        tcp_port: int,
+        authenticate: bool,
        encrypt: bool,
    ):
        self.channel = channel
@@ -245,6 +246,7 @@ class ClientBridge:
        self.address = address
        self.tcp_host = tcp_host
        self.tcp_port = tcp_port
+        self.authenticate = authenticate
        self.encrypt = encrypt
        self.device: Optional[Device] = None
        self.connection: Optional[Connection] = None
@@ -274,6 +276,11 @@ class ClientBridge:
        print(color(f"@@@ Bluetooth connection: {self.connection}", "blue"))
        self.connection.on("disconnection", self.on_disconnection)

+        if self.authenticate:
+            print(color("@@@ Authenticating Bluetooth connection", "blue"))
+            await self.connection.authenticate()
+            print(color("@@@ Bluetooth connection authenticated", "blue"))
+
        if self.encrypt:
            print(color("@@@ Encrypting Bluetooth connection", "blue"))
            await self.connection.encrypt()
@@ -491,8 +498,9 @@ def server(context, tcp_host, tcp_port):
@click.argument("bluetooth-address")
@click.option("--tcp-host", help="TCP host", default="_")
@click.option("--tcp-port", help="TCP port", default=DEFAULT_CLIENT_TCP_PORT)
+@click.option("--authenticate", is_flag=True, help="Authenticate the connection")
@click.option("--encrypt", is_flag=True, help="Encrypt the connection")
-def client(context, bluetooth_address, tcp_host, tcp_port, encrypt):
+def client(context, bluetooth_address, tcp_host, tcp_port, authenticate, encrypt):
    bridge = ClientBridge(
        context.obj["channel"],
        context.obj["uuid"],
@@ -500,6 +508,7 @@ def client(context, bluetooth_address, tcp_host, tcp_port, encrypt):
        bluetooth_address,
        tcp_host,
        tcp_port,
+        authenticate,
        encrypt,
    )
    asyncio.run(run(context.obj["device_config"], context.obj["hci_transport"], bridge))
--- a/docs/mkdocs/src/apps_and_tools/bench.md
+++ b/docs/mkdocs/src/apps_and_tools/bench.md
@@ -11,32 +11,44 @@ Usage: bumble-bench [OPTIONS] COMMAND [ARGS]...

 Options:
  --device-config FILENAME        Device configuration file
-  --role [sender|receiver|ping|pong]
+  --scenario [send|receive|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]
  --extended-data-length TEXT     Request a data length upon connection,
                                  specified as tx_octets/tx_time
-  --rfcomm-channel INTEGER        RFComm channel to use
+  --role-switch [central|peripheral]
+                                  Request role switch upon connection (central
+                                  or peripheral)
+  --rfcomm-channel INTEGER        RFComm channel to use (specify 0 for channel
+                                  discovery via SDP)
  --rfcomm-uuid TEXT              RFComm service UUID to use (ignored if
                                  --rfcomm-channel is not 0)
+  --rfcomm-l2cap-mtu INTEGER      RFComm L2CAP MTU
+  --rfcomm-max-frame-size INTEGER
+                                  RFComm maximum frame size
+  --rfcomm-initial-credits INTEGER
+                                  RFComm initial credits
+  --rfcomm-max-credits INTEGER    RFComm max credits
+  --rfcomm-credits-threshold INTEGER
+                                  RFComm credits threshold
  --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
+  -s, --packet-size SIZE          Packet size (send or ping scenario)
+                                  [8<=x<=8192]
+  -c, --packet-count COUNT        Packet count (send or ping scenario)
+  -sd, --start-delay SECONDS      Start delay (send or ping scenario)
+  --repeat N                      Repeat the run N times (send and ping
+                                  scenario)(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)
+  --linger                        Don't exit at the end of a run (receive and
+                                  pong scenarios)
  --help                          Show this message and exit.

 Commands:
@@ -71,19 +83,19 @@ using the ``--peripheral`` option. The address will be printed by the Peripheral
 it starts.

 Independently of whether the device is the Central or Peripheral, each device selects a
-``mode`` and and ``role`` to run as. The ``mode`` and ``role`` of the Central and Peripheral
+``mode`` and and ``scenario`` to run as. The ``mode`` and ``scenario`` of the Central and Peripheral
 must be compatible.

-Device 1 mode     | Device 2 mode
+Device 1 scenario | Device 2 scenario
 ------------------|------------------
 ``gatt-client``   | ``gatt-server``
 ``l2cap-client``  | ``l2cap-server``
 ``rfcomm-client`` | ``rfcomm-server``

-Device 1 role | Device 2 role
--------------|--------------
-``sender``    | ``receiver``
-``ping``      | ``pong``
+Device 1 scenario | Device 2 scenario
+------------------|--------------
+``send``          | ``receive``
+``ping``          | ``pong``


 # Examples
@@ -92,7 +104,7 @@ In the following examples, we have two USB Bluetooth controllers, one on `usb:0`
 the other on `usb:1`, and two consoles/terminals. We will run a command in each.

 !!! example "GATT Throughput"
-    Using the default mode and role for the Central and Peripheral.
+    Using the default mode and scenario for the Central and Peripheral.

    In the first console/terminal:
    ```
@@ -137,12 +149,12 @@ the other on `usb:1`, and two consoles/terminals. We will run a command in each.
 !!! example "Ping/Pong Latency"
    In the first console/terminal:
    ```
-    $ bumble-bench --role pong peripheral usb:0
+    $ bumble-bench --scenario pong peripheral usb:0
    ```

    In the second console/terminal:
    ```
-    $ bumble-bench --role ping central usb:1
+    $ bumble-bench --scenario ping central usb:1
    ```

 !!! example "Reversed modes with GATT and custom connection interval"
@@ -167,13 +179,13 @@ the other on `usb:1`, and two consoles/terminals. We will run a command in each.
    $ bumble-bench --mode l2cap-server central --phy 2m usb:1
    ```

-!!! example "Reversed roles with L2CAP"
+!!! example "Reversed scenarios with L2CAP"
    In the first console/terminal:
    ```
-    $ bumble-bench --mode l2cap-client --role sender peripheral usb:0
+    $ bumble-bench --mode l2cap-client --scenario send peripheral usb:0
    ```

    In the second console/terminal:
    ```
-    $ bumble-bench --mode l2cap-server --role receiver central usb:1
+    $ bumble-bench --mode l2cap-server --scenario receive central usb:1
    ```
--- a/examples/mobly/bench/one_device_bench_test.py
+++ b/examples/mobly/bench/one_device_bench_test.py
@@ -0,0 +1,47 @@
+from mobly import base_test
+from mobly import test_runner
+from mobly.controllers import android_device
+
+
+class OneDeviceBenchTest(base_test.BaseTestClass):
+
+    def setup_class(self):
+        self.ads = self.register_controller(android_device)
+        self.dut = self.ads[0]
+        self.dut.load_snippet("bench", "com.github.google.bumble.btbench")
+
+    def test_rfcomm_client_ping(self):
+        runner = self.dut.bench.runRfcommClient(
+            "ping", "DC:E5:5B:E5:51:2C", 100, 970, 100
+        )
+        print("### Initial status:", runner)
+        final_status = self.dut.bench.waitForRunnerCompletion(runner["id"])
+        print("### Final status:", final_status)
+
+    def test_rfcomm_client_send(self):
+        runner = self.dut.bench.runRfcommClient(
+            "send", "DC:E5:5B:E5:51:2C", 100, 970, 0
+        )
+        print("### Initial status:", runner)
+        final_status = self.dut.bench.waitForRunnerCompletion(runner["id"])
+        print("### Final status:", final_status)
+
+    def test_l2cap_client_ping(self):
+        runner = self.dut.bench.runL2capClient(
+            "ping", "4B:2A:67:76:2B:E3", 128, True, 100, 970, 100
+        )
+        print("### Initial status:", runner)
+        final_status = self.dut.bench.waitForRunnerCompletion(runner["id"])
+        print("### Final status:", final_status)
+
+    def test_l2cap_client_send(self):
+        runner = self.dut.bench.runL2capClient(
+            "send", "7E:90:D0:F2:7A:11", 131, True, 100, 970, 0
+        )
+        print("### Initial status:", runner)
+        final_status = self.dut.bench.waitForRunnerCompletion(runner["id"])
+        print("### Final status:", final_status)
+
+
+if __name__ == "__main__":
+    test_runner.main()
--- a/examples/mobly/bench/sample_config.yml
+++ b/examples/mobly/bench/sample_config.yml
@@ -0,0 +1,9 @@
+TestBeds:
+  - Name: BenchTestBed
+    Controllers:
+        AndroidDevice:
+          - serial: 37211FDJG000DJ
+            local_bt_address: 94:45:60:5E:03:B0
+
+          - serial: 23071FDEE001F7
+            local_bt_address: DC:E5:5B:E5:51:2C
--- a/examples/mobly/bench/two_devices_bench_test.py
+++ b/examples/mobly/bench/two_devices_bench_test.py
@@ -0,0 +1,38 @@
+import time
+
+from mobly import base_test
+from mobly import test_runner
+from mobly.controllers import android_device
+
+
+class TwoDevicesBenchTest(base_test.BaseTestClass):
+    def setup_class(self):
+        self.ads = self.register_controller(android_device)
+        self.dut1 = self.ads[0]
+        self.dut1.load_snippet("bench", "com.github.google.bumble.btbench")
+        self.dut2 = self.ads[1]
+        self.dut2.load_snippet("bench", "com.github.google.bumble.btbench")
+
+    def test_rfcomm_client_send_receive(self):
+        print("### Starting Receiver")
+        receiver = self.dut2.bench.runRfcommServer("receive")
+        receiver_id = receiver["id"]
+        print("--- Receiver status:", receiver)
+        while not receiver["model"]["running"]:
+            print("--- Waiting for Receiver to be running...")
+            time.sleep(1)
+            receiver = self.dut2.bench.getRunner(receiver_id)
+
+        print("### Starting Sender")
+        sender = self.dut1.bench.runRfcommClient(
+            "send", "DC:E5:5B:E5:51:2C", 100, 970, 100
+        )
+        print("--- Sender status:", sender)
+
+        print("--- Waiting for Sender to complete...")
+        sender_result = self.dut1.bench.waitForRunnerCompletion(sender["id"])
+        print("--- Sender result:", sender_result)
+
+
+if __name__ == "__main__":
+    test_runner.main()
--- a/extras/android/BtBench/app/build.gradle.kts
+++ b/extras/android/BtBench/app/build.gradle.kts
@@ -60,6 +60,8 @@ dependencies {
    implementation(libs.ui.graphics)
    implementation(libs.ui.tooling.preview)
    implementation(libs.material3)
+    implementation(libs.mobly.snippet)
+    implementation(libs.androidx.core)
    testImplementation(libs.junit)
    androidTestImplementation(libs.androidx.test.ext.junit)
    androidTestImplementation(libs.espresso.core)
--- a/extras/android/BtBench/app/src/main/AndroidManifest.xml
+++ b/extras/android/BtBench/app/src/main/AndroidManifest.xml
@@ -23,6 +23,9 @@
        android:supportsRtl="true"
        android:theme="@style/Theme.BTBench"
        >
+        <meta-data
+            android:name="mobly-snippets"
+            android:value="com.github.google.bumble.btbench.AutomationSnippet"/>
        <activity
            android:name=".MainActivity"
            android:exported="true"
@@ -35,5 +38,7 @@
        </activity>
 <!--        <profileable android:shell="true"/>-->
    </application>
-
-</manifest>
+    <instrumentation
+        android:name="com.google.android.mobly.snippet.SnippetRunner"
+        android:targetPackage="com.github.google.bumble.btbench" />
+</manifest>
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/AutomationSnippet.java
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/AutomationSnippet.java
@@ -0,0 +1,289 @@
+// 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.
+
+package com.github.google.bumble.btbench;
+
+import android.bluetooth.BluetoothAdapter;
+import android.bluetooth.BluetoothManager;
+import android.content.Context;
+
+import androidx.test.core.app.ApplicationProvider;
+
+import com.google.android.mobly.snippet.Snippet;
+import com.google.android.mobly.snippet.rpc.Rpc;
+
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+
+import java.security.InvalidParameterException;
+import java.util.ArrayList;
+import java.util.UUID;
+
+class Runner {
+    public UUID mId;
+    private final Mode mMode;
+    private final String mModeName;
+    private final String mScenario;
+    private final AppViewModel mModel;
+
+    Runner(Mode mode, String modeName, String scenario, AppViewModel model) {
+        this.mId = UUID.randomUUID();
+        this.mMode = mode;
+        this.mModeName = modeName;
+        this.mScenario = scenario;
+        this.mModel = model;
+    }
+
+    public JSONObject toJson() throws JSONException {
+        JSONObject result = new JSONObject();
+        result.put("id", mId.toString());
+        result.put("mode", mModeName);
+        result.put("scenario", mScenario);
+        result.put("model", AutomationSnippet.modelToJson(mModel));
+
+        return result;
+    }
+
+    public void stop() {
+        mModel.abort();
+    }
+
+    public void waitForCompletion() {
+        mMode.waitForCompletion();
+    }
+}
+
+public class AutomationSnippet implements Snippet {
+    private static final String TAG = "btbench.snippet";
+    private final BluetoothAdapter mBluetoothAdapter;
+    private final Context mContext;
+    private final ArrayList<Runner> mRunners = new ArrayList<>();
+
+    public AutomationSnippet() {
+        mContext = ApplicationProvider.getApplicationContext();
+        BluetoothManager bluetoothManager = mContext.getSystemService(BluetoothManager.class);
+        mBluetoothAdapter = bluetoothManager.getAdapter();
+        if (mBluetoothAdapter == null) {
+            throw new RuntimeException("bluetooth not supported");
+        }
+    }
+
+    private Runner runScenario(AppViewModel model, String mode, String scenario) {
+        Mode runnable;
+        switch (mode) {
+            case "rfcomm-client":
+                runnable = new RfcommClient(model, mBluetoothAdapter,
+                                          (PacketIO packetIO) -> createIoClient(model, scenario,
+                                                                                packetIO));
+                break;
+
+            case "rfcomm-server":
+                runnable = new RfcommServer(model, mBluetoothAdapter,
+                                          (PacketIO packetIO) -> createIoClient(model, scenario,
+                                                                                packetIO));
+                break;
+
+            case "l2cap-client":
+                runnable = new L2capClient(model, mBluetoothAdapter, mContext,
+                                         (PacketIO packetIO) -> createIoClient(model, scenario,
+                                                                               packetIO));
+                break;
+
+            case "l2cap-server":
+                runnable = new L2capServer(model, mBluetoothAdapter,
+                                         (PacketIO packetIO) -> createIoClient(model, scenario,
+                                                                               packetIO));
+                break;
+
+            default:
+                return null;
+        }
+
+        runnable.run();
+        Runner runner = new Runner(runnable, mode, scenario, model);
+        mRunners.add(runner);
+        return runner;
+    }
+
+    private IoClient createIoClient(AppViewModel model, String scenario, PacketIO packetIO) {
+        switch (scenario) {
+            case "send":
+                return new Sender(model, packetIO);
+
+            case "receive":
+                return new Receiver(model, packetIO);
+
+            case "ping":
+                return new Pinger(model, packetIO);
+
+            case "pong":
+                return new Ponger(model, packetIO);
+
+            default:
+                return null;
+        }
+    }
+
+    public static JSONObject modelToJson(AppViewModel model) throws JSONException {
+        JSONObject result = new JSONObject();
+        result.put("status", model.getStatus());
+        result.put("running", model.getRunning());
+        result.put("l2cap_psm", model.getL2capPsm());
+        if (model.getStatus().equals("OK")) {
+            JSONObject stats = new JSONObject();
+            result.put("stats", stats);
+            stats.put("throughput", model.getThroughput());
+            JSONObject rttStats = new JSONObject();
+            stats.put("rtt", rttStats);
+            rttStats.put("compound", model.getStats());
+        } else {
+            result.put("last_error", model.getLastError());
+        }
+
+        return result;
+    }
+
+    private Runner findRunner(String runnerId) {
+        for (Runner runner : mRunners) {
+            if (runner.mId.toString().equals(runnerId)) {
+                return runner;
+            }
+        }
+
+        return null;
+    }
+
+    @Rpc(description = "Run a scenario in RFComm Client mode")
+    public JSONObject runRfcommClient(String scenario, String peerBluetoothAddress, int packetCount,
+                                      int packetSize, int packetInterval) throws JSONException {
+        assert (mBluetoothAdapter != null);
+
+        // We only support "send" and "ping" for this mode for now
+        if (!(scenario.equals("send") || scenario.equals("ping"))) {
+            throw new InvalidParameterException("only 'send' and 'ping' are supported for this mode");
+        }
+
+        AppViewModel model = new AppViewModel();
+        model.setPeerBluetoothAddress(peerBluetoothAddress);
+        model.setSenderPacketCount(packetCount);
+        model.setSenderPacketSize(packetSize);
+        model.setSenderPacketInterval(packetInterval);
+
+        Runner runner = runScenario(model, "rfcomm-client", scenario);
+        assert runner != null;
+        return runner.toJson();
+    }
+
+    @Rpc(description = "Run a scenario in RFComm Server mode")
+    public JSONObject runRfcommServer(String scenario) throws JSONException {
+        assert (mBluetoothAdapter != null);
+
+        // We only support "receive" and "pong" for this mode for now
+        if (!(scenario.equals("receive") || scenario.equals("pong"))) {
+            throw new InvalidParameterException("only 'receive' and 'pong' are supported for this mode");
+        }
+
+        AppViewModel model = new AppViewModel();
+
+        Runner runner = runScenario(model, "rfcomm-server", scenario);
+        assert runner != null;
+        return runner.toJson();
+    }
+
+    @Rpc(description = "Run a scenario in L2CAP Client mode")
+    public JSONObject runL2capClient(String scenario, String peerBluetoothAddress, int psm,
+                                     boolean use_2m_phy, int packetCount, int packetSize,
+                                     int packetInterval) throws JSONException {
+        assert (mBluetoothAdapter != null);
+
+        // We only support "send" and "ping" for this mode for now
+        if (!(scenario.equals("send") || scenario.equals("ping"))) {
+            throw new InvalidParameterException("only 'send' and 'ping' are supported for this mode");
+        }
+
+        AppViewModel model = new AppViewModel();
+        model.setPeerBluetoothAddress(peerBluetoothAddress);
+        model.setL2capPsm(psm);
+        model.setUse2mPhy(use_2m_phy);
+        model.setSenderPacketCount(packetCount);
+        model.setSenderPacketSize(packetSize);
+        model.setSenderPacketInterval(packetInterval);
+
+        Runner runner = runScenario(model, "l2cap-client", scenario);
+        assert runner != null;
+        return runner.toJson();
+    }
+
+    @Rpc(description = "Run a scenario in L2CAP Server mode")
+    public JSONObject runL2capServer(String scenario) throws JSONException {
+        assert (mBluetoothAdapter != null);
+
+        // We only support "receive" and "pong" for this mode for now
+        if (!(scenario.equals("receive") || scenario.equals("pong"))) {
+            throw new InvalidParameterException("only 'receive' and 'pong' are supported for this mode");
+        }
+
+        AppViewModel model = new AppViewModel();
+
+        Runner runner = runScenario(model, "l2cap-server", scenario);
+        assert runner != null;
+        return runner.toJson();
+    }
+
+    @Rpc(description = "Stop a Runner")
+    public JSONObject stopRunner(String runnerId) throws JSONException {
+        Runner runner = findRunner(runnerId);
+        if (runner == null) {
+            return new JSONObject();
+        }
+        runner.stop();
+        return runner.toJson();
+    }
+
+    @Rpc(description = "Wait for a Runner to complete")
+    public JSONObject waitForRunnerCompletion(String runnerId) throws JSONException {
+        Runner runner = findRunner(runnerId);
+        if (runner == null) {
+            return new JSONObject();
+        }
+        runner.waitForCompletion();
+        return runner.toJson();
+    }
+
+    @Rpc(description = "Get a Runner by ID")
+    public JSONObject getRunner(String runnerId) throws JSONException {
+        Runner runner = findRunner(runnerId);
+        if (runner == null) {
+            return new JSONObject();
+        }
+        return runner.toJson();
+    }
+
+    @Rpc(description = "Get all Runners")
+    public JSONObject getRunners() throws JSONException {
+        JSONObject result = new JSONObject();
+        JSONArray runners = new JSONArray();
+        result.put("runners", runners);
+        for (Runner runner: mRunners) {
+            runners.put(runner.toJson());
+        }
+
+        return result;
+    }
+
+    @Override
+    public void shutdown() {
+    }
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/IoClient.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/IoClient.kt
@@ -0,0 +1,20 @@
+// 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.
+
+package com.github.google.bumble.btbench
+
+interface IoClient {
+    fun run()
+    fun abort()
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/L2capClient.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/L2capClient.kt
@@ -29,10 +29,13 @@ private val Log = Logger.getLogger("btbench.l2cap-client")
 class L2capClient(
    private val viewModel: AppViewModel,
    private val bluetoothAdapter: BluetoothAdapter,
-    private val context: Context
-) {
+    private val context: Context,
+    private val createIoClient: (packetIo: PacketIO) -> IoClient
+) : Mode {
+    private var socketClient: SocketClient? = null
+
    @SuppressLint("MissingPermission")
-    fun run() {
+    override fun run() {
        viewModel.running = true
        val addressIsPublic = viewModel.peerBluetoothAddress.endsWith("/P")
        val address = viewModel.peerBluetoothAddress.take(17)
@@ -75,6 +78,7 @@ class L2capClient(
                ) {
                    if (gatt != null && newState == BluetoothProfile.STATE_CONNECTED) {
                        if (viewModel.use2mPhy) {
+                            Log.info("requesting 2M PHY")
                            gatt.setPreferredPhy(
                                BluetoothDevice.PHY_LE_2M_MASK,
                                BluetoothDevice.PHY_LE_2M_MASK,
@@ -95,7 +99,11 @@ class L2capClient(

        val socket = remoteDevice.createInsecureL2capChannel(viewModel.l2capPsm)

-        val client = SocketClient(viewModel, socket)
-        client.run()
+        socketClient = SocketClient(viewModel, socket, createIoClient)
+        socketClient!!.run()
    }
-}
+
+    override fun waitForCompletion() {
+        socketClient?.waitForCompletion()
+    }
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/L2capServer.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/L2capServer.kt
@@ -27,11 +27,17 @@ import kotlin.concurrent.thread

 private val Log = Logger.getLogger("btbench.l2cap-server")

-class L2capServer(private val viewModel: AppViewModel, private val bluetoothAdapter: BluetoothAdapter) {
+class L2capServer(
+    private val viewModel: AppViewModel,
+    private val bluetoothAdapter: BluetoothAdapter,
+    private val createIoClient: (packetIo: PacketIO) -> IoClient
+) : Mode {
+    private var socketServer: SocketServer? = null
+
    @SuppressLint("MissingPermission")
-    fun run() {
+    override fun run() {
        // Advertise so that the peer can find us and connect.
-        val callback = object: AdvertiseCallback() {
+        val callback = object : AdvertiseCallback() {
            override fun onStartFailure(errorCode: Int) {
                Log.warning("failed to start advertising: $errorCode")
            }
@@ -55,7 +61,14 @@ class L2capServer(private val viewModel: AppViewModel, private val bluetoothAdap
        viewModel.l2capPsm = serverSocket.psm
        Log.info("psm = $serverSocket.psm")

-        val server = SocketServer(viewModel, serverSocket)
-        server.run({ advertiser.stopAdvertising(callback) }, { advertiser.startAdvertising(advertiseSettings, advertiseData, scanData, callback) })
+        socketServer = SocketServer(viewModel, serverSocket, createIoClient)
+        socketServer!!.run(
+            { advertiser.stopAdvertising(callback) },
+            { advertiser.startAdvertising(advertiseSettings, advertiseData, scanData, callback) }
+        )
    }
-}
+
+    override fun waitForCompletion() {
+        socketServer?.waitForCompletion()
+    }
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/MainActivity.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/MainActivity.kt
@@ -34,12 +34,15 @@ import androidx.compose.foundation.layout.fillMaxSize
 import androidx.compose.foundation.layout.fillMaxWidth
 import androidx.compose.foundation.layout.padding
 import androidx.compose.foundation.rememberScrollState
+import androidx.compose.foundation.selection.selectable
+import androidx.compose.foundation.selection.selectableGroup
 import androidx.compose.foundation.text.KeyboardActions
 import androidx.compose.foundation.text.KeyboardOptions
 import androidx.compose.foundation.verticalScroll
 import androidx.compose.material3.Button
 import androidx.compose.material3.Divider
 import androidx.compose.material3.MaterialTheme
+import androidx.compose.material3.RadioButton
 import androidx.compose.material3.Slider
 import androidx.compose.material3.Surface
 import androidx.compose.material3.Switch
@@ -54,6 +57,7 @@ import androidx.compose.ui.focus.FocusRequester
 import androidx.compose.ui.focus.focusRequester
 import androidx.compose.ui.platform.LocalFocusManager
 import androidx.compose.ui.platform.LocalSoftwareKeyboardController
+import androidx.compose.ui.semantics.Role
 import androidx.compose.ui.text.font.FontWeight
 import androidx.compose.ui.text.input.ImeAction
 import androidx.compose.ui.text.input.KeyboardType
@@ -69,6 +73,9 @@ private val Log = Logger.getLogger("bumble.main-activity")
 const val PEER_BLUETOOTH_ADDRESS_PREF_KEY = "peer_bluetooth_address"
 const val SENDER_PACKET_COUNT_PREF_KEY = "sender_packet_count"
 const val SENDER_PACKET_SIZE_PREF_KEY = "sender_packet_size"
+const val SENDER_PACKET_INTERVAL_PREF_KEY = "sender_packet_interval"
+const val SCENARIO_PREF_KEY = "scenario"
+const val MODE_PREF_KEY = "mode"

 class MainActivity : ComponentActivity() {
    private val appViewModel = AppViewModel()
@@ -139,10 +146,7 @@ class MainActivity : ComponentActivity() {
            MainView(
                appViewModel,
                ::becomeDiscoverable,
-                ::runRfcommClient,
-                ::runRfcommServer,
-                ::runL2capClient,
-                ::runL2capServer,
+                ::runScenario
            )
        }

@@ -159,37 +163,54 @@ class MainActivity : ComponentActivity() {
        if (packetSize > 0) {
            appViewModel.senderPacketSize = packetSize
        }
+        val packetInterval = intent.getIntExtra("packet-interval", 0)
+        if (packetInterval > 0) {
+            appViewModel.senderPacketInterval = packetInterval
+        }
        appViewModel.updateSenderPacketSizeSlider()
+        intent.getStringExtra("scenario")?.let {
+            when (it) {
+                "send" -> appViewModel.scenario = SEND_SCENARIO
+                "receive" -> appViewModel.scenario = RECEIVE_SCENARIO
+                "ping" -> appViewModel.scenario = PING_SCENARIO
+                "pong" -> appViewModel.scenario = PONG_SCENARIO
+            }
+        }
+        intent.getStringExtra("mode")?.let {
+            when (it) {
+                "rfcomm-client" -> appViewModel.mode = RFCOMM_CLIENT_MODE
+                "rfcomm-server" -> appViewModel.mode = RFCOMM_SERVER_MODE
+                "l2cap-client" -> appViewModel.mode = L2CAP_CLIENT_MODE
+                "l2cap-server" -> appViewModel.mode = L2CAP_SERVER_MODE
+            }
+        }
        intent.getStringExtra("autostart")?.let {
            when (it) {
-                "rfcomm-client" -> runRfcommClient()
-                "rfcomm-server" -> runRfcommServer()
-                "l2cap-client" -> runL2capClient()
-                "l2cap-server" -> runL2capServer()
+                "run-scenario" -> runScenario()
                "scan-start" -> runScan(true)
                "stop-start" -> runScan(false)
            }
        }
    }

-    private fun runRfcommClient() {
-        val rfcommClient = bluetoothAdapter?.let { RfcommClient(appViewModel, it) }
-        rfcommClient?.run()
-    }
+    private fun runScenario() {
+        if (bluetoothAdapter == null) {
+            return
+        }

-    private fun runRfcommServer() {
-        val rfcommServer = bluetoothAdapter?.let { RfcommServer(appViewModel, it) }
-        rfcommServer?.run()
-    }
-
-    private fun runL2capClient() {
-        val l2capClient = bluetoothAdapter?.let { L2capClient(appViewModel, it, baseContext) }
-        l2capClient?.run()
-    }
-
-    private fun runL2capServer() {
-        val l2capServer = bluetoothAdapter?.let { L2capServer(appViewModel, it) }
-        l2capServer?.run()
+        val runner = when (appViewModel.mode) {
+            RFCOMM_CLIENT_MODE -> RfcommClient(appViewModel, bluetoothAdapter!!, ::createIoClient)
+            RFCOMM_SERVER_MODE -> RfcommServer(appViewModel, bluetoothAdapter!!, ::createIoClient)
+            L2CAP_CLIENT_MODE -> L2capClient(
+                appViewModel,
+                bluetoothAdapter!!,
+                baseContext,
+                ::createIoClient
+            )
+            L2CAP_SERVER_MODE -> L2capServer(appViewModel, bluetoothAdapter!!, ::createIoClient)
+            else -> throw IllegalStateException()
+        }
+        runner.run()
    }

    private fun runScan(startScan: Boolean) {
@@ -197,6 +218,17 @@ class MainActivity : ComponentActivity() {
        scan?.run(startScan)
    }

+    private fun createIoClient(packetIo: PacketIO): IoClient {
+        return when (appViewModel.scenario) {
+            SEND_SCENARIO -> Sender(appViewModel, packetIo)
+            RECEIVE_SCENARIO -> Receiver(appViewModel, packetIo)
+            PING_SCENARIO -> Pinger(appViewModel, packetIo)
+            PONG_SCENARIO -> Ponger(appViewModel, packetIo)
+            else -> throw IllegalStateException()
+        }
+    }
+
+
    @SuppressLint("MissingPermission")
    fun becomeDiscoverable() {
        val discoverableIntent = Intent(BluetoothAdapter.ACTION_REQUEST_DISCOVERABLE)
@@ -210,10 +242,7 @@ class MainActivity : ComponentActivity() {
 fun MainView(
    appViewModel: AppViewModel,
    becomeDiscoverable: () -> Unit,
-    runRfcommClient: () -> Unit,
-    runRfcommServer: () -> Unit,
-    runL2capClient: () -> Unit,
-    runL2capServer: () -> Unit,
+    runScenario: () -> Unit,
 ) {
    BTBenchTheme {
        val scrollState = rememberScrollState()
@@ -239,7 +268,9 @@ fun MainView(
                        Text(text = "Peer Bluetooth Address")
                    },
                    value = appViewModel.peerBluetoothAddress,
-                    modifier = Modifier.fillMaxWidth().focusRequester(focusRequester),
+                    modifier = Modifier
+                        .fillMaxWidth()
+                        .focusRequester(focusRequester),
                    keyboardOptions = KeyboardOptions.Default.copy(
                        keyboardType = KeyboardType.Ascii, imeAction = ImeAction.Done
                    ),
@@ -249,14 +280,18 @@ fun MainView(
                    keyboardActions = KeyboardActions(onDone = {
                        keyboardController?.hide()
                        focusManager.clearFocus()
-                    })
+                    }),
+                    enabled = (appViewModel.mode == RFCOMM_CLIENT_MODE) or (appViewModel.mode == L2CAP_CLIENT_MODE)
                )
                Divider()
-                TextField(label = {
-                    Text(text = "L2CAP PSM")
-                },
+                TextField(
+                    label = {
+                        Text(text = "L2CAP PSM")
+                    },
                    value = appViewModel.l2capPsm.toString(),
-                    modifier = Modifier.fillMaxWidth().focusRequester(focusRequester),
+                    modifier = Modifier
+                        .fillMaxWidth()
+                        .focusRequester(focusRequester),
                    keyboardOptions = KeyboardOptions.Default.copy(
                        keyboardType = KeyboardType.Number, imeAction = ImeAction.Done
                    ),
@@ -271,7 +306,8 @@ fun MainView(
                    keyboardActions = KeyboardActions(onDone = {
                        keyboardController?.hide()
                        focusManager.clearFocus()
-                    })
+                    }),
+                    enabled = (appViewModel.mode == L2CAP_CLIENT_MODE)
                )
                Divider()
                Slider(
@@ -290,6 +326,32 @@ fun MainView(
                )
                Text(text = "Packet Size: " + appViewModel.senderPacketSize.toString())
                Divider()
+                TextField(
+                    label = {
+                        Text(text = "Packet Interval (ms)")
+                    },
+                    value = appViewModel.senderPacketInterval.toString(),
+                    modifier = Modifier
+                        .fillMaxWidth()
+                        .focusRequester(focusRequester),
+                    keyboardOptions = KeyboardOptions.Default.copy(
+                        keyboardType = KeyboardType.Number, imeAction = ImeAction.Done
+                    ),
+                    onValueChange = {
+                        if (it.isNotEmpty()) {
+                            val interval = it.toIntOrNull()
+                            if (interval != null) {
+                                appViewModel.updateSenderPacketInterval(interval)
+                            }
+                        }
+                    },
+                    keyboardActions = KeyboardActions(onDone = {
+                        keyboardController?.hide()
+                        focusManager.clearFocus()
+                    }),
+                    enabled = (appViewModel.scenario == PING_SCENARIO)
+                )
+                Divider()
                ActionButton(
                    text = "Become Discoverable", onClick = becomeDiscoverable, true
                )
@@ -300,25 +362,78 @@ fun MainView(
                    Text(text = "2M PHY")
                    Spacer(modifier = Modifier.padding(start = 8.dp))
                    Switch(
+                        enabled = (appViewModel.mode == L2CAP_CLIENT_MODE || appViewModel.mode == L2CAP_SERVER_MODE),
                        checked = appViewModel.use2mPhy,
                        onCheckedChange = { appViewModel.use2mPhy = it }
                    )

                }
                Row {
-                    ActionButton(
-                        text = "RFCOMM Client", onClick = runRfcommClient, !appViewModel.running
-                    )
-                    ActionButton(
-                        text = "RFCOMM Server", onClick = runRfcommServer, !appViewModel.running
-                    )
+                    Column(Modifier.selectableGroup()) {
+                        listOf(
+                            RFCOMM_CLIENT_MODE,
+                            RFCOMM_SERVER_MODE,
+                            L2CAP_CLIENT_MODE,
+                            L2CAP_SERVER_MODE
+                        ).forEach { text ->
+                            Row(
+                                Modifier
+                                    .selectable(
+                                        selected = (text == appViewModel.mode),
+                                        onClick = { appViewModel.updateMode(text) },
+                                        role = Role.RadioButton
+                                    )
+                                    .padding(horizontal = 16.dp),
+                                verticalAlignment = Alignment.CenterVertically
+                            ) {
+                                RadioButton(
+                                    selected = (text == appViewModel.mode),
+                                    onClick = null
+                                )
+                                Text(
+                                    text = text,
+                                    style = MaterialTheme.typography.bodyLarge,
+                                    modifier = Modifier.padding(start = 16.dp)
+                                )
+                            }
+                        }
+                    }
+                    Column(Modifier.selectableGroup()) {
+                        listOf(
+                            SEND_SCENARIO,
+                            RECEIVE_SCENARIO,
+                            PING_SCENARIO,
+                            PONG_SCENARIO
+                        ).forEach { text ->
+                            Row(
+                                Modifier
+                                    .selectable(
+                                        selected = (text == appViewModel.scenario),
+                                        onClick = { appViewModel.updateScenario(text) },
+                                        role = Role.RadioButton
+                                    )
+                                    .padding(horizontal = 16.dp),
+                                verticalAlignment = Alignment.CenterVertically
+                            ) {
+                                RadioButton(
+                                    selected = (text == appViewModel.scenario),
+                                    onClick = null
+                                )
+                                Text(
+                                    text = text,
+                                    style = MaterialTheme.typography.bodyLarge,
+                                    modifier = Modifier.padding(start = 16.dp)
+                                )
+                            }
+                        }
+                    }
                }
                Row {
                    ActionButton(
-                        text = "L2CAP Client", onClick = runL2capClient, !appViewModel.running
+                        text = "Start", onClick = runScenario, enabled = !appViewModel.running
                    )
                    ActionButton(
-                        text = "L2CAP Server", onClick = runL2capServer, !appViewModel.running
+                        text = "Stop", onClick = appViewModel::abort, enabled = appViewModel.running
                    )
                }
                Divider()
@@ -328,6 +443,12 @@ fun MainView(
                Text(
                    text = if (appViewModel.rxPhy != 0 || appViewModel.txPhy != 0) "PHY: tx=${appViewModel.txPhy}, rx=${appViewModel.rxPhy}" else ""
                )
+                Text(
+                    text = "Status: ${appViewModel.status}"
+                )
+                Text(
+                    text = "Last Error: ${appViewModel.lastError}"
+                )
                Text(
                    text = "Packets Sent: ${appViewModel.packetsSent}"
                )
@@ -337,9 +458,8 @@ fun MainView(
                Text(
                    text = "Throughput: ${appViewModel.throughput}"
                )
-                Divider()
-                ActionButton(
-                    text = "Abort", onClick = appViewModel::abort, appViewModel.running
+                Text(
+                    text = "Stats: ${appViewModel.stats}"
                )
            }
        }
@@ -351,4 +471,4 @@ fun ActionButton(text: String, onClick: () -> Unit, enabled: Boolean) {
    Button(onClick = onClick, enabled = enabled) {
        Text(text = text)
    }
-}
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Mode.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Mode.kt
@@ -0,0 +1,20 @@
+// 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.
+
+package com.github.google.bumble.btbench
+
+interface Mode {
+    fun run()
+    fun waitForCompletion()
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Model.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Model.kt
@@ -27,10 +27,25 @@ val DEFAULT_RFCOMM_UUID: UUID = UUID.fromString("E6D55659-C8B4-4B85-96BB-B1143AF
 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_SENDER_PACKET_INTERVAL = 100
 const val DEFAULT_PSM = 128

+const val L2CAP_CLIENT_MODE = "L2CAP Client"
+const val L2CAP_SERVER_MODE = "L2CAP Server"
+const val RFCOMM_CLIENT_MODE = "RFCOMM Client"
+const val RFCOMM_SERVER_MODE = "RFCOMM Server"
+
+const val SEND_SCENARIO = "Send"
+const val RECEIVE_SCENARIO = "Receive"
+const val PING_SCENARIO = "Ping"
+const val PONG_SCENARIO = "Pong"
+
 class AppViewModel : ViewModel() {
    private var preferences: SharedPreferences? = null
+    var status by mutableStateOf("")
+    var lastError by mutableStateOf("")
+    var mode by mutableStateOf(RFCOMM_SERVER_MODE)
+    var scenario by mutableStateOf(RECEIVE_SCENARIO)
    var peerBluetoothAddress by mutableStateOf(DEFAULT_PEER_BLUETOOTH_ADDRESS)
    var l2capPsm by mutableIntStateOf(DEFAULT_PSM)
    var use2mPhy by mutableStateOf(true)
@@ -41,9 +56,11 @@ class AppViewModel : ViewModel() {
    var senderPacketSizeSlider by mutableFloatStateOf(0.0F)
    var senderPacketCount by mutableIntStateOf(DEFAULT_SENDER_PACKET_COUNT)
    var senderPacketSize by mutableIntStateOf(DEFAULT_SENDER_PACKET_SIZE)
+    var senderPacketInterval by mutableIntStateOf(DEFAULT_SENDER_PACKET_INTERVAL)
    var packetsSent by mutableIntStateOf(0)
    var packetsReceived by mutableIntStateOf(0)
    var throughput by mutableIntStateOf(0)
+    var stats by mutableStateOf("")
    var running by mutableStateOf(false)
    var aborter: (() -> Unit)? = null

@@ -66,6 +83,21 @@ class AppViewModel : ViewModel() {
            senderPacketSize = savedSenderPacketSize
        }
        updateSenderPacketSizeSlider()
+
+        val savedSenderPacketInterval = preferences.getInt(SENDER_PACKET_INTERVAL_PREF_KEY, -1)
+        if (savedSenderPacketInterval != -1) {
+            senderPacketInterval = savedSenderPacketInterval
+        }
+
+        val savedMode = preferences.getString(MODE_PREF_KEY, null)
+        if (savedMode != null) {
+            mode = savedMode
+        }
+
+        val savedScenario = preferences.getString(SCENARIO_PREF_KEY, null)
+        if (savedScenario != null) {
+            scenario = savedScenario
+        }
    }

    fun updatePeerBluetoothAddress(peerBluetoothAddress: String) {
@@ -164,6 +196,42 @@ class AppViewModel : ViewModel() {
        }
    }

+    fun updateSenderPacketInterval(senderPacketInterval: Int) {
+        this.senderPacketInterval = senderPacketInterval
+        with(preferences!!.edit()) {
+            putInt(SENDER_PACKET_INTERVAL_PREF_KEY, senderPacketInterval)
+            apply()
+        }
+    }
+
+    fun updateScenario(scenario: String) {
+        this.scenario = scenario
+        with(preferences!!.edit()) {
+            putString(SCENARIO_PREF_KEY, scenario)
+            apply()
+        }
+    }
+
+    fun updateMode(mode: String) {
+        this.mode = mode
+        with(preferences!!.edit()) {
+            putString(MODE_PREF_KEY, mode)
+            apply()
+        }
+    }
+
+    fun clear() {
+        status = ""
+        lastError = ""
+        mtu = 0
+        rxPhy = 0
+        txPhy = 0
+        packetsSent = 0
+        packetsReceived = 0
+        throughput = 0
+        stats = ""
+    }
+
    fun abort() {
        aborter?.let { it() }
    }
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Packet.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Packet.kt
@@ -74,13 +74,13 @@ abstract class PacketSink {
    fun onPacket(packet: Packet) {
        when (packet) {
            is ResetPacket -> onResetPacket()
-            is AckPacket -> onAckPacket()
+            is AckPacket -> onAckPacket(packet)
            is SequencePacket -> onSequencePacket(packet)
        }
    }

    abstract fun onResetPacket()
-    abstract fun onAckPacket()
+    abstract fun onAckPacket(packet: AckPacket)
    abstract fun onSequencePacket(packet: SequencePacket)
 }

@@ -175,4 +175,4 @@ class SocketDataSource(
        } while (true)
        Log.info("end of stream")
    }
-}
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Pinger.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Pinger.kt
@@ -0,0 +1,104 @@
+// 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.
+
+package com.github.google.bumble.btbench
+
+import java.util.concurrent.Semaphore
+import java.util.logging.Logger
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.TimeSource
+
+private const val DEFAULT_STARTUP_DELAY = 3000
+
+private val Log = Logger.getLogger("btbench.pinger")
+
+class Pinger(private val viewModel: AppViewModel, private val packetIO: PacketIO) : IoClient,
+    PacketSink() {
+    private val pingTimes: ArrayList<TimeSource.Monotonic.ValueTimeMark> = ArrayList()
+    private val rtts: ArrayList<Long> = ArrayList()
+    private val done = Semaphore(0)
+
+    init {
+        packetIO.packetSink = this
+    }
+
+    override fun run() {
+        viewModel.clear()
+
+        Log.info("startup delay: $DEFAULT_STARTUP_DELAY")
+        Thread.sleep(DEFAULT_STARTUP_DELAY.toLong());
+        Log.info("running")
+
+        Log.info("sending reset")
+        packetIO.sendPacket(ResetPacket())
+
+        val packetCount = viewModel.senderPacketCount
+        val packetSize = viewModel.senderPacketSize
+
+        val startTime = TimeSource.Monotonic.markNow()
+        for (i in 0..<packetCount) {
+            val now = TimeSource.Monotonic.markNow()
+            val targetTime = startTime + (i * viewModel.senderPacketInterval).milliseconds
+            val delay = targetTime - now
+            if (delay.isPositive()) {
+                Log.info("sleeping ${delay.inWholeMilliseconds} ms")
+                Thread.sleep(delay.inWholeMilliseconds)
+            }
+            pingTimes.add(TimeSource.Monotonic.markNow())
+            packetIO.sendPacket(
+                SequencePacket(
+                    if (i < packetCount - 1) 0 else Packet.LAST_FLAG,
+                    i,
+                    ByteArray(packetSize - 6)
+                )
+            )
+            viewModel.packetsSent = i + 1
+        }
+
+        // Wait for the last ACK
+        Log.info("waiting for last ACK")
+        done.acquire()
+        Log.info("got last ACK")
+    }
+
+    override fun abort() {
+        done.release()
+    }
+
+    override fun onResetPacket() {
+    }
+
+    override fun onAckPacket(packet: AckPacket) {
+        val now = TimeSource.Monotonic.markNow()
+        viewModel.packetsReceived += 1
+        if (packet.sequenceNumber < pingTimes.size) {
+            val rtt = (now - pingTimes[packet.sequenceNumber]).inWholeMilliseconds
+            rtts.add(rtt)
+            Log.info("received ACK ${packet.sequenceNumber}, RTT=$rtt")
+        } else {
+            Log.warning("received ACK with unexpected sequence ${packet.sequenceNumber}")
+        }
+
+        if (packet.flags and Packet.LAST_FLAG != 0) {
+            Log.info("last packet received")
+            val stats = "RTTs: min=${rtts.min()}, max=${rtts.max()}, avg=${rtts.sum() / rtts.size}"
+            Log.info(stats)
+            viewModel.stats = stats
+            done.release()
+        }
+    }
+
+    override fun onSequencePacket(packet: SequencePacket) {
+    }
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Ponger.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Ponger.kt
@@ -0,0 +1,62 @@
+// 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.
+
+package com.github.google.bumble.btbench
+
+import java.util.logging.Logger
+import kotlin.time.TimeSource
+
+private val Log = Logger.getLogger("btbench.receiver")
+
+class Ponger(private val viewModel: AppViewModel, private val packetIO: PacketIO) : IoClient, PacketSink() {
+    private var startTime: TimeSource.Monotonic.ValueTimeMark = TimeSource.Monotonic.markNow()
+    private var lastPacketTime: TimeSource.Monotonic.ValueTimeMark = TimeSource.Monotonic.markNow()
+    private var expectedSequenceNumber: Int = 0
+
+    init {
+        packetIO.packetSink = this
+    }
+
+    override fun run() {
+        viewModel.clear()
+    }
+
+    override fun abort() {}
+
+    override fun onResetPacket() {
+        startTime = TimeSource.Monotonic.markNow()
+        lastPacketTime = startTime
+        expectedSequenceNumber = 0
+        viewModel.packetsSent = 0
+        viewModel.packetsReceived = 0
+        viewModel.stats = ""
+    }
+
+    override fun onAckPacket(packet: AckPacket) {
+    }
+
+    override fun onSequencePacket(packet: SequencePacket) {
+        val now = TimeSource.Monotonic.markNow()
+        lastPacketTime = now
+        viewModel.packetsReceived += 1
+
+        if (packet.sequenceNumber != expectedSequenceNumber) {
+            Log.warning("unexpected packet sequence number (expected ${expectedSequenceNumber}, got ${packet.sequenceNumber})")
+        }
+        expectedSequenceNumber += 1
+
+        packetIO.sendPacket(AckPacket(packet.flags, packet.sequenceNumber))
+        viewModel.packetsSent += 1
+    }
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Receiver.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Receiver.kt
@@ -20,7 +20,7 @@ import kotlin.time.TimeSource

 private val Log = Logger.getLogger("btbench.receiver")

-class Receiver(private val viewModel: AppViewModel, private val packetIO: PacketIO) : PacketSink() {
+class Receiver(private val viewModel: AppViewModel, private val packetIO: PacketIO) : IoClient, PacketSink() {
    private var startTime: TimeSource.Monotonic.ValueTimeMark = TimeSource.Monotonic.markNow()
    private var lastPacketTime: TimeSource.Monotonic.ValueTimeMark = TimeSource.Monotonic.markNow()
    private var bytesReceived = 0
@@ -29,6 +29,12 @@ class Receiver(private val viewModel: AppViewModel, private val packetIO: Packet
        packetIO.packetSink = this
    }

+    override fun run() {
+        viewModel.clear()
+    }
+
+    override fun abort() {}
+
    override fun onResetPacket() {
        startTime = TimeSource.Monotonic.markNow()
        lastPacketTime = startTime
@@ -36,9 +42,10 @@ class Receiver(private val viewModel: AppViewModel, private val packetIO: Packet
        viewModel.throughput = 0
        viewModel.packetsSent = 0
        viewModel.packetsReceived = 0
+        viewModel.stats = ""
    }

-    override fun onAckPacket() {
+    override fun onAckPacket(packet: AckPacket) {

    }

--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/RfcommClient.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/RfcommClient.kt
@@ -16,22 +16,30 @@ package com.github.google.bumble.btbench

 import android.annotation.SuppressLint
 import android.bluetooth.BluetoothAdapter
-import java.io.IOException
 import java.util.logging.Logger
-import kotlin.concurrent.thread

 private val Log = Logger.getLogger("btbench.rfcomm-client")

-class RfcommClient(private val viewModel: AppViewModel, val bluetoothAdapter: BluetoothAdapter) {
+class RfcommClient(
+    private val viewModel: AppViewModel,
+    private val bluetoothAdapter: BluetoothAdapter,
+    private val createIoClient: (packetIo: PacketIO) -> IoClient
+) : Mode {
+    private var socketClient: SocketClient? = null
+
    @SuppressLint("MissingPermission")
-    fun run() {
+    override fun run() {
        val address = viewModel.peerBluetoothAddress.take(17)
        val remoteDevice = bluetoothAdapter.getRemoteDevice(address)
        val socket = remoteDevice.createInsecureRfcommSocketToServiceRecord(
            DEFAULT_RFCOMM_UUID
        )

-        val client = SocketClient(viewModel, socket)
-        client.run()
+        socketClient = SocketClient(viewModel, socket, createIoClient)
+        socketClient!!.run()
+    }
+
+    override fun waitForCompletion() {
+        socketClient?.waitForCompletion()
    }
 }
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/RfcommServer.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/RfcommServer.kt
@@ -16,20 +16,27 @@ package com.github.google.bumble.btbench

 import android.annotation.SuppressLint
 import android.bluetooth.BluetoothAdapter
-import java.io.IOException
 import java.util.logging.Logger
-import kotlin.concurrent.thread

 private val Log = Logger.getLogger("btbench.rfcomm-server")

-class RfcommServer(private val viewModel: AppViewModel, val bluetoothAdapter: BluetoothAdapter) {
+class RfcommServer(
+    private val viewModel: AppViewModel,
+    private val bluetoothAdapter: BluetoothAdapter,
+    private val createIoClient: (packetIo: PacketIO) -> IoClient
+) : Mode {
+    private var socketServer: SocketServer? = null
+
    @SuppressLint("MissingPermission")
-    fun run() {
+    override fun run() {
        val serverSocket = bluetoothAdapter.listenUsingInsecureRfcommWithServiceRecord(
            "BumbleBench", DEFAULT_RFCOMM_UUID
        )
-
-        val server = SocketServer(viewModel, serverSocket)
-        server.run({}, {})
+        socketServer = SocketServer(viewModel, serverSocket, createIoClient)
+        socketServer!!.run({}, {})
    }
-}
+
+    override fun waitForCompletion() {
+        socketServer?.waitForCompletion()
+    }
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Scan.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Scan.kt
@@ -35,4 +35,4 @@ class Scan(val bluetoothAdapter: BluetoothAdapter) {
            bluetoothLeScanner?.stopScan(scanCallback)
        }
    }
-}
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Sender.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Sender.kt
@@ -19,9 +19,12 @@ import java.util.logging.Logger
 import kotlin.time.DurationUnit
 import kotlin.time.TimeSource

+private const val DEFAULT_STARTUP_DELAY = 3000
+
 private val Log = Logger.getLogger("btbench.sender")

-class Sender(private val viewModel: AppViewModel, private val packetIO: PacketIO) : PacketSink() {
+class Sender(private val viewModel: AppViewModel, private val packetIO: PacketIO) : IoClient,
+    PacketSink() {
    private var startTime: TimeSource.Monotonic.ValueTimeMark = TimeSource.Monotonic.markNow()
    private var bytesSent = 0
    private val done = Semaphore(0)
@@ -30,10 +33,12 @@ class Sender(private val viewModel: AppViewModel, private val packetIO: PacketIO
        packetIO.packetSink = this
    }

-    fun run() {
-        viewModel.packetsSent = 0
-        viewModel.packetsReceived = 0
-        viewModel.throughput = 0
+    override fun run() {
+        viewModel.clear()
+
+        Log.info("startup delay: $DEFAULT_STARTUP_DELAY")
+        Thread.sleep(DEFAULT_STARTUP_DELAY.toLong());
+        Log.info("running")

        Log.info("sending reset")
        packetIO.sendPacket(ResetPacket())
@@ -63,14 +68,14 @@ class Sender(private val viewModel: AppViewModel, private val packetIO: PacketIO
        Log.info("got ACK")
    }

-    fun abort() {
+    override fun abort() {
        done.release()
    }

    override fun onResetPacket() {
    }

-    override fun onAckPacket() {
+    override fun onAckPacket(packet: AckPacket) {
        Log.info("received ACK")
        val elapsed = TimeSource.Monotonic.markNow() - startTime
        val throughput = (bytesSent / elapsed.toDouble(DurationUnit.SECONDS)).toInt()
@@ -81,4 +86,4 @@ class Sender(private val viewModel: AppViewModel, private val packetIO: PacketIO

    override fun onSequencePacket(packet: SequencePacket) {
    }
-}
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/SocketClient.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/SocketClient.kt
@@ -22,16 +22,20 @@ import kotlin.concurrent.thread

 private val Log = Logger.getLogger("btbench.socket-client")

-private const val DEFAULT_STARTUP_DELAY = 3000
+class SocketClient(
+    private val viewModel: AppViewModel,
+    private val socket: BluetoothSocket,
+    private val createIoClient: (packetIo: PacketIO) -> IoClient
+) {
+    private var clientThread: Thread? = null

-class SocketClient(private val viewModel: AppViewModel, private val socket: BluetoothSocket) {
    @SuppressLint("MissingPermission")
    fun run() {
        viewModel.running = true
        val socketDataSink = SocketDataSink(socket)
        val streamIO = StreamedPacketIO(socketDataSink)
        val socketDataSource = SocketDataSource(socket, streamIO::onData)
-        val sender = Sender(viewModel, streamIO)
+        val ioClient = createIoClient(streamIO)

        fun cleanup() {
            socket.close()
@@ -39,9 +43,9 @@ class SocketClient(private val viewModel: AppViewModel, private val socket: Blue
            viewModel.running = false
        }

-        thread(name = "SocketClient") {
+        clientThread = thread(name = "SocketClient") {
            viewModel.aborter = {
-                sender.abort()
+                ioClient.abort()
                socket.close()
            }
            Log.info("connecting to remote")
@@ -49,27 +53,37 @@ class SocketClient(private val viewModel: AppViewModel, private val socket: Blue
                socket.connect()
            } catch (error: IOException) {
                Log.warning("connection failed")
+                viewModel.status = "ABORTED"
+                viewModel.lastError = "CONNECTION_FAILED"
                cleanup()
                return@thread
            }
            Log.info("connected")

-            thread {
+            val sourceThread = thread {
                socketDataSource.receive()
                socket.close()
-                sender.abort()
+                ioClient.abort()
            }

-            Log.info("Startup delay: $DEFAULT_STARTUP_DELAY")
-            Thread.sleep(DEFAULT_STARTUP_DELAY.toLong());
-            Log.info("Starting to send")
-
            try {
-                sender.run()
+                ioClient.run()
+                socket.close()
+                viewModel.status = "OK"
            } catch (error: IOException) {
                Log.info("run ended abruptly")
+                viewModel.status = "ABORTED"
+                viewModel.lastError = "IO_ERROR"
            }
+
+            Log.info("waiting for source thread to finish")
+            sourceThread.join()
+
            cleanup()
        }
    }
-}
+
+    fun waitForCompletion() {
+        clientThread?.join()
+    }
+}
--- a/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/SocketServer.kt
+++ b/extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/SocketServer.kt
@@ -21,7 +21,13 @@ import kotlin.concurrent.thread

 private val Log = Logger.getLogger("btbench.socket-server")

-class SocketServer(private val viewModel: AppViewModel, private val serverSocket: BluetoothServerSocket) {
+class SocketServer(
+    private val viewModel: AppViewModel,
+    private val serverSocket: BluetoothServerSocket,
+    private val createIoClient: (packetIo: PacketIO) -> IoClient
+) {
+    private var serverThread: Thread? = null
+
    fun run(onConnected: () -> Unit, onDisconnected: () -> Unit) {
        var aborted = false
        viewModel.running = true
@@ -31,7 +37,7 @@ class SocketServer(private val viewModel: AppViewModel, private val serverSocket
            viewModel.running = false
        }

-        thread(name = "SocketServer") {
+        serverThread = thread(name = "SocketServer") {
            while (!aborted) {
                viewModel.aborter = {
                    serverSocket.close()
@@ -46,6 +52,8 @@ class SocketServer(private val viewModel: AppViewModel, private val serverSocket
                    return@thread
                }
                Log.info("got connection from ${socket.remoteDevice.address}")
+                Log.info("maxReceivePacketSize=${socket.maxReceivePacketSize}")
+                Log.info("maxTransmitPacketSize=${socket.maxTransmitPacketSize}")
                onConnected()

                viewModel.aborter = {
@@ -57,11 +65,22 @@ class SocketServer(private val viewModel: AppViewModel, private val serverSocket
                val socketDataSink = SocketDataSink(socket)
                val streamIO = StreamedPacketIO(socketDataSink)
                val socketDataSource = SocketDataSource(socket, streamIO::onData)
-                val receiver = Receiver(viewModel, streamIO)
+
+                val ioThread = thread(name = "IoClient") {
+                    val ioClient = createIoClient(streamIO)
+                    ioClient.run()
+                }
+
                socketDataSource.receive()
                socket.close()
+                ioThread.join()
            }
            cleanup()
        }
+
    }
-}
+
+    fun waitForCompletion() {
+        serverThread?.join()
+    }
+}
--- a/extras/android/BtBench/gradle/libs.versions.toml
+++ b/extras/android/BtBench/gradle/libs.versions.toml
@@ -1,5 +1,5 @@
 [versions]
-agp = "8.2.0"
+agp = "8.4.0"
 kotlin = "1.9.0"
 core-ktx = "1.12.0"
 junit = "4.13.2"
@@ -8,6 +8,8 @@ espresso-core = "3.5.1"
 lifecycle-runtime-ktx = "2.6.2"
 activity-compose = "1.7.2"
 compose-bom = "2023.08.00"
+mobly-snippet = "1.4.0"
+core = "1.6.1"

 [libraries]
 core-ktx = { group = "androidx.core", name = "core-ktx", version.ref = "core-ktx" }
@@ -24,6 +26,8 @@ ui-tooling-preview = { group = "androidx.compose.ui", name = "ui-tooling-preview
 ui-test-manifest = { group = "androidx.compose.ui", name = "ui-test-manifest" }
 ui-test-junit4 = { group = "androidx.compose.ui", name = "ui-test-junit4" }
 material3 = { group = "androidx.compose.material3", name = "material3" }
+mobly-snippet = { group = "com.google.android.mobly", name = "mobly-snippet-lib", version.ref = "mobly.snippet" }
+androidx-core = { group = "androidx.test", name = "core", version.ref = "core" }

 [plugins]
 androidApplication = { id = "com.android.application", version.ref = "agp" }
--- a/extras/android/BtBench/gradle/wrapper/gradle-wrapper.properties
+++ b/extras/android/BtBench/gradle/wrapper/gradle-wrapper.properties
@@ -1,6 +1,6 @@
 #Wed Oct 25 07:40:52 PDT 2023
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.4-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.6-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
--- a/extras/android/RemoteHCI/app/src/main/java/com/github/google/bumble/remotehci/HciProxy.java
+++ b/extras/android/RemoteHCI/app/src/main/java/com/github/google/bumble/remotehci/HciProxy.java
@@ -23,6 +23,7 @@ public class HciProxy {
        HciHal hciHal = HciHal.create(new HciHalCallback() {
            @Override
            public void onPacket(HciPacket.Type type, byte[] packet) {
+                Log.d(TAG, String.format("CONTROLLER->HOST: type=%s, size=%d", type, packet.length));
                mServer.sendPacket(type, packet);

                switch (type) {
@@ -83,7 +84,7 @@ public class HciProxy {

            @Override
            public void onPacket(HciPacket.Type type, byte[] packet) {
-                Log.d(TAG, String.format("onPacket: type=%s, size=%d", type, packet.length));
+                Log.d(TAG, String.format("HOST->CONTROLLER: type=%s, size=%d", type, packet.length));
                hciHal.sendPacket(type, packet);

                switch (type) {
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -55,7 +55,7 @@ extend-exclude = '''
 '''

 [tool.mypy]
-exclude = ['bumble/transport/grpc_protobuf']
+exclude = ['bumble/transport/grpc_protobuf', 'examples/mobly/bench']

 [[tool.mypy.overrides]]
 module = "bumble.transport.grpc_protobuf.*"
--- a/setup.cfg
+++ b/setup.cfg
@@ -91,6 +91,7 @@ development =
    black == 24.3
    grpcio-tools >= 1.62.1
    invoke >= 1.7.3
+    mobly >= 1.12.2
    mypy == 1.10.0
    nox >= 2022
    pylint == 3.1.0