From 33dfcb27b083b7d07389174feec010b3e3873d79 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Sat, 10 Jun 2023 18:49:00 +0200 Subject: [PATCH] fix: datachannel deadlocks & dispose crashes (#87) - Fix datachannel deadlock on dispose - Fix audio device being disposed on the wrong thread - Allow multiple RtcRuntime to be created (Useful for unit tests where we use multiple PeerConnectionFactory) --- livekit-webrtc/src/peer_connection.rs | 22 ++++---- webrtc-sys/include/livekit/data_channel.h | 3 +- webrtc-sys/include/livekit/peer_connection.h | 2 + .../include/livekit/peer_connection_factory.h | 3 ++ webrtc-sys/include/livekit/webrtc.h | 8 +-- webrtc-sys/src/data_channel.cpp | 8 +-- webrtc-sys/src/peer_connection.cpp | 9 +++- webrtc-sys/src/peer_connection_factory.cpp | 22 +++++--- webrtc-sys/src/webrtc.cpp | 52 ++++++++++++++++--- 9 files changed, 90 insertions(+), 39 deletions(-) diff --git a/livekit-webrtc/src/peer_connection.rs b/livekit-webrtc/src/peer_connection.rs index e43af53..95d9fd6 100644 --- a/livekit-webrtc/src/peer_connection.rs +++ b/livekit-webrtc/src/peer_connection.rs @@ -247,13 +247,9 @@ mod tests { use log::trace; use tokio::sync::mpsc; - fn init_log() { - let _ = env_logger::builder().is_test(true).try_init(); - } - #[tokio::test] async fn create_pc() { - init_log(); + let _ = env_logger::builder().is_test(true).try_init(); let factory = PeerConnectionFactory::default(); let config = RtcConfiguration { @@ -269,20 +265,20 @@ mod tests { let bob = factory.create_peer_connection(config.clone()).unwrap(); let alice = factory.create_peer_connection(config.clone()).unwrap(); - let (bob_ice_tx, mut bob_ice_rx) = mpsc::channel::(16); - let (alice_ice_tx, mut alice_ice_rx) = mpsc::channel::(16); - let (alice_dc_tx, mut alice_dc_rx) = mpsc::channel::(16); + let (bob_ice_tx, mut bob_ice_rx) = mpsc::unbounded_channel::(); + let (alice_ice_tx, mut alice_ice_rx) = mpsc::unbounded_channel::(); + let (alice_dc_tx, mut alice_dc_rx) = mpsc::unbounded_channel::(); bob.on_ice_candidate(Some(Box::new(move |candidate| { - bob_ice_tx.blocking_send(candidate).unwrap(); + bob_ice_tx.send(candidate).unwrap(); }))); alice.on_ice_candidate(Some(Box::new(move |candidate| { - alice_ice_tx.blocking_send(candidate).unwrap(); + alice_ice_tx.send(candidate).unwrap(); }))); alice.on_data_channel(Some(Box::new(move |dc| { - alice_dc_tx.blocking_send(dc).unwrap(); + alice_dc_tx.send(dc).unwrap(); }))); let bob_dc = bob @@ -305,11 +301,11 @@ mod tests { bob.add_ice_candidate(alice_ice).await.unwrap(); alice.add_ice_candidate(bob_ice).await.unwrap(); - let (data_tx, mut data_rx) = mpsc::channel::(1); + let (data_tx, mut data_rx) = mpsc::unbounded_channel::(); let alice_dc = alice_dc_rx.recv().await.unwrap(); alice_dc.on_message(Some(Box::new(move |buffer| { data_tx - .blocking_send(String::from_utf8_lossy(buffer.data).to_string()) + .send(String::from_utf8_lossy(buffer.data).to_string()) .unwrap(); }))); diff --git a/webrtc-sys/include/livekit/data_channel.h b/webrtc-sys/include/livekit/data_channel.h index 9c64aba..9052ec5 100644 --- a/webrtc-sys/include/livekit/data_channel.h +++ b/webrtc-sys/include/livekit/data_channel.h @@ -40,6 +40,7 @@ class DataChannel { explicit DataChannel( std::shared_ptr rtc_runtime, rtc::scoped_refptr data_channel); + ~DataChannel(); void register_observer(rust::Box observer) const; void unregister_observer() const; @@ -64,8 +65,6 @@ class NativeDataChannelObserver : public webrtc::DataChannelObserver { NativeDataChannelObserver(rust::Box observer, const DataChannel* dc); - ~NativeDataChannelObserver(); - void OnStateChange() override; void OnMessage(const webrtc::DataBuffer& buffer) override; void OnBufferedAmountChange(uint64_t sent_data_size) override; diff --git a/webrtc-sys/include/livekit/peer_connection.h b/webrtc-sys/include/livekit/peer_connection.h index d3d6bf9..1fe6700 100644 --- a/webrtc-sys/include/livekit/peer_connection.h +++ b/webrtc-sys/include/livekit/peer_connection.h @@ -48,6 +48,8 @@ class PeerConnection { std::unique_ptr observer, rtc::scoped_refptr peer_connection); + ~PeerConnection(); + void create_offer( RtcOfferAnswerOptions options, rust::Box ctx, diff --git a/webrtc-sys/include/livekit/peer_connection_factory.h b/webrtc-sys/include/livekit/peer_connection_factory.h index bb110e5..a80469d 100644 --- a/webrtc-sys/include/livekit/peer_connection_factory.h +++ b/webrtc-sys/include/livekit/peer_connection_factory.h @@ -17,6 +17,8 @@ #pragma once #include "api/peer_connection_interface.h" +#include "api/scoped_refptr.h" +#include "livekit/audio_device.h" #include "media_stream.h" #include "peer_connection.h" #include "rtp_parameters.h" @@ -57,6 +59,7 @@ class PeerConnectionFactory { private: std::shared_ptr rtc_runtime_; + rtc::scoped_refptr audio_device_; rtc::scoped_refptr peer_factory_; }; diff --git a/webrtc-sys/include/livekit/webrtc.h b/webrtc-sys/include/livekit/webrtc.h index 78e28ee..88ad102 100644 --- a/webrtc-sys/include/livekit/webrtc.h +++ b/webrtc-sys/include/livekit/webrtc.h @@ -80,6 +80,8 @@ class RtcRuntime : public std::enable_shared_from_this { // have one livekit::VideoTrack associated with it). // The only reason we to do that is to allow to add states inside our // wrappers (e.g: the sinks_ member inside AudioTrack) + // DataChannel and the PeerConnectionFactory don't need to do this (There's no + // way to retrieve them after creation) webrtc::Mutex mutex_; std::vector> media_stream_tracks_; // We don't have additonal state in RtpReceiver and RtpSender atm.. @@ -87,9 +89,9 @@ class RtcRuntime : public std::enable_shared_from_this { // std::vector> rtp_senders_; #ifdef WEBRTC_WIN - rtc::WinsockInitializer winsock_; - rtc::PhysicalSocketServer ss_; - rtc::AutoSocketServerThread main_thread_{&ss_}; + // rtc::WinsockInitializer winsock_; + // rtc::PhysicalSocketServer ss_; + // rtc::AutoSocketServerThread main_thread_{&ss_}; #endif }; diff --git a/webrtc-sys/src/data_channel.cpp b/webrtc-sys/src/data_channel.cpp index d6863f7..0eb1bcb 100644 --- a/webrtc-sys/src/data_channel.cpp +++ b/webrtc-sys/src/data_channel.cpp @@ -47,6 +47,10 @@ DataChannel::DataChannel( rtc::scoped_refptr data_channel) : rtc_runtime_(rtc_runtime), data_channel_(std::move(data_channel)) {} +DataChannel::~DataChannel() { + unregister_observer(); +} + void DataChannel::register_observer( rust::Box observer) const { webrtc::MutexLock lock(&mutex_); @@ -86,10 +90,6 @@ NativeDataChannelObserver::NativeDataChannelObserver( const DataChannel* dc) : observer_(std::move(observer)), dc_(dc) {} -NativeDataChannelObserver::~NativeDataChannelObserver() { - dc_->unregister_observer(); -} - void NativeDataChannelObserver::OnStateChange() { observer_->on_state_change(dc_->state()); } diff --git a/webrtc-sys/src/peer_connection.cpp b/webrtc-sys/src/peer_connection.cpp index cb7c8ce..507f4e4 100644 --- a/webrtc-sys/src/peer_connection.cpp +++ b/webrtc-sys/src/peer_connection.cpp @@ -25,6 +25,7 @@ #include "livekit/media_stream.h" #include "livekit/rtc_error.h" #include "livekit/rtp_transceiver.h" +#include "rtc_base/logging.h" #include "webrtc-sys/src/peer_connection.rs.h" #include "webrtc-sys/src/rtc_error.rs.h" @@ -50,7 +51,13 @@ PeerConnection::PeerConnection( rtc::scoped_refptr peer_connection) : rtc_runtime_(rtc_runtime), observer_(std::move(observer)), - peer_connection_(std::move(peer_connection)) {} + peer_connection_(std::move(peer_connection)) { + RTC_LOG(LS_VERBOSE) << "PeerConnection::PeerConnection()"; +} + +PeerConnection::~PeerConnection() { + RTC_LOG(LS_VERBOSE) << "PeerConnection::~PeerConnection()"; +} void PeerConnection::create_offer( RtcOfferAnswerOptions options, diff --git a/webrtc-sys/src/peer_connection_factory.cpp b/webrtc-sys/src/peer_connection_factory.cpp index 14653f6..6829d1f 100644 --- a/webrtc-sys/src/peer_connection_factory.cpp +++ b/webrtc-sys/src/peer_connection_factory.cpp @@ -70,7 +70,7 @@ webrtc::PeerConnectionInterface::RTCConfiguration to_native_rtc_configuration( PeerConnectionFactory::PeerConnectionFactory( std::shared_ptr rtc_runtime) : rtc_runtime_(rtc_runtime) { - RTC_LOG(LS_INFO) << "PeerConnectionFactory::PeerConnectionFactory()"; + RTC_LOG(LS_VERBOSE) << "PeerConnectionFactory::PeerConnectionFactory()"; webrtc::PeerConnectionFactoryDependencies dependencies; dependencies.network_thread = rtc_runtime_->network_thread(); @@ -86,12 +86,14 @@ PeerConnectionFactory::PeerConnectionFactory( cricket::MediaEngineDependencies media_deps; media_deps.task_queue_factory = dependencies.task_queue_factory.get(); - media_deps.adm = rtc_runtime_->worker_thread() - ->Invoke>( - RTC_FROM_HERE, [&] { - return rtc::make_ref_counted( - media_deps.task_queue_factory); - }); + audio_device_ = rtc_runtime_->worker_thread() + ->Invoke>( + RTC_FROM_HERE, [&] { + return rtc::make_ref_counted( + media_deps.task_queue_factory); + }); + + media_deps.adm = audio_device_; media_deps.video_encoder_factory = std::move(std::make_unique()); @@ -114,7 +116,11 @@ PeerConnectionFactory::PeerConnectionFactory( } PeerConnectionFactory::~PeerConnectionFactory() { - RTC_LOG(LS_INFO) << "PeerConnectionFactory::~PeerConnectionFactory()"; + RTC_LOG(LS_VERBOSE) << "PeerConnectionFactory::~PeerConnectionFactory()"; + + peer_factory_ = nullptr; + rtc_runtime_->worker_thread()->Invoke( + RTC_FROM_HERE, [this] { audio_device_ = nullptr; }); } std::shared_ptr PeerConnectionFactory::create_peer_connection( diff --git a/webrtc-sys/src/webrtc.cpp b/webrtc-sys/src/webrtc.cpp index eacd3bb..92182b1 100644 --- a/webrtc-sys/src/webrtc.cpp +++ b/webrtc-sys/src/webrtc.cpp @@ -16,6 +16,9 @@ #include "livekit/webrtc.h" +#include +#include +#include #include #include "livekit/audio_track.h" @@ -27,10 +30,34 @@ #include "rtc_base/logging.h" #include "rtc_base/synchronization/mutex.h" +#ifdef WEBRTC_WIN +#include "rtc_base/win32.h" +#endif + namespace livekit { + +static webrtc::Mutex g_mutex{}; +// Can't be atomic, we're using a Mutex because we need to wait for the +// execution of the first init +static uint32_t g_release_counter(0); + RtcRuntime::RtcRuntime() { - RTC_LOG(LS_INFO) << "RtcRuntime()"; - RTC_CHECK(rtc::InitializeSSL()) << "Failed to InitializeSSL()"; + rtc::LogMessage::LogToDebug(rtc::LS_INFO); + RTC_LOG(LS_VERBOSE) << "RtcRuntime()"; + + { + // Not the best way to do it... + webrtc::MutexLock lock(&g_mutex); + if (g_release_counter == 0) { + RTC_CHECK(rtc::InitializeSSL()) << "Failed to InitializeSSL()"; + +#ifdef WEBRTC_WIN + WSADATA data; + WSAStartup(MAKEWORD(1, 0), &data); +#endif + } + g_release_counter++; + } network_thread_ = rtc::Thread::CreateWithSocketServer(); network_thread_->SetName("network_thread", &network_thread_); @@ -44,14 +71,23 @@ RtcRuntime::RtcRuntime() { } RtcRuntime::~RtcRuntime() { - RTC_LOG(LS_INFO) << "~RtcRuntime()"; + RTC_LOG(LS_VERBOSE) << "~RtcRuntime()"; - rtc::ThreadManager::Instance()->SetCurrentThread(nullptr); - RTC_CHECK(rtc::CleanupSSL()) << "Failed to CleanupSSL()"; + worker_thread_->Stop(); + signaling_thread_->Stop(); + network_thread_->Stop(); - worker_thread_->Quit(); - signaling_thread_->Quit(); - network_thread_->Quit(); + { + webrtc::MutexLock lock(&g_mutex); + g_release_counter--; + if (g_release_counter == 0) { + RTC_CHECK(rtc::CleanupSSL()) << "Failed to CleanupSSL()"; + +#ifdef WEBRTC_WIN + WSACleanup(); +#endif + } + } } rtc::Thread* RtcRuntime::network_thread() const {