diff --git a/.changeset/platform-audio-stability.md b/.changeset/platform-audio-stability.md new file mode 100644 index 000000000..69fa9ca18 --- /dev/null +++ b/.changeset/platform-audio-stability.md @@ -0,0 +1,14 @@ +--- +livekit: minor +libwebrtc: minor +livekit-ffi: patch +webrtc-sys: patch +--- + +Fix platform ADM teardown races on macOS and release FFI handles on dispose. + +- Clear remaining FFI handles during `FfiServer::dispose` so native resources are released across repeated initialize/shutdown cycles. +- Stop and detach platform/synthetic audio I/O before peer connection factory teardown, preventing `CaptureWorkerThread` from delivering into destroyed transports. Audio I/O is stopped (joining the worker threads) before the audio callback is detached, since `AudioDeviceBuffer` refuses callback changes while media is active. +- Close rooms before dropping FFI track handles and stop platform capture when releasing the platform ADM reference. +- Add `LkRuntime::shutdown_audio_io()` and `PeerConnectionFactoryExt::shutdown_audio_io()` for explicit audio I/O shutdown during runtime teardown. +- Serialize runtime lifecycles: `LkRuntime::instance()` now waits (bounded, 10s) for a previous runtime's teardown to fully complete before constructing a new peer connection factory/ADM. This closes a race where factory/ADM/transport destruction from one initialize/shutdown cycle could overlap the next cycle's startup. diff --git a/libwebrtc/src/native/peer_connection_factory.rs b/libwebrtc/src/native/peer_connection_factory.rs index b48f2bd83..406d9116f 100644 --- a/libwebrtc/src/native/peer_connection_factory.rs +++ b/libwebrtc/src/native/peer_connection_factory.rs @@ -316,6 +316,14 @@ impl PeerConnectionFactory { pub fn is_platform_adm_active(&self) -> bool { self.sys_handle.audio_device().is_platform_adm_active() } + + /// Stops platform/synthetic audio I/O and detaches WebRTC callbacks. + /// + /// Call before tearing down peer connections so capture worker threads + /// cannot deliver frames into transports that are being destroyed. + pub fn shutdown_audio_io(&self) { + self.sys_handle.shutdown_audio_io(); + } } #[cfg(test)] diff --git a/libwebrtc/src/peer_connection_factory.rs b/libwebrtc/src/peer_connection_factory.rs index d4e7cb8d9..6fdbcf486 100644 --- a/libwebrtc/src/peer_connection_factory.rs +++ b/libwebrtc/src/peer_connection_factory.rs @@ -156,6 +156,9 @@ pub mod native { fn release_platform_adm(&self); fn platform_adm_ref_count(&self) -> i32; fn is_platform_adm_active(&self) -> bool; + + /// Stops platform/synthetic audio I/O before runtime teardown. + fn shutdown_audio_io(&self); } impl PeerConnectionFactoryExt for PeerConnectionFactory { @@ -298,5 +301,9 @@ pub mod native { fn is_platform_adm_active(&self) -> bool { self.handle.is_platform_adm_active() } + + fn shutdown_audio_io(&self) { + self.handle.shutdown_audio_io(); + } } } diff --git a/livekit-ffi/src/server/mod.rs b/livekit-ffi/src/server/mod.rs index 1689bdbbe..f616045b5 100644 --- a/livekit-ffi/src/server/mod.rs +++ b/livekit-ffi/src/server/mod.rs @@ -176,8 +176,16 @@ impl FfiServer { self.logger.set_capture_logs(false); - // Drop all handles *self.config.lock() = None; // Invalidate the config + self.ffi_handles.clear(); + self.handle_dropped_txs.clear(); + + // Note: clearing the handles above releases the last FFI-held + // references to the shared LkRuntime, but its teardown may complete + // asynchronously on another thread. dispose() intentionally does not + // block on it; `LkRuntime::instance()` waits for any in-flight + // teardown before constructing the next runtime, so a subsequent + // initialize cannot overlap it. } pub fn send_event(&self, message: proto::ffi_event::Message) -> FfiResult<()> { diff --git a/livekit-ffi/src/server/room.rs b/livekit-ffi/src/server/room.rs index fd7aefe63..ab52f1ee1 100644 --- a/livekit-ffi/src/server/room.rs +++ b/livekit-ffi/src/server/room.rs @@ -361,6 +361,10 @@ impl FfiRoom { /// Close the room and stop the tasks pub async fn close(&self, server: &'static FfiServer, reason: DisconnectReason) { + // Close the room first so local tracks are unpublished and WebRTC + // stops platform capture before FFI track handles are dropped. + let _ = self.inner.room.close_with_reason(reason.into()).await; + // drop associated track handles for (_, &handle) in self.inner.track_handle_lookup.lock().iter() { if server.drop_handle(handle) { @@ -369,8 +373,6 @@ impl FfiRoom { } } - let _ = self.inner.room.close_with_reason(reason.into()).await; - let handle = self.handle.lock().await.take(); if let Some(handle) = handle { let _ = handle.close_tx.send(()); diff --git a/livekit/src/platform_audio/mod.rs b/livekit/src/platform_audio/mod.rs index 7af17dfc2..63e9c90f3 100644 --- a/livekit/src/platform_audio/mod.rs +++ b/livekit/src/platform_audio/mod.rs @@ -304,8 +304,9 @@ struct PlatformAdmHandle { impl Drop for PlatformAdmHandle { fn drop(&mut self) { log::debug!("PlatformAdmHandle dropped - releasing Platform ADM"); - // Release Platform ADM reference - // When ref_count reaches 0, the Platform ADM is terminated + // Stop platform capture before releasing the ADM reference so the + // capture worker cannot deliver frames during teardown. + let _ = self.runtime.stop_recording(); self.runtime.release_platform_adm(); log::info!( "PlatformAdmHandle: released Platform ADM (ref_count now: {})", diff --git a/livekit/src/rtc_engine/lk_runtime.rs b/livekit/src/rtc_engine/lk_runtime.rs index 81ad95e70..e29fd875d 100644 --- a/livekit/src/rtc_engine/lk_runtime.rs +++ b/livekit/src/rtc_engine/lk_runtime.rs @@ -15,21 +15,59 @@ use std::{ fmt::{Debug, Formatter}, sync::{Arc, Weak}, + time::Duration, }; use lazy_static::lazy_static; use libwebrtc::prelude::*; -use parking_lot::Mutex; +use parking_lot::{Condvar, Mutex}; #[cfg(not(target_arch = "wasm32"))] use libwebrtc::peer_connection_factory::native::PeerConnectionFactoryExt; lazy_static! { static ref LK_RUNTIME: Mutex> = Mutex::new(Weak::new()); + + /// Number of [`LkRuntime`] instances created but not yet fully dropped. + /// + /// Unlike [`LK_RUNTIME`] (whose [`Weak`] stops upgrading the instant the + /// strong count hits zero, i.e. *before* `Drop` runs), this only reaches + /// zero once teardown - peer connection factory destruction, ADM + /// termination, capture/render worker-thread joins - has fully completed. + static ref LK_RUNTIME_TEARDOWN_GATE: (Mutex, Condvar) = + (Mutex::new(0), Condvar::new()); +} + +/// How long to wait for a previous runtime's teardown before giving up. +const RUNTIME_TEARDOWN_TIMEOUT: Duration = Duration::from_secs(10); + +/// Tracks a live [`LkRuntime`] in [`LK_RUNTIME_TEARDOWN_GATE`]. +/// +/// Must be the *last* field of [`LkRuntime`]: struct fields drop in +/// declaration order, so this guard's `Drop` (which opens the gate) only runs +/// after `pc_factory` has been fully destroyed. +struct RuntimeTeardownGuard; + +impl RuntimeTeardownGuard { + fn new() -> Self { + *LK_RUNTIME_TEARDOWN_GATE.0.lock() += 1; + Self + } +} + +impl Drop for RuntimeTeardownGuard { + fn drop(&mut self) { + let (lock, cv) = &*LK_RUNTIME_TEARDOWN_GATE; + let mut live = lock.lock(); + *live = live.saturating_sub(1); + cv.notify_all(); + } } pub struct LkRuntime { pc_factory: PeerConnectionFactory, + /// Keep last so it drops after `pc_factory`; see [`RuntimeTeardownGuard`]. + _teardown_guard: RuntimeTeardownGuard, } impl Debug for LkRuntime { @@ -44,13 +82,47 @@ impl LkRuntime { if let Some(lk_runtime) = lk_runtime_ref.upgrade() { lk_runtime } else { + // The previous runtime's strong count may have just reached zero + // while its `Drop` (factory + ADM teardown, including joining the + // platform capture/render worker threads) is still running on + // another thread. Creating a new factory/ADM concurrently with + // that teardown races platform audio/device state, so wait for it + // to finish first. Holding the `LK_RUNTIME` lock here is safe: the + // dropping thread only touches the teardown gate, never this lock. + if !Self::wait_for_teardown(RUNTIME_TEARDOWN_TIMEOUT) { + log::error!( + "LkRuntime::instance() timed out after {RUNTIME_TEARDOWN_TIMEOUT:?} waiting \ + for a previous runtime to tear down; continuing anyway - audio I/O from the \ + old runtime may still be shutting down" + ); + } log::debug!("LkRuntime::new()"); - let new_runtime = Arc::new(Self { pc_factory: PeerConnectionFactory::default() }); + let new_runtime = Arc::new(Self { + pc_factory: PeerConnectionFactory::default(), + _teardown_guard: RuntimeTeardownGuard::new(), + }); *lk_runtime_ref = Arc::downgrade(&new_runtime); new_runtime } } + /// Blocks until every previously created runtime has finished tearing down, + /// or `timeout` elapses. + /// + /// Returns `true` once no runtime teardown is in flight, `false` on + /// timeout. Used by [`instance`](Self::instance) before constructing a new + /// runtime, so that no factory/ADM teardown from a previous lifecycle + /// overlaps the next one's startup. + pub(crate) fn wait_for_teardown(timeout: Duration) -> bool { + let (lock, cv) = &*LK_RUNTIME_TEARDOWN_GATE; + let mut live = lock.lock(); + if *live == 0 { + return true; + } + log::debug!("waiting for {} previous LkRuntime teardown(s) to complete", *live); + !cv.wait_while_for(&mut live, |live| *live > 0, timeout).timed_out() + } + pub fn pc_factory(&self) -> &PeerConnectionFactory { &self.pc_factory } @@ -275,10 +347,18 @@ impl LkRuntime { pub(crate) fn is_platform_adm_active(&self) -> bool { self.pc_factory.is_platform_adm_active() } + + /// Stops platform/synthetic audio I/O before runtime teardown. + #[cfg(not(target_arch = "wasm32"))] + pub fn shutdown_audio_io(&self) { + self.pc_factory.shutdown_audio_io(); + } } impl Drop for LkRuntime { fn drop(&mut self) { log::debug!("LkRuntime::drop()"); + #[cfg(not(target_arch = "wasm32"))] + self.shutdown_audio_io(); } } diff --git a/webrtc-sys/include/livekit/adm_proxy.h b/webrtc-sys/include/livekit/adm_proxy.h index d099f916e..6e0fe3c51 100644 --- a/webrtc-sys/include/livekit/adm_proxy.h +++ b/webrtc-sys/include/livekit/adm_proxy.h @@ -93,6 +93,13 @@ class AdmProxy : public webrtc::AudioDeviceModule { /// Returns true if Platform ADM is currently active (ref_count > 0). bool is_platform_adm_active() const; + /// Stops platform and synthetic audio I/O and detaches the WebRTC + /// AudioTransport callback. + /// + /// Call before tearing down the peer connection factory so capture worker + /// threads cannot deliver frames into transports that are being destroyed. + void StopAudioIO(); + // =========================================================================== // Recording/Playout Control // =========================================================================== @@ -227,6 +234,10 @@ class AdmProxy : public webrtc::AudioDeviceModule { // Must be called with mutex_ held. void SwitchRecordingAdmIfNeeded(); + // Stops platform capture/playout and detaches its callback without touching + // synthetic mode. Must be called with mutex_ held. + void StopPlatformAudioIO(); + #if defined(__ANDROID__) // Lazily creates the Platform ADM on Android. // Must be called with mutex_ held. diff --git a/webrtc-sys/include/livekit/peer_connection_factory.h b/webrtc-sys/include/livekit/peer_connection_factory.h index 371d5313a..048bb600c 100644 --- a/webrtc-sys/include/livekit/peer_connection_factory.h +++ b/webrtc-sys/include/livekit/peer_connection_factory.h @@ -68,6 +68,10 @@ class PeerConnectionFactory { RtpCapabilities rtp_receiver_capabilities(MediaType type) const; + /// Stops platform/synthetic audio I/O and detaches callbacks on the worker + /// thread before peer connection factory teardown. + void shutdown_audio_io() const; + std::shared_ptr rtc_runtime() const { return rtc_runtime_; } std::shared_ptr audio_device() const; diff --git a/webrtc-sys/src/adm_proxy.cpp b/webrtc-sys/src/adm_proxy.cpp index 209f39d53..a34d4ea2d 100644 --- a/webrtc-sys/src/adm_proxy.cpp +++ b/webrtc-sys/src/adm_proxy.cpp @@ -71,6 +71,9 @@ AdmProxy::AdmProxy(const webrtc::Environment& env, webrtc::Thread* worker_thread AdmProxy::~AdmProxy() { RTC_LOG(LS_VERBOSE) << "AdmProxy::~AdmProxy()"; + StopAudioIO(); + + webrtc::MutexLock lock(&mutex_); if (synthetic_adm_) { synthetic_adm_->Terminate(); synthetic_adm_ = nullptr; @@ -180,6 +183,7 @@ void AdmProxy::ReleasePlatformAdm() { // Note: We do NOT terminate the Platform ADM - it stays alive until destructor. // This avoids iOS KVO race conditions from re-creating the ADM. if (platform_adm_ref_count_ == 0) { + StopPlatformAudioIO(); SwitchPlayoutModeIfNeeded(); SwitchRecordingAdmIfNeeded(); } @@ -273,6 +277,50 @@ void AdmProxy::SwitchRecordingAdmIfNeeded() { } } +void AdmProxy::StopPlatformAudioIO() { + recording_ = false; + + if (platform_adm_) { + // Stop capture/playout first: StopRecording()/StopPlayout() join the + // platform worker threads. Only then detach the callback - + // AudioDeviceBuffer refuses RegisterAudioCallback() while media is active, + // so detaching before stopping would silently fail and leave the transport + // attached. + platform_adm_->StopRecording(); + platform_adm_->StopPlayout(); + platform_adm_->RegisterAudioCallback(nullptr); + // platform_adm_ is kept alive for re-acquire and iOS compatibility; see + // ReleasePlatformAdm(). + } +} + +void AdmProxy::StopAudioIO() { + webrtc::MutexLock lock(&mutex_); + + recording_ = false; + playing_ = false; + recording_initialized_ = false; + playout_initialized_ = false; + + // Stop before detaching: AudioDeviceBuffer refuses RegisterAudioCallback() + // while media is active, so the detach only takes effect once + // capture/playout worker threads have been stopped (and joined). + if (platform_adm_) { + platform_adm_->StopRecording(); + platform_adm_->StopPlayout(); + platform_adm_->RegisterAudioCallback(nullptr); + } + + if (synthetic_adm_) { + synthetic_adm_->StopRecording(); + synthetic_adm_->StopPlayout(); + synthetic_adm_->RegisterAudioCallback(nullptr); + // synthetic_adm_ is kept alive until ~AdmProxy() / Terminate(). + } + + audio_transport_ = nullptr; +} + // ============================================================================= // AudioDeviceModule Interface Implementation // ============================================================================= @@ -306,8 +354,9 @@ int32_t AdmProxy::Init() { } int32_t AdmProxy::Terminate() { - webrtc::MutexLock lock(&mutex_); + StopAudioIO(); + webrtc::MutexLock lock(&mutex_); int32_t result = 0; if (synthetic_adm_) { result = synthetic_adm_->Terminate(); @@ -554,11 +603,20 @@ int32_t AdmProxy::StopRecording() { webrtc::MutexLock lock(&mutex_); recording_ = false; - auto* adm = recording_adm(); - if (adm) { - return adm->StopRecording(); + int32_t result = 0; + if (platform_adm_) { + const int32_t platform_result = platform_adm_->StopRecording(); + if (result == 0) { + result = platform_result; + } } - return 0; + if (synthetic_adm_) { + const int32_t synthetic_result = synthetic_adm_->StopRecording(); + if (result == 0) { + result = synthetic_result; + } + } + return result; } bool AdmProxy::Recording() const { diff --git a/webrtc-sys/src/peer_connection_factory.cpp b/webrtc-sys/src/peer_connection_factory.cpp index cfb2d0c06..b34ed7cdf 100644 --- a/webrtc-sys/src/peer_connection_factory.cpp +++ b/webrtc-sys/src/peer_connection_factory.cpp @@ -90,6 +90,8 @@ PeerConnectionFactory::PeerConnectionFactory( PeerConnectionFactory::~PeerConnectionFactory() { RTC_LOG(LS_VERBOSE) << "PeerConnectionFactory::~PeerConnectionFactory()"; + shutdown_audio_io(); + peer_factory_ = nullptr; audio_device_ = nullptr; rtc_runtime_->worker_thread()->BlockingCall( @@ -163,6 +165,14 @@ std::shared_ptr PeerConnectionFactory::audio_device() con return audio_device_; } +void PeerConnectionFactory::shutdown_audio_io() const { + rtc_runtime_->worker_thread()->BlockingCall([this] { + if (adm_proxy_) { + adm_proxy_->StopAudioIO(); + } + }); +} + std::shared_ptr create_peer_connection_factory() { return std::make_shared(RtcRuntime::create()); } diff --git a/webrtc-sys/src/peer_connection_factory.rs b/webrtc-sys/src/peer_connection_factory.rs index 16794f6bf..5f5ed1242 100644 --- a/webrtc-sys/src/peer_connection_factory.rs +++ b/webrtc-sys/src/peer_connection_factory.rs @@ -89,6 +89,8 @@ pub mod ffi { fn create_peer_connection_factory() -> SharedPtr; + fn shutdown_audio_io(self: &PeerConnectionFactory); + fn create_peer_connection( self: &PeerConnectionFactory, config: RtcConfiguration,