got-feedBack/feedBack-desktop
1
0
Fork 0
mirror of https://github.com/got-feedBack/feedBack-desktop.git synced 2026-09-10 23:04:09 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

refactor(audio): extract StreamSink (phase 2)

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
OmikronApex
2026-07-13 23:53:27 +02:00
co-authored by Claude Fable 5
parent 70f3316094
commit 797501e5ff
5 changed files with 431 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/audio/AudioEngine.cpp
+12 -259
View File
@@ -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<float>& guitarMix,
const juce::AudioBuffer<float>* backingBuf,
int backingFrames, float backingVol,
const juce::AudioBuffer<float>* 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<float> 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,
src/audio/AudioEngine.h
+13 -85
View File
@@ -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 <juce_audio_devices/juce_audio_devices.h>
@@ -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<float>& mixBuf, juce::AudioBuffer<float>& 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<kOutputRingFrames> ring;
std::atomic<uint64_t> underflowCount{0};
std::atomic<uint64_t> overflowCount{0};
std::atomic<bool> active{false};
std::atomic<double> sampleRate{48000.0};
std::atomic<int> blockSize{256};
std::vector<float> 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<bool> streamBusIncludeBacking{true};
std::atomic<bool> streamBusIncludeGuitar{true};
std::atomic<float> streamBusGain{1.0f};
std::atomic<float> 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<float> streamGuitarScratch;
juce::AudioBuffer<float> 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<float>& guitarMix,
const juce::AudioBuffer<float>* backingBuf,
int backingFrames, float backingVol,
const juce::AudioBuffer<float>* rendererBuf,
int rendererFrames, int numSamples);
JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR(AudioEngine)
};
src/audio/CMakeLists.txt
+1
View File
@@ -6,6 +6,7 @@ set(AUDIO_SOURCES
NoiseGate.cpp
TonePolish.cpp
AudioEngine.cpp
engine/StreamSink.cpp
SourceChain.cpp
SignalChain.cpp
VSTHost.cpp
src/audio/engine/StreamSink.cpp
+263
View File
@@ -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 <cmath>
#include <cstdio>
namespace slopsmith {
void StreamSink::publish(const juce::AudioBuffer<float>& guitarMix,
const juce::AudioBuffer<float>* backingBuf, int backingFrames, float backingVol,
const juce::AudioBuffer<float>* 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<float> 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
src/audio/engine/StreamSink.h
+142
View File
@@ -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 <juce_audio_devices/juce_audio_devices.h>
#include <atomic>
#include <cstdint>
#include <vector>
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<float>& guitarMix,
const juce::AudioBuffer<float>* backingBuf, int backingFrames, float backingVol,
const juce::AudioBuffer<float>* 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<kRingFrames> ring;
std::atomic<uint64_t> underflowCount{0};
std::atomic<uint64_t> overflowCount{0};
std::atomic<bool> active{false};
std::atomic<double> sinkSampleRate{48000.0};
std::atomic<int> sinkBlockSize{256};
std::vector<float> pullScratchL, pullScratchR; // sized in deviceAboutToStart
bool callbackRegistered = false;
bool initialised = false;
std::atomic<bool> busIncludeBacking{true};
std::atomic<bool> busIncludeGuitar{true};
std::atomic<float> busGain{1.0f};
std::atomic<float> level{0.0f};
// Producer-side composed submix scratch — fixed ring capacity, see
// prepareProducerScratch().
juce::AudioBuffer<float> 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