From 797501e5ff5b8c3005d099227586ba4f61bb5079 Mon Sep 17 00:00:00 2001 From: OmikronApex Date: Mon, 13 Jul 2026 23:53:27 +0200 Subject: [PATCH] refactor(audio): extract StreamSink (phase 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Promotes the streamer-mix output sink to a class owning its AudioDeviceManager, drain callback, ring, scratches, submix compose (publish, was composeAndPushStreamMix), and open/close/clear/reopen lifecycle — moved verbatim into src/audio/engine/StreamSink.{h,cpp}. Bus flags (includeBacking/includeGuitar/gain) and the level meter move in; engine sample rate / output block size are read through the bound EngineState&. AudioEngine keeps thin facades so the NodeAddon surface is unchanged; the guitar-snapshot scratch stays on the engine (it snapshots the engine's own mix). Compose-matrix unit tests are deferred: they need juce::AudioBuffer, which the JUCE-free engine_units harness doesn't link — covered meanwhile by the stream under/overflow counters + level meter over IPC and the OBS-capture manual smoke. Co-Authored-By: Claude Fable 5 --- src/audio/AudioEngine.cpp | 271 ++------------------------------ src/audio/AudioEngine.h | 98 ++---------- src/audio/CMakeLists.txt | 1 + src/audio/engine/StreamSink.cpp | 263 +++++++++++++++++++++++++++++++ src/audio/engine/StreamSink.h | 142 +++++++++++++++++ 5 files changed, 431 insertions(+), 344 deletions(-) create mode 100644 src/audio/engine/StreamSink.cpp create mode 100644 src/audio/engine/StreamSink.h diff --git a/src/audio/AudioEngine.cpp b/src/audio/AudioEngine.cpp index ecb060f..e79c83c 100644 --- a/src/audio/AudioEngine.cpp +++ b/src/audio/AudioEngine.cpp @@ -1243,7 +1243,7 @@ void AudioEngine::startAudio() // Restore the streamer-mix output device too (same intent-survives-restart // pattern). Best-effort — a failure leaves the sink inactive, never blocks. - reopenDesiredStreamSink(); + streamSink.reopenDesired(); } void AudioEngine::stopAudio() @@ -1277,7 +1277,7 @@ void AudioEngine::stopAudio() // 2nd output device keeps running and underflowing while the engine is "stopped", // and the destructor (which calls stopAudio()) would be the only path that ever // closes it — closing the device here also makes that destructor teardown safe. - closeStreamSinkDevice(); + streamSink.close(); audioRunning.store(false, std::memory_order_relaxed); currentBackingLevel.store(0.0f); } @@ -1836,7 +1836,7 @@ void AudioEngine::audioDeviceAboutToStart(juce::AudioIODevice* device) // once: it can never realloc under a live producer, for any later block size. constexpr int streamScratchCap = (int) kOutputRingFrames; if (streamGuitarScratch.getNumSamples() < streamScratchCap) streamGuitarScratch.setSize(2, streamScratchCap, false, false, true); - if (streamMixScratch.getNumSamples() < streamScratchCap) streamMixScratch.setSize(2, streamScratchCap, false, false, true); + streamSink.prepareProducerScratch(); if (rendererBusPullScratch.getNumSamples() < streamScratchCap) rendererBusPullScratch.setSize(2, streamScratchCap, false, false, true); // Prepare each ACTIVE PRIMARY-device source's DSP and reset its rings for a @@ -1927,7 +1927,7 @@ void AudioEngine::audioOutputAboutToStart(juce::AudioIODevice* device) // note in audioDeviceAboutToStart(). constexpr int streamScratchCap = (int) kOutputRingFrames; if (streamGuitarScratch.getNumSamples() < streamScratchCap) streamGuitarScratch.setSize(2, streamScratchCap, false, false, true); - if (streamMixScratch.getNumSamples() < streamScratchCap) streamMixScratch.setSize(2, streamScratchCap, false, false, true); + streamSink.prepareProducerScratch(); if (rendererBusPullScratch.getNumSamples() < streamScratchCap) rendererBusPullScratch.setSize(2, streamScratchCap, false, false, true); // NOTE: outputBackingBuffer is sized by audioDeviceAboutToStart() from the // INPUT device's block size — it's the split-input DSP scratch, not an @@ -1991,264 +1991,17 @@ void AudioEngine::audioOutputStopped() // ring invariants being re-established, which is harder to reason about. } -// ── Streamer mix output sink (PR1) ────────────────────────────────────────── -// Producer (primary/output callback): compose the stream submix and pack it into -// the sink's ring. Consumer (streamSinkCallback): drain + write to the device. -// Mirrors the InputDeviceSlot ring discipline, inverted to the output side. - -void AudioEngine::composeAndPushStreamMix(const juce::AudioBuffer& guitarMix, - const juce::AudioBuffer* backingBuf, - int backingFrames, float backingVol, - const juce::AudioBuffer* rendererBuf, - int rendererFrames, int numSamples) -{ - if (! streamSink.active.load(std::memory_order_acquire)) return; - // A block larger than the entire ring can't be published atomically (it would - // wrap and overwrite unread slots before writeIndex is bumped). Skip it and - // count an overflow. Checked FIRST (before the scratch guard) so an oversized - // block is always counted — the fixed-size scratch is exactly the ring, so an - // oversized block also trips the scratch guard below and would otherwise be - // dropped silently. The split path already rejects oversized devices at setup; - // this guards the duplex path, whose block size we don't pre-validate. - if (numSamples > (int) kOutputRingFrames) - { - streamSink.overflowCount.fetch_add(1, std::memory_order_relaxed); - return; - } - // Scratch not yet sized to the full ring (cold start before the producer's - // about-to-start ran, or a transient reconfig) — skip rather than alloc on RT. - // After warm-up the scratch is exactly the ring, so for an in-range block this - // never trips. - if (streamMixScratch.getNumSamples() < numSamples) return; - - const bool ig = streamBusIncludeGuitar.load(std::memory_order_relaxed); - const bool ib = streamBusIncludeBacking.load(std::memory_order_relaxed); - const float gain = streamBusGain.load(std::memory_order_relaxed); - - streamMixScratch.clear(0, 0, numSamples); - streamMixScratch.clear(1, 0, numSamples); - if (ig) - for (int ch = 0; ch < 2; ++ch) - streamMixScratch.addFrom(ch, 0, guitarMix, - juce::jmin(ch, guitarMix.getNumChannels() - 1), 0, numSamples); - if (ib && backingBuf != nullptr && backingFrames > 0) - { - const int n = juce::jmin(backingFrames, numSamples); - for (int ch = 0; ch < 2; ++ch) - streamMixScratch.addFrom(ch, 0, *backingBuf, - juce::jmin(ch, backingBuf->getNumChannels() - 1), 0, n, backingVol); - } - // Renderer-fed song audio (stems / element / loopback riding the renderer - // bus) is song audio for the streamer too — without this the stream mix - // carries guitar only whenever the song bypasses the native transport - // (multi-stem under exclusive/ASIO output). Bus gain is already applied by - // pullRendererBus; only the stream gain below shapes it further. Backing - // transport and renderer bus are mutually exclusive song paths in - // practice, so this never double-carries. - if (ib && rendererBuf != nullptr && rendererFrames > 0) - { - const int n = juce::jmin(rendererFrames, numSamples); - for (int ch = 0; ch < 2; ++ch) - streamMixScratch.addFrom(ch, 0, *rendererBuf, - juce::jmin(ch, rendererBuf->getNumChannels() - 1), 0, n); - } - streamMixScratch.applyGain(0, 0, numSamples, gain); - streamMixScratch.applyGain(1, 0, numSamples, gain); - - const float peak = juce::jmax(streamMixScratch.getMagnitude(0, 0, numSamples), - streamMixScratch.getMagnitude(1, 0, numSamples)); - streamSinkLevel.store(peak, std::memory_order_relaxed); - - streamSink.ring.push(streamMixScratch.getReadPointer(0), - streamMixScratch.getReadPointer(1), numSamples); -} - -void AudioEngine::streamSinkCallback(float* const* outputData, int numOutputChannels, int numSamples) -{ - const juce::ScopedNoDenormals noDenormals; - if (numOutputChannels <= 0) return; - juce::AudioBuffer buffer(outputData, numOutputChannels, numSamples); - buffer.clear(); - - const int scratchCap = (int) streamSink.pullScratchL.size(); - const int outSamples = juce::jmin(numSamples, scratchCap); - - auto& ring = streamSink.ring; - uint64_t r = ring.readIndex.load(std::memory_order_relaxed); - const uint64_t w = ring.writeIndex.load(std::memory_order_acquire); - ring.resyncIfIndicesReset(r, w); - if (ring.catchUpIfLapped(r, w)) - streamSink.overflowCount.fetch_add(1, std::memory_order_relaxed); - const uint64_t available = w - r; - const int pullCount = juce::jmin(outSamples, (int) available); - const int consumeCount = juce::jmin(numSamples, (int) available); - const int copyChannels = juce::jmin(numOutputChannels, 2); - for (int i = 0; i < pullCount; ++i) - { - float l, rr; - ring.readFrame(r + (uint64_t) i, l, rr); - buffer.setSample(0, i, l); - if (copyChannels > 1) buffer.setSample(1, i, rr); - } - if (pullCount < outSamples) - streamSink.underflowCount.fetch_add(1, std::memory_order_relaxed); - ring.commitRead(r + (uint64_t) consumeCount); -} - -void AudioEngine::streamSinkAboutToStart(juce::AudioIODevice* device) -{ - if (device == nullptr) return; - const int bs = device->getCurrentBufferSizeSamples(); - double sr = device->getCurrentSampleRate(); - if (sr <= 0.0) sr = currentSampleRate.load(std::memory_order_relaxed); - streamSink.blockSize.store(bs, std::memory_order_relaxed); - streamSink.sampleRate.store(sr, std::memory_order_relaxed); - const int cap = juce::jmax(bs, 2048); - if ((int) streamSink.pullScratchL.size() < cap) streamSink.pullScratchL.assign((size_t) cap, 0.0f); - if ((int) streamSink.pullScratchR.size() < cap) streamSink.pullScratchR.assign((size_t) cap, 0.0f); - streamSink.ring.reset(); - streamSink.underflowCount.store(0, std::memory_order_relaxed); - streamSink.overflowCount.store(0, std::memory_order_relaxed); -} - -void AudioEngine::streamSinkStopped() -{ - // Fires on any stop of the stream device — an unplanned loss (unplug / driver - // reset) as well as our own teardown. Mark inactive so the producer stops - // filling a now-consumer-less ring and the meter clears; desiredTypeName/Name - // are deliberately left intact so reopenDesiredStreamSink() can restore it. - streamSink.active.store(false, std::memory_order_release); - streamSinkLevel.store(0.0f, std::memory_order_relaxed); -} +// ── Streamer mix output sink — moved to engine/StreamSink.{h,cpp} (TLC +// phase 2). Facades below keep the NodeAddon-visible surface unchanged. juce::String AudioEngine::setStreamOutputDevice(const juce::String& typeName, const juce::String& deviceName) { - // Control-thread only. Opens an OUTPUT-only device on the stream sink's own - // AudioDeviceManager and attaches the drain callback. Mirrors applySplitSetup's - // output open. v1 requires the sink's nominal SR to match the engine rate (no - // async resampler yet — that's PR3); a mismatch is rejected with a clear error. - streamSink.desiredTypeName = typeName; - streamSink.desiredDeviceName = deviceName; - - // Stop the producer from writing the ring while we (re)configure the device: - // setAudioDeviceSetup() below drives streamSinkAboutToStart(), which resets the - // ring indices and slots. Clearing `active` first stops the main callback from - // STARTING new pushes. A producer block already past the active-check can still - // finish one push, but the device close/reopen takes far longer than a single - // audio block, so that in-flight push completes well before about-to-start runs. - // Worst case is therefore one imperfect block on the STREAM bus (never the local - // monitor) during a manual device switch — atomic, no data race, no UAF. We only - // re-arm `active` after a fully clean open. - streamSink.active.store(false, std::memory_order_release); - - // Any failure below: detach/close the half-open device AND drop the desired - // intent. A deterministic failure (e.g. SR mismatch) is then NOT retried on every - // startAudio(), and the engine never reports active with no device behind it. The - // renderer keeps its own persisted choice and re-applies it, so nothing is lost. - auto fail = [this](const juce::String& msg) -> juce::String { - closeStreamSinkDevice(); - streamSink.desiredTypeName = {}; - streamSink.desiredDeviceName = {}; - return msg; - }; - - if (! streamSink.initialised) - { - streamSink.manager.initialise(0, 2, nullptr, false); - streamSink.initialised = true; - } - - juce::AudioIODeviceType* outType = nullptr; - for (auto* t : streamSink.manager.getAvailableDeviceTypes()) - if (t->getTypeName() == typeName) { outType = t; break; } - if (! outType) return fail("Stream output device type not found: " + typeName); - - try { - if (auto* cur = streamSink.manager.getCurrentDeviceTypeObject()) - { - if (cur->getTypeName() != typeName) - streamSink.manager.setCurrentAudioDeviceType(typeName, true); - } - else streamSink.manager.setCurrentAudioDeviceType(typeName, true); - } catch (...) { return fail("setCurrentAudioDeviceType threw for stream output type '" + typeName + "'"); } - - juce::String resolved = deviceName; - if (resolved.isEmpty()) - { - auto names = outType->getDeviceNames(false); - if (names.size() > 0) resolved = names[0]; - } - - juce::AudioDeviceManager::AudioDeviceSetup setup; - setup.inputDeviceName = ""; - setup.outputDeviceName = resolved; - setup.sampleRate = currentSampleRate.load(std::memory_order_relaxed); - setup.bufferSize = outputBlockSize.load(std::memory_order_relaxed); - setup.useDefaultInputChannels = false; - setup.useDefaultOutputChannels = false; - setup.inputChannels.clear(); - setup.outputChannels.setRange(0, 2, true); - - juce::String err; - try { err = streamSink.manager.setAudioDeviceSetup(setup, true); } - catch (...) { return fail("stream output setAudioDeviceSetup threw"); } - if (err.isNotEmpty()) return fail("stream output setup: " + err); - - auto* dev = streamSink.manager.getCurrentAudioDevice(); - if (! dev) return fail("no stream output device after setup"); - const double devSr = dev->getCurrentSampleRate(); - const double engineSr = currentSampleRate.load(std::memory_order_relaxed); - if (engineSr > 0.0 && std::abs(devSr - engineSr) > 0.5) - return fail("Stream output sample rate (" + juce::String(devSr) - + ") must match the engine rate (" + juce::String(engineSr) - + "). Pick a device that supports " + juce::String(engineSr) + " Hz."); - - streamSink.callback.engine = this; - if (! streamSink.callbackRegistered) - { - streamSink.manager.addAudioCallback(&streamSink.callback); - streamSink.callbackRegistered = true; - } - streamSink.active.store(true, std::memory_order_release); - fprintf(stderr, "[AudioEngine] stream output active: %s (%s)\n", - resolved.toRawUTF8(), typeName.toRawUTF8()); - return {}; -} - -void AudioEngine::closeStreamSinkDevice() -{ - // Detach the drain callback and close the device, leaving desiredTypeName/Name - // intact so startAudio()/reopenDesiredStreamSink() can restore it. active=false - // first so the producer stops pushing; removeAudioCallback() then blocks until - // the consumer callback is no longer in flight, so the ring/device go quiescent - // before close. Idempotent — safe when nothing is open. - streamSink.active.store(false, std::memory_order_release); - if (streamSink.callbackRegistered) - { - streamSink.manager.removeAudioCallback(&streamSink.callback); - streamSink.callbackRegistered = false; - } - try { streamSink.manager.closeAudioDevice(); } catch (...) {} - streamSinkLevel.store(0.0f, std::memory_order_relaxed); + return streamSink.open(typeName, deviceName); } void AudioEngine::clearStreamOutput() { - closeStreamSinkDevice(); - streamSink.desiredTypeName = {}; - streamSink.desiredDeviceName = {}; -} - -void AudioEngine::reopenDesiredStreamSink() -{ - // Copy the intent first: setStreamOutputDevice() mutates desiredTypeName/Name - // (and clears them on failure), so don't pass the members in by reference. - const juce::String t = streamSink.desiredTypeName; - const juce::String d = streamSink.desiredDeviceName; - if (d.isEmpty() && t.isEmpty()) return; - const juce::String err = setStreamOutputDevice(t, d); - if (err.isNotEmpty()) - fprintf(stderr, "[AudioEngine] reopenDesiredStreamSink failed: %s\n", err.toRawUTF8()); + streamSink.clear(); } void AudioEngine::audioDeviceIOCallbackWithContext( @@ -2335,7 +2088,7 @@ void AudioEngine::audioDeviceIOCallbackWithContext( { // Stream sink (producer): snapshot the guitar monitor mix BEFORE backing is // added, so the stream submix can carry guitar independent of the local mix. - const bool streamActive = streamSink.active.load(std::memory_order_acquire); + const bool streamActive = streamSink.isActive(); int streamBackingFrames = 0; float streamBackingVol = 0.0f; bool streamBackingOn = false; @@ -2383,7 +2136,7 @@ void AudioEngine::audioDeviceIOCallbackWithContext( // backing + renderer-bus song audio) BEFORE the local master output // gain, so the stream level is independent. if (streamActive) - composeAndPushStreamMix(streamGuitarScratch, + streamSink.publish(streamGuitarScratch, streamBackingOn ? &backingBuffer : nullptr, streamBackingFrames, streamBackingVol, rendererFrames > 0 ? &rendererBusPullScratch : nullptr, @@ -2959,7 +2712,7 @@ void AudioEngine::audioOutputCallback(const float* const* /*inputData*/, // Stream sink (producer, split clock): snapshot the full guitar mix (primary // ring + extra inputs) BEFORE backing is added. - const bool streamActive = streamSink.active.load(std::memory_order_acquire); + const bool streamActive = streamSink.isActive(); int streamBackingFrames = 0; float streamBackingVol = 0.0f; bool streamBackingOn = false; @@ -3013,7 +2766,7 @@ void AudioEngine::audioOutputCallback(const float* const* /*inputData*/, // path. When the tryLock failed (streamBackingOn=false) we pass a null backing // pointer and never touch backingBuffer, so there is nothing to protect. if (streamActive) - composeAndPushStreamMix(streamGuitarScratch, + streamSink.publish(streamGuitarScratch, streamBackingOn ? &backingBuffer : nullptr, streamBackingFrames, streamBackingVol, rendererFrames > 0 ? &rendererBusPullScratch : nullptr, diff --git a/src/audio/AudioEngine.h b/src/audio/AudioEngine.h index 81e580c..aab50a0 100644 --- a/src/audio/AudioEngine.h +++ b/src/audio/AudioEngine.h @@ -4,6 +4,7 @@ #include "engine/PackedStereoRing.h" #include "engine/EngineState.h" #include "engine/RendererBus.h" +#include "engine/StreamSink.h" #include "BackingLeveler.h" #include "signalsmith-stretch.h" #include @@ -252,18 +253,13 @@ public: // setStreamOutputDevice returns "" on success or an error string. juce::String setStreamOutputDevice(const juce::String& typeName, const juce::String& deviceName); void clearStreamOutput(); - bool isStreamOutputActive() const { return streamSink.active.load(std::memory_order_acquire); } - juce::String getStreamOutputDeviceName() const { return streamSink.desiredDeviceName; } + bool isStreamOutputActive() const { return streamSink.isActive(); } + juce::String getStreamOutputDeviceName() const { return streamSink.getDesiredDeviceName(); } // Bus content: include the backing/game, include the guitar monitor mix, and a // linear output gain. All atomic — safe to set live. Gain is sanitised // (finite, clamped 0..8) so a NaN/Inf from JS can never reach the stream ring. - void setStreamBus(bool includeBacking, bool includeGuitar, float gain) - { - streamBusIncludeBacking.store(includeBacking, std::memory_order_relaxed); - streamBusIncludeGuitar.store(includeGuitar, std::memory_order_relaxed); - streamBusGain.store(sanitizeStreamGain(gain), std::memory_order_relaxed); - } - void setStreamBusGain(float gain) { streamBusGain.store(sanitizeStreamGain(gain), std::memory_order_relaxed); } + void setStreamBus(bool includeBacking, bool includeGuitar, float gain) { streamSink.setBus(includeBacking, includeGuitar, gain); } + void setStreamBusGain(float gain) { streamSink.setBusGain(gain); } // ── Renderer-audio bus (Phase 2: WebAudio master → engine output) ───────── // The renderer pushes its WebAudio master mix here (via IPC) so song/stem @@ -285,11 +281,11 @@ public: }; RendererBusMetrics getRendererBusMetrics() const; - float getStreamSinkLevel() const { return streamSinkLevel.load(std::memory_order_relaxed); } - uint64_t getStreamUnderflowCount() const { return streamSink.underflowCount.load(std::memory_order_relaxed); } + float getStreamSinkLevel() const { return streamSink.getLevel(); } + uint64_t getStreamUnderflowCount() const { return streamSink.getUnderflowCount(); } // Producer overflow (drop-oldest): the consumer fell a full ring behind and // frames were skipped. Exposed alongside underflow for stream drift diagnosis. - uint64_t getStreamOverflowCount() const { return streamSink.overflowCount.load(std::memory_order_relaxed); } + uint64_t getStreamOverflowCount() const { return streamSink.getOverflowCount(); } // Latency double getLatencyMs() const; @@ -658,84 +654,16 @@ private: juce::AudioBuffer& mixBuf, juce::AudioBuffer& monitorScratch, int effectiveOutputChannels, int numSamples); - // ── Streamer mix output sink (PR1) ─────────────────────────────────────── - // A second OUTPUT AudioDeviceManager on its OWN clock that drains a dedicated - // SPSC ring fed by the main output path's composed stream submix. This mirrors - // the InputDeviceSlot pattern INVERTED to the output side: the PRODUCER is the - // primary/output callback (composeAndPushStreamMix), the CONSUMER is this extra - // output device's callback (streamSinkCallback). Default off → no behaviour change. - struct StreamSinkCallback : juce::AudioIODeviceCallback - { - AudioEngine* engine = nullptr; - void audioDeviceIOCallbackWithContext(const float* const* inputData, int numInputChannels, - float* const* outputData, int numOutputChannels, - int numSamples, - const juce::AudioIODeviceCallbackContext&) override - { - juce::ignoreUnused(inputData, numInputChannels); - if (engine) engine->streamSinkCallback(outputData, numOutputChannels, numSamples); - } - void audioDeviceAboutToStart(juce::AudioIODevice* d) override { if (engine) engine->streamSinkAboutToStart(d); } - void audioDeviceStopped() override { if (engine) engine->streamSinkStopped(); } - }; - struct StreamSink - { - StreamSinkCallback callback; - slopsmith::PackedStereoRing ring; - std::atomic underflowCount{0}; - std::atomic overflowCount{0}; - std::atomic active{false}; - std::atomic sampleRate{48000.0}; - std::atomic blockSize{256}; - std::vector pullScratchL, pullScratchR; // sized in streamSinkAboutToStart - bool callbackRegistered = false; - bool initialised = false; - // Declared LAST so it DESTRUCTS FIRST (members tear down in reverse - // declaration order): the manager's dtor closes the device and detaches - // `callback` while `callback`/`ring` are still alive — no use-after-free - // even if an explicit teardown path is ever missed. stopAudio() / - // closeStreamSinkDevice() also tear it down explicitly before this. - juce::AudioDeviceManager manager; - // Persistent INTENT (control thread only): the device the user chose. - // Survives a stop/restart so reopenDesiredStreamSink() can re-open it. - juce::String desiredTypeName; - juce::String desiredDeviceName; - }; - StreamSink streamSink; - std::atomic streamBusIncludeBacking{true}; - std::atomic streamBusIncludeGuitar{true}; - std::atomic streamBusGain{1.0f}; - std::atomic streamSinkLevel{0.0f}; - // Producer-side scratch (written by the primary/output callback): the guitar - // monitor-mix snapshot (pre-backing) and the composed stream submix. Sized in + // ── Streamer mix output sink — moved to engine/StreamSink.{h,cpp} (TLC + // phase 2). Declared after `state` (bound by reference). + slopsmith::StreamSink streamSink{state}; + // Producer-side guitar monitor-mix snapshot (pre-backing), written by the + // primary/output callback and handed to streamSink.publish(). Sized in // audioDeviceAboutToStart / audioOutputAboutToStart alongside the other scratch. juce::AudioBuffer streamGuitarScratch; - juce::AudioBuffer streamMixScratch; // Clamp a requested stream gain to a finite, sane range so a NaN/Inf (or a // wild value) from the JS bridge can never be packed into the stream ring. static float sanitizeStreamGain(float g) { return slopsmith::sanitizeStreamGain(g); } - - void streamSinkCallback(float* const* outputData, int numOutputChannels, int numSamples); - void streamSinkAboutToStart(juce::AudioIODevice* device); - void streamSinkStopped(); - void reopenDesiredStreamSink(); - // Detach + close the stream-sink device but KEEP desiredTypeName/Name, so a - // stopAudio()/startAudio() cycle re-opens it (intent survives, like extra - // inputs). Also the single teardown used by the dtor and clearStreamOutput(). - void closeStreamSinkDevice(); - // Compose the stream submix from the captured guitar mix + the just-rendered - // backing block + the just-pulled renderer-bus block and pack it into the - // stream ring. Called from both output callbacks after backing render. - // `backingBuf` / `rendererBuf` may be null (not playing / bus gated). - // The renderer bus rides the includeBacking flag: it IS song audio, just - // fed from the renderer instead of the native transport (bus gain already - // applied by pullRendererBus). - void composeAndPushStreamMix(const juce::AudioBuffer& guitarMix, - const juce::AudioBuffer* backingBuf, - int backingFrames, float backingVol, - const juce::AudioBuffer* rendererBuf, - int rendererFrames, int numSamples); - JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR(AudioEngine) }; diff --git a/src/audio/CMakeLists.txt b/src/audio/CMakeLists.txt index b4870d6..7e6175e 100644 --- a/src/audio/CMakeLists.txt +++ b/src/audio/CMakeLists.txt @@ -6,6 +6,7 @@ set(AUDIO_SOURCES NoiseGate.cpp TonePolish.cpp AudioEngine.cpp + engine/StreamSink.cpp SourceChain.cpp SignalChain.cpp VSTHost.cpp diff --git a/src/audio/engine/StreamSink.cpp b/src/audio/engine/StreamSink.cpp new file mode 100644 index 0000000..a052b71 --- /dev/null +++ b/src/audio/engine/StreamSink.cpp @@ -0,0 +1,263 @@ +// StreamSink implementation — moved verbatim from AudioEngine.cpp (TLC plan +// phase 2 / §2.5); member names lose their streamSink./streamBus prefixes, +// logic is unchanged. See StreamSink.h for the design rationale. + +#include "StreamSink.h" + +#include +#include + +namespace slopsmith { + +void StreamSink::publish(const juce::AudioBuffer& guitarMix, + const juce::AudioBuffer* backingBuf, int backingFrames, float backingVol, + const juce::AudioBuffer* rendererBuf, int rendererFrames, + int numSamples) +{ + if (! active.load(std::memory_order_acquire)) return; + // A block larger than the entire ring can't be published atomically (it would + // wrap and overwrite unread slots before writeIndex is bumped). Skip it and + // count an overflow. Checked FIRST (before the scratch guard) so an oversized + // block is always counted — the fixed-size scratch is exactly the ring, so an + // oversized block also trips the scratch guard below and would otherwise be + // dropped silently. The split path already rejects oversized devices at setup; + // this guards the duplex path, whose block size we don't pre-validate. + if (numSamples > kRingFrames) + { + overflowCount.fetch_add(1, std::memory_order_relaxed); + return; + } + // Scratch not yet sized to the full ring (cold start before the producer's + // about-to-start ran, or a transient reconfig) — skip rather than alloc on RT. + // After warm-up the scratch is exactly the ring, so for an in-range block this + // never trips. + if (mixScratch.getNumSamples() < numSamples) return; + + const bool ig = busIncludeGuitar.load(std::memory_order_relaxed); + const bool ib = busIncludeBacking.load(std::memory_order_relaxed); + const float gain = busGain.load(std::memory_order_relaxed); + + mixScratch.clear(0, 0, numSamples); + mixScratch.clear(1, 0, numSamples); + if (ig) + for (int ch = 0; ch < 2; ++ch) + mixScratch.addFrom(ch, 0, guitarMix, + juce::jmin(ch, guitarMix.getNumChannels() - 1), 0, numSamples); + if (ib && backingBuf != nullptr && backingFrames > 0) + { + const int n = juce::jmin(backingFrames, numSamples); + for (int ch = 0; ch < 2; ++ch) + mixScratch.addFrom(ch, 0, *backingBuf, + juce::jmin(ch, backingBuf->getNumChannels() - 1), 0, n, backingVol); + } + // Renderer-fed song audio (stems / element / loopback riding the renderer + // bus) is song audio for the streamer too — without this the stream mix + // carries guitar only whenever the song bypasses the native transport + // (multi-stem under exclusive/ASIO output). Bus gain is already applied by + // pullRendererBus; only the stream gain below shapes it further. Backing + // transport and renderer bus are mutually exclusive song paths in + // practice, so this never double-carries. + if (ib && rendererBuf != nullptr && rendererFrames > 0) + { + const int n = juce::jmin(rendererFrames, numSamples); + for (int ch = 0; ch < 2; ++ch) + mixScratch.addFrom(ch, 0, *rendererBuf, + juce::jmin(ch, rendererBuf->getNumChannels() - 1), 0, n); + } + mixScratch.applyGain(0, 0, numSamples, gain); + mixScratch.applyGain(1, 0, numSamples, gain); + + const float peak = juce::jmax(mixScratch.getMagnitude(0, 0, numSamples), + mixScratch.getMagnitude(1, 0, numSamples)); + level.store(peak, std::memory_order_relaxed); + + ring.push(mixScratch.getReadPointer(0), mixScratch.getReadPointer(1), numSamples); +} + +void StreamSink::deviceCallback(float* const* outputData, int numOutputChannels, int numSamples) +{ + const juce::ScopedNoDenormals noDenormals; + if (numOutputChannels <= 0) return; + juce::AudioBuffer buffer(outputData, numOutputChannels, numSamples); + buffer.clear(); + + const int scratchCap = (int) pullScratchL.size(); + const int outSamples = juce::jmin(numSamples, scratchCap); + + uint64_t r = ring.readIndex.load(std::memory_order_relaxed); + const uint64_t w = ring.writeIndex.load(std::memory_order_acquire); + ring.resyncIfIndicesReset(r, w); + if (ring.catchUpIfLapped(r, w)) + overflowCount.fetch_add(1, std::memory_order_relaxed); + const uint64_t available = w - r; + const int pullCount = juce::jmin(outSamples, (int) available); + const int consumeCount = juce::jmin(numSamples, (int) available); + const int copyChannels = juce::jmin(numOutputChannels, 2); + for (int i = 0; i < pullCount; ++i) + { + float l, rr; + ring.readFrame(r + (uint64_t) i, l, rr); + buffer.setSample(0, i, l); + if (copyChannels > 1) buffer.setSample(1, i, rr); + } + if (pullCount < outSamples) + underflowCount.fetch_add(1, std::memory_order_relaxed); + ring.commitRead(r + (uint64_t) consumeCount); +} + +void StreamSink::deviceAboutToStart(juce::AudioIODevice* device) +{ + if (device == nullptr) return; + const int bs = device->getCurrentBufferSizeSamples(); + double sr = device->getCurrentSampleRate(); + if (sr <= 0.0) sr = state.currentSampleRate.load(std::memory_order_relaxed); + sinkBlockSize.store(bs, std::memory_order_relaxed); + sinkSampleRate.store(sr, std::memory_order_relaxed); + const int cap = juce::jmax(bs, 2048); + if ((int) pullScratchL.size() < cap) pullScratchL.assign((size_t) cap, 0.0f); + if ((int) pullScratchR.size() < cap) pullScratchR.assign((size_t) cap, 0.0f); + ring.reset(); + underflowCount.store(0, std::memory_order_relaxed); + overflowCount.store(0, std::memory_order_relaxed); +} + +void StreamSink::deviceStopped() +{ + // Fires on any stop of the stream device — an unplanned loss (unplug / driver + // reset) as well as our own teardown. Mark inactive so the producer stops + // filling a now-consumer-less ring and the meter clears; desiredTypeName/Name + // are deliberately left intact so reopenDesired() can restore it. + active.store(false, std::memory_order_release); + level.store(0.0f, std::memory_order_relaxed); +} + +juce::String StreamSink::open(const juce::String& typeName, const juce::String& deviceName) +{ + // Control-thread only. Opens an OUTPUT-only device on the stream sink's own + // AudioDeviceManager and attaches the drain callback. Mirrors applySplitSetup's + // output open. v1 requires the sink's nominal SR to match the engine rate (no + // async resampler yet — that's PR3); a mismatch is rejected with a clear error. + desiredTypeName = typeName; + desiredDeviceName = deviceName; + + // Stop the producer from writing the ring while we (re)configure the device: + // setAudioDeviceSetup() below drives deviceAboutToStart(), which resets the + // ring indices and slots. Clearing `active` first stops the main callback from + // STARTING new pushes. A producer block already past the active-check can still + // finish one push, but the device close/reopen takes far longer than a single + // audio block, so that in-flight push completes well before about-to-start runs. + // Worst case is therefore one imperfect block on the STREAM bus (never the local + // monitor) during a manual device switch — atomic, no data race, no UAF. We only + // re-arm `active` after a fully clean open. + active.store(false, std::memory_order_release); + + // Any failure below: detach/close the half-open device AND drop the desired + // intent. A deterministic failure (e.g. SR mismatch) is then NOT retried on every + // startAudio(), and the engine never reports active with no device behind it. The + // renderer keeps its own persisted choice and re-applies it, so nothing is lost. + auto fail = [this](const juce::String& msg) -> juce::String { + close(); + desiredTypeName = {}; + desiredDeviceName = {}; + return msg; + }; + + if (! initialised) + { + manager.initialise(0, 2, nullptr, false); + initialised = true; + } + + juce::AudioIODeviceType* outType = nullptr; + for (auto* t : manager.getAvailableDeviceTypes()) + if (t->getTypeName() == typeName) { outType = t; break; } + if (! outType) return fail("Stream output device type not found: " + typeName); + + try { + if (auto* cur = manager.getCurrentDeviceTypeObject()) + { + if (cur->getTypeName() != typeName) + manager.setCurrentAudioDeviceType(typeName, true); + } + else manager.setCurrentAudioDeviceType(typeName, true); + } catch (...) { return fail("setCurrentAudioDeviceType threw for stream output type '" + typeName + "'"); } + + juce::String resolved = deviceName; + if (resolved.isEmpty()) + { + auto names = outType->getDeviceNames(false); + if (names.size() > 0) resolved = names[0]; + } + + juce::AudioDeviceManager::AudioDeviceSetup setup; + setup.inputDeviceName = ""; + setup.outputDeviceName = resolved; + setup.sampleRate = state.currentSampleRate.load(std::memory_order_relaxed); + setup.bufferSize = state.outputBlockSize.load(std::memory_order_relaxed); + setup.useDefaultInputChannels = false; + setup.useDefaultOutputChannels = false; + setup.inputChannels.clear(); + setup.outputChannels.setRange(0, 2, true); + + juce::String err; + try { err = manager.setAudioDeviceSetup(setup, true); } + catch (...) { return fail("stream output setAudioDeviceSetup threw"); } + if (err.isNotEmpty()) return fail("stream output setup: " + err); + + auto* dev = manager.getCurrentAudioDevice(); + if (! dev) return fail("no stream output device after setup"); + const double devSr = dev->getCurrentSampleRate(); + const double engineSr = state.currentSampleRate.load(std::memory_order_relaxed); + if (engineSr > 0.0 && std::abs(devSr - engineSr) > 0.5) + return fail("Stream output sample rate (" + juce::String(devSr) + + ") must match the engine rate (" + juce::String(engineSr) + + "). Pick a device that supports " + juce::String(engineSr) + " Hz."); + + if (! callbackRegistered) + { + manager.addAudioCallback(&callback); + callbackRegistered = true; + } + active.store(true, std::memory_order_release); + fprintf(stderr, "[AudioEngine] stream output active: %s (%s)\n", + resolved.toRawUTF8(), typeName.toRawUTF8()); + return {}; +} + +void StreamSink::close() +{ + // Detach the drain callback and close the device, leaving desiredTypeName/Name + // intact so startAudio()/reopenDesired() can restore it. active=false + // first so the producer stops pushing; removeAudioCallback() then blocks until + // the consumer callback is no longer in flight, so the ring/device go quiescent + // before close. Idempotent — safe when nothing is open. + active.store(false, std::memory_order_release); + if (callbackRegistered) + { + manager.removeAudioCallback(&callback); + callbackRegistered = false; + } + try { manager.closeAudioDevice(); } catch (...) {} + level.store(0.0f, std::memory_order_relaxed); +} + +void StreamSink::clear() +{ + close(); + desiredTypeName = {}; + desiredDeviceName = {}; +} + +void StreamSink::reopenDesired() +{ + // Copy the intent first: open() mutates desiredTypeName/Name (and clears + // them on failure), so don't pass the members in by reference. + const juce::String t = desiredTypeName; + const juce::String d = desiredDeviceName; + if (d.isEmpty() && t.isEmpty()) return; + const juce::String err = open(t, d); + if (err.isNotEmpty()) + fprintf(stderr, "[AudioEngine] reopenDesiredStreamSink failed: %s\n", err.toRawUTF8()); +} + +} // namespace slopsmith diff --git a/src/audio/engine/StreamSink.h b/src/audio/engine/StreamSink.h new file mode 100644 index 0000000..3d7f03a --- /dev/null +++ b/src/audio/engine/StreamSink.h @@ -0,0 +1,142 @@ +#pragma once + +// StreamSink — the streamer-mix output sink (TLC plan phase 2 / §2.5, was +// "PR1" inside AudioEngine). A second OUTPUT AudioDeviceManager on its OWN +// clock that drains a dedicated SPSC ring fed by the main output path's +// composed stream submix (publish()). Mirrors the InputDeviceSlot pattern +// INVERTED to the output side: the PRODUCER is the primary/output callback, +// the CONSUMER is this extra output device's callback. Default off → no +// behaviour change. +// +// Moved verbatim from AudioEngine; the engine keeps thin facades +// (setStreamOutputDevice / clearStreamOutput / setStreamBus / metrics +// getters) so the NodeAddon surface is unchanged. Engine sample rate / output +// block size are read through the EngineState& bound at construction. + +#include "PackedStereoRing.h" +#include "EngineState.h" +#include "../GainSanitize.h" + +#include + +#include +#include +#include + +namespace slopsmith { + +class StreamSink +{ +public: + // Must match the main engine ring capacity: publish() rejects blocks + // larger than one ring (they can't be published atomically). + static constexpr int kRingFrames = 4096; + + explicit StreamSink(EngineState& engineState) : state(engineState) + { + callback.sink = this; + } + + // ── Control thread ──────────────────────────────────────────────────── + // Open an OUTPUT-only device and attach the drain callback. Empty error + // string = success. v1 requires the sink's nominal SR to match the engine + // rate (no async resampler yet); a mismatch is rejected with a clear error. + juce::String open(const juce::String& typeName, const juce::String& deviceName); + // Detach + close but KEEP desiredTypeName/Name so reopenDesired() can + // restore it after a stop/restart (intent survives). Idempotent. + void close(); + // close() + drop the desired intent (a user "no stream output"). + void clear(); + void reopenDesired(); + + // Size the producer-side scratches to the FIXED ring capacity — call from + // the engine's about-to-start hooks (device-management thread), never RT. + // Fixed cap so a hotplug about-to-start can never realloc under a live + // producer on the other clock; allocates exactly once. + void prepareProducerScratch() + { + if (mixScratch.getNumSamples() < kRingFrames) + mixScratch.setSize(2, kRingFrames, false, false, true); + } + + // Bus content: include the backing/game, include the guitar monitor mix, + // and a linear output gain. All atomic — safe to set live. Gain sanitised + // (finite, 0..8) so a NaN/Inf from JS can never reach the stream ring. + void setBus(bool includeBacking, bool includeGuitar, float gain) + { + busIncludeBacking.store(includeBacking, std::memory_order_relaxed); + busIncludeGuitar.store(includeGuitar, std::memory_order_relaxed); + busGain.store(sanitizeStreamGain(gain), std::memory_order_relaxed); + } + void setBusGain(float gain) { busGain.store(sanitizeStreamGain(gain), std::memory_order_relaxed); } + + bool isActive() const { return active.load(std::memory_order_acquire); } + juce::String getDesiredDeviceName() const { return desiredDeviceName; } + float getLevel() const { return level.load(std::memory_order_relaxed); } + uint64_t getUnderflowCount() const { return underflowCount.load(std::memory_order_relaxed); } + uint64_t getOverflowCount() const { return overflowCount.load(std::memory_order_relaxed); } + + // ── Producer (primary/output callback, RT) ──────────────────────────── + // Compose the stream submix (guitar/backing/renderer × include flags × + // gain) into the fixed scratch and pack it into the ring. + void publish(const juce::AudioBuffer& guitarMix, + const juce::AudioBuffer* backingBuf, int backingFrames, float backingVol, + const juce::AudioBuffer* rendererBuf, int rendererFrames, + int numSamples); + +private: + // Forwards the sink device's callbacks (the sink's own clock). + struct Callback : juce::AudioIODeviceCallback + { + StreamSink* sink = nullptr; + void audioDeviceIOCallbackWithContext(const float* const* inputData, int numInputChannels, + float* const* outputData, int numOutputChannels, + int numSamples, + const juce::AudioIODeviceCallbackContext&) override + { + juce::ignoreUnused(inputData, numInputChannels); + if (sink) sink->deviceCallback(outputData, numOutputChannels, numSamples); + } + void audioDeviceAboutToStart(juce::AudioIODevice* d) override { if (sink) sink->deviceAboutToStart(d); } + void audioDeviceStopped() override { if (sink) sink->deviceStopped(); } + }; + + // Consumer side (the sink device's thread). + void deviceCallback(float* const* outputData, int numOutputChannels, int numSamples); + void deviceAboutToStart(juce::AudioIODevice* device); + void deviceStopped(); + + EngineState& state; + + Callback callback; + PackedStereoRing ring; + std::atomic underflowCount{0}; + std::atomic overflowCount{0}; + std::atomic active{false}; + std::atomic sinkSampleRate{48000.0}; + std::atomic sinkBlockSize{256}; + std::vector pullScratchL, pullScratchR; // sized in deviceAboutToStart + bool callbackRegistered = false; + bool initialised = false; + + std::atomic busIncludeBacking{true}; + std::atomic busIncludeGuitar{true}; + std::atomic busGain{1.0f}; + std::atomic level{0.0f}; + // Producer-side composed submix scratch — fixed ring capacity, see + // prepareProducerScratch(). + juce::AudioBuffer mixScratch; + + // Declared LAST so it DESTRUCTS FIRST (members tear down in reverse + // declaration order): the manager's dtor closes the device and detaches + // `callback` while `callback`/`ring` are still alive — no use-after-free + // even if an explicit teardown path is ever missed. close() also tears it + // down explicitly before this. + juce::AudioDeviceManager manager; + // Persistent INTENT (control thread only): the device the user chose. + // Survives a stop/restart so reopenDesired() can re-open it. + juce::String desiredTypeName; + juce::String desiredDeviceName; +}; + +} // namespace slopsmith