A transport's priming threshold is part of the interface, and a queue that stops moving is not full

Put a wrapper between AudioOut and a queued transport -- a tee, a logger,
a volume shim -- and the audio stopped after about a tenth of a second,
with no exception and no message. The backpressure cap has to clear the
device's priming threshold or the two mechanisms deadlock, and it read
that threshold off a PRIVATE attribute by getattr with a default of 0:
a wrapper that forwarded every documented part of PCMOutput still missed
it, and 0 is the value that wedges.

So prebuffer_bytes is a documented property on the device interface, and
the two wrappers this repository ships forward it -- ChannelAdapter
scaled to its own frames like queued_size, PaceOutput straight through.
ChannelAdapter is not hypothetical: it is what a mono graph on a stereo
device gets, and it answered 0.

The second half is worth more than the first. A cap is a latency guard,
not a fact about the device, so the skip is only believed while the queue
is MOVING; a queue that has not fallen for eight consecutive skips gets
fed on the ordinary schedule until it does, or until a second of audio is
queued, which is a dead sink rather than a priming one. That also catches
a device that stopped consuming for reasons nobody wrote a guard for.
AudioOut.stalls counts the episodes.

Proven on a real SDL device through tests/pump_probes/device.py: a
wrapper forwarding the interface renders all 192000 bytes in 0.99 s with
matching digests, where the measurement on the issue was 12288 bytes and
a 20 s timeout; a wrapper still hiding the threshold now completes too,
on the stall guard. Three planted faults, each failing only its own test.

Closes #54

Author: bdbarnett

lib/audiodev/__init__.py

@@ -534,6 +534,27 @@ def queued_size(self):
         """Bytes still waiting to play or capture. Default 0."""
         return 0
 
+    @property
+    def prebuffer_bytes(self):
+        """Bytes that must be queued before this device starts consuming.
+
+        Part of the interface, not an implementation detail, because a
+        producer has to clear it or the two mechanisms deadlock: a freshly
+        primed device stays paused until this many bytes are queued, while a
+        producer that caps itself lower has already decided the queue is full
+        enough. Each waits on the other and the audio stops without an
+        exception (pydevices#54).
+
+        It used to be readable only as ``_prebuffer_bytes``, by ``getattr``
+        with a default of 0 -- so a wrapper that forwarded every *documented*
+        part of this interface still missed it, and 0 is the value that
+        wedges. A wrapper forwards this the way it forwards `queued_size`.
+
+        0 means the device starts on the first byte, which is the honest
+        answer for a device with no queue at all.
+        """
+        return getattr(self, "_prebuffer_bytes", 0)
+
     def is_active(self):
         """True while PCM remains queued. Default False."""
         return False
@@ -912,6 +933,26 @@ def queued_size(self):
             return 0
         return inner_q * self.format.frame_size // inner_frame
 
+    @property
+    def prebuffer_bytes(self):
+        """The inner device's threshold, in this wrapper's frames.
+
+        Scaled exactly like `queued_size`, and for the same reason: the
+        producer compares the two, so they have to be counted in the same
+        bytes. A mono graph on a stereo device writes half the bytes the
+        wire does, and an unscaled threshold would be twice what the
+        producer has to clear.
+
+        Not forwarding this at all is pydevices#54: the pump capped itself
+        below the inner device's priming threshold, the device stayed
+        paused, and the audio stopped after about a tenth of a second with
+        no exception and no message.
+        """
+        inner_frame = self._inner.format.frame_size
+        if inner_frame <= 0:
+            return 0
+        return self._inner.prebuffer_bytes * self.format.frame_size // inner_frame
+
     def is_active(self):
         return self._inner.is_active()
 
@@ -1085,6 +1126,16 @@ def close(self):
     def queued_size(self):
         return self._inner.queued_size() + self._held
 
+    @property
+    def prebuffer_bytes(self):
+        """The inner device's threshold, unscaled: same format either side.
+
+        Forwarded rather than inherited, for pydevices#54's reason -- a
+        wrapper that answers 0 here caps the producer below what the device
+        needs to start, and both sides wait.
+        """
+        return self._inner.prebuffer_bytes
+
     def is_active(self):
         return self._held > 0 or self._inner.is_active()
 

lib/audiodev/sample_out.py

