thread-safety-itc

# Skill: Thread Safety & Inter-Thread Communication Three threads interact in this codebase. Every line of code that crosses a thread boundary must use the correct primitive or it is a bug. **When in doubt about which ITC primitive to use → go to the Decision Table below first.** --- ## The Three Threads | Thread | Alias | Runs | |---|---|---| | React Native JS thread | "JS thread" | User code, HostObject methods, `scheduleAudioEvent` calls | | Audio thread | "audio thread" | `processNode()` — driven by Oboe (Android) / CoreAudio (iOS) | | Worker threads | "off-thread" | FFmpeg decoding, file I/O, `TaskOffloader` tasks | **Audio thread is real-time.** It has a hard deadline (~3ms at 44100 Hz, 128 frames). Missing it causes audible glitches. --- ## Audio Thread Contract `processNode()` **MUST NOT**: - Allocate or free memory (`new`, `delete`, `malloc`, `free`, any `push_back` that grows) - Acquire any mutex (`std::mutex`, `std::lock_guard`, `std::unique_lock`) - Make blocking syscalls (file I/O, socket I/O, `sleep`, `wait`) - Call into JavaScript (no JSI calls, no `callInvoker_->invokeSync()`) - Throw exceptions **Preallocate everything in the constructor (JS thread).** The audio thread only uses what was already allocated. --- ## JS → Audio: `CrossThreadEventScheduler` The standard way to send property updates from JS to the audio thread. ```cpp // JS thread (HostObject setter): auto oscillatorNode = std::static_pointer_cast<OscillatorNode>(node_); auto event = [oscillatorNode, type](BaseAudioContext &) { oscillatorNode->setType(type); // runs on audio thread }; oscillatorNode->scheduleAudioEvent(std::move(event)); ``` `scheduleAudioEvent()` is defined on `AudioNode`. It enqueues a lambda into the node's `CrossThreadEventScheduler<BaseAudioContext>`. The audio thread drains the queue at the start of each render cycle. **Never assume immediate consistency** — by the time the audio thread processes the event, several render quanta may have passed. --- ## Audio → JS: `IAudioEventHandlerRegistry` and `EventCaller` Send events from the audio thread back to JS (e.g. `ended`, `loopEnded`, `positionChanged`). **Prefer `EventCaller<AudioEvent::X>`** — a small RAII helper templated on the event type. `dispatch()` requires a payload matching `EventPayloadFor<AudioEvent::X, Payload>` (see `AudioEventPayloadMapping.hpp`). ```cpp // Node member (composition — one EventCaller per event) EventCaller<AudioEvent::ENDED> onEndedEvent_{context->getAudioEventHandlerRegistry()}; // JS thread: HostObject setter forwards to a named method void assignOnEndedCallbackId(uint64_t id) { onEndedEvent_.assignCallbackId(id); } // Audio thread: fire when playback ends (EmptyPayload only for ENDED) onEndedEvent_.dispatchEmpty(); // With typed payload — compile error if payload does not match the event errorEvent_.dispatch(StringPayload{.name = "message", .reason = message}); positionChangedEvent_.dispatch(DoubleValuePayload{.value = position}); ``` Throttled events (e.g. `positionChanged`) use `PositionChangedDispatcher`, which **contains** an `EventCaller` plus interval/flush logic — still composition, not node inheritance from `EventCaller`. **Unregister lifecycle:** - `EventCaller::~EventCaller()` unregisters if `callbackId != 0` when the node is destroyed - `assignCallbackId(0)` unregisters when JS sets `onX = null` - **HostObject destructor must call `assignOnXCallbackId(0)`** for every event wired on that node — the C++ node can outlive the HostObject, and the registry JSI function may be invalid after GC - 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 `EventCaller::deferEmpty(context->getDeferredEvents(), dueTime)`. The context keeps a preallocated `DeferredEventQueue` 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 audioEventHandlerRegistry_->dispatchEvent(AudioEvent::ENDED, callbackId, AudioEventPayload{...}); ``` Internally calls `callInvoker_->invokeAsync()` — safe to call from the audio thread via `dispatchEvent`. --- ## Graph Mutations: `AudioGraphManager` Connect/disconnect operations queue via `AudioGraphManager` (its own internal SPSC channel). The audio thread calls `graphManager_->preProcessGraph()` before each render pass to apply pending changes. Do not call `AudioGraphManager` directly — go through `AudioNode::connect()` / `disconnect()`. ### Processable state is audio-thread-only Per-quantum processable state (`ALWAYS_`/`CONDITIONAL_`/`NOT_PROCESSABLE`) is derived exclusively on the audio thread by `AudioGraph::settleProcessableState()` (a reverse-topological pull run inside `Graph::process()`), using only audio-thread-owned data: the topo-sorted node array, `InputPool` input lists, and `link_head` processable-links. **Pitfall (fixed):** earlier, `HostGraph` AGEvents (`addEdge`/`removeEdge`/`removeAllEdges`) walked `HostGraph::Node::{inputs,outputs,linkedNodes}` on the audio thread to mark processable state incrementally. Those vectors mutate on the JS thread under `nodesMutex_` — a cross-thread race. AGEvents must never read HostGraph adjacency for processable state; they only mirror structural edges/links onto the audio graph. The host-side `linkedNodes` list is now kept solely so links can be scrubbed when a linked node is disposed. --- ## Decision Table | Scenario | Correct pattern | |---|---| | JS sets a property → audio thread reads it | Shadow state in HostObject + `scheduleAudioEvent` | | Audio thread fires an event → JS callback | `EventCaller::dispatch()` / `dispatchEmpty()` | | JS connects/disconnects nodes | `AudioNode::connect()` → `AudioGraphManager` | | Property written by audio thread, JS reads it | `std::atomic<T>` on C++ node; getter reads directly | | Non-primitive, can be written by audio thread | Triple buffer (see `AnalyserNode` for reference) | | CPU-heavy work, must not block JS or audio | `TaskOffloader` on a dedicated worker thread | | Context lifecycle (`resume`/`suspend`/`close`) | `scheduleContextPromise` → `pendingPromisesOffloader_` | | Platform code must reach the live recorder without going through JS | Process-global handle (`ActiveRecorderHandle` — recursive mutex + `weak_ptr`; `tryStart` / `stopAndReturnInfo` / `stopAndReturnState` / `pause` / `resume` take the handle mutex before `AudioRecorder`). Android: static-JNI `NativeRecorderControl` (no HybridData). iOS: `AudioAPIModule.setAudioSessionActivity(false)` calls `stopAndReturnInfo()` before deactivating the session. HostObject `start` / `stop` go through the handle; HostObject `pause` / `resume` still call `AudioRecorder` directly. Android blocking calls run on a Kotlin executor (`goAsync()` in receivers), never a detached `std::thread` | --- ## Off-Thread Work: `TaskOffloader` For work that would block both the JS thread and the audio thread (decoding, file writing): ```cpp TaskOffloader<MyWorkItem> offloader([](MyWorkItem item) { // runs on dedicated worker thread — allocs OK, blocking I/O OK item.process(); }); offloader.scheduleTask(std::move(workItem)); ``` See the `utilities` skill for full API. **Pitfall — file writer / recorder shutdown:** `TaskOffloader::shutdown()` drains the SPSC queue before joining the worker, and it drains by *running the task* for each pending item. So drain **while the file still counts as open**. `runWriterTask()` gates its encode on `isFileOpen_`, so clearing that flag first makes every drained buffer take the false branch and be dropped, costing a rotated or closed M4A segment seconds of buffered audio. `finishCurrentFile()` therefore destroys the offloader first and clears the flag after, not the other way round. Types with a `.slot` member use `slot == size_t max` as the shutdown sentinel. That flag is **not** what keeps the audio thread out of the pool being freed, and it cannot be: `writeAudioData()` reads it and then dereferences the pool pointers, so no ordering of those two lines closes the window between the two steps. The caller closes it. The audio thread only enters the writer under the recorder's `fileWriterMutex_`, and every close path either moves the writer out of `fileWriter_` under that mutex first (`disableFileOutput`, `detachSideEffects`) or holds it across the call (the iOS format-change path). That includes the platform recorder destructors: both call `stop()` first, which detaches under the mutexes, and only then close the native stream. A writer driven without that discipline would need a guard of its own. **Android ownership cycle.** `AndroidAudioRecorder::openAudioStream()` hands Oboe `shared_from_this()`, and Oboe keeps that `shared_ptr` for the lifetime of the stream *object* (`AudioStreamBase::mSharedDataCallback`), not just while the stream is open. Since `mStream_` is a member, an opened recorder owns itself through its stream until `cleanup()` resets `mStream_`. Two consequences. The destructor can only run after `cleanup()` has already closed the stream, so no callback can race it — which is why the pre-`stop()` destructor got away with an unlocked `closeFile()`. And `stop()` only calls `requestStop()`, never `cleanup()`, so once JS drops a normally stopped recorder nothing breaks the cycle: the recorder and its AAudio stream leak until a disconnect error, and a recorder dropped mid-recording keeps recording. Any change that breaks the cycle (closing in `stop()`, releasing from the HostObject, or a weak-pointer proxy as Oboe's callback) makes the destructor order above load-bearing. Never hold a lock the worker takes while joining it: `finishCurrentFile()` drains and joins, and only then takes `fileMutex_` to retire the encoder — otherwise the join deadlocks against whatever the worker is doing under that lock. **iOS lock order: engine lock before consumer mutexes.** `AudioEngine` invokes `onInputConfigurationChange` while holding `_engineLock` (interruption end, engine rebuild), and that handler takes the recorder's consumer mutexes (`fileWriterMutex_`, `callbackMutex_`, `adapterNodeMutex_`) one at a time. So no recorder method may call into the native recorder (`start`, `stop`, `detachInputNode`, anything that takes `_engineLock`) while holding a consumer mutex. `IOSAudioRecorder::stop()` therefore stores `Idle`, disarms and stops the native side first, and only then locks to detach the side effects; a handler that fires in between sees `Idle` and returns. The same gate is why `resume()` must store `Recording` *before* the native resume: the native side re-arms input only through that handler, and the handler arms only while the state reads `Recording` (issue #1294). **Pattern — reacting to audio without blocking on it.** When a decision depends on data the audio thread produced but costs blocking work to make (rotating a recording once its file outgrows a cap), do not make it in `writeAudioData()`. Measuring a file can be a `stat`, and acting on it closes and opens an encoder — both forbidden on the audio thread. `AudioFileWriter` makes that decision **on its own worker thread**, in the task that just encoded a buffer (`rotateIfFileOutgrowsCap()`), and only when the file properties carry a non-zero `rotateIntervalBytes`, so a session that never rotates never pays for the `stat`. One mutex (`fileMutex_`) guards the open `RecordingFile` (encoder, path, and the frames it holds — a plain counter, since only the worker encodes and only under this lock) together with the session bookkeeping the rotation advances (file count, finished-file totals); the JS thread reads those under the same lock. `RecordingFile` itself is not thread-safe; rotation is swapping one for the next. The same lock guards the session's file list, which `closeFile()` hands back after joining the worker, so the recorder needs no lock or callback of its own to learn which files a rotation produced. Anything that reaches back out of the writer — the error event — is invoked **after** that lock is released, so a handler may call straight back in (`getFilePath()`) without re-entering a non-recursive mutex. The join-versus-lock ordering is the shutdown pitfall above: `finishCurrentFile()` stops the worker first, then takes the lock to retire the encoder. Swapping the output file must not restart the worker: a rotation exchanges the encoder underneath the live offloader, so a rotating recording costs one thread per file open, not one per segment. Only `reprepareStreamFormat()` (a stream-format change) stops the worker, because the buffer pool is sized from the format. It does **not** swap the file: the file's settings come from the file properties, not the input, so the same encoder is pointed at the new input through `AudioEncoder::changeInputFormat()`, whose protected `reprepareInput` hook only iOS overrides (`IOSEncoder::reprepareInput` rebuilds the `AVAudioConverter`; Android's stream keeps its format, so `AudioEncoder` has no such virtual). Joining the worker first is what makes the in-place switch safe: no `encode()` is in flight when the converter is replaced. Duration bookkeeping needs care here, since `framesWritten_` counts in the current stream rate: the earlier formats' share is folded into `currentFileEarlierFormatsDurationSec_`, which the encoder's whole-file report on close must not double count. **Test seam — inject, do not derive.** `resolveOsFilePath` and `createOsEncoder` both fail in the desktop test build, so a real writer cannot open a file there. Those steps are bundled into a `PlatformFileBackend` struct of `std::function`s (output spec, path, encoder). The writer borrows it by `const &` — the constructor overload taking a temporary is deleted, so it cannot dangle — and defaults to `osFileBackend()`, one static instance shared by every writer. The test fixture owns a backend returning bare file names and a fake `AudioEncoder`, declared before the writer that borrows it; everything else — pool, worker, rotation, totals — runs for real, on a real `AudioFileWriter` rather than a subclass. For name-collision tests the fake backend prefixes a scratch directory, so `stat()` sees real files. Prefer this shape over `protected virtual` hooks for a class the production code instantiates directly. Virtual hooks make the class polymorphic, which then forces a virtual destructor for anyone holding it by base pointer, and they let a test assert against a subclass instead of the real type. With injection the writer is `final` with a non-virtual destructor, and a misconstructed backend is the one thing to guard: calling an empty `std::function` throws, and on the worker thread that terminates, so `openEncoderForNextFile()` checks both slots and returns an error instead. Tests stay deterministic by writing at most a pool's worth of buffers (32) per open and asserting only after `closeFile()`, which drains and joins the worker. **Pitfall — the task type cannot be a nested struct.** `TaskOffloader<T>` constrains `T` with `std::default_initializable`. A task struct carrying default member initializers (which the `.slot` sentinel requires) does *not* satisfy that constraint while its enclosing class is still incomplete, so `using Offloader = TaskOffloader<NestedTask, …>;` inside the class fails to compile with "constraints not satisfied". Making the struct `public` does not help — it is not an access problem. Declare the task type at **namespace scope** instead (`PendingFileWrite`, `PendingCallbackFrames`). Dropping the initializers to satisfy the constraint is worse: `T{}` would then produce `slot == 0`, a valid slot index, making the shutdown sentinel indistinguishable from real work. --- ## Driver synchronization (layered model) Control-plane synchronization uses two layers — both are non-recursive `std::mutex`, never held on the audio thread. | Layer | Location | Protects | |---|---|---| | Context | `BaseAudioContext::driverMutex_` (`AudioContext` + `OfflineAudioContext`) | `start` / `resume` / `suspend` / `close` (live); `resume` / `suspend` / `startRendering` (offline) — JS thread vs promise-pool | | Engine | `AudioEngine` mutex (iOS only) | Process-wide `AVAudioEngine` graph: attach/detach, engine start/stop, interruptions, recorder paths | `AudioContext::initialize()`, `createMediaElementSource()`, and `isDriverRunning()` are JS-thread-only — do not take `driverMutex_`. `BaseAudioContext::getState()` reads the atomic control-thread state and, unless `CLOSED`, reports `SUSPENDED` when `isDriverRunning()` is false (`BaseAudioContext.cpp`); it must not acquire `driverMutex_`, so the player's `isRunning()` and latency getters are reached from the JS thread without it. On Android, `AndroidAudioPlayer::onErrorAfterClose` also takes `driverMutex_` because Oboe error callbacks bypass `AudioContext`. `AndroidAudioPlayer` has a second, non-recursive `streamMutex_` that exists only for those unlocked JS-thread readers: every mutation of `mStream_` already runs under `driverMutex_` (promise worker or error thread). Lock order is `driverMutex_` → `streamMutex_`; public methods lock once and delegate to `Locked` internals (`openAudioStreamLocked`, `rebuildStreamLocked`, `startStreamLocked`, `cleanupLocked`, `isRunningLocked`), which must never call a public locking method. `onErrorAfterClose` releases `streamMutex_` before `context->onStreamFail()`, because that calls back into the public `stop()`. **Live `AudioContext` render quiescence:** `currentRenders_` on `AudioContext` is incremented at the start of each platform I/O callback (`IOSAudioPlayer::deliverOutputBuffers` / `AndroidAudioPlayer::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()`. `NodeAudioPlayer` (WPT harness) receives the same reference but never opens a render scope.