diff --git a/docs/bledev-internals.md b/docs/bledev-internals.md index c80723e5..a9e51ed3 100644 --- a/docs/bledev-internals.md +++ b/docs/bledev-internals.md @@ -181,6 +181,27 @@ no difference to nus throughput. After a disconnect bleak sets `client.services` to `None`, so every client call checks the connection first and raises `DisconnectedError`. +## Reconnecting fast + +The REPL gate once failed one run in five: bleak couldn't subscribe on a +connection made 1.5 s after the last one closed. A loop of rapid reconnects +from the laptop (`tests/bledev_board/reconnect_loop.py`: log in, run a line, +close, wait 1.5 s) caught it at about one cycle in fifty, always the same +way. bleak's connect returned in 70-440 ms where a real one takes 1.7 s or +more, Windows reported the device connected and the GATT session active, the +board never saw a connection (its IRQ log shows it advertising throughout), +and the first GATT operation hung until Windows reported a drop about nine +seconds later. + +It went away with the REPL's output fixes (below): against `bledev-host`'s +`repl.py` the loop failed 7 times in 465 cycles; against the fixed one, 0 in +270, 150 of them with the retry switched off. We haven't found how the +board's old timer handling produced a connection Windows believed in and the +board never saw, so `bledev.connect_and_set_up()` stays as the defence: it +bounds the setup at 4 s and tries again, up to three times. `nus.connect()` +and `midi.connect()` use it, and the contract checks both the retry and the +giving up. + ## How the REPL gets there The REPL has to work when nothing runs asyncio, at the `>>>` prompt, so the @@ -190,9 +211,26 @@ with `gap_advertise`, and does everything else from an IRQ handler it adds to aioble's dispatcher (or sets itself, with no aioble). `os.dupterm` reads a `_Stream` whose `readinto` returns `None` when empty, because 0 means end of stream and detaches it. Output goes out as notifications of `mtu - 3` bytes, -and what the controller can't take yet is retried from a one-shot soft timer. -A program printing faster than the link carries waits in `write()` up to two -seconds, then drops the excess, because `print()` can't fail. +and what the controller can't take yet is retried every 5 ms by one periodic +`machine.Timer`. A program printing faster than the link carries waits in +`write()` up to two seconds, then drops the excess, because `print()` can't +fail. + +Two ways that retry went wrong, both found by printing a 16,890-character +list over the link (the REPL gate now does this): + +- **A doubled chunk and a lost one.** The timer's callback runs between the + main program's bytecodes, so it could land inside a `flush()` that + `print()` had started: both sent the same chunk, then both trimmed it, and + the next chunk never went out. Five prints out of five came back corrupt. + `flush()` now returns at once if it's already running. +- **A panic.** The retry used to be a one-shot timer re-armed on every write. + On the esp32, `Timer.init()` on a virtual timer whose alarm is being + dispatched clears its handler first, so `esp_timer`'s task on the other + core calls a NULL function (`InstrFetchProhibited` in + `timer_process_alarm`). It took the board down twice in about 45 long + prints. The timer is now started once and never re-armed; 60 long prints + afterwards, no panic and every byte intact. Ctrl-C works because the IRQ calls `os.dupterm_notify()` whenever a write contains 0x03; dupterm then reads it and raises `KeyboardInterrupt`, even in diff --git a/lib/bledev/__init__.py b/lib/bledev/__init__.py index 49d81a62..a5a8bac6 100644 --- a/lib/bledev/__init__.py +++ b/lib/bledev/__init__.py @@ -870,3 +870,38 @@ def _want_response(self, response): def is_adapter(obj): """True for a bledev :class:`BLE`; False for a raw ``bluetooth.BLE`` or anything else.""" return isinstance(obj, BLE) + + +async def connect_and_set_up( + ble, setup, *, name=None, service=None, device=None, timeout_ms=10000, attempts=3, setup_timeout_ms=4000, **options +): + """Find a device, connect, and run ``await setup(connection)``; returns + ``(connection, what setup returned)``. + + If the link drops or stalls before ``setup`` finishes, this disconnects + and tries again, up to ``attempts`` times. Profiles use it (``nus``, + ``midi``), because a link can come up dead: about one reconnect in fifty + from Windows, the OS reports a connection the peripheral never saw, and + the first GATT operation on it hangs until Windows gives up nine seconds + later. ``setup_timeout_ms`` bounds that wait. ``options`` go to + ``device.connect()``. + """ + error = None + for _ in range(attempts): + target = device + if target is None: + target = await ble.find(name=name, service=service, timeout_ms=timeout_ms) + connection = await target.connect(timeout_ms=timeout_ms, **options) + try: + return connection, await wait_ms(setup(connection), setup_timeout_ms, "setting up the connection") + except (DisconnectedError, BLETimeoutError) as e: + error = e + except BaseException: + if connection.is_connected(): + await connection.disconnect() + raise + try: + await connection.disconnect() + except BLEError: + pass + raise error diff --git a/lib/bledev/nus.py b/lib/bledev/nus.py index 1d6109b0..07f7388c 100644 --- a/lib/bledev/nus.py +++ b/lib/bledev/nus.py @@ -38,6 +38,7 @@ Service, UnsupportedError, UUID, + connect_and_set_up, is_adapter, sleep_ms, ) @@ -351,24 +352,23 @@ async def connect(ble, name=None, *, device=None, timeout_ms=10000, mtu=MTU, **c """ ble = _adapter(ble) _set_mtu(ble, mtu) - if device is None: - device = await ble.find(name=name, service=SERVICE, timeout_ms=timeout_ms) - connection = await device.connect(timeout_ms=timeout_ms, **connect_options) - try: + + async def setup(connection): try: await connection.exchange_mtu(mtu) except (UnsupportedError, BLETimeoutError): pass service = await connection.service(SERVICE) if service is None: - raise BLEError("{!r} has no Nordic UART service".format(device)) + raise BLEError("{!r} has no Nordic UART service".format(connection.device)) rx = await service.characteristic(RX) tx = await service.characteristic(TX) if rx is None or tx is None: - raise BLEError("{!r}: Nordic UART service is missing RX or TX".format(device)) + raise BLEError("{!r}: Nordic UART service is missing RX or TX".format(connection.device)) await tx.subscribe(notify=True) - except BaseException: - if connection.is_connected(): - await connection.disconnect() - raise + return rx, tx + + connection, (rx, tx) = await connect_and_set_up( + ble, setup, name=name, service=SERVICE, device=device, timeout_ms=timeout_ms, **connect_options + ) return _ClientLink(connection, rx, tx) diff --git a/lib/bledev/repl.py b/lib/bledev/repl.py index b07f6b94..9401f40e 100644 --- a/lib/bledev/repl.py +++ b/lib/bledev/repl.py @@ -34,6 +34,9 @@ from . import nus PROMPT = b"Password: " + +# How often the server retries output the controller couldn't take yet. +_TICK_MS = 5 BANNER = b"\r\nbledev REPL connected\r\n>>> " DENIED = b"\r\nAccess denied\r\n" @@ -190,10 +193,12 @@ def __init__(self, password, name, interval_us, dupterm_index): self.rx = bytearray() self.rx_pos = 0 self.tx = bytearray() + self.flushing = False self.dropped_out = 0 self.prev_term = None self.stream = _Stream(self) self.timer = None + self.hang_up_at = None self.early_mtu = None # -- setup @@ -236,8 +241,12 @@ def start(self): # Appending: several writes between two IRQs arrive together. ble.gatts_set_buffer(self.rx_handle, 1024, True) self._forget_other_services() + # One periodic timer for the life of the server, never re-armed: on + # the esp32, re-initialising a virtual Timer while its alarm is being + # dispatched calls a NULL handler and panics the board (seen as a + # Guru Meditation in esp_timer's task during long prints). self.timer = machine.Timer(-1) - self.one_shot = machine.Timer.ONE_SHOT + self.timer.init(mode=machine.Timer.PERIODIC, period=_TICK_MS, callback=self._tick) self.adv = pack_advertisement(self.name, [nus.SERVICE]) self.running = True self.advertise() @@ -292,6 +301,7 @@ def on_connect(self, conn): self.mtu = self.early_mtu[1] self.early_mtu = None self.login = _Gate(self.password) + self.hang_up_at = None self.authed = False self.rx = bytearray() self.rx_pos = 0 @@ -305,6 +315,7 @@ def on_disconnect(self, conn): return self._detach() self.conn = None + self.hang_up_at = None self.login = None self.authed = False self.rx = bytearray() @@ -321,7 +332,9 @@ def on_write(self, conn, data): self.notify_now(reply) self.login.refused = True # Let "Access denied" reach the client, then hang up. - self.timer.init(mode=self.one_shot, period=300, callback=self._hang_up) + import time + + self.hang_up_at = time.ticks_add(time.ticks_ms(), 300) return if verdict is None: if reply: @@ -373,7 +386,18 @@ def feed_repl(self, data): # -- sending - def _hang_up(self, _timer=None): + def _tick(self, _timer=None): + # Every _TICK_MS while the server runs, as a scheduled callback. + if self.hang_up_at is not None: + import time + + if time.ticks_diff(time.ticks_ms(), self.hang_up_at) >= 0: + self.hang_up_at = None + self._hang_up() + if self.tx and self.conn is not None: + self.flush() + + def _hang_up(self): if self.conn is not None and not self.authed: try: self.ble.gap_disconnect(self.conn) @@ -409,23 +433,26 @@ def send(self, data): break time.sleep_ms(2) self.flush() - if self.tx: - self.timer.init(mode=self.one_shot, period=5, callback=self._flush_later) - - def _flush_later(self, _timer=None): - self.flush() - if self.tx and self.conn is not None: - self.timer.init(mode=self.one_shot, period=5, callback=self._flush_later) def flush(self): - size = self.mtu - 3 - while self.tx and self.conn is not None: - chunk = self.tx[:size] - try: - self.ble.gatts_notify(self.conn, self.tx_handle, chunk) - except OSError: - return # the controller's buffers are full; the timer retries - self.tx = self.tx[len(chunk) :] + # The timer's callback runs between the main program's + # bytecodes, so it can land inside a flush() that print() started. + # Two flushes interleaved sent one chunk twice and dropped the next, + # so the second one leaves the work to the first. + if self.flushing: + return + self.flushing = True + try: + size = self.mtu - 3 + while self.tx and self.conn is not None: + chunk = self.tx[:size] + try: + self.ble.gatts_notify(self.conn, self.tx_handle, chunk) + except OSError: + return # the controller's buffers are full; _tick retries + self.tx = self.tx[len(chunk) :] + finally: + self.flushing = False class _Gate(Login): diff --git a/tests/bledev_board/reconnect_loop.py b/tests/bledev_board/reconnect_loop.py new file mode 100644 index 00000000..21ebb308 --- /dev/null +++ b/tests/bledev_board/reconnect_loop.py @@ -0,0 +1,72 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Rapid reconnects from a laptop against the REPL gate's board. + +Needs ``repl_server.py`` serving (from ``/main.py``). Each cycle logs in +through ``bledev.repl.connect()``, runs one line, closes, and waits ``GAP`` +seconds before the next, which is how the REPL gate once failed: bleak +couldn't subscribe on a connection made 1.5 s after the last one closed. +Prints each cycle, how many connection attempts it took, and a total:: + + PYTHONPATH="$(wslpath -w lib)" python.exe "$(wslpath -w tests/bledev_board/reconnect_loop.py)" 100 1.5 + +``--no-retry`` sets ``attempts=1``, the old behaviour, to show the failure. +""" +import asyncio +import sys +import time + +import bledev +import bledev.bleak +import bledev.nus as nus +import bledev.repl as repl + +args = [a for a in sys.argv[1:] if not a.startswith("--")] +N = int(args[0]) if args else 20 +GAP = float(args[1]) if len(args) > 1 else 1.5 +RETRY = "--no-retry" not in sys.argv + +attempts = [] +_helper = nus.connect_and_set_up + + +async def counting(ble, setup, **kwargs): + async def counted(connection): + attempts[-1] += 1 + return await setup(connection) + + if not RETRY: + kwargs["attempts"] = 1 + return await _helper(ble, counted, **kwargs) + + +nus.connect_and_set_up = counting + + +async def main(): + ble = bledev.bleak.BleakBLE() + fails = 0 + retried = 0 + for i in range(N): + attempts.append(0) + t0 = time.monotonic() + try: + link = await repl.connect(ble, "bledev-gate-pw", name="bledev-repl", timeout_ms=15000) + out = await repl.run(link, "%d * 3" % i) + ok = out.strip() == str(i * 3) + await link.close() + status = "OK" if ok else "BAD OUTPUT" + fails += 0 if ok else 1 + except Exception as e: + fails += 1 + status = "FAIL {} {}".format(type(e).__name__, e) + if attempts[-1] > 1: + retried += 1 + print("cycle", i, status, "attempts", attempts[-1], "%.1fs" % (time.monotonic() - t0), flush=True) + await asyncio.sleep(GAP) + print("RESULT {}/{} failed, {} needed a second attempt".format(fails, N, retried)) + return fails + + +sys.exit(1 if asyncio.run(main()) else 0) diff --git a/tests/bledev_board/repl_client.py b/tests/bledev_board/repl_client.py index c285e272..41c2cdba 100644 --- a/tests/bledev_board/repl_client.py +++ b/tests/bledev_board/repl_client.py @@ -7,6 +7,7 @@ board (mpble); see gatt_central.py for the command line. Checks, in order: 1. The right password gets a prompt, and ``123 * 456`` evaluates to 56088. + A 16,890-character print arrives byte for byte. 2. A wrong password, sent together with a line of code that would create ``/pwned``, gets ``Access denied`` and a disconnect, and never a prompt. 3. Logged in again, ``/pwned`` does not exist: the code never ran. @@ -65,6 +66,11 @@ async def main(): link = await repl.connect(ble, PASSWORD, name=NAME, timeout_ms=15000) out = await repl.run(link, "123 * 456") check(out.strip() == "56088", "the right password gets a working prompt", repr(out)) + # A list prints as hundreds of small writes, which is what once let the + # retry timer interleave two flushes: a chunk sent twice, the next lost. + out = await repl.run(link, "print(list(range(3000)))", timeout_ms=30000) + want = str(list(range(3000))) + check(out.strip() == want, "a long print arrives intact", "{} chars, want {}".format(len(out.strip()), len(want))) await repl.run(link, "import os") await repl.run(link, "exec(\"try:\\n os.remove('/pwned')\\nexcept OSError:\\n pass\")") out = await repl.run(link, "'pwned' in os.listdir('/')") diff --git a/tests/bledev_contract.py b/tests/bledev_contract.py index b671c219..8ae35d87 100644 --- a/tests/bledev_contract.py +++ b/tests/bledev_contract.py @@ -756,6 +756,55 @@ async def improv_scan(air): # ---------------------------------------------------------------- auto +@check +async def setup_retries_a_link_that_drops(air): + # Windows sometimes reports a connection the peripheral never saw; the + # first GATT operation then fails. connect_and_set_up tries again. + board, laptop = _two(air) + attempts = [] + + async def setup(connection): + attempts.append(connection) + if len(attempts) == 1: + raise bledev.DisconnectedError("the first link came up dead") + return "ready" + + served = asyncio.create_task(_serve_twice(board)) + connection, result = await bledev.connect_and_set_up(laptop, setup, name="board", timeout_ms=2000) + equal(result, "ready", "setup's result") + equal(len(attempts), 2, "attempts") + expect(not attempts[0].is_connected(), "the dead link was closed") + expect(connection.is_connected(), "the second link is up") + served.cancel() + + +def _two(air): + board = FakeBLE(air=air) + laptop = FakeBLE(air=air, peripheral=False) + board.register_services(gatt()[0]) + return board, laptop + + +async def _serve_twice(board): + for _ in range(2): + await board.advertise(name="board", timeout_ms=2000) + + +@check +async def setup_gives_up_after_its_attempts(air): + board, laptop = _two(air) + served = asyncio.create_task(_serve_twice(board)) + + async def setup(connection): + await asyncio.sleep(10) + + await raises( + BLETimeoutError, + bledev.connect_and_set_up(laptop, setup, name="board", timeout_ms=2000, attempts=2, setup_timeout_ms=100), + ) + served.cancel() + + @check async def auto_picks_or_explains(air): import bledev.auto as auto