diff --git a/.claude/skills/thread-safety-itc/SKILL.md b/.claude/skills/thread-safety-itc/SKILL.md index 86d1c4bd6..1e13b88b4 100644 --- a/.claude/skills/thread-safety-itc/SKILL.md +++ b/.claude/skills/thread-safety-itc/SKILL.md @@ -95,6 +95,8 @@ Throttled events (e.g. `positionChanged`) use `PositionChangedDispatcher`, which - Clear callbacks in the **most-derived HostObject first** (each layer clears its own events; base destructors run afterward and clear parent events) - Do **not** inherit `EventCaller` on multiple node bases (ambiguous API); add member `EventCaller` fields per event instead +**Time-delayed dispatch outside the node's lifetime** (e.g. firing `ended` at a context time the render path will not reach for this node): use `BaseAudioContext::deferEmptyEventDispatch(event, callbackId, dueTime)`. The context keeps a preallocated pending list drained each quantum in `processAudioEvents()`; the context clock is the timer (suspend pauses it, offline fires at render speed) and the context holds no node reference. Never use a detached timer thread for this — spawning a thread from an audio-event body runs on the audio thread (allocation + syscalls), and every capture choice has been a real bug in `AudioScheduledSourceNode::stop()`: raw `this` was a use-after-free, `weak_from_this().lock()` ran the node destructor on the timer thread (bypassing the graph's disposal path), and `std::move`-ing the `EventCaller` member raced the JS/GC-thread `assignCallbackId` paths (the registry `shared_ptr` inside `EventCaller` is only safe for concurrent access while it is never mutated — which is why `EventCaller` is non-movable). + Low-level registry API (used internally by `EventCaller`): ```cpp @@ -166,7 +168,7 @@ On Android, `AudioPlayer::onErrorAfterClose` also takes `driverMutex_` because O **Live `AudioContext` render quiescence:** `currentRenders_` on `AudioContext` is incremented at the start of each platform I/O callback (`IOSAudioPlayer::deliverOutputBuffers` / `AudioPlayer::onAudioReady`) via a reference passed in `initialize()`, and decremented when the callback returns (RAII scope). `suspend()` and `close()` call `waitForRenderQuiescence()` (under `driverMutex_`) before `processAudioEvents()` / `cleanup()`. Platform drivers share the `CommonPlayer` abstract base (`common/cpp/audioapi/core/CommonPlayer.h`). -**Graph Channel A producer self-drain:** `Graph::setProducerSelfDrain(true)` makes the JS/main producer drain Channel A after each enqueue. Enable only when there is no audio/render consumer (realtime: construction + after `suspend`/`close` quiescence; offline: before `startRendering` and after a scheduled suspend). Before disabling for `start`/`resume`/`renderAudio`, call `processEvents()` once (still as sole consumer) so the bounded channel is empty, then disable, then start the audio/render consumer; re-enable if start/resume fails. After enabling, call `processEvents()` once to flush backlog (avoids `WAIT_ON_FULL` deadlock if the channel was already full). +**Graph producer self-drain:** `Graph::enableProducerSelfDrain()` makes producer threads drain the event channels themselves after each enqueue; `disableProducerSelfDrain()` hands consumption back to the audio/render thread. Both flush the channels internally (no separate `processEvents()` call needed) and serialize with in-flight drains via `selfDrainMutex_`, because two producers can drain concurrently: the JS thread (mutations) and the GC finalizer thread (`removeNode`, which self-drains after its Channel B orphan send — otherwise a finalizer burst with no consumer fills the bounded channel and blocks forever, e.g. at process exit). Enable only when there is no audio/render consumer (realtime: construction + after `suspend`/`close` quiescence; offline: before `startRendering` and after a scheduled suspend); disable *before* starting the audio/render consumer; re-enable if start/resume fails. **Context lifecycle promises:** HostObjects wrap JSI `Promise`s via `ContextPromiseResolver` (`jsi/ContextPromiseResolver.hpp`); tasks are queued as `ContextPromiseTask` @@ -177,8 +179,15 @@ ops (`resume` / `suspend` / `close` / offline start) are **control messages** on SPSC single-producer). A dedicated `TaskOffloader` worker thread drains the queue under `driverMutex_`, runs `collectDisposedNodes()` (host-graph ghost cleanup — never on the audio thread), then executes the lifecycle body and settles the promise on the CallInvoker. The -`TaskOffloader` destructor joins the worker and drains any queued control messages. When the driver -is stopped, `scheduleAudioEvent` still drains already-queued SPSC events then runs +`TaskOffloader` destructor joins the worker and drains any queued control messages — which is why +`~AudioContext` / `~OfflineAudioContext` MUST call `joinPendingPromiseWorker()` as their first +statement (and never while holding `driverMutex_`): default member destruction order would join the +worker last, letting drained tasks run against an already-destroyed player/graph/disposer. The +offline render thread is likewise owned (`renderThread_` + `stopRendering_` flag, joined in the +destructor, detach only when the last context reference drops on the render thread itself); the +render lambda must release its `resumePromise` right after resolving, since resolver callbacks +capture the context and would otherwise make the render thread its own context's last owner. When +the driver is stopped, `scheduleAudioEvent` still drains already-queued SPSC events then runs the new event synchronously under `driverMutex_` (FIFO with prior messages). Control-message bodies must not re-lock `driverMutex_` or `waitForRenderQuiescence()` while `currentRenders_ > 0` (audio-callback self-deadlock). Apply the visible `state` attribute and settle the promise diff --git a/packages/react-native-audio-api/common/cpp/audioapi/core/AudioContext.cpp b/packages/react-native-audio-api/common/cpp/audioapi/core/AudioContext.cpp index fed66eb8a..12f8d1546 100644 --- a/packages/react-native-audio-api/common/cpp/audioapi/core/AudioContext.cpp +++ b/packages/react-native-audio-api/common/cpp/audioapi/core/AudioContext.cpp @@ -18,12 +18,13 @@ AudioContext::AudioContext( const std::shared_ptr &audioEventHandlerRegistry) : BaseAudioContext(sampleRate, audioEventHandlerRegistry), isInitialized_(false) { // Context starts SUSPENDED with no audio-thread consumer. Let the producer - // drain Channel A itself until start()/resume() hands draining to the + // drain the channels itself until start()/resume() hands draining to the // audio callback (same pattern as OfflineAudioContext before rendering). - getGraph()->setProducerSelfDrain(true); + getGraph()->enableProducerSelfDrain(); } AudioContext::~AudioContext() { + joinPendingPromiseWorker(); if (getState() != ContextState::CLOSED) { std::scoped_lock lock(driverMutex_); close(nullptr); @@ -65,11 +66,7 @@ bool AudioContext::tryStartDriver() { return false; } - // Flush while we are still the sole consumer, then hand the channel to the - // audio callback. flushing first avoids blocking forever if the bounded - // channel was full when self-drain is turned off. - getGraph()->processEvents(); - getGraph()->setProducerSelfDrain(false); + getGraph()->disableProducerSelfDrain(); if (audioPlayer_->start()) { isInitialized_.store(true, std::memory_order_release); @@ -81,7 +78,7 @@ bool AudioContext::tryStartDriver() { return true; } - getGraph()->setProducerSelfDrain(true); + getGraph()->enableProducerSelfDrain(); return false; } @@ -96,8 +93,7 @@ void AudioContext::close(const std::shared_ptr> &pr // No audio-thread consumer after stop; allow producer self-drain for any // remaining graph mutations (and flush events already queued). - getGraph()->setProducerSelfDrain(true); - getGraph()->processEvents(); + getGraph()->enableProducerSelfDrain(); processAudioEvents(); audioPlayer_->cleanup(); @@ -117,12 +113,11 @@ bool AudioContext::resume(const std::shared_ptr> &p bool result = false; if (isInitialized_.load(std::memory_order_acquire)) { - getGraph()->processEvents(); - getGraph()->setProducerSelfDrain(false); + getGraph()->disableProducerSelfDrain(); if (audioPlayer_->resume()) { result = true; } else { - getGraph()->setProducerSelfDrain(true); + getGraph()->enableProducerSelfDrain(); } } else { result = tryStartDriver(); @@ -150,8 +145,7 @@ bool AudioContext::suspend(const std::shared_ptr> & // Audio callback is no longer the consumer; enable self-drain so graph // mutations while suspended cannot fill the bounded channel and block. - getGraph()->setProducerSelfDrain(true); - getGraph()->processEvents(); + getGraph()->enableProducerSelfDrain(); processAudioEvents(); } diff --git a/packages/react-native-audio-api/common/cpp/audioapi/core/BaseAudioContext.cpp b/packages/react-native-audio-api/common/cpp/audioapi/core/BaseAudioContext.cpp index d9336c957..952f7b7ca 100644 --- a/packages/react-native-audio-api/common/cpp/audioapi/core/BaseAudioContext.cpp +++ b/packages/react-native-audio-api/common/cpp/audioapi/core/BaseAudioContext.cpp @@ -35,7 +35,39 @@ BaseAudioContext::BaseAudioContext( gcAudioEventScheduler_(GC_AUDIO_SCHEDULER_CAPACITY), disposer_( std::make_unique>(AUDIO_SCHEDULER_CAPACITY)), - graph_(std::make_shared(AUDIO_SCHEDULER_CAPACITY, disposer_.get())) {} + graph_(std::make_shared(AUDIO_SCHEDULER_CAPACITY, disposer_.get())) { + deferredEmptyEvents_.reserve(DEFERRED_EMPTY_EVENTS_CAPACITY); +} + +void BaseAudioContext::deferEmptyEventDispatch( + AudioEvent event, + uint64_t callbackId, + double dueTime) { + if (callbackId == 0) { + return; + } + + deferredEmptyEvents_.push_back({.dueTime = dueTime, .event = event, .callbackId = callbackId}); +} + +void BaseAudioContext::dispatchDueDeferredEvents() { + if (deferredEmptyEvents_.empty()) { + return; + } + + const auto now = getCurrentTime(); + std::erase_if(deferredEmptyEvents_, [&](const DeferredEmptyEvent &deferred) { + if (deferred.dueTime > now) { + return false; + } + + if (audioEventHandlerRegistry_ != nullptr) { + audioEventHandlerRegistry_->dispatchEventFromAudioThread( + deferred.event, deferred.callbackId, AudioEventPayload{EmptyPayload{}}); + } + return true; + }); +} void BaseAudioContext::initialize(const AudioDestinationNode *destination) { destination_ = destination; diff --git a/packages/react-native-audio-api/common/cpp/audioapi/core/BaseAudioContext.h b/packages/react-native-audio-api/common/cpp/audioapi/core/BaseAudioContext.h index 221200c0f..d270800b2 100644 --- a/packages/react-native-audio-api/common/cpp/audioapi/core/BaseAudioContext.h +++ b/packages/react-native-audio-api/common/cpp/audioapi/core/BaseAudioContext.h @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -15,6 +16,7 @@ #include #include #include +#include #include #include #include @@ -67,8 +69,15 @@ class BaseAudioContext : public std::enable_shared_from_this { // finalization that clears the callback id). audioEventScheduler_.processAllEvents(*this); gcAudioEventScheduler_.processAllEvents(*this); + dispatchDueDeferredEvents(); } + /// @brief Schedules a one-shot, payload-less event dispatch for when + /// `currentTime` reaches `dueTime`. + /// @note Runs on the render-serialized domain (audio thread, or the + /// synchronous `scheduleAudioEvent` path) — same as audio event bodies. + void deferEmptyEventDispatch(AudioEvent event, uint64_t callbackId, double dueTime); + template bool scheduleAudioEvent(F &&event) noexcept { // NOLINT(cppcoreguidelines-missing-std-forward) std::scoped_lock lock(driverMutex_); @@ -123,6 +132,16 @@ class BaseAudioContext : public std::enable_shared_from_this { mutable std::mutex driverMutex_; std::atomic state_; + /// @brief Joins the pending-promises worker after draining any queued + /// lifecycle tasks. Idempotent. + /// + /// Derived-class destructors MUST call this as their first teardown step: + /// the drained task bodies lock `driverMutex_` and touch the player, graph, + /// and disposer, all of which start being destroyed once the destructor bodies return. + void joinPendingPromiseWorker() { + pendingPromisesOffloader_->shutdown(); + } + /// Debug-only: `driverMutex_` must already be held by the calling thread. void assertDriverMutexHeld() const { #ifndef NDEBUG @@ -164,6 +183,22 @@ class BaseAudioContext : public std::enable_shared_from_this { std::unique_ptr> disposer_; std::shared_ptr graph_; + struct DeferredEmptyEvent { + double dueTime; + AudioEvent event; + uint64_t callbackId; + }; + + /// Reserved up front so `deferEmptyEventDispatch` stays allocation-free on + /// the audio thread until this many dispatches are pending at once. + static constexpr size_t DEFERRED_EMPTY_EVENTS_CAPACITY = 16; + + /// Render-serialized only (audio thread or the synchronous + /// `scheduleAudioEvent` path) — no lock, so no other thread may touch it. + std::vector deferredEmptyEvents_; + + void dispatchDueDeferredEvents(); + [[nodiscard]] virtual bool isDriverRunning() const = 0; }; diff --git a/packages/react-native-audio-api/common/cpp/audioapi/core/OfflineAudioContext.cpp b/packages/react-native-audio-api/common/cpp/audioapi/core/OfflineAudioContext.cpp index 3e1565215..f75b104c3 100644 --- a/packages/react-native-audio-api/common/cpp/audioapi/core/OfflineAudioContext.cpp +++ b/packages/react-native-audio-api/common/cpp/audioapi/core/OfflineAudioContext.cpp @@ -32,7 +32,26 @@ OfflineAudioContext::OfflineAudioContext( // graph-event channel, so a large graph would otherwise fill it and block // the producer forever. Let the producing thread drain it itself for now; // renderAudio() hands draining back to the render thread. - getGraph()->setProducerSelfDrain(true); + getGraph()->enableProducerSelfDrain(); +} + +OfflineAudioContext::~OfflineAudioContext() { + // Join the promise worker first: a queued resume() task could spawn a fresh + // render thread after the join below. Both must be gone before base-class + // members (graph, disposer, driverMutex_) are destroyed. + joinPendingPromiseWorker(); + stopRendering_.store(true, std::memory_order_release); + if (renderThread_.joinable()) { + if (renderThread_.get_id() == std::this_thread::get_id()) { + // The render thread can drop the context's last reference itself (a + // drained event lambda holding the final shared_ptr). Self-join would + // throw; the thread is already past its render loop, so let it finish + // unobserved. + renderThread_.detach(); + } else { + renderThread_.join(); + } + } } void OfflineAudioContext::resume(const std::shared_ptr> &promise) { @@ -81,15 +100,22 @@ bool OfflineAudioContext::suspend( void OfflineAudioContext::renderAudio( const std::shared_ptr> &resumePromise) { - // Flush while we are still the sole consumer, then hand the channel to the - // render thread. - getGraph()->processEvents(); - getGraph()->setProducerSelfDrain(false); + getGraph()->disableProducerSelfDrain(); - std::thread([this, resumePromise = resumePromise]() mutable { + // first wait for the previous render thread to finish + // can be in the middle because of suspend and then resume call again + if (renderThread_.joinable()) { + renderThread_.join(); + } + + renderThread_ = std::thread([this, resumePromise = resumePromise]() mutable { ContextPromiseResolver::resolve(resumePromise); + // The resolver's callbacks (transitively) own this context — both + // `startRendering` and the HostObject resume path capture it. Release the + // reference so the render thread never keeps its own context alive: + resumePromise.reset(); - while (currentSampleFrame_ < length_) { + while (!stopRendering_.load(std::memory_order_acquire) && currentSampleFrame_ < length_) { Locker locker(driverMutex_); int framesToProcess = std::min(static_cast(length_ - currentSampleFrame_), RENDER_QUANTUM_SIZE); @@ -111,16 +137,19 @@ void OfflineAudioContext::renderAudio( // The render thread is about to exit; with no consumer again, let the // producer self-drain any graph mutations made from the suspend // callback until resume() restarts rendering. - getGraph()->setProducerSelfDrain(true); - getGraph()->processEvents(); + getGraph()->enableProducerSelfDrain(); locker.unlock(); ContextPromiseResolver::resolve(promise); return; } } + if (stopRendering_.load(std::memory_order_acquire)) { + return; + } OfflineAudioContextResultPromise::resolve(resultPromise_, resultBuffer_); - }).detach(); + resultPromise_.reset(); + }); } void OfflineAudioContext::startRendering( diff --git a/packages/react-native-audio-api/common/cpp/audioapi/core/OfflineAudioContext.h b/packages/react-native-audio-api/common/cpp/audioapi/core/OfflineAudioContext.h index 5cd7352a8..0705e8309 100644 --- a/packages/react-native-audio-api/common/cpp/audioapi/core/OfflineAudioContext.h +++ b/packages/react-native-audio-api/common/cpp/audioapi/core/OfflineAudioContext.h @@ -5,7 +5,9 @@ #include #include +#include #include +#include #include namespace audioapi { @@ -17,7 +19,7 @@ class OfflineAudioContext : public BaseAudioContext { size_t length, float sampleRate, const std::shared_ptr &audioEventHandlerRegistry); - ~OfflineAudioContext() override = default; + ~OfflineAudioContext() override; DELETE_COPY_AND_MOVE(OfflineAudioContext); void resume(const std::shared_ptr> &promise); @@ -35,6 +37,11 @@ class OfflineAudioContext : public BaseAudioContext { std::shared_ptr audioBuffer_; std::shared_ptr resultBuffer_; + /// Render worker. Owned: it captures `this`, so the destructor must be able + /// to stop and join it before members are torn down. + std::thread renderThread_; + std::atomic stopRendering_{false}; + void renderAudio(const std::shared_ptr> &resumePromise); bool isDriverRunning() const override; diff --git a/packages/react-native-audio-api/common/cpp/audioapi/core/sources/AudioFileSourceNode.cpp b/packages/react-native-audio-api/common/cpp/audioapi/core/sources/AudioFileSourceNode.cpp index c98b2936c..e837e8a6f 100644 --- a/packages/react-native-audio-api/common/cpp/audioapi/core/sources/AudioFileSourceNode.cpp +++ b/packages/react-native-audio-api/common/cpp/audioapi/core/sources/AudioFileSourceNode.cpp @@ -125,6 +125,8 @@ bool AudioFileSourceNode::initDecoder( std::move(frameSender), frameReceiver_); + seekDecoderThread_ = std::thread(std::move(*seekDecoderDaemon_)); + if (!decoderState_->isReady.load(std::memory_order_acquire)) { return false; } @@ -393,11 +395,6 @@ void AudioFileSourceNode::start(double when) { endOfStreamStopPending_ = false; endOfStreamDrainPending_ = false; positionChanged_.requestFlush(); - - if (seekDecoderDaemon_) { - seekDecoderThread_ = std::thread(std::move(*seekDecoderDaemon_)); - seekDecoderDaemon_.reset(); - } } void AudioFileSourceNode::bindMediaElementSource(uint64_t bindingId) { diff --git a/packages/react-native-audio-api/common/cpp/audioapi/core/sources/AudioScheduledSourceNode.cpp b/packages/react-native-audio-api/common/cpp/audioapi/core/sources/AudioScheduledSourceNode.cpp index 327c51c8c..33704505f 100644 --- a/packages/react-native-audio-api/common/cpp/audioapi/core/sources/AudioScheduledSourceNode.cpp +++ b/packages/react-native-audio-api/common/cpp/audioapi/core/sources/AudioScheduledSourceNode.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include #include #if !RN_AUDIO_API_TEST @@ -8,10 +9,8 @@ #endif // RN_AUDIO_API_TEST #include -#include #include #include -#include namespace audioapi { @@ -46,13 +45,9 @@ void AudioScheduledSourceNode::stop(double when) { if (stopFrame <= startFrame) { playbackState_ = PlaybackState::FINISHED; AudioNode::disable(); - // Fire-and-forget: defer event dispatch off the caller thread. - static constexpr double kMillisecondsPerSecond = 1000.0; - std::thread([this]() { - std::this_thread::sleep_for( - std::chrono::milliseconds(static_cast(stopTime_ * kMillisecondsPerSecond))); - onEndedEvent_.dispatchEmpty(); - }).detach(); + // The node never plays, so the render path will not fire onended. Defer a + // dispatch until the context clock reaches the requested stop time + context->deferEmptyEventDispatch(AudioEvent::ENDED, onEndedEvent_.getCallbackId(), when); return; } } diff --git a/packages/react-native-audio-api/common/cpp/audioapi/core/utils/graph/Graph.cpp b/packages/react-native-audio-api/common/cpp/audioapi/core/utils/graph/Graph.cpp index b81e1632f..c62a73a7b 100644 --- a/packages/react-native-audio-api/common/cpp/audioapi/core/utils/graph/Graph.cpp +++ b/packages/react-native-audio-api/common/cpp/audioapi/core/utils/graph/Graph.cpp @@ -103,6 +103,7 @@ Graph::Res Graph::removeNode(HNode *node) { // could possibly reference this node lives at index < barrier. auto barrier = eventSender_.sendCursor(); gcEventSender_.send(OrphanEnvelope{.barrier = barrier, .action = std::move(event)}); + drainProducedEventsIfSelfDraining(); return NoneType{}; }); } diff --git a/packages/react-native-audio-api/common/cpp/audioapi/core/utils/graph/Graph.h b/packages/react-native-audio-api/common/cpp/audioapi/core/utils/graph/Graph.h index 03408c3fb..5e6726d39 100644 --- a/packages/react-native-audio-api/common/cpp/audioapi/core/utils/graph/Graph.h +++ b/packages/react-native-audio-api/common/cpp/audioapi/core/utils/graph/Graph.h @@ -12,6 +12,7 @@ #include #include #include +#include #include namespace audioapi::utils::graph { @@ -156,11 +157,12 @@ class Graph { void collectDisposedNodes(); - /// @brief Controls whether the producing (main/JS) thread drains Channel A - /// itself right after enqueuing an event. + /// @brief Enters self-drain mode: producing (JS / GC finalizer) threads + /// drain both event channels themselves right after enqueuing, then flushes + /// any backlog already in the channels. /// - /// The event channel is a bounded SPSC queue with a `WAIT_ON_FULL` + - /// `ATOMIC_WAIT` sender: once full, `send()` blocks until a consumer + /// The event channels are bounded SPSC queues with `WAIT_ON_FULL` + + /// `ATOMIC_WAIT` senders: once full, `send()` blocks until a consumer /// advances the receive cursor. /// /// Enable self-drain only while there is **no** audio/render consumer: @@ -168,19 +170,26 @@ class Graph { /// scheduled suspend until `resume()` restarts the render thread. /// - Realtime `AudioContext`: while SUSPENDED / stopped (construction, /// after `suspend()` / `close()` quiescence). With no callback draining - /// the channel, a large graph would otherwise fill it and block. + /// the channels, a large graph would otherwise fill them and block. /// - /// It MUST be disabled again *before* the audio/render thread starts so that - /// thread becomes the single consumer (the producer must not race it on the - /// receiver). Call `processEvents()` once immediately before disabling so the - /// bounded channel is empty (otherwise a full queue could block forever with - /// no consumer). If start/resume fails, re-enable. + /// While enabled, drains from the JS thread and the GC finalizer thread are + /// serialized by `selfDrainMutex_`, so exactly one thread consumes at a time. + void enableProducerSelfDrain() { + std::scoped_lock lock(selfDrainMutex_); + producerSelfDrain_.store(true, std::memory_order_release); + processEvents(); + } + + /// @brief Leaves self-drain mode so the audio/render thread can become the + /// single consumer. Flushes both channels first (while still serialized with + /// any in-flight self-drain), so the consumer starts on empty channels. /// - /// @note Toggle only from the thread that owns graph construction, and only - /// while no other thread is consuming the channel. After enabling, call - /// `processEvents()` once to flush any backlog already in the channel. - void setProducerSelfDrain(bool enabled) { - producerSelfDrain_.store(enabled, std::memory_order_release); + /// Call *before* starting the audio/render thread; once it runs, producers + /// must not touch the receivers. If start/resume fails, re-enable. + void disableProducerSelfDrain() { + std::scoped_lock lock(selfDrainMutex_); + processEvents(); + producerSelfDrain_.store(false, std::memory_order_release); } private: @@ -214,17 +223,31 @@ class Graph { std::uint32_t poolCapacity_; ///< Pool capacity we have ensured std::uint32_t nodeCapacity_; ///< Node vector capacity we have ensured - /// @brief When set, the producer thread drains Channel A right after each - /// enqueue (see setProducerSelfDrain). Default off — realtime contexts rely - /// on the audio thread as the consumer. + /// @brief When set, producer threads drain the channels right after each + /// enqueue (see enableProducerSelfDrain). Default off — realtime contexts + /// rely on the audio thread as the consumer. std::atomic producerSelfDrain_{false}; + /// @brief Serializes self-drain consumption between producer threads (JS + /// thread mutations vs GC-finalizer `removeNode`) and the mode toggles. + std::mutex selfDrainMutex_; + /// @brief Drains any pending produced events on the calling (producer) /// thread when self-drain is enabled. No-op otherwise. + /// + /// The flag is re-checked under `selfDrainMutex_`: a concurrent + /// `disableProducerSelfDrain()` may have handed consumption to the + /// audio/render thread between the fast-path check and the lock, and + /// consuming past that point would race the new consumer. void drainProducedEventsIfSelfDraining() { - if (producerSelfDrain_.load(std::memory_order_acquire)) { - processEvents(); + if (!producerSelfDrain_.load(std::memory_order_acquire)) { + return; + } + std::scoped_lock lock(selfDrainMutex_); + if (!producerSelfDrain_.load(std::memory_order_acquire)) { + return; } + processEvents(); } /// @brief Pre-grows the InputPool when the edge count approaches capacity. diff --git a/packages/react-native-audio-api/common/cpp/audioapi/events/EventCaller.hpp b/packages/react-native-audio-api/common/cpp/audioapi/events/EventCaller.hpp index 7f5c5fd8b..63fb1f133 100644 --- a/packages/react-native-audio-api/common/cpp/audioapi/events/EventCaller.hpp +++ b/packages/react-native-audio-api/common/cpp/audioapi/events/EventCaller.hpp @@ -20,21 +20,14 @@ class EventCaller { : eventHandlerRegistry_(audioEventHandlerRegistry) {} ~EventCaller() { - const auto callbackId = getCallbackId(); - if (callbackId == 0) { - return; - } - - unregisterCallback(callbackId); - callbackId_.store(0, std::memory_order_release); + unregisterCallback(); } DELETE_COPY_AND_MOVE(EventCaller); void assignCallbackId(uint64_t callbackId) noexcept { - const auto previousCallbackId = getCallbackId(); - if (previousCallbackId != callbackId) { - unregisterCallback(previousCallbackId); + if (getCallbackId() != callbackId) { + unregisterCallback(); } callbackId_.store(callbackId, std::memory_order_release); @@ -48,12 +41,14 @@ class EventCaller { return getCallbackId() != 0; } - void unregisterCallback(uint64_t callbackId) const { - if (eventHandlerRegistry_ == nullptr || callbackId == 0) { + void unregisterCallback() { + auto id = getCallbackId(); + callbackId_.store(0, std::memory_order_release); + if (eventHandlerRegistry_ == nullptr || id == 0) { return; } - eventHandlerRegistry_->unregisterHandler(Event, callbackId); + eventHandlerRegistry_->unregisterHandler(Event, id); } bool dispatchEmpty() const noexcept diff --git a/packages/react-native-audio-api/common/cpp/test/src/core/sources/AudioScheduledSourceTest.cpp b/packages/react-native-audio-api/common/cpp/test/src/core/sources/AudioScheduledSourceTest.cpp index 2cb41e0e9..3060d9381 100644 --- a/packages/react-native-audio-api/common/cpp/test/src/core/sources/AudioScheduledSourceTest.cpp +++ b/packages/react-native-audio-api/common/cpp/test/src/core/sources/AudioScheduledSourceTest.cpp @@ -1,7 +1,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -129,4 +131,28 @@ TEST_F(AudioScheduledSourceTest, IsFinishedStateSetCorrectly) { EXPECT_TRUE(sourceNode.isFinished()); } +TEST_F(AudioScheduledSourceTest, StopBeforeStartFiresEndedWhenContextTimeReachesStopTime) { + static constexpr uint64_t ENDED_CALLBACK_ID = 42; + auto sourceNode = TestableAudioScheduledSourceNode(context); + sourceNode.assignOnEndedCallbackId(ENDED_CALLBACK_ID); + + sourceNode.start(2 * RENDER_QUANTUM_TIME); + sourceNode.stop(RENDER_QUANTUM_TIME); + EXPECT_TRUE(sourceNode.isFinished()); + + EXPECT_CALL( + *eventRegistry, + dispatchEventFromAudioThread(AudioEvent::ENDED, ENDED_CALLBACK_ID, testing::_)) + .Times(0); + sourceNode.playFrames(RENDER_QUANTUM); // context time is still before the stop time + + EXPECT_CALL( + *eventRegistry, + dispatchEventFromAudioThread(AudioEvent::ENDED, ENDED_CALLBACK_ID, testing::_)) + .WillOnce(testing::Return(true)); + sourceNode.playFrames(RENDER_QUANTUM); // context time reaches the stop time + + EXPECT_CALL(*eventRegistry, unregisterHandler(AudioEvent::ENDED, ENDED_CALLBACK_ID)).Times(1); +} + // NOLINTEND diff --git a/packages/react-native-audio-api/common/cpp/test/src/events/EventCallerTest.cpp b/packages/react-native-audio-api/common/cpp/test/src/events/EventCallerTest.cpp index e78ede522..3be847179 100644 --- a/packages/react-native-audio-api/common/cpp/test/src/events/EventCallerTest.cpp +++ b/packages/react-native-audio-api/common/cpp/test/src/events/EventCallerTest.cpp @@ -71,11 +71,28 @@ TEST(EventCallerTest, DispatchForwardsPayload) { TEST(EventCallerTest, UnregisterForwardsEventAndCallbackId) { auto registry = std::make_shared(); EventCaller eventCaller(registry); + eventCaller.assignCallbackId(POSITION_CALLBACK_ID); EXPECT_CALL(*registry, unregisterHandler(AudioEvent::POSITION_CHANGED, POSITION_CALLBACK_ID)) .Times(1); - eventCaller.unregisterCallback(POSITION_CALLBACK_ID); + eventCaller.unregisterCallback(); + + EXPECT_FALSE(eventCaller.hasCallback()); +} + +TEST(EventCallerTest, ReassigningSameCallbackIdKeepsHandlerRegistered) { + auto registry = std::make_shared(); + EventCaller eventCaller(registry); + eventCaller.assignCallbackId(ENDED_CALLBACK_ID); + + EXPECT_CALL(*registry, unregisterHandler(testing::_, testing::_)).Times(0); + + eventCaller.assignCallbackId(ENDED_CALLBACK_ID); + + EXPECT_EQ(eventCaller.getCallbackId(), ENDED_CALLBACK_ID); + + EXPECT_CALL(*registry, unregisterHandler(AudioEvent::ENDED, ENDED_CALLBACK_ID)).Times(1); } TEST(EventCallerTest, AssignCallbackIdUnregistersPreviousCallback) {