@@ -170,6 +170,29 @@ def __init__(self, transport, *, chunk_ms=40, lookahead_chunks=2,
         # bytes-per-len(buf) unit for the current sample's get_buffer()
         # results; calibrated on the first pull (see _pump_locked).
         self._buf_len_scale = None
+        #: How many consecutive backpressure skips with a queue that has not
+        #: fallen before the cap stops being believed. See `_pump_locked`.
+        #:
+        #: Eight is a judgement about which way to be wrong, not a
+        #: measurement. A false positive costs one chunk of extra latency;
+        #: a false negative costs the whole stream, silently. It is set
+        #: above the five ticks `test_backpressure_skips_pull_and_drains_
+        #: backlog` holds a queue saturated deliberately, so an ordinary
+        #: full queue is never mistaken for a stall and nothing that
+        #: worked before behaves differently -- a real backlogged transport
+        #: drains, and this only fires for one that does not.
+        self._stall_ticks = 8
+        #: The queued size at the last skip, how many skips have seen it not
+        #: fall, and whether we are currently feeding past the cap. -1 means
+        #: "not skipping".
+        self._stall_queued = -1
+        self._stalled = 0
+        self._pushing = False
+        #: How many times the pump started feeding past the cap because the
+        #: transport had stopped consuming -- episodes, not ticks. A number
+        #: above 0 means the sink was not draining: a wrapper that does not
+        #: forward `prebuffer_bytes` (pydevices#54), or a device that died.
+        self.stalls = 0
         # --- the audio pump, if this firmware has one --------------------
         # ``pump=False`` is for a caller that has a reason (a test proving
         # the two paths agree, a board whose peripheral someone else owns).
@@ -776,14 +799,61 @@ def _pump_locked(self):
         # each side waiting on the other, measured as exactly that.
         chunk_bytes = chunk_frames * frame_size
         cap = (self._lookahead_chunks + 1) * chunk_bytes
-        prebuffer = getattr(self.transport, "_prebuffer_bytes", 0)
+        # The PUBLIC name, because a wrapper can only forward what it can see.
+        # `_prebuffer_bytes` stays as a fallback for a transport written
+        # before `prebuffer_bytes` existed, but a wrapper that forwards the
+        # documented interface now gets this right by doing nothing special.
+        prebuffer = getattr(self.transport, "prebuffer_bytes", None)
+        if prebuffer is None:
+            prebuffer = getattr(self.transport, "_prebuffer_bytes", 0)
         if prebuffer and prebuffer + chunk_bytes > cap:
             cap = prebuffer + chunk_bytes
         queued = self.transport.queued_size()
         if queued > cap:
-            self._played_frames = target_frames
-            self.transport.service()
-            return
+            # The cap is a latency guard, not a fact about the device, and
+            # believing it forever is how the audio stops in silence. A
+            # transport that cannot say what it needs to prime gets capped
+            # below its own threshold, stays paused, and never consumes --
+            # so `queued` never falls and every later tick skips too.
+            #
+            # So the skip is only trusted while the queue is MOVING. When it
+            # has not fallen for `_stall_ticks` consecutive skips, feed it on
+            # the ordinary schedule instead: more bytes is exactly what a
+            # device waiting to prime needs, and for a sink that has genuinely
+            # stopped it is the one observable difference between "full" and
+            # "dead". `stalls` counts the episodes, so an app can see it.
+            #
+            # A CHUNK a tick is not enough, and that was measured rather than
+            # reasoned: SDL wants 96 000 B before it unpauses, which at one
+            # 1 920 B chunk per eight ticks is sixteen seconds of trickle --
+            # long enough that the probe ran the SDL queue out of memory
+            # before the device ever started. Feeding the normal schedule is
+            # bounded by `max_catchup_chunks` and primes it in about ten
+            # ticks.
+            if queued < self._stall_queued or self._stall_queued < 0:
+                self._stall_queued = queued
+                self._stalled = 0
+                self._pushing = False
+            else:
+                self._stalled += 1
+            # One second of audio queued with nothing consumed is a dead sink,
+            # not a priming one -- no device on this bench asks for more than
+            # that -- so stop feeding rather than filling a board's memory.
+            # Derived from the format rather than picked, and it sits above
+            # every priming threshold this repository ships.
+            if queued >= int(rate) * frame_size:
+                self._pushing = False
+            elif self._stalled >= self._stall_ticks and not self._pushing:
+                self._pushing = True
+                self.stalls += 1
+            if not self._pushing:
+                self._played_frames = target_frames
+                self.transport.service()
+                return
+        else:
+            self._stall_queued = -1
+            self._stalled = 0
+            self._pushing = False
 
         bytes_needed = frames_needed * frame_size
         pulled = 0

tests/pump_probes/device.py

@@ -95,14 +95,22 @@ def __init__(self, inner):
         self.seen = bytearray()
         self.calls = 0
         self.short_writes = 0
-        # AudioOut's backpressure cap reads this off the transport, and a
-        # wrapper that hides it DEADLOCKS: the cap stays at one lookahead
-        # (11 520 B here) while the SDL device waits for 96 000 B before it
-        # unpauses, so each side waits for the other and the stream stops
-        # after 12 writes. Found the hard way by this probe's first run --
-        # the instrument planted the fault, which is also the clearest
-        # demonstration that the guard in `_pump_locked` is load-bearing.
-        self._prebuffer_bytes = getattr(inner, "_prebuffer_bytes", 0)
+
+    # AudioOut's backpressure cap has to clear this, and a wrapper that hides
+    # it DEADLOCKS: the cap stays at one lookahead (11 520 B here) while the
+    # SDL device waits for 96 000 B before it unpauses, so each side waits for
+    # the other and the stream stops after 12 writes. Found the hard way by
+    # this probe's first run -- the instrument planted the fault, and it was
+    # filed as pydevices#54.
+    #
+    # It was `self._prebuffer_bytes = getattr(inner, "_prebuffer_bytes", 0)`
+    # here: a wrapper reaching for a private attribute because the public
+    # interface did not carry one. It does now, so this is an ordinary
+    # forward like `queued_size` below, and a wrapper that forwards what it
+    # is documented to forward no longer has to know why this one matters.
+    @property
+    def prebuffer_bytes(self):
+        return self.inner.prebuffer_bytes
 
     # -- what AudioOut asks of a transport ---------------------------------
     @property

tests/test_audiodev.py

@@ -583,6 +583,177 @@ def queued_size(self):
             len(transport.data), transport._prebuffer_bytes,
             "pump stopped feeding a still-priming transport (deadlock)")
 
