diff --git a/docs/workers.md b/docs/workers.md index 62585a2..b62bd7f 100644 --- a/docs/workers.md +++ b/docs/workers.md @@ -51,6 +51,14 @@ startWorker({ ..., ports: { audio: audio.workerPort } }); resumeOnGesture(audio.context); // browsers start audio on the first click or key ``` +For more than one audio worker, each rendering part of the app on its own +core, give each its own stream: `ports: { audio: audio.addStream() }`. The +worklet sums the streams. A worker that's about to be busy for a while +(building an instrument, say) can post `{pause: true}` on its audio port, and +`{pause: false}` when it's done: the worklet plays its stream as silence +meanwhile instead of counting dropouts. Keep the number of streams to what +you need: each one is asked for audio every 2.7 ms. + 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 @@ -71,6 +79,13 @@ In the worker, `midiports.webmidi.relay_ports(ports.midi)` gives and what you write goes to the page's outputs. Call `midi.port()` once per worker that should hear MIDI. +## Talking to the page + +Things only the page can do, like the browser's file pickers, go through app +messages. From Python, `js.self.postMessage({"type": "app", ...})` arrives in +the page's `handle.onmessage`. The page answers with `handle.send({...})`, +which Python receives by setting `js.self.onPageMessage` to a function. + ## Serving it Workers load as modules and Pyodide fetches its files, so serve the site diff --git a/workers/audio_worklet.js b/workers/audio_worklet.js index c3aa491..57d967c 100644 --- a/workers/audio_worklet.js +++ b/workers/audio_worklet.js @@ -1,25 +1,60 @@ -// The AudioWorklet end of a Python worker's audio (audiodev.worklet_audio). +// The AudioWorklet end of Python workers' audio (audiodev.worklet_audio). // -// It plays 256-frame blocks of int16 PCM the worker sends and asks for more -// over a MessagePort, keeping a queue of `target` frames: +// One or more workers each send a stream of 256-frame int16 blocks (stereo +// interleaved when channels is 2). The worklet sums the streams, applies the +// master gain and a peak limiter, and plays the result. Several workers can +// share the rendering of one app this way, each on its own core. Per stream, +// over its MessagePort: // // worklet -> worker {need: blocks, asked: frame} send that many blocks // worker -> worklet {pcm: Int16Array, asked} (transferred, interleaved) // worklet -> worker {offset} rendered frame k plays at // context frame k + offset -// worklet -> worker {underrun: count} underruns since the last one +// 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 its low point stays above 512 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. +// 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 node's own port (to the page) answers {type: "stats"} and, when a test -// asks with {type: "tap", on: true}, posts a copy of every frame played. +// 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} +// (a test copy of every frame played). const BLOCK = 256; +class Stream { + constructor(port, owner, index) { + this.port = port; + this.index = index; + this.queue = []; + this.head = 0; + this.queued = 0; + this.requested = 0; + this.played = 0; + this.offset = null; + this.started = false; + this.underruns = 0; + this.renderMs = 0; + this.paused = false; + port.onmessage = (ev) => { + if (ev.data.pause !== undefined) { + this.paused = !!ev.data.pause; + if (!this.paused) { this.queue = []; this.head = 0; this.queued = 0; this.requested = 0; this.started = false; } + return; + } + const pcm = ev.data.pcm; + if (!pcm) return; + const frames = pcm.length / owner.channels; + this.queue.push(pcm); + this.queued += frames; + this.requested = Math.max(0, this.requested - frames); + }; + } +} + class PythonWorkerOutput extends AudioWorkletProcessor { constructor(options) { super(); @@ -28,19 +63,17 @@ class PythonWorkerOutput extends AudioWorkletProcessor { this.minFrames = o.minFrames || 768; this.maxFrames = o.maxFrames || 4096; this.target = o.startFrames || 1536; - this.queue = []; - this.head = 0; // sample index into queue[0] - this.queued = 0; // frames queued - this.requested = 0; // frames asked for, not yet here - this.played = 0; // queued frames played so far - this.offset = null; - this.worker = null; - this.started = false; + this.streams = []; + this.gain = 1.0; + this.env = 0; // limiter's peak follower + this.ceiling = 0.97; + this.release = Math.exp(-1 / (0.05 * sampleRate)); this.underruns = 0; this.windowLow = Infinity; this.windowStart = 0; this.holdUntil = 0; - this.low = Infinity; // low point since the last stats request + this.low = Infinity; + this.peak = 0; this.tap = null; this.tapAt = 0; this.tapStart = 0; @@ -50,32 +83,30 @@ class PythonWorkerOutput extends AudioWorkletProcessor { onPage(e) { const m = e.data; if (m.type === "worker") { - this.worker = e.ports[0]; - this.worker.onmessage = (ev) => { - const pcm = ev.data.pcm; - if (!pcm) return; - const frames = pcm.length / this.channels; - this.queue.push(pcm); - this.queued += frames; - this.requested = Math.max(0, this.requested - frames); - }; - this.ask(); + const s = new Stream(e.ports[0], this, this.streams.length); + this.streams.push(s); + this.ask(s); + } else if (m.type === "master") { + this.gain = m.gain; } else if (m.type === "stats") { this.port.postMessage({ type: "stats", underruns: this.underruns, target: this.target, - queued: this.queued, low: this.low, frame: currentFrame }); + low: this.low, peak: this.peak, frame: currentFrame, + streams: this.streams.map((s) => ({ queued: s.queued, underruns: s.underruns, offset: s.offset })) }); this.low = Infinity; + this.peak = 0; } else if (m.type === "tap") { this.tap = m.on ? new Float32Array(Math.floor(sampleRate / 2) * this.channels) : null; this.tapAt = 0; } } - ask() { - const short = this.target - this.queued - this.requested; - if (short > 0 && this.worker) { + ask(s) { + if (s.paused) return; + const short = this.target - s.queued - s.requested; + if (short > 0) { const blocks = Math.ceil(short / BLOCK); - this.requested += blocks * BLOCK; - this.worker.postMessage({ need: blocks, asked: currentFrame }); + s.requested += blocks * BLOCK; + s.port.postMessage({ need: blocks, asked: currentFrame }); } } @@ -88,9 +119,8 @@ class PythonWorkerOutput extends AudioWorkletProcessor { this.windowLow = Infinity; return; } - if (this.queued < this.windowLow) this.windowLow = this.queued; if (now - this.windowStart >= 2 * sampleRate) { - if (now >= this.holdUntil && this.windowLow > 512 && this.target > this.minFrames) { + if (now >= this.holdUntil && this.windowLow > 384 && this.target > this.minFrames) { this.target = Math.max(this.minFrames, this.target - BLOCK); } this.windowStart = now; @@ -98,45 +128,83 @@ class PythonWorkerOutput extends AudioWorkletProcessor { } } - process(inputs, outputs) { - const out = outputs[0]; + // Add one stream's next `frames` frames into `out`; true if it ran short. + mix(s, out, frames, first) { const ch = this.channels; - const frames = out[0].length; + if (s.paused) { + if (first) for (let c = 0; c < out.length; c++) out[c].fill(0); + return false; + } let i = 0; - while (i < frames && this.queue.length) { - const blk = this.queue[0]; - const n = Math.min(frames - i, (blk.length - this.head) / ch); - for (let k = 0; k < n; k++) { - for (let c = 0; c < ch; c++) out[c][i + k] = blk[this.head + k * ch + c] / 32768; + while (i < frames && s.queue.length) { + const blk = s.queue[0]; + const n = Math.min(frames - i, (blk.length - s.head) / ch); + for (let c = 0; c < ch; c++) { + const o = out[c]; + if (first) for (let k = 0; k < n; k++) o[i + k] = blk[s.head + k * ch + c] / 32768; + else for (let k = 0; k < n; k++) o[i + k] += blk[s.head + k * ch + c] / 32768; } i += n; - this.head += n * ch; - this.queued -= n; - if (this.head >= blk.length) { this.queue.shift(); this.head = 0; } + s.head += n * ch; + s.queued -= n; + if (s.head >= blk.length) { s.queue.shift(); s.head = 0; } } - let underran = false; + if (first && i < frames) for (let c = 0; c < out.length; c++) out[c].fill(0, i); + s.played += i; + let short = false; if (i < frames) { - for (let c = 0; c < out.length; c++) out[c].fill(0, i); - if (this.started) { - underran = true; - this.underruns++; - if (this.worker) this.worker.postMessage({ underrun: 1 }); + if (s.started) { + short = true; + s.underruns++; + s.port.postMessage({ underrun: 1 }); } } else { - this.started = true; + s.started = true; } - this.played += i; - if (this.started) { - const offset = currentFrame + frames - this.played; - if (offset !== this.offset && this.worker) { - this.offset = offset; - this.worker.postMessage({ offset }); + if (s.started) { + const offset = currentFrame + frames - s.played; + if (offset !== s.offset) { + s.offset = offset; + s.port.postMessage({ offset }); } - if (this.queued < this.low) this.low = this.queued; - this.adapt(underran); } + return short; + } + + process(inputs, outputs) { + const out = outputs[0]; + const frames = out[0].length; + let underran = false; + if (!this.streams.length) { + for (let c = 0; c < out.length; c++) out[c].fill(0); + } + for (let j = 0; j < this.streams.length; j++) { + if (this.mix(this.streams[j], out, frames, j === 0)) underran = true; + } + // Master gain and a peak limiter: fast attack, 50 ms release. + const ch = this.channels; + for (let k = 0; k < frames; k++) { + let p = 0; + for (let c = 0; c < ch; c++) { + out[c][k] *= this.gain; + const a = Math.abs(out[c][k]); + if (a > p) p = a; + } + this.env = p > this.env ? p : this.env * this.release; + if (p > this.peak) this.peak = p; + if (this.env > this.ceiling) { + const g = this.ceiling / this.env; + for (let c = 0; c < ch; c++) out[c][k] *= g; + } + } + 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.low) this.low = s.queued; + } + if (this.streams.some((s) => s.started)) this.adapt(underran); if (this.tap) this.record(out, frames); - this.ask(); + for (const s of this.streams) this.ask(s); return true; } diff --git a/workers/host.js b/workers/host.js index f58c8f9..8038eb7 100644 --- a/workers/host.js +++ b/workers/host.js @@ -45,7 +45,8 @@ export function startWorker({ name = "python", main, packages = [], files = {}, offscreen = canvas.transferControlToOffscreen(); transfer.push(offscreen); } - const handle = { worker, name, loaded: null, ready: null, error: null }; + const handle = { worker, name, loaded: null, ready: null, error: null, onmessage: null, + send: (msg, transfer) => worker.postMessage({ type: "app", ...msg }, transfer || []) }; handle.loaded = new Promise((resolveLoaded) => { handle.ready = new Promise((resolve, reject) => { worker.onmessage = (e) => { @@ -53,7 +54,7 @@ export function startWorker({ name = "python", main, packages = [], files = {}, if (m.type === "loaded") resolveLoaded(m); else if (m.type === "ready") resolve(m); else if (m.type === "error") { handle.error = m.error; console.error(`[${name}] ${m.error}`); reject(new Error(m.error)); } - else if (handle.onmessage) handle.onmessage(m); + else if (handle.onmessage) handle.onmessage(m); // e.g. {type: "app", ...} from Python }; worker.onerror = (e) => { handle.error = e.message; reject(new Error(e.message || "worker failed to load")); }; }); @@ -97,8 +98,10 @@ export function relayInput(canvas, worker) { // -- Audio ---------------------------------------------------------------------- -// An AudioContext and the worklet that plays a worker's blocks. Hand -// `workerPort` to the audio worker (ports: {audio: ...}). +// An AudioContext and the worklet that plays workers' blocks. Hand +// `workerPort` to the audio worker (ports: {audio: ...}); for more workers +// rendering in parallel, give each its own `addStream()` port and the worklet +// sums them. export async function createAudio({ channels = 1, sampleRate = 48000, minFrames = 768, startFrames = 1536, maxFrames = 4096 } = {}) { const context = new AudioContext({ latencyHint: "interactive", sampleRate }); await context.audioWorklet.addModule(new URL("./audio_worklet.js", import.meta.url)); @@ -107,8 +110,11 @@ export async function createAudio({ channels = 1, sampleRate = 48000, minFrames processorOptions: { channels, minFrames, startFrames, maxFrames }, }); node.connect(context.destination); - const channel = new MessageChannel(); - node.port.postMessage({ type: "worker" }, [channel.port2]); + const addStream = () => { + const channel = new MessageChannel(); + node.port.postMessage({ type: "worker" }, [channel.port2]); + return channel.port1; + }; let statsWaiter = null; const tapChunks = []; node.port.onmessage = (e) => { @@ -116,7 +122,8 @@ export async function createAudio({ channels = 1, sampleRate = 48000, minFrames else if (e.data.type === "tap") tapChunks.push(e.data); }; return { - context, node, workerPort: channel.port1, tapChunks, + context, node, workerPort: addStream(), addStream, tapChunks, + setGain: (gain) => node.port.postMessage({ type: "master", gain }), stats: () => new Promise((r) => { statsWaiter = r; node.port.postMessage({ type: "stats" }); }), tap: (on) => node.port.postMessage({ type: "tap", on }), }; diff --git a/workers/python_worker.js b/workers/python_worker.js index 37f319e..aa075f5 100644 --- a/workers/python_worker.js +++ b/workers/python_worker.js @@ -5,7 +5,8 @@ // OffscreenCanvas to hand over. Python sees them as `js.ports.` and // `js.display_canvas` (board_configs/workerdisplay reads the latter). // Pointer, wheel and key events the page relays go to -// displaydev.workerdisplay once the app has imported it. +// displaydev.workerdisplay once the app has imported it. Messages of type +// "app" go to the Python function set as js.self.onPageMessage. import { loadPyodide } from "../vendor/pyscript/pyodide/pyodide.mjs"; @@ -70,5 +71,11 @@ self.onmessage = (e) => { .catch((err) => self.postMessage({ type: "error", error: String((err && err.stack) || err) })); } else if (msg.type === "input" || msg.type === "resize") { relayInput(msg); + } else if (msg.type === "app" && typeof self.onPageMessage === "function") { + // Anything else the page sends for the app: Python sets + // js.self.onPageMessage to receive it, and answers with + // js.self.postMessage({type: "app", ...}), which arrives in the page's + // handle.onmessage. + self.onPageMessage(msg); } };