A fault the pump has been torn down for stops turning players away (#49)

* A fault the pump has already been torn down for stops turning players away

`Pump._fault` was both the sentence saying what happened and the refusal
that stopped every later `play()`, and only a fresh boot cleared it. So one
released node took the pump away for the rest of the boot: the speaker came
back on `machine.I2S`, `AudioOut.pumped` read False, the DMA counter never
moved and nothing anywhere said why (#45).

They come apart. `fault()` is the record and survives; `blocked()` is
whether it is still in the way, and the teardown that repairs the pump --
no clients, no thread, no driver, no channel -- clears it, as does the new
`clear_fault()`. A `play()` that fails with nothing left sounding tears the
pump down itself rather than leaving a refusal nobody can see. And a player
that does fall back because the pump is still holding something now says so
once per fault, the way a board with no `wire=` does.

`play(..., raises=True)` is for the caller that asked for the pump:
`audiobusio.I2SOut.retarget` raises a `ValueError` naming which refusal it
is -- a rate change, or a sample that is not signed 16-bit -- and flattening
it to a sentence was all a caller could ever see (audioif#2). `refusal()`
hands the same object to callers that take the `False`; `attach_stream()`
now raises the driver's exception instead of a `RuntimeError` quoting it.

Each of the three new tests was watched failing against its own planted
defect first.

* tests: the board probe that pulled an MP3 through the pump

#40 asked for a prefetcher in front of `MP3Decoder` and an MP3
played on a board through `audiodev`. The prefetcher was already there --
`FILE_BACKED` names `MP3Decoder` beside `WaveFile` -- and what had never
happened is anybody running it.

The probe reports the four things the issue needs: whether the pump took it,
whether a Prefetch was built, whether the DMA clocked the bytes, and what it
sounded like as a digest. `digest()` beside it is the reproducible one, over
a fixed 128 blocks, because the pump's own STATUS_DIGEST covers whatever it
happened to pull before somebody stopped it and moves between runs.

Its comments carry the three traps that each produced a confident wrong
reading first: the status digest is published at run-end, the status block
belongs to the I2SOut and goes when `stop()` closes it, and `play()` retunes
the bus and zeroes the DMA counter under a baseline taken a line earlier.

* pump: a teardown that raises must not replace the refusal it was cleaning up after

`_play`'s except path tears the pump down when nothing is left sounding.
If that teardown raised, the exception the caller actually needs -- the
driver's own `ValueError` naming which refusal it is -- would be replaced by
whatever went wrong on the way out, and under `raises=True` that is the one
they would catch.

Author: bdbarnett

lib/audiodev/pump.py

@@ -1003,12 +1003,21 @@ def attach_stream(fmt, *, driver=None, frames=256, capacity=6):
     Returns a :class:`PumpOutput` the caller writes PCM into, already
     registered as a pump client -- so an ``AudioOut`` graph and this stream
     are summed by a root Mixer and come out of one peripheral.
+
+    This caller DID ask for the pump, so it plays with ``raises=True`` and a
+    refusal arrives as the driver's own exception -- the ``ValueError`` that
+    says whether it was the rate or the sample width -- rather than as a
+    ``RuntimeError`` carrying a copy of its wording.
     """
     stream = PumpOutput(fmt, frames=frames, capacity=capacity)
     _enter()
     try:
-        took = owner().play(stream, stream.ring, driver=driver)
-    finally:
+        took = owner().play(stream, stream.ring, driver=driver, raises=True)
+    except Exception:
+        _leave()
+        stream.deinit()
+        raise
+    else:
         _leave()
     if not took:
         stream.deinit()
@@ -1054,6 +1063,15 @@ def __init__(self):
         self._retired = None
         self._spawned = False
         self._fault = None
+        # The sentence is what happened; this is whether it is still in the
+        # way. They came apart because a fault that had already been torn
+        # down went on refusing every later `play()` for the rest of the
+        # boot, silently, and the speaker came back on `machine.I2S`
+        # (pydevices#45). A pump with no clients, no thread and no driver has
+        # nothing broken left to carry, so `_shutdown` clears this and keeps
+        # the sentence for whoever wants to report it.
+        self._blocked = False
+        self._refusal = None
         self._events = None
 
     # --- the clock -------------------------------------------------------
@@ -1106,16 +1124,29 @@ def _find(self, owner):
     def clients(self):
         return len(self._clients)
 
-    def play(self, owner, sample, driver=None, loop=False):
-        """Put *owner*'s *sample* on the pump. Returns True when it sounds."""
+    def play(self, owner, sample, driver=None, loop=False, raises=False):
+        """Put *owner*'s *sample* on the pump. Returns True when it sounds.
+
+        A refusal is ``False`` and a sentence on :meth:`fault`, because the
+        usual caller is a player that never asked for the pump and cannot be
+        made to handle its problems.
+
+        ``raises=True`` is for the caller that DID ask. The driver's own
+        exception comes out instead of being flattened into that sentence --
+        `audiobusio.I2SOut.retarget` raises a ``ValueError`` naming *which*
+        refusal it is, a rate change or a sample that is not signed 16-bit,
+        and until this keyword existed there was no way to catch it
+        (audioif#2). :meth:`refusal` is the same object for the callers that
+        take the ``False``.
+        """
         _enter()
         try:
-            return self._play(owner, sample, driver, loop)
+            return self._play(owner, sample, driver, loop, raises)
         finally:
             _leave()
 
-    def _play(self, owner, sample, driver, loop):
-        if self._fault is not None:
+    def _play(self, owner, sample, driver, loop, raises=False):
+        if self._blocked:
             return False
         client = self._find(owner)
         if client is None:
@@ -1131,7 +1162,23 @@ def _play(self, owner, sample, driver, loop):
             self._retarget()
         except Exception as exc:          # noqa: BLE001 - reported, not raised
             self._clients.remove(client)
-            self._fault = str(exc)
+            self._note(exc)
+            if not self._clients:
+                # Nothing is sounding, so there is nothing half-swapped to
+                # protect: tear the pump down, which is also what clears the
+                # block. The next play() gets a fresh driver and a fresh
+                # channel rather than a refusal it cannot see.
+                #
+                # Guarded, because a teardown that raises here would replace
+                # the refusal the caller actually needs to see with whatever
+                # went wrong on the way out -- and under `raises=True` that is
+                # the exception they would get.
+                try:
+                    self._shutdown()
+                except Exception:    # noqa: BLE001 - the refusal outranks it
+                    pass
+            if raises:
+                raise
             return False
         return True
 
@@ -1155,7 +1202,7 @@ def _stop(self, owner):
         except Exception as exc:    # noqa: BLE001 - a client leaving must not
             # take the others' audio out with it, and it must not raise inside
             # somebody else's close().
-            self._fault = str(exc)
+            self._note(exc)
             self.shutdown()
 
     def pause(self, owner):
@@ -1311,11 +1358,51 @@ def died(self):
         return why or "the pump stopped (error %d, fault %d)" % (w[5], w[24])
 
     def fault(self):
-        """The reason the pump is not usable for the rest of this process."""
+        """The last reason the pump stopped or refused a graph, or None.
+
+        It is a record, not a verdict: it survives the teardown that repairs
+        the pump, so an app can still say what happened after the speaker has
+        come back. :meth:`blocked` is the one to ask before deciding whether
+        to reach for the pump at all.
+        """
         return self._fault
 
+    def refusal(self):
+        """The exception behind :meth:`fault`, or None.
+
+        `audiobusio.I2SOut.retarget` raises a ``ValueError`` naming which
+        refusal it is; flattening it to `str()` was the whole of what a
+        caller could see (audioif#2). A caller that took the ``False`` can
+        re-raise this or test its type.
+        """
+        return self._refusal
+
+    def blocked(self):
+        """True while the pump will refuse every :meth:`play`.
+
+        Set when a fault lands on a pump that is still holding something --
+        a thread, a channel, a half-swapped tail. Cleared by the teardown,
+        which is the repair, or by :meth:`clear_fault`.
+        """
+        return self._blocked
+
+    def clear_fault(self):
+        """Forget the last fault and let the next :meth:`play` try again."""
+        was = self._fault
+        self._fault = None
+        self._refusal = None
+        self._blocked = False
+        return was
+
+    def _note(self, exc):
+        self._fault = str(exc)
+        self._refusal = exc
+        self._blocked = True
+
     def note_fault(self, why):
         self._fault = why
+        self._refusal = None
+        self._blocked = True
 
     def shutdown(self):
         _enter()
@@ -1340,6 +1427,12 @@ def _shutdown(self):
                 mod.shutdown()
         self._spawned = False
         self._driver = None
+        # The teardown IS the repair. Whatever the fault was, what is left
+        # after this has no clients, no thread, no driver and no channel, so
+        # there is nothing for the next play() to trip over. The sentence
+        # stays on `fault()` for whoever wants to report it; what goes is the
+        # refusal that used to outlive it (pydevices#45).
+        self._blocked = False
         if driver is not None:
             driver.close()
 

lib/audiodev/sample_out.py

@@ -62,6 +62,11 @@ def ticks_diff(a, b):
 #: than a global so the check costs no `global` statement on the play path.
 _WARNED = []
 
+#: The fault sentence the "the pump is still down" note was last said about.
+#: Once per fault rather than once per boot: a second, different fault later
+#: in the same boot is news, and the same one said on every play() is not.
+_SAID = []
+
 #: `audiodev.pump`, imported once, lazily. Lazily because importing it from
 #: module scope re-enters `audiodev/__init__` while THIS module is halfway
 #: through being imported by it; once because the latch below is read on
@@ -315,7 +320,19 @@ def _attach_pump(self, sample):
         if not _pump_mod.available():
             return
         engine = _pump_mod.owner()
-        if engine.fault() is not None:
+        if engine.blocked():
+            # The pump is still holding whatever broke. Falling back is
+            # right; falling back SILENTLY is what cost a board sitting an
+            # hour -- the speaker works, `pumped` is False, the DMA counter
+            # never moves and nothing anywhere says why (pydevices#45). Same
+            # shape as the no-`wire=` note below, said once per fault.
+            self._pump_refused = engine.fault()
+            if not _SAID or _SAID[-1] != self._pump_refused:
+                del _SAID[:]
+                _SAID.append(self._pump_refused)
+                print("audiodev: the audio pump is still down -",
+                      self._pump_refused,
+                      "- playing on the interpreter thread instead")
             return
         try:
             tail = sample

tests/pump_probes/mp3_pump.py

@@ -0,0 +1,129 @@
+"""pydevices#40: an MP3 pulled through the pump, with a prefetcher in front.
+
+`audiodev.pump.FILE_BACKED` already names `MP3Decoder` beside `WaveFile`, so
+`AudioOut._attach_pump` should build a `Prefetch` for it exactly as it does
+for a WAV. Nobody had ever run it. This plays one MP3 through
+`board_peripherals.audio_out()` -- the factory a user calls -- and asks four
+things the issue needs answered:
+
+  * did the pump take it, or did it fall back to machine.I2S
+  * is there a Prefetch in front of the decoder
+  * did the DMA actually clock the audio out
+  * what did it sound like, as the pump's own FNV-1a digest over every byte
+
+Volume is low on purpose: nobody is listening and the numbers are counters.
+"""
+import gc
+import struct
+import time
+
+import audiomp3
+import audiopump
+import _audioif
+import board_peripherals as bp
+from audiodev import pump as apump
+
+PATH = "/beat.mp3"
+VOLUME = 20
+
+
+def _dma():
+    return _audioif.i2s_dma_bytes()
+
+
+def play(path=PATH, seconds=6.0, volume=VOLUME):
+    gc.collect()
+    dev = bp.audio_out()
+    dev.set_volume(volume)
+    fh = open(path, "rb")
+    dec = audiomp3.MP3Decoder(fh)
+    print("decoder: %d Hz, %d ch, %d bits"
+          % (dec.sample_rate, dec.channel_count, dec.bits_per_sample))
+    print("pumpable(decoder):", apump.pumpable(dec))
+
+    dev.play(dec)
+    # AFTER the play, not before it. `play()` retunes the bus to the sample's
+    # rate, which re-opens the channel and zeroes the DMA counter -- so a
+    # baseline taken a line earlier makes the difference come out NEGATIVE
+    # and the run look like a board that clocked nothing. Both readings have
+    # to sit inside one open.
+    before = _dma()
+    prefetch = dev._prefetch
+    print("pumped=%s  prefetch=%s  refused=%r"
+          % (dev.pumped, prefetch is not None, dev.pump_refused))
+
+    end = time.ticks_add(time.ticks_ms(), int(seconds * 1000))
+    ticks = 0
+    while time.ticks_diff(end, time.ticks_ms()) > 0:
+        dev.service()
+        ticks += 1
+        time.sleep_ms(20)
+        if not dev.playing:
+            break
+    clocked = _dma() - before
+    starved = (_audioif.i2s_starved_bytes()
+               if hasattr(_audioif, "i2s_starved_bytes") else -1)
+    pumped = dev.pumped
+    playing = dev.playing
+    wrote = prefetch.wrote if prefetch is not None else 0
+    fed = prefetch.fed() if prefetch is not None else None
+    # Two traps, one line apart. STATUS_DIGEST is published by the pump's
+    # RUN-END, so reading the block while it plays gives a truthful zero that
+    # looks exactly like a pump that hashed nothing. And on a board the block
+    # belongs to the `I2SOut`, which `stop()` closes -- so asking for it after
+    # the stop gets `audiodev`'s own spare, which nothing ever wrote, and
+    # reads as zeros again. Hold the REFERENCE across the stop: the C side
+    # writes the digest into this very bytearray on its way out.
+    st = apump.owner().status()
+    dev.stop()
+    w = struct.unpack("<%dQ" % audiopump.STATUS_WORDS, st)
+    print("blocks=%d bytes=%d digest=%016x" % (w[0], w[1], w[2]))
+    print("DMA +%d bytes  starved=%d  sink_timeouts=%d  err=%d fault=%d"
+          % (clocked, starved, w[14], w[5], w[24]))
+    print("prefetch: wrote=%d fed=%s" % (wrote, fed))
+    print("pumped=%s playing=%s after %d ticks" % (pumped, playing, ticks))
+    dev.close()
+    fh.close()
+    gc.collect()
+    ok = bool(pumped) and clocked > 0 and w[0] > 0 and w[5] == 0 and w[24] == 0
+    print("MP3 %s" % ("OK" if ok else "FAILED"))
+    return ok
+
+
+def digest(path=PATH, blocks=128):
+    """The decode's own identity: sha256 over a FIXED number of blocks.
+
+    The pump's `STATUS_DIGEST` covers everything it happened to pull before
+    somebody stopped it, so two runs of the same file give two different
+    numbers -- 1038 blocks and 1039 blocks are not the same audio. This is
+    the reproducible one, over the same first `blocks` blocks, pulled
+    straight through `audiocore` the way `Prefetch` pulls it.
+
+    sha256 rather than the pump's FNV-1a on purpose: the same hash in Python
+    is a per-byte loop, and 128 blocks of it outran `exec`'s quiet timeout on
+    this board without printing anything. The C hash is the one that returns.
+    """
+    import audiocore
+    import binascii
+    import hashlib
+    gc.collect()
+    fh = open(path, "rb")
+    dec = audiomp3.MP3Decoder(fh)
+    audiocore.reset_buffer(dec)
+    h = hashlib.sha256()
+    got = 0
+    n = 0
+    while n < blocks:
+        result, buf = audiocore.get_buffer(dec)
+        if buf is None or len(buf) == 0:
+            break
+        h.update(bytes(buf))
+        got += len(buf)
+        n += 1
+        if result == 0:                       # GET_BUFFER_DONE
+            break
+    fh.close()
+    out = binascii.hexlify(h.digest()).decode()[:16]
+    gc.collect()
+    print("offline: %d blocks, %d bytes, digest=%s" % (n, got, out))
+    return out