diff --git a/src/audio/AudioEngine.cpp b/src/audio/AudioEngine.cpp index 2be1ca2..595e0b3 100644 --- a/src/audio/AudioEngine.cpp +++ b/src/audio/AudioEngine.cpp @@ -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(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 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 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::listSources() const { std::vector 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 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& mixBuf, juce::AudioBuffer& 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 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 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 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 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 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 lock(sourcesMutex); - for (auto& src : sources) - if (src->isActive() && src->getDeviceKey() == dk) - src->setActive(false); + pool.withDeviceSources(dk, [](SourceChain& src) { src.setActive(false); }); } } } diff --git a/src/audio/AudioEngine.h b/src/audio/AudioEngine.h index 0f3912b..e75266c 100644 --- a/src/audio/AudioEngine.h +++ b/src/audio/AudioEngine.h @@ -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 @@ -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, 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 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, 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 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 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& mixBuf, juce::AudioBuffer& 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). diff --git a/src/audio/CMakeLists.txt b/src/audio/CMakeLists.txt index 629e1af..954e764 100644 --- a/src/audio/CMakeLists.txt +++ b/src/audio/CMakeLists.txt @@ -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 diff --git a/src/audio/engine/SourcePool.cpp b/src/audio/engine/SourcePool.cpp new file mode 100644 index 0000000..0346973 --- /dev/null +++ b/src/audio/engine/SourcePool.cpp @@ -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 +#include + +namespace slopsmith { + +int SourcePool::addResolved(int inputChannel, int deviceKey, + bool deviceReady, double sr, int bs, double latencyDeltaSec) +{ + std::lock_guard 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 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::list() const +{ + std::vector 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 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 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& mixBuf, juce::AudioBuffer& 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 diff --git a/src/audio/engine/SourcePool.h b/src/audio/engine/SourcePool.h new file mode 100644 index 0000000..59ad5da --- /dev/null +++ b/src/audio/engine/SourcePool.h @@ -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 +#include +#include +#include +#include + +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( + 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& mixBuf, juce::AudioBuffer& 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 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 void forEach(Fn&& fn) + { + for (auto& s : sources) + if (s) fn(*s); + } + template 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 void withDeviceSources(int deviceKey, Fn&& fn) + { + std::lock_guard 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 lock(mutex); + reclaimPendingLocked(); + } + +private: + void reclaimPendingLocked(); + + std::array, 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, 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 pendingRelease{}; +}; + +} // namespace slopsmith