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 SourcePool (phase 5a)

Browse Source 
Moves the fixed SourceChain pool, add/remove/reclaim lifecycle, the
per-deviceKey callbacksInFlight quiescence handshake, deferred-release
parking, and mixSourcesForDevice verbatim into
src/audio/engine/SourcePool.{h,cpp}. Device callbacks now hold an RAII
CallbackGuard (identical increment/decrement points — no early returns
existed between them) and call pool.mixForDevice(); the device hooks use
prepare/releaseDeviceSources and withDeviceSources, preserving each site's
original locking (the primary about-to-start prepare loop stays deliberately
lockless, as before).

addSource's extra-device resolution (registry reads) stays on the engine
facade, which passes resolved readiness/format/latency into
pool.addResolved() — the pool has no dependency on the InputDeviceSlot
registry, which phase 5b extracts next.

Threaded storm unit test deferred (SourceChain is JUCE-linked; TSAN
unavailable on MSVC) — multi-source.test.js covers the pool through the
addon and is green against the rebuilt binary.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
OmikronApex
2026-07-14 01:23:55 +02:00
co-authored by Claude Fable 5
parent 6ace5a209a
commit 827f02b4b4
5 changed files with 427 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/audio/AudioEngine.cpp
+37 -262
View File
@@ -34,15 +34,7 @@ AudioEngine::AudioEngine()
// always has a live thread to pull decoded audio on. It sleeps while idle and
// costs nothing until a track is loaded.
// Construct the full source pool up front so addSource/removeSource never
// reassign a pointer the audio thread reads — they only flip `active`. Each
// chain reads the engine's audioRunning / currentSampleRate atomics by
// reference (both already constructed as members before this body runs).
// sources[0] is the permanent default input, active from the start; the rest
// are inactive (no threads — NoteVerifier's worker only starts in prepare()).
for (int i = 0; i < kMaxSources; ++i)
sources[(size_t) i] = std::make_unique<SourceChain>(i, audioRunning, currentSampleRate);
sources[0]->setActive(true);
// The source pool (chains, quiescence handshake) lives on SourcePool now.
// Phase 2: tag each additional-input slot with its identity so its JUCE
// callback can route back to the engine. deviceKey = slot index + 1 (0 is the
@@ -675,9 +667,7 @@ void AudioEngine::resetPeaks()
{
// Input peak is per-source — clear EVERY active source (getSourceLevels() exposes
// each one's peak), not just source 0, or extra-device peaks latch forever.
for (auto& src : sources)
if (src->isActive())
src->resetInputPeak();
pool.forEachActive([](SourceChain& s) { s.resetInputPeak(); });
outputPeak.store(0.0f);
}
@@ -705,121 +695,26 @@ int AudioEngine::addSource(int inputChannel, int deviceKey)
return -1;
}
std::lock_guard<std::mutex> lock(sourcesMutex);
reclaimPendingReleases(); // free up any slot whose release was deferred
// Find a free pooled slot (slot 0 is the permanent default). Skip a slot whose
// release is still pending — its chain/worker hasn't been torn down yet, so
// re-preparing it would double-start the verifier thread.
int slot = -1;
for (int i = 1; i < kMaxSources; ++i)
if (! sources[(size_t) i]->isActive() && ! pendingRelease[(size_t) i]) { slot = i; break; }
if (slot < 0)
return -1; // pool full
SourceChain& src = *sources[(size_t) slot];
src.setInputChannel(inputChannel);
src.setDeviceKey(deviceKey);
// Inherit the bound device's capture-latency correction (0 for the primary).
// The device usually started before the source was created (the user picks the
// device, then enables detect), so its delta is already known.
if (deviceKey >= 1 && deviceKey <= kMaxExtraInputDevices)
src.setVerifierAutoOffset(extraInputs[(size_t) (deviceKey - 1)].latencyDeltaSec.load(std::memory_order_relaxed));
else
src.setVerifierAutoOffset(0.0);
// Clear any MANUAL offset left on this pooled chain by a previous player — a
// freshly added source starts with no user fine-tune (the renderer re-applies
// its own via setSourceVerifierOffset). releaseResources() doesn't touch it.
src.setVerifierUserOffset(0.0);
// Likewise clear stale meters so this source doesn't briefly report the previous
// player's level/peak through getSourceLevels() until fresh audio arrives.
src.resetInputMeters();
// Prepare fully BEFORE making it visible to the audio thread, so the first
// callback that observes active==true sees a ready chain + rings. When audio
// isn't running yet, audioDeviceAboutToStart (primary) / extraInputAboutToStart
// (extra) prepares it later. An EXTRA-device source must be prepared with ITS
// device's sample rate / block size, not the primary's — the two can differ.
// Resolve device readiness/format/latency for the pool (the extra-device
// registry is engine-owned; the pool takes resolved values).
bool deviceReady = audioRunning.load(std::memory_order_relaxed);
double sr = currentSampleRate.load(std::memory_order_relaxed);
int bs = inputBlockSize.load(std::memory_order_relaxed);
double latencyDeltaSec = 0.0;
if (deviceKey >= 1 && deviceKey <= kMaxExtraInputDevices)
{
const InputDeviceSlot& es = extraInputs[(size_t) (deviceKey - 1)];
deviceReady = es.active.load(std::memory_order_acquire);
sr = es.sampleRate.load(std::memory_order_relaxed);
bs = es.blockSize.load(std::memory_order_relaxed);
latencyDeltaSec = es.latencyDeltaSec.load(std::memory_order_relaxed);
}
if (deviceReady && sr > 0.0 && bs > 0)
src.prepare(sr, bs);
src.setActive(true); // release-store: now picked up by the audio callback
return slot;
return pool.addResolved(inputChannel, deviceKey, deviceReady, sr, bs, latencyDeltaSec);
}
bool AudioEngine::removeSource(int id)
{
if (id <= 0 || id >= kMaxSources)
return false; // 0 is permanent; out-of-range rejected
std::lock_guard<std::mutex> lock(sourcesMutex);
reclaimPendingReleases(); // opportunistically reclaim earlier deferrals
SourceChain& src = *sources[(size_t) id];
if (! src.isActive())
return false;
// Hide it from the audio callback first; subsequent blocks snapshot active
// once and skip it. It is logically removed from here on, regardless of when
// its resources are reclaimed.
src.setActive(false);
// Reclaim now if we can confirm THIS SOURCE's device callback is not executing.
// Only the callback for the source's own deviceKey can touch it; that counter is
// decremented at the callback's real exit (release store), so observing 0
// (acquire) proves it is not inside processBlock right now; any callback that
// starts afterwards snapshots active and skips this (now-inactive) source — so
// releasing cannot race the audio thread. Keying on the source's deviceKey (not a
// global all-callbacks-idle check) is what lets removals reclaim during steady
// multi-device playback, when callbacks on independent clocks are never all idle
// at once. Bounded so a wedged device can't hang this thread.
const size_t dk = (size_t) src.getDeviceKey();
for (int spins = 0; spins < 200; ++spins) // ~200 ms cap
{
if (callbacksInFlight[dk].load(std::memory_order_acquire) == 0)
{
src.releaseResources(); // stops its threads + releases its chain
return true;
}
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
// A callback stayed wedged in-flight past the wait (a >200 ms block would be a
// catastrophic stall). Do NOT force a release that could race it — DEFER it.
// The source is already inactive so no future callback touches it; reclaim it
// later (next add/removeSource, or audioDeviceStopped) when the body is quiet.
pendingRelease[(size_t) id] = true;
return true;
}
void AudioEngine::reclaimPendingReleases()
{
// Caller holds sourcesMutex. A deferred source is inactive (future callbacks skip
// it); releasing it is safe once the callback for ITS deviceKey is not in a body.
// We key per-deviceKey (decremented by each callback at its real exit) so a
// pending release frees as soon as its OWN device is quiescent — not only when
// every device callback happens to be idle simultaneously (which, on independent
// clocks during steady multi-device playback, may never occur and would strand
// the slot until full stop). On the audioDeviceStopped path the relevant callback
// already left its count at 0.
for (int i = 1; i < kMaxSources; ++i)
{
if (! pendingRelease[(size_t) i]) continue;
const size_t dk = (size_t) sources[(size_t) i]->getDeviceKey();
if (callbacksInFlight[dk].load(std::memory_order_acquire) != 0)
continue; // this source's device is mid-body — try again later
sources[(size_t) i]->releaseResources();
pendingRelease[(size_t) i] = false;
}
return pool.remove(id);
}
// Lifetime/threading of getSource() + the NodeAddon *Source* methods:
@@ -839,10 +734,7 @@ void AudioEngine::reclaimPendingReleases()
// new race class versus the original single-source engine.
SourceChain* AudioEngine::getSource(int id)
{
if (id < 0 || id >= kMaxSources)
return nullptr;
SourceChain& src = *sources[(size_t) id];
return (id == 0 || src.isActive()) ? &src : nullptr;
return pool.get(id);
}
void AudioEngine::setMlNoteDetectionEnabled(bool e)
@@ -851,24 +743,14 @@ void AudioEngine::setMlNoteDetectionEnabled(bool e)
// activated later inherits the current arm state instead of silently
// staying dormant. Each MlNoteDetector::setEnabled is a cheap atomic + a
// cold-state clear on a real transition.
for (int i = 0; i < kMaxSources; ++i)
sources[(size_t) i]->getMlNoteDetector().setEnabled(e);
pool.forEach([e](SourceChain& s) { s.getMlNoteDetector().setEnabled(e); });
}
std::vector<AudioEngine::SourceInfo> AudioEngine::listSources() const
{
std::vector<SourceInfo> out;
for (int i = 0; i < kMaxSources; ++i)
{
const SourceChain& src = *sources[(size_t) i];
if (! src.isActive()) continue;
SourceInfo info;
info.id = src.getId();
info.inputChannel = src.getInputChannel();
info.deviceKey = src.getDeviceKey();
info.active = true;
out.push_back(info);
}
for (const auto& i : pool.list())
out.push_back({ i.id, i.inputChannel, i.deviceKey, i.active });
return out;
}
@@ -934,9 +816,7 @@ void AudioEngine::audioDeviceAboutToStart(juce::AudioIODevice* device)
// device sources (deviceKey > 0) are prepared by their own extraInputAboutToStart
// with THAT device's format — a primary restart must not clobber them with the
// primary's sample rate / block size (they run on a different hardware clock).
for (auto& src : sources)
if (src->isActive() && src->getDeviceKey() == 0)
src->prepare(sr, bs);
pool.forEachActive([&](SourceChain& s) { if (s.getDeviceKey() == 0) s.prepare(sr, bs); });
// Split mode preps the backing stretcher in audioOutputAboutToStart
// instead — that callback owns the device the backing audio actually
@@ -957,17 +837,9 @@ void AudioEngine::audioDeviceStopped()
if (slopsmith_vst_trace::isEnabled())
fprintf(stderr, "[diag] audioDeviceStopped (audioRunning cleared; callbacks stay attached for JUCE auto-restart)\n");
audioRunning.store(false, std::memory_order_relaxed);
{
std::lock_guard<std::mutex> lock(sourcesMutex);
// Release each ACTIVE primary-device source's chain and zero its rings.
// Inactive pooled chains were never prepared.
for (auto& src : sources)
if (src->isActive() && src->getDeviceKey() == 0)
src->releaseResources();
// Reclaim deferred removals only when ALL callback bodies are quiescent
// (an extra-device callback could still be mid-block).
reclaimPendingReleases();
}
// Release each ACTIVE primary-device source's chain and zero its rings;
// retries deferred removals now that the primary body is quiescent.
pool.releaseDeviceSources(0, false, false);
outputRing.resetIndices();
currentBackingLevel.store(0.0f);
@@ -1078,7 +950,8 @@ void AudioEngine::audioDeviceIOCallbackWithContext(
// Publish that the callback body is executing so removeSource() and deferred-
// release reclamation know when no source is being processed (the body is
// quiescent) and a removed source can be safely released. Index 0 = primary.
const int inFlightBefore = callbacksInFlight[0].fetch_add(1, std::memory_order_acq_rel);
const slopsmith::SourcePool::CallbackGuard cbGuard(pool, 0);
const int inFlightBefore = cbGuard.previousInFlight;
// DIAG: two primary callback bodies at once = the input callback is
// registered twice on the device manager (the half-speed/garble bug) or a
@@ -1138,8 +1011,8 @@ void AudioEngine::audioDeviceIOCallbackWithContext(
// clock — never here. With no extra device every source is deviceKey 0, so
// this is the single-device path unchanged. See mixSourcesForDevice for the
// fast-path / multi-source rationale; it is shared with each extra callback.
mixSourcesForDevice(0, inputData, numInputChannels, buffer, sourceMonitorScratch,
effectiveOutputChannels, numSamples);
pool.mixForDevice(0, inputData, numInputChannels, buffer, sourceMonitorScratch,
effectiveOutputChannels, numSamples);
// Duplex mixes backing + applies output gain + meters here.
// Split defers all three to OutputCallback (output device's clock).
@@ -1225,65 +1098,7 @@ void AudioEngine::audioDeviceIOCallbackWithContext(
outputRing.push(buffer.getReadPointer(0), buffer.getReadPointer(1), numSamples);
}
// Body done — this callback is no longer processing a source. Pairs with
// removeSource()/reclaimPendingReleases() acquire loads. Index 0 = primary.
callbacksInFlight[0].fetch_sub(1, std::memory_order_acq_rel);
}
int AudioEngine::mixSourcesForDevice(int deviceKey, const float* const* inputData, int numInputChannels,
juce::AudioBuffer<float>& mixBuf, juce::AudioBuffer<float>& monitorScratch,
int effectiveOutputChannels, int numSamples)
{
// Snapshot each source's active flag ONCE so the count and the process/mix
// passes are consistent within this block — a concurrent add/removeSource
// flipping a flag between two reads must not change which branch runs.
// removeSource waits for all callback bodies to drain before releasing, so a
// source snapshotted active here is safe even if deactivated an instant later.
bool act[kMaxSources];
int firstActive = -1, activeCount = 0;
for (int i = 0; i < kMaxSources; ++i)
{
act[i] = sources[(size_t) i]->isActive()
&& sources[(size_t) i]->getDeviceKey() == deviceKey;
if (act[i]) { ++activeCount; if (firstActive < 0) firstActive = i; }
}
if (activeCount == 0)
{
// No source on this device → silence. An extra device with no bound source
// contributes nothing to the output sum. The primary always has sources[0]
// (deviceKey 0, active from construction), so it never reaches this branch.
for (int ch = 0; ch < effectiveOutputChannels; ++ch)
mixBuf.clear(ch, 0, numSamples);
return 0;
}
if (activeCount == 1)
{
// Fast path — exactly one source: process in place on mixBuf, byte-
// identical to the single-pipeline engine (channel select / mono mix +
// input gain, metering, ML + ring feed, gate, YIN, tone chain, monitor).
sources[(size_t) firstActive]
->processBlock(inputData, numInputChannels, mixBuf, effectiveOutputChannels, numSamples);
return 1;
}
// Multi-source: each renders its own 2-channel monitor into monitorScratch
// (each builds its mono from its bound channel + feeds its own rings /
// detectors / verifier), summed to STEREO (0/1). A >2-channel output keeps
// channels 2+ silent in multi-source mode (the fast path still broadcasts).
for (int ch = 0; ch < effectiveOutputChannels; ++ch)
mixBuf.clear(ch, 0, numSamples);
const int mixCh = juce::jmin(effectiveOutputChannels, 2);
const int n = juce::jmin(numSamples, monitorScratch.getNumSamples());
for (int i = 0; i < kMaxSources; ++i)
{
if (! act[i]) continue;
sources[(size_t) i]->processBlock(inputData, numInputChannels, monitorScratch, 2, n);
for (int ch = 0; ch < mixCh; ++ch)
mixBuf.addFrom(ch, 0, monitorScratch, ch, 0, n);
}
return activeCount;
// Body done (CallbackGuard dtor) — pairs with remove()/reclaim acquire loads.
}
void AudioEngine::extraInputCallback(int slot, const float* const* inputData, int numInputChannels, int numSamples)
@@ -1292,7 +1107,7 @@ void AudioEngine::extraInputCallback(int slot, const float* const* inputData, in
InputDeviceSlot& s = extraInputs[(size_t) slot];
if (! s.active.load(std::memory_order_acquire)) return;
callbacksInFlight[(size_t) s.deviceKey].fetch_add(1, std::memory_order_acq_rel);
const slopsmith::SourcePool::CallbackGuard cbGuard(pool, s.deviceKey);
// Clamp to the per-slot scratch sized in extraInputAboutToStart so the hot
// loop never allocates if a reconfig race delivers a larger block.
@@ -1301,10 +1116,8 @@ void AudioEngine::extraInputCallback(int slot, const float* const* inputData, in
juce::AudioBuffer<float> mix;
mix.setDataToReferTo(s.fanScratch.getArrayOfWritePointers(), 2, numSamples);
mixSourcesForDevice(s.deviceKey, inputData, numInputChannels, mix, s.monitorScratch, 2, numSamples);
pool.mixForDevice(s.deviceKey, inputData, numInputChannels, mix, s.monitorScratch, 2, numSamples);
s.ring.push(mix.getReadPointer(0), mix.getReadPointer(1), numSamples);
callbacksInFlight[(size_t) s.deviceKey].fetch_sub(1, std::memory_order_acq_rel);
}
void AudioEngine::extraInputAboutToStart(int slot, juce::AudioIODevice* device)
@@ -1348,15 +1161,7 @@ void AudioEngine::extraInputAboutToStart(int slot, juce::AudioIODevice* device)
// Prepare each source bound to this device so its verifier/detectors run, and
// apply the latency correction.
{
std::lock_guard<std::mutex> lock(sourcesMutex);
for (auto& src : sources)
if (src->isActive() && src->getDeviceKey() == s.deviceKey)
{
src->prepare(sr, bs);
src->setVerifierAutoOffset(deltaSec);
}
}
pool.prepareDeviceSources(s.deviceKey, sr, bs, deltaSec, true);
s.active.store(true, std::memory_order_release);
}
@@ -1368,34 +1173,13 @@ void AudioEngine::extraInputStopped(int slot)
// slot's body is quiescent. Hide it from the output sum, then release ITS
// sources (no other callback touches them — they all filter by deviceKey).
s.active.store(false, std::memory_order_release);
{
std::lock_guard<std::mutex> lock(sourcesMutex);
// PERMANENT unbind (the user removed this device): the slot will never
// re-open, so DEACTIVATE its sources too — leaving them "active" would strand
// pooled slots no callback can ever service (a ghost detector in listSources).
// A TRANSIENT close (stopAudio/reconfigure/unplug) only releases them, so
// startAudio()'s re-open resumes them in place. Read the atomic flag (set by
// the control-thread unbind) rather than the juce::String desiredDeviceName,
// which this device-thread path must not race on.
const bool permanent = s.permanentUnbind.load(std::memory_order_acquire);
for (auto& src : sources)
if (src->isActive() && src->getDeviceKey() == s.deviceKey)
{
src->releaseResources();
// Zero the meters so getSourceLevels() reports silence while the
// device is gone — otherwise the renderer's per-source silence gate
// treats a stopped/unplugged input as still hearing audio (the last
// non-zero level latches). releaseResources() doesn't touch them.
src->resetInputMeters();
if (permanent) src->setActive(false);
}
// Retry any removeSource() cleanup that was deferred waiting for callbacks to
// drain. With this slot's callback now stopped, callbacksInFlight may finally
// be 0; audioDeviceStopped() (primary) might never observe that window during
// multi-device playback, so reclaim here too or a pending slot can leak until
// the next add/remove.
reclaimPendingReleases();
}
// PERMANENT unbind (user removed this device) deactivates its sources too;
// a TRANSIENT close (stopAudio/reconfigure/unplug) only releases them so
// startAudio()'s re-open resumes them in place. Read the atomic flag (set
// by the control-thread unbind) rather than the juce::String
// desiredDeviceName, which this device-thread path must not race on.
pool.releaseDeviceSources(s.deviceKey, true,
s.permanentUnbind.load(std::memory_order_acquire));
s.ring.resetIndices();
}
@@ -1586,13 +1370,10 @@ bool AudioEngine::unbindInputDevice(int deviceKey)
// pool slots and showing in listSources().
if (! closeExtraInputDevice(slot))
{
std::lock_guard<std::mutex> lock(sourcesMutex);
for (auto& src : sources)
if (src->isActive() && src->getDeviceKey() == deviceKey)
{
src->releaseResources();
src->setActive(false);
}
pool.withDeviceSources(deviceKey, [](SourceChain& s) {
s.releaseResources();
s.setActive(false);
});
}
return true;
}
@@ -1638,10 +1419,7 @@ void AudioEngine::reopenDesiredExtraInputs()
{
if (extraInputs[(size_t) (dk - 1)].desiredDeviceName.isEmpty())
continue;
std::lock_guard<std::mutex> lock(sourcesMutex);
for (auto& src : sources)
if (src->isActive() && src->getDeviceKey() == dk)
src->resetInputMeters();
pool.withDeviceSources(dk, [](SourceChain& s) { s.resetInputMeters(); });
}
return;
}
@@ -1659,10 +1437,7 @@ void AudioEngine::reopenDesiredExtraInputs()
// deactivate them so they do not linger as ghost sources stranding pool
// slots. The renderer re-binds + re-adds if the device returns.
s.desiredDeviceName = {};
std::lock_guard<std::mutex> lock(sourcesMutex);
for (auto& src : sources)
if (src->isActive() && src->getDeviceKey() == dk)
src->setActive(false);
pool.withDeviceSources(dk, [](SourceChain& src) { src.setActive(false); });
}
}
}
src/audio/AudioEngine.h
+16 -51
View File
@@ -7,6 +7,7 @@
#include "engine/StreamSink.h"
#include "engine/BackingPlayer.h"
#include "engine/DeviceSetup.h"
#include "engine/SourcePool.h"
#include "BackingLeveler.h"
#include "signalsmith-stretch.h"
#include <juce_audio_devices/juce_audio_devices.h>
@@ -174,8 +175,7 @@ public:
// off the control thread is race-free. Default off; see SourceChain.
void setMonitorKill(bool kill)
{
for (auto& s : sources)
if (s) s->setMonitorKill(kill);
pool.forEach([kill](SourceChain& s) { s.setMonitorKill(kill); });
}
bool isMonitorKilled() const { return source0().isMonitorKilled(); }
@@ -349,8 +349,8 @@ public:
private:
// sources[0] is the legacy default input chain; always present + active.
SourceChain& source0() { return *sources[0]; }
const SourceChain& source0() const { return *sources[0]; }
SourceChain& source0() { return pool.chain0(); }
const SourceChain& source0() const { return pool.chain0(); }
// Input-device callback. In duplex it writes outputData directly; in split
// it pushes processed stereo into outputRing for OutputCallback.
void audioDeviceIOCallbackWithContext(const float* const* inputData,
@@ -409,47 +409,18 @@ private:
// Probe/apply/teardown component (TLC phase 4). Holds references only.
slopsmith::DeviceSetup deviceSetup{ inputDeviceManager, outputDeviceManager, state };
// Per-input capture+detect+monitor chains. A FIXED pool, all constructed up
// front, so adding/removing a source never reassigns a pointer the audio
// thread is reading — addSource/removeSource only flip an atomic `active`
// flag (and prepare/release the chain). sources[0] is the legacy default,
// active from construction and bound to the primary input device. The audio
// callback fans device channels out to each active source and fans their
// monitor signals into the output mix. SourceChain reads the engine's
// audioRunning / currentSampleRate atomics through references bound at
// construction.
static constexpr int kMaxSources = 8;
// Max ADDITIONAL input devices (beyond the primary). Declared here — ahead of the
// members that size arrays by it (e.g. callbacksInFlight) — though the extra-input
// slot registry that uses it lives further below.
static constexpr int kMaxExtraInputDevices = 3;
std::array<std::unique_ptr<SourceChain>, kMaxSources> sources;
// Serialises addSource/removeSource (control threads only — never the audio
// thread, which just reads each slot's atomic `active`).
std::mutex sourcesMutex;
// Audio-thread scratch for the multi-source mix: each active source renders
// its 2-channel monitor here in turn, then it is summed into the output.
// Pre-sized in audioDeviceAboutToStart so the hot loop never allocates.
// Per-input capture+detect+monitor chains + the add/remove/reclaim
// lifecycle + per-deviceKey quiescence handshake — moved to
// engine/SourcePool (TLC phase 5). Constants mirrored for the members
// that size arrays by them (extraInputs, and NodeAddon range checks).
static constexpr int kMaxSources = slopsmith::SourcePool::kMaxSources;
static constexpr int kMaxExtraInputDevices = slopsmith::SourcePool::kMaxExtraInputDevices;
slopsmith::SourcePool pool{ state };
// Audio-thread scratch for the multi-source mix on the PRIMARY callback:
// each active source renders its 2-channel monitor here in turn, then it
// is summed into the output. Pre-sized in audioDeviceAboutToStart so the
// hot loop never allocates. (Extra devices carry their own scratch.)
juce::AudioBuffer<float> sourceMonitorScratch;
// Count of device callback bodies currently executing, PER deviceKey (index 0 =
// primary input, 1..kMaxExtraInputDevices = each extra-input slot). Each device
// callback increments its own key on entry and decrements at its real exit.
// removeSource() flips a source inactive (future callbacks snapshot active once
// and skip it), then waits to observe THIS SOURCE's deviceKey count == 0 — at
// that instant no callback that could touch this source is inside processBlock,
// so it is safe to release. Keying per-deviceKey (not a single global counter) is
// essential: with the primary + extra inputs on independent clocks they are
// rarely ALL idle at once, so a global check would strand removals during steady
// multi-device playback. A wedged callback past the bounded wait DEFERS the
// release via pendingRelease[], reclaimed later when that key's body is quiescent.
std::array<std::atomic<int>, kMaxExtraInputDevices + 1> callbacksInFlight{};
// Sources whose release was deferred (handshake timed out). Reclaimed under
// sourcesMutex by reclaimPendingReleases() at the next add/removeSource and on
// device stop, once it is safe (audio stopped or no callback in flight).
std::array<bool, kMaxSources> pendingRelease{};
// Release any deferred sources that are now safe to reclaim. Caller holds
// sourcesMutex (or is the device-stop path, where the callback is gone).
void reclaimPendingReleases();
// Master output (post-mix) — engine-global, not per-source.
std::atomic<float> outputGain{1.0f};
@@ -584,13 +555,7 @@ private:
bool closeExtraInputDevice(int slot);
void reopenDesiredExtraInputs();
// Shared fan-out used by both the primary and each extra device's callback:
// mix every active source bound to `deviceKey` into `mixBuf` (using the
// caller-owned `monitorScratch` for the N>1 render so concurrent device
// threads never share scratch). Returns the active source count for that key.
int mixSourcesForDevice(int deviceKey, const float* const* inputData, int numInputChannels,
juce::AudioBuffer<float>& mixBuf, juce::AudioBuffer<float>& monitorScratch,
int effectiveOutputChannels, int numSamples);
// (mixSourcesForDevice moved to SourcePool::mixForDevice — TLC phase 5.)
// ── Streamer mix output sink — moved to engine/StreamSink.{h,cpp} (TLC
// phase 2). Declared after `state` (bound by reference).
src/audio/CMakeLists.txt
+1
View File
@@ -9,6 +9,7 @@ set(AUDIO_SOURCES
engine/StreamSink.cpp
engine/BackingPlayer.cpp
engine/DeviceSetup.cpp
engine/SourcePool.cpp
SourceChain.cpp
SignalChain.cpp
VSTHost.cpp
src/audio/engine/SourcePool.cpp
+225
View File
@@ -0,0 +1,225 @@
// SourcePool implementation — moved verbatim from AudioEngine.cpp (TLC plan
// phase 5 / §2.2). The extra-device resolution that addSource performed
// in-line (reading the InputDeviceSlot registry) stays on the AudioEngine
// facade, which passes the resolved values into addResolved().
#include "SourcePool.h"
#include <chrono>
#include <thread>
namespace slopsmith {
int SourcePool::addResolved(int inputChannel, int deviceKey,
bool deviceReady, double sr, int bs, double latencyDeltaSec)
{
std::lock_guard<std::mutex> lock(mutex);
reclaimPendingLocked(); // free up any slot whose release was deferred
// Find a free pooled slot (slot 0 is the permanent default). Skip a slot whose
// release is still pending — its chain/worker hasn't been torn down yet, so
// re-preparing it would double-start the verifier thread.
int slot = -1;
for (int i = 1; i < kMaxSources; ++i)
if (! sources[(size_t) i]->isActive() && ! pendingRelease[(size_t) i]) { slot = i; break; }
if (slot < 0)
return -1; // pool full
SourceChain& src = *sources[(size_t) slot];
src.setInputChannel(inputChannel);
src.setDeviceKey(deviceKey);
// Inherit the bound device's capture-latency correction (0 for the primary).
src.setVerifierAutoOffset(latencyDeltaSec);
// Clear any MANUAL offset left on this pooled chain by a previous player — a
// freshly added source starts with no user fine-tune (the renderer re-applies
// its own via setSourceVerifierOffset). releaseResources() doesn't touch it.
src.setVerifierUserOffset(0.0);
// Likewise clear stale meters so this source doesn't briefly report the previous
// player's level/peak through getSourceLevels() until fresh audio arrives.
src.resetInputMeters();
// Prepare fully BEFORE making it visible to the audio thread, so the first
// callback that observes active==true sees a ready chain + rings. When audio
// isn't running yet, the relevant about-to-start hook prepares it later. An
// EXTRA-device source must be prepared with ITS device's sample rate / block
// size, not the primary's — the caller resolved those.
if (deviceReady && sr > 0.0 && bs > 0)
src.prepare(sr, bs);
src.setActive(true); // release-store: now picked up by the audio callback
return slot;
}
bool SourcePool::remove(int id)
{
if (id <= 0 || id >= kMaxSources)
return false; // 0 is permanent; out-of-range rejected
std::lock_guard<std::mutex> lock(mutex);
reclaimPendingLocked(); // opportunistically reclaim earlier deferrals
SourceChain& src = *sources[(size_t) id];
if (! src.isActive())
return false;
// Hide it from the audio callback first; subsequent blocks snapshot active
// once and skip it. It is logically removed from here on, regardless of when
// its resources are reclaimed.
src.setActive(false);
// Reclaim now if we can confirm THIS SOURCE's device callback is not executing.
// Only the callback for the source's own deviceKey can touch it; that counter is
// decremented at the callback's real exit (release store), so observing 0
// (acquire) proves it is not inside processBlock right now; any callback that
// starts afterwards snapshots active and skips this (now-inactive) source — so
// releasing cannot race the audio thread. Keying on the source's deviceKey (not a
// global all-callbacks-idle check) is what lets removals reclaim during steady
// multi-device playback, when callbacks on independent clocks are never all idle
// at once. Bounded so a wedged device can't hang this thread.
const size_t dk = (size_t) src.getDeviceKey();
for (int spins = 0; spins < 200; ++spins) // ~200 ms cap
{
if (callbacksInFlight[dk].load(std::memory_order_acquire) == 0)
{
src.releaseResources(); // stops its threads + releases its chain
return true;
}
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
// A callback stayed wedged in-flight past the wait (a >200 ms block would be a
// catastrophic stall). Do NOT force a release that could race it — DEFER it.
// The source is already inactive so no future callback touches it; reclaim it
// later (next add/remove, or a device-stopped hook) when the body is quiet.
pendingRelease[(size_t) id] = true;
return true;
}
void SourcePool::reclaimPendingLocked()
{
// Caller holds `mutex`. A deferred source is inactive (future callbacks skip
// it); releasing it is safe once the callback for ITS deviceKey is not in a body.
// We key per-deviceKey (decremented by each callback at its real exit) so a
// pending release frees as soon as its OWN device is quiescent — not only when
// every device callback happens to be idle simultaneously (which, on independent
// clocks during steady multi-device playback, may never occur and would strand
// the slot until full stop). On the device-stopped path the relevant callback
// already left its count at 0.
for (int i = 1; i < kMaxSources; ++i)
{
if (! pendingRelease[(size_t) i]) continue;
const size_t dk = (size_t) sources[(size_t) i]->getDeviceKey();
if (callbacksInFlight[dk].load(std::memory_order_acquire) != 0)
continue; // this source's device is mid-body — try again later
sources[(size_t) i]->releaseResources();
pendingRelease[(size_t) i] = false;
}
}
std::vector<SourcePool::Info> SourcePool::list() const
{
std::vector<Info> out;
for (int i = 0; i < kMaxSources; ++i)
{
const SourceChain& src = *sources[(size_t) i];
if (! src.isActive()) continue;
Info info;
info.id = src.getId();
info.inputChannel = src.getInputChannel();
info.deviceKey = src.getDeviceKey();
info.active = true;
out.push_back(info);
}
return out;
}
void SourcePool::prepareDeviceSources(int deviceKey, double sr, int bs,
double verifierAutoOffsetSec, bool applyOffset)
{
std::lock_guard<std::mutex> lock(mutex);
for (auto& src : sources)
if (src->isActive() && src->getDeviceKey() == deviceKey)
{
src->prepare(sr, bs);
if (applyOffset) src->setVerifierAutoOffset(verifierAutoOffsetSec);
}
}
void SourcePool::releaseDeviceSources(int deviceKey, bool resetMeters, bool deactivate)
{
std::lock_guard<std::mutex> lock(mutex);
for (auto& src : sources)
if (src->isActive() && src->getDeviceKey() == deviceKey)
{
src->releaseResources();
// Zero the meters so getSourceLevels() reports silence while the
// device is gone — otherwise the renderer's per-source silence gate
// treats a stopped/unplugged input as still hearing audio (the last
// non-zero level latches). releaseResources() doesn't touch them.
if (resetMeters) src->resetInputMeters();
// PERMANENT unbind: the slot will never re-open, so DEACTIVATE its
// sources too — leaving them "active" would strand pooled slots no
// callback can ever service (a ghost detector in listSources).
if (deactivate) src->setActive(false);
}
// Retry any remove() cleanup deferred waiting for callbacks to drain. With
// this device's callback now stopped, callbacksInFlight may finally be 0.
reclaimPendingLocked();
}
int SourcePool::mixForDevice(int deviceKey, const float* const* inputData, int numInputChannels,
juce::AudioBuffer<float>& mixBuf, juce::AudioBuffer<float>& monitorScratch,
int effectiveOutputChannels, int numSamples) noexcept
{
// Snapshot each source's active flag ONCE so the count and the process/mix
// passes are consistent within this block — a concurrent add/remove flipping
// a flag between two reads must not change which branch runs. remove() waits
// for all callback bodies to drain before releasing, so a source snapshotted
// active here is safe even if deactivated an instant later.
bool act[kMaxSources];
int firstActive = -1, activeCount = 0;
for (int i = 0; i < kMaxSources; ++i)
{
act[i] = sources[(size_t) i]->isActive()
&& sources[(size_t) i]->getDeviceKey() == deviceKey;
if (act[i]) { ++activeCount; if (firstActive < 0) firstActive = i; }
}
if (activeCount == 0)
{
// No source on this device → silence. An extra device with no bound source
// contributes nothing to the output sum. The primary always has chain 0
// (deviceKey 0, active from construction), so it never reaches this branch.
for (int ch = 0; ch < effectiveOutputChannels; ++ch)
mixBuf.clear(ch, 0, numSamples);
return 0;
}
if (activeCount == 1)
{
// Fast path — exactly one source: process in place on mixBuf, byte-
// identical to the single-pipeline engine (channel select / mono mix +
// input gain, metering, ML + ring feed, gate, YIN, tone chain, monitor).
sources[(size_t) firstActive]
->processBlock(inputData, numInputChannels, mixBuf, effectiveOutputChannels, numSamples);
return 1;
}
// Multi-source: each renders its own 2-channel monitor into monitorScratch
// (each builds its mono from its bound channel + feeds its own rings /
// detectors / verifier), summed to STEREO (0/1). A >2-channel output keeps
// channels 2+ silent in multi-source mode (the fast path still broadcasts).
for (int ch = 0; ch < effectiveOutputChannels; ++ch)
mixBuf.clear(ch, 0, numSamples);
const int mixCh = juce::jmin(effectiveOutputChannels, 2);
const int n = juce::jmin(numSamples, monitorScratch.getNumSamples());
for (int i = 0; i < kMaxSources; ++i)
{
if (! act[i]) continue;
sources[(size_t) i]->processBlock(inputData, numInputChannels, monitorScratch, 2, n);
for (int ch = 0; ch < mixCh; ++ch)
mixBuf.addFrom(ch, 0, monitorScratch, ch, 0, n);
}
return activeCount;
}
} // namespace slopsmith
src/audio/engine/SourcePool.h
+148
View File
@@ -0,0 +1,148 @@
#pragma once
// SourcePool — the fixed pool of per-input SourceChains plus the add/remove/
// reclaim lifecycle and the per-deviceKey callback-quiescence handshake (TLC
// plan phase 5 / §2.2). Moved verbatim from AudioEngine.
//
// Pool invariants (unchanged):
// - ALL chains are constructed up front; add/removeSource never reassigns a
// pointer the audio thread reads — they only flip an atomic `active` flag.
// - chain 0 is the permanent legacy default input, active from construction.
// - removal uses the per-deviceKey callbacksInFlight counter handshake;
// wedged callbacks defer the release (pendingRelease[]) instead of
// blocking, reclaimed when that key's body is quiescent.
//
// Boundary: device callbacks hold a CallbackGuard for their body and call
// mixForDevice(); control threads use add/remove/list/get and the
// per-deviceKey prepare/release helpers the device hooks need.
#include "EngineState.h"
#include "../SourceChain.h"
#include <array>
#include <atomic>
#include <memory>
#include <mutex>
#include <vector>
namespace slopsmith {
class SourcePool
{
public:
static constexpr int kMaxSources = 8;
// Max ADDITIONAL input devices (beyond the primary); sizes the per-key
// in-flight counters (key 0 = primary).
static constexpr int kMaxExtraInputDevices = 3;
explicit SourcePool(EngineState& engineState)
{
// Construct the full pool up front so the audio thread never observes
// a pointer swap. Each chain reads deviceRunning / currentSampleRate
// by reference. Chain 0 active from the start; the rest inactive (no
// threads — NoteVerifier's worker only starts in prepare()).
for (int i = 0; i < kMaxSources; ++i)
sources[(size_t) i] = std::make_unique<SourceChain>(
i, engineState.deviceRunning, engineState.currentSampleRate);
sources[0]->setActive(true);
}
// ── RT side ───────────────────────────────────────────────────────────
// Publishes that a device callback body is executing for `deviceKey`, so
// remove()/reclaim know when that key is quiescent. previousInFlight is
// exposed for the duplicate-registration diagnostic.
struct CallbackGuard
{
CallbackGuard(SourcePool& p, int deviceKey)
: pool(p), key((size_t) deviceKey),
previousInFlight(p.callbacksInFlight[key].fetch_add(1, std::memory_order_acq_rel)) {}
~CallbackGuard() { pool.callbacksInFlight[key].fetch_sub(1, std::memory_order_acq_rel); }
SourcePool& pool;
const size_t key;
const int previousInFlight;
};
// Mix every active source bound to `deviceKey` into `mixBuf` (using the
// caller-owned `monitorScratch` for the N>1 render so concurrent device
// threads never share scratch). Returns the active source count.
int mixForDevice(int deviceKey, const float* const* inputData, int numInputChannels,
juce::AudioBuffer<float>& mixBuf, juce::AudioBuffer<float>& monitorScratch,
int effectiveOutputChannels, int numSamples) noexcept;
// ── Control threads ───────────────────────────────────────────────────
// Activate a pooled chain (device info pre-resolved by the engine facade,
// which owns the extra-device registry). Returns the slot id or -1.
int addResolved(int inputChannel, int deviceKey,
bool deviceReady, double sr, int bs, double latencyDeltaSec);
// Deactivate + release (id != 0). Defers the release when the source's
// device callback stays in-flight past the bounded wait.
bool remove(int id);
SourceChain* get(int id)
{
if (id < 0 || id >= kMaxSources) return nullptr;
SourceChain& src = *sources[(size_t) id];
return (id == 0 || src.isActive()) ? &src : nullptr;
}
SourceChain& chain0() { return *sources[0]; }
const SourceChain& chain0() const { return *sources[0]; }
struct Info { int id = -1; int inputChannel = -1; int deviceKey = 0; bool active = false; };
std::vector<Info> list() const;
// Fan an operation to every pooled chain (active or not) — plain atomic
// stores on fixed pointers, race-free off the control thread.
template <typename Fn> void forEach(Fn&& fn)
{
for (auto& s : sources)
if (s) fn(*s);
}
template <typename Fn> void forEachActive(Fn&& fn)
{
for (auto& s : sources)
if (s && s->isActive()) fn(*s);
}
// Run `fn` on every active source bound to `deviceKey`, under the pool
// lock — for the extra-device close/unbind paths' bespoke sequences.
template <typename Fn> void withDeviceSources(int deviceKey, Fn&& fn)
{
std::lock_guard<std::mutex> lock(mutex);
for (auto& s : sources)
if (s->isActive() && s->getDeviceKey() == deviceKey) fn(*s);
}
// Device hooks: prepare / release every active source bound to a key,
// under the pool lock. Mirrors the per-device halves of the old
// about-to-start / stopped handlers; release also retries deferred
// reclamation (that key's callback is now quiescent).
void prepareDeviceSources(int deviceKey, double sr, int bs, double verifierAutoOffsetSec,
bool applyOffset);
void releaseDeviceSources(int deviceKey, bool resetMeters, bool deactivate);
// Retry deferred releases whose device is quiescent. Public form takes the
// pool lock (used by the primary device-stopped path).
void reclaimPending()
{
std::lock_guard<std::mutex> lock(mutex);
reclaimPendingLocked();
}
private:
void reclaimPendingLocked();
std::array<std::unique_ptr<SourceChain>, kMaxSources> sources;
// Serialises add/remove (control threads only — never the audio thread,
// which just reads each slot's atomic `active`).
std::mutex mutex;
// How many device-callback bodies are currently executing per deviceKey
// (0 = primary, 1.. = extras). Incremented/decremented by CallbackGuard at
// the body's real entry/exit; remove() observing 0 (acquire) proves the
// source's device is not inside processBlock.
std::array<std::atomic<int>, kMaxExtraInputDevices + 1> callbacksInFlight{};
// A remove() that timed out waiting for quiescence parks the release here;
// reclaimed under `mutex` at the next add/remove and on device-stop paths.
std::array<bool, kMaxSources> pendingRelease{};
};
} // namespace slopsmith