Worklet: come back down quickly after a stall

docs/workers.md

@@ -63,8 +63,9 @@ In the worker, `from js import ports` and hand `ports.audio` to
 `audiodev.worklet_audio` (pydevices): set `WorkletOutput(ports.audio).source`
 to a function that renders blocks, or use its `pcm_out(port=ports.audio)` like
 any other audio output. The worklet keeps about 32 ms queued at first and
-works down to about 16 ms when the machine keeps up, backing off after any
-dropout.
+works down to about 16 ms when the machine keeps up. After a dropout it
+queues more for a couple of seconds, then comes back down in one step once
+the workers keep up again; dropouts that keep coming hold it up longer.
 
 ## Web MIDI
 

tests/test_template.py

@@ -1,5 +1,6 @@
 import json
 import re
+import shutil
 import subprocess
 from pathlib import Path
 import unittest
@@ -19,6 +20,13 @@ def _local_asset_paths_from_html(source):
     return paths
 
 
+def _worklet_sim(*args):
+    """Run tests/worklet_sim.mjs; one dict per tenth of a second."""
+    out = subprocess.run(["node", str(ROOT / "tests" / "worklet_sim.mjs"), *map(str, args)],
+                         capture_output=True, text=True, check=True, timeout=120).stdout
+    return [json.loads(line) for line in out.splitlines()]
+
+
 class TemplateTests(unittest.TestCase):
     def test_manifest_and_config_are_valid_json(self):
         manifest = json.loads((ROOT / "manifest.json").read_text(encoding="utf-8"))
@@ -109,6 +117,26 @@ def test_referenced_local_assets_exist_after_vendoring(self):
 
         self.assertFalse(missing, f"referenced assets missing from tree: {missing}")
 
+    @unittest.skipUnless(shutil.which("node"), "needs node")
+    def test_worklet_queue_recovers_after_a_stall(self):
+        # A worker stalls for 0.3 s at 3 s: the queue rises once, not to the
+        # maximum, and is back near its floor a few seconds later.
+        rows = _worklet_sim(3, 0.3, 15)
+        peak = max(r["target"] for r in rows if r["t"] > 3)
+        back = next((r["t"] for r in rows if r["t"] > 3.3 and r["target"] <= 1024), None)
+        self.assertLessEqual(peak, 2048, rows)
+        self.assertIsNotNone(back)
+        self.assertLess(back - 3.3, 3.0)
+
+    @unittest.skipUnless(shutil.which("node"), "needs node")
+    def test_worklet_queue_holds_up_when_a_worker_keeps_running_late(self):
+        # Replies late at random for a minute: the queue grows and stays up,
+        # and dropouts stay few.
+        rows = _worklet_sim(100, 0, 60, 0.3)
+        late = [r["target"] for r in rows if r["t"] >= 10]
+        self.assertGreater(sum(late) / len(late), 1536)
+        self.assertLess(rows[-1]["underruns"], 40)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/worklet_sim.mjs

@@ -0,0 +1,51 @@
+// Runs workers/audio_worklet.js outside a browser: a fake audio thread asks
+// for 128 frames per quantum and a fake worker answers every request at once,
+// except while `stall` says it can't. Prints the queue target over time as
+// JSON lines: {t, target, underruns}.
+//
+//   node tests/worklet_sim.mjs [stallStart stallSeconds total answer]
+import { readFileSync } from "node:fs";
+import { fileURLToPath } from "node:url";
+
+const [stallStart = 3, stallSeconds = 0.3, total = 12, answer = 1] = process.argv.slice(2).map(Number);
+// `answer`: the chance the worker answers in a given quantum (1 = always);
+// below 1 its replies come late at random, as a busy machine's do.
+let seed = 1;
+const random = () => ((seed = (seed * 16807) % 2147483647) / 2147483647);
+const sampleRate = 48000;
+let Processor = null;
+globalThis.sampleRate = sampleRate;
+globalThis.currentFrame = 0;
+globalThis.AudioWorkletProcessor = class { constructor() { this.port = { postMessage() {}, onmessage: null }; } };
+globalThis.registerProcessor = (name, cls) => { Processor = cls; };
+const src = readFileSync(process.env.WORKLET || fileURLToPath(new URL("../workers/audio_worklet.js", import.meta.url)), "utf8");
+new Function(src)();
+
+const node = new Processor({ processorOptions: { channels: 1, minFrames: 768, startFrames: 1536, maxFrames: 4096 } });
+const pending = [];
+const workerPort = {
+  onmessage: null,
+  postMessage(m) { if (m.need) pending.push(m.need); },
+};
+node.onPage({ data: { type: "worker" }, ports: [workerPort] });
+const stream = node.streams[0];
+const quanta = Math.round(total * sampleRate / 128);
+let lastPrint = -1;
+for (let q = 0; q < quanta; q++) {
+  const t = q * 128 / sampleRate;
+  const stalled = t >= stallStart && t < stallStart + stallSeconds;
+  if (!stalled && random() < answer) {
+    while (pending.length) {
+      const blocks = pending.shift();
+      for (let b = 0; b < blocks; b++) stream.port.onmessage({ data: { pcm: new Int16Array(256) } });
+    }
+  }
+  const out = [new Float32Array(128)];
+  node.process([], [out]);
+  globalThis.currentFrame += 128;
+  const tenth = Math.floor(t * 10);
+  if (tenth !== lastPrint) {
+    lastPrint = tenth;
+    console.log(JSON.stringify({ t: tenth / 10, target: node.target, underruns: node.underruns }));
+  }
+}

workers/audio_worklet.js

@@ -12,12 +12,15 @@
 //                                                     context frame k + offset
 //   worklet -> worker  {underrun: count}              underruns since the last one
 //
-// The queue adapts: it starts at `startFrames`, steps down 256 frames at a
-// time towards `minFrames` while every stream's low point stays above 384
-// frames for two seconds, and steps up 512 frames (to at most `maxFrames`)
-// on any underrun, then holds for ten seconds. The browser's audio thread
-// asks for audio about every 10 ms, so a queue much under 768 frames can't
-// keep up.
+// The queue adapts: it starts at `startFrames` and steps up 512 frames (to
+// at most `maxFrames`) on an underrun, once per quarter second however many
+// quanta run short, then holds for two seconds. After that it comes back
+// down quickly: whenever every stream's low point over two seconds leaves
+// 512 frames or more to spare above 384, the target drops by what was spare,
+// and the extra audio queued plays out within a few milliseconds. Near the
+// floor it steps down 256 frames at a time towards `minFrames` while the low
+// point stays above 384 for two seconds. The browser's audio thread asks for
+// audio about every 10 ms, so a queue much under 768 frames can't keep up.
 //
 // The node's own port (to the page) takes {type: "worker"} with a port per
 // stream, {type: "master", gain}, {type: "stats"}, and {type: "tap", on}
@@ -71,7 +74,12 @@ class PythonWorkerOutput extends AudioWorkletProcessor {
     this.underruns = 0;
     this.windowLow = Infinity;
     this.windowStart = 0;
+    this.fastLow = Infinity;
+    this.fastStart = 0;
     this.holdUntil = 0;
+    this.raisedAt = -Infinity;
+    this.lastShort = -Infinity;
+    this.hold = 2 * sampleRate;
     this.low = Infinity;
     this.peak = 0;
     this.tap = null;
@@ -113,12 +121,34 @@ class PythonWorkerOutput extends AudioWorkletProcessor {
   adapt(underran) {
     const now = currentFrame;
     if (underran) {
-      this.target = Math.min(this.maxFrames, this.target + 512);
-      this.holdUntil = now + 10 * sampleRate;
-      this.windowStart = now;
-      this.windowLow = Infinity;
+      // Short quanta less than a second apart are one stall. After a stall,
+      // hold two seconds; when a new one comes within ten seconds of the
+      // last, the machine is struggling: hold ten.
+      const quiet = now - this.lastShort;
+      if (quiet > sampleRate) this.hold = (quiet < 10 * sampleRate ? 10 : 2) * sampleRate;
+      this.lastShort = now;
+      // Within a stall, raise at most every quarter second: a stall of a few
+      // milliseconds would otherwise push the target straight to the maximum.
+      if (now - this.raisedAt >= sampleRate / 4) {
+        this.target = Math.min(this.maxFrames, this.target + 512);
+        this.raisedAt = now;
+      }
+      this.holdUntil = now + this.hold;
+      this.windowStart = this.fastStart = now;
+      this.windowLow = this.fastLow = Infinity;
       return;
     }
+    if (now - this.fastStart >= 2 * sampleRate) {
+      // Plenty to spare: drop by what was spare, all at once.
+      const spare = Math.floor((this.fastLow - 384) / BLOCK) * BLOCK;
+      if (now >= this.holdUntil && spare >= 2 * BLOCK && this.target > this.minFrames) {
+        this.target = Math.max(this.minFrames, this.target - spare);
+        this.windowStart = now;
+        this.windowLow = Infinity;
+      }
+      this.fastStart = now;
+      this.fastLow = Infinity;
+    }
     if (now - this.windowStart >= 2 * sampleRate) {
       if (now >= this.holdUntil && this.windowLow > 384 && this.target > this.minFrames) {
         this.target = Math.max(this.minFrames, this.target - BLOCK);
@@ -200,6 +230,7 @@ class PythonWorkerOutput extends AudioWorkletProcessor {
     if (underran) this.underruns++;
     for (const s of this.streams) {
       if (s.started && s.queued < this.windowLow) this.windowLow = s.queued;
+      if (s.started && s.queued < this.fastLow) this.fastLow = s.queued;
       if (s.started && s.queued < this.low) this.low = s.queued;
     }
     if (this.streams.some((s) => s.started)) this.adapt(underran);