+    def test_a_wrapper_that_forwards_the_interface_does_not_deadlock(self):
+        # pydevices#54. Put anything between AudioOut and a queued transport
+        # -- a tee, a logger, a volume shim -- and the audio used to stop
+        # after about a tenth of a second, with no exception and no message,
+        # because the backpressure cap read the priming threshold off a
+        # PRIVATE attribute the wrapper had no reason to know about.
+        #
+        # `prebuffer_bytes` is part of the interface now, so a wrapper that
+        # forwards what it is documented to forward gets this right without
+        # knowing why it matters.
+        class PrimingTransport(FakePCMOutput):
+            _prebuffer_bytes = 4000
+
+            def queued_size(self):
+                return len(self.data)      # priming: nothing consumed yet
+
+        class Wrapper:
+            """Forwards the documented PCMOutput interface and nothing else."""
+
+            def __init__(self, inner):
+                self.inner = inner
+
+            format = property(lambda self: self.inner.format)
+            codec = property(lambda self: self.inner.codec)
+            volume = property(lambda self: self.inner.volume)
+            muted = property(lambda self: self.inner.muted)
+            prebuffer_bytes = property(lambda self: self.inner.prebuffer_bytes)
+
+            def set_volume(self, percent):
+                return self.inner.set_volume(percent)
+
+            def mute(self, value=True):
+                return self.inner.mute(value)
+
+            def open(self):
+                return self.inner.open()
+
+            def close(self):
+                return self.inner.close()
+
+            def service(self):
+                return self.inner.service()
+
+            def queued_size(self):
+                return self.inner.queued_size()
+
+            def write(self, buf):
+                return self.inner.write(buf)
+
+        inner = PrimingTransport(self.fmt)
+        out = self.sample_out.AudioOut(Wrapper(inner), chunk_ms=40,
+                                       lookahead_chunks=2)
+        out.play(FakeSample([bytes(200) for _ in range(200)]))
+        for _ in range(10):
+            self.clock.advance(40)
+            out.service()
+        self.assertGreater(
+            len(inner.data), inner._prebuffer_bytes,
+            "a wrapper forwarding the whole interface still deadlocked")
+        self.assertEqual(0, out.stalls,
+                         "it should not have needed the stall guard")
+
+    def test_a_transport_that_stops_consuming_is_pushed_past_the_cap(self):
+        # The other half of pydevices#54, and the more valuable one: a cap is
+        # a latency guard, not a fact about the device. A wrapper that cannot
+        # say what it needs to prime -- every wrapper written before
+        # `prebuffer_bytes` existed -- still caps below the threshold and
+        # still wedges. So the skip is only believed while the queue is
+        # MOVING, and a queue that has not fallen for `_stall_ticks` skips
+        # gets fed anyway. That also catches a sink that has died for reasons
+        # nobody wrote a guard for, which is the same observable signature.
+        class PrimingTransport(FakePCMOutput):
+            _prebuffer_bytes = 4000
+
+            def queued_size(self):
+                return len(self.data)
+
+        class Deaf:
+            """A wrapper that hides the threshold: the old, naive shape."""
+
+            def __init__(self, inner):
+                self.inner = inner
+
+            format = property(lambda self: self.inner.format)
+            codec = property(lambda self: self.inner.codec)
+            volume = property(lambda self: self.inner.volume)
+            muted = property(lambda self: self.inner.muted)
+
+            def set_volume(self, percent):
+                return self.inner.set_volume(percent)
+
+            def mute(self, value=True):
+                return self.inner.mute(value)
+
+            def open(self):
+                return self.inner.open()
+
+            def close(self):
+                return self.inner.close()
+
+            def service(self):
+                return self.inner.service()
+
+            def queued_size(self):
+                return self.inner.queued_size()
+
+            def write(self, buf):
+                return self.inner.write(buf)
+
+        inner = PrimingTransport(self.fmt)
+        out = self.sample_out.AudioOut(Deaf(inner), chunk_ms=40,
+                                       lookahead_chunks=2)
+        out.play(FakeSample([bytes(200) for _ in range(200)]))
+        wedged = len(inner.data)
+        for _ in range(30):
+            self.clock.advance(40)
+            out.service()
+        self.assertGreater(out.stalls, 0,
+                           "the pump never noticed the queue had stopped")
+        self.assertGreater(
+            len(inner.data), wedged,
+            "a transport that stopped consuming was never fed again")
+        self.assertGreater(
+            len(inner.data), inner._prebuffer_bytes,
+            "the stall guard did not push it past its priming threshold")
+
+    def test_the_stall_guard_leaves_a_draining_queue_alone(self):
+        # The control. A suite of only failures proves a guard always fires:
+        # a transport whose queue actually falls must never be pushed past
+        # the cap, however long it stays above it.
+        class Draining(FakePCMOutput):
+            def __init__(self, fmt):
+                super().__init__(fmt)
+                self.backlog = 20000
+
+            def queued_size(self):
+                return self.backlog
+
+            def service(self):
+                self.backlog -= 200      # a real device consumes
+
+        transport = Draining(self.fmt)
+        out = self.sample_out.AudioOut(transport, chunk_ms=40,
+                                       lookahead_chunks=2)
+        out.play(FakeSample([bytes(200) for _ in range(200)]))
+        written = len(transport.data)
+        for _ in range(30):
+            self.clock.advance(40)
+            out.service()
+        self.assertEqual(0, out.stalls,
+                         "a draining queue was mistaken for a stalled one")
+        self.assertEqual(written, len(transport.data),
+                         "the pump pulled against a draining backlog")
+
+    def test_channel_adapter_forwards_the_priming_threshold(self):
+        # `ChannelAdapter` is not hypothetical: this repository ships it and
+        # calls it "the producer-facing PCMOutput". It is what a mono graph
+        # on a stereo device gets, and before pydevices#54 it answered 0.
+        stereo = AudioFormat(rate=48000, channels=2, bits=16)
+        mono = AudioFormat(rate=48000, channels=1, bits=16)
+
+        class PrimingTransport(FakePCMOutput):
+            _prebuffer_bytes = 4000
+
+        inner = PrimingTransport(stereo)
+        adapter = audiodev.adapt_channels(inner, mono)
+        self.assertIsNot(adapter, inner, "no wrapper was built at all")
+        # Counted in the wrapper's own frames, like queued_size: a mono
+        # writer needs half the bytes to fill the same wire.
+        self.assertEqual(2000, adapter.prebuffer_bytes)
+
     def test_attach_callback_tolerates_timer_arg(self):
         # appdev.App._dispatch_tick always calls a subscribed callback with
         # one positional arg (the timer object) -- attach() must adapt