Create a new worker thread for RTP module.

The worker thread needs its own run loop to periodically run some tasks
in the background. Main WebAssembly thread (media worker) already has
its loop for JavaScript events.

Author: jianjunz

talk/owt/BUILD.gn

@@ -337,11 +337,12 @@ if (is_wasm) {
     name = "owt"
     deps = [
       "//third_party/boringssl",
+      "//third_party/webrtc/api/rtc_event_log:rtc_event_log_factory",
       "//third_party/webrtc/call",
+      "//third_party/webrtc/logging:fake_rtc_event_log",
       "//third_party/webrtc/modules/rtp_rtcp:rtp_rtcp",
       "//third_party/webrtc/rtc_base:rtc_json",
       "//third_party/webrtc/rtc_base:rtc_task_queue_stdlib",
-      "//third_party/webrtc/api/rtc_event_log:rtc_event_log_factory",
     ]
     sources = [
       "sdk/wasm/binding.h",

talk/owt/sdk/wasm/gn/wasm.gni

@@ -42,19 +42,13 @@ template("wasm_lib") {
     _target_ldflags = [
       "-s",
       "WASM=1",
-      # "-s",
-      # "DISABLE_EXCEPTION_CATCHING=1",
       "-s",
       "INITIAL_MEMORY=33554432",
       "-s",
       "ALLOW_TABLE_GROWTH=1",
-      # "-s",
-      # "WASM_ASYNC_COMPILATION=0",
       "-s",
       "USE_PTHREADS=1",
       "-s",
-      "PROXY_TO_PTHREAD=1",
-      "-s",
       "LLD_REPORT_UNDEFINED",
       "-s",
       "EXPORTED_RUNTIME_METHODS=" + _exports,
@@ -74,15 +68,6 @@ template("wasm_lib") {
 
       "--bind",
 
-      #"-fno-stack-protector",
-
-      # This is to prevent that two different wasm modules end up generating
-      # JS that overrides the same global variable (var Module = ...)
-      # "-s",
-      # "EXPORT_NAME=${target_name}",
-
-      # "-lworkerfs.js",  # For FS.filesystems.WORKERFS
-
       "-s",
       "PTHREAD_POOL_SIZE=8",
       "-s",
@@ -92,16 +77,8 @@ template("wasm_lib") {
     ]
     if (is_debug) {
       _target_ldflags += [
-        # "-s",
-        # "ASSERTIONS=2",
-        # "-s",
-        # "SAFE_HEAP=1",
-        # "-s",
-        # "STACK_OVERFLOW_CHECK=1",
         "-O0",
         "-gsource-map",
-        "--source-map-base",
-        "http://jianjunz-nuc-ubuntu.sh.intel.com:9900/out/wasm/",
       ]
     } else {
       _target_ldflags += [
@@ -127,7 +104,6 @@ template("wasm_lib") {
     }
 
     _vars_to_forward = [
-      #"cflags",
       "defines",
       "deps",
       "includes",

talk/owt/sdk/wasm/media_session.cc

@@ -3,16 +3,19 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #include "talk/owt/sdk/wasm/media_session.h"
+#include <emscripten/threading.h>
 #include "third_party/webrtc/api/rtc_event_log/rtc_event_log_factory.h"
 #include "third_party/webrtc/api/transport/field_trial_based_config.h"
+#include "third_party/webrtc/logging/rtc_event_log/fake_rtc_event_log_factory.h"
+#include "third_party/webrtc/rtc_base/event.h"
 #include "third_party/webrtc/rtc_base/task_queue_stdlib.h"
 
 namespace owt {
 namespace wasm {
 MediaSession::MediaSession()
-    : task_queue_factory_(webrtc::CreateTaskQueueStdlibFactory()),
-      event_log_factory_(std::make_unique<webrtc::RtcEventLogFactory>(
-          task_queue_factory_.get())),
+    : worker_thread_(rtc::Thread::Create()),
+      task_queue_factory_(webrtc::CreateTaskQueueStdlibFactory()),
+      event_log_factory_(std::make_unique<webrtc::FakeRtcEventLogFactory>()),
       event_log_(event_log_factory_->CreateRtcEventLog(
           webrtc::RtcEventLog::EncodingType::NewFormat)),
       web_transport_session_(std::make_unique<WebTransportSession>()),
@@ -22,32 +25,34 @@ MediaSession::MediaSession()
       receiver_process_thread_(nullptr),
       video_receiver_(nullptr),
       rtcp_callback_(emscripten::val::null()) {
-  rtc::ThreadManager::Instance()->WrapCurrentThread();
-  receiver_process_thread_ =
-      webrtc::ProcessThread::Create("ReceiverProcessThread");
-  webrtc::Call::Config config(event_log_.get());
-  config.task_queue_factory = task_queue_factory_.get();
-  webrtc::FieldTrialBasedConfig trials;
-  config.trials = &trials;
-  call_ = std::unique_ptr<webrtc::Call>(webrtc::Call::Create(config));
+  RTC_CHECK(worker_thread_);
+  worker_thread_->Start();
+  worker_thread_->Invoke<void>(RTC_FROM_HERE, [&] {
+    receiver_process_thread_ =
+        webrtc::ProcessThread::Create("ReceiverProcessThread");
+    webrtc::Call::Config config(event_log_.get());
+    config.task_queue_factory = task_queue_factory_.get();
+    webrtc::FieldTrialBasedConfig trials;
+    config.trials = &trials;
+    call_ = std::unique_ptr<webrtc::Call>(webrtc::Call::Create(config));
+  });
 }
 
 std::shared_ptr<RtpVideoReceiver> MediaSession::CreateRtpVideoReceiver(
     uint32_t remote_ssrc) {
-  webrtc::VideoReceiveStream::Config config(this);
-  config.rtp.remote_ssrc = remote_ssrc;
-  config.rtp.local_ssrc = 1;
-  // Same as `kNackRtpHistoryMs` in third_party/webrtc/media/engine/webrtc_voice_engine.cc.
-  config.rtp.nack.rtp_history_ms = 5000;
-  // TODO: Use call_->worker_thread() when libwebrtc is rolled to a newer version.
-  webrtc::TaskQueueBase* current = webrtc::TaskQueueBase::Current();
-  if (!current)
-    current = rtc::ThreadManager::Instance()->CurrentThread();
-  RTC_DCHECK(current);
-  video_receiver_ = std::make_shared<RtpVideoReceiver>(
-      current, this, &config, receive_statistics_.get(),
-      receiver_process_thread_.get());
-  return video_receiver_;
+  return worker_thread_->Invoke<std::shared_ptr<RtpVideoReceiver>>(
+      RTC_FROM_HERE, [&] {
+        webrtc::VideoReceiveStream::Config config(this);
+        config.rtp.remote_ssrc = remote_ssrc;
+        config.rtp.local_ssrc = 1;
+        // Same as `kNackRtpHistoryMs` in
+        // third_party/webrtc/media/engine/webrtc_voice_engine.cc.
+        config.rtp.nack.rtp_history_ms = 5000;
+        video_receiver_ = std::make_shared<RtpVideoReceiver>(
+            worker_thread_.get(), this, &config, receive_statistics_.get(),
+            receiver_process_thread_.get());
+        return video_receiver_;
+      });
 }
 
 void MediaSession::SetRtcpCallback(emscripten::val callback) {
@@ -64,10 +69,20 @@ bool MediaSession::SendRtcp(const uint8_t* packet, size_t length) {
   if (!rtcp_callback_) {
     return false;
   }
-  emscripten::val buffer(emscripten::typed_memory_view(length, packet));
-  rtcp_callback_(buffer);
+  emscripten_sync_run_in_main_runtime_thread(
+      EM_FUNC_SIG_VIII, MediaSession::RunRtcpCallback, this, packet, length);
   return true;
 }
 
+void MediaSession::RunRtcpCallback(MediaSession* session,
+                                   const uint8_t* packet,
+                                   size_t length) {
+  if (!session->rtcp_callback_) {
+    return;
+  }
+  emscripten::val buffer(emscripten::typed_memory_view(length, packet));
+  session->rtcp_callback_(buffer);
+}
+
 }  // namespace wasm
 }  // namespace owt

talk/owt/sdk/wasm/media_session.h

@@ -10,6 +10,7 @@
 #include "talk/owt/sdk/wasm/rtp_video_receiver.h"
 #include "talk/owt/sdk/wasm/web_transport_session.h"
 #include "third_party/webrtc/api/rtc_event_log/rtc_event_log_factory_interface.h"
+#include "third_party/webrtc/rtc_base/thread.h"
 
 namespace owt {
 namespace wasm {
@@ -29,6 +30,12 @@ class MediaSession : public webrtc::Transport {
   bool SendRtcp(const uint8_t* packet, size_t length) override;
 
  private:
+  // Expected to run on main WebAssembly thread.
+  static void RunRtcpCallback(MediaSession* session,
+                              const uint8_t* packet,
+                              size_t length);
+
+  std::unique_ptr<rtc::Thread> worker_thread_;
   std::unique_ptr<webrtc::TaskQueueFactory> task_queue_factory_;
   std::unique_ptr<webrtc::RtcEventLogFactoryInterface> event_log_factory_;
   std::unique_ptr<webrtc::RtcEventLog> event_log_;

talk/owt/sdk/wasm/rtp_video_receiver.cc

@@ -3,21 +3,25 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #include "talk/owt/sdk/wasm/rtp_video_receiver.h"
+#include <emscripten/threading.h>
 #include <cstdint>
+#include "third_party/webrtc/rtc_base/location.h"
 #include "third_party/webrtc/rtc_base/logging.h"
 
 namespace owt {
 namespace wasm {
 
 RtpVideoReceiver::RtpVideoReceiver(
-    webrtc::TaskQueueBase* current_queue,
+    rtc::Thread* worker_thread,
     webrtc::Transport* transport,
     const webrtc::VideoReceiveStream::Config* config,
     webrtc::ReceiveStatistics* rtp_receive_statistics,
     webrtc::ProcessThread* process_thread)
-    : receiver_(nullptr), complete_frame_callback_(emscripten::val::null()) {
+    : worker_thread_(worker_thread),
+      receiver_(nullptr),
+      complete_frame_callback_(emscripten::val::null()) {
   receiver_ = std::make_unique<webrtc::RtpVideoStreamReceiver2>(
-      current_queue, webrtc::Clock::GetRealTimeClock(), transport, nullptr,
+      worker_thread, webrtc::Clock::GetRealTimeClock(), transport, nullptr,
       nullptr, config, rtp_receive_statistics, nullptr, nullptr, process_thread,
       this, nullptr, this, nullptr, nullptr);
   webrtc::VideoCodec codec;
@@ -29,24 +33,33 @@ RtpVideoReceiver::RtpVideoReceiver(
   receiver_->StartReceive();
 }
 
-bool RtpVideoReceiver::OnRtpPacket(uintptr_t packet_ptr, size_t packet_size) {
-  const uint8_t* packet = reinterpret_cast<uint8_t*>(packet_ptr);
-  // TODO: Create `rtp_packet_received` from `packet`.
-  webrtc::RtpPacketReceived rtp_packet_received;
-  rtp_packet_received.Parse(packet, packet_size);
-  receiver_->OnRtpPacket(rtp_packet_received);
-  return false;
+void RtpVideoReceiver::OnRtpPacket(uintptr_t packet_ptr, size_t packet_size) {
+  worker_thread_->Invoke<void>(RTC_FROM_HERE, [&] {
+    const uint8_t* packet = reinterpret_cast<uint8_t*>(packet_ptr);
+    // TODO: Create `rtp_packet_received` from `packet`.
+    webrtc::RtpPacketReceived rtp_packet_received;
+    rtp_packet_received.Parse(packet, packet_size);
+    receiver_->OnRtpPacket(rtp_packet_received);
+  });
+}
+
+void RtpVideoReceiver::RunCompleteFrameCallback(
+    RtpVideoReceiver* receiver,
+    uint8_t* buffer,
+    size_t size) {
+  emscripten::val value(emscripten::typed_memory_view(size, buffer));
+  receiver->complete_frame_callback_(value);
 }
 
 void RtpVideoReceiver::OnCompleteFrame(
     std::unique_ptr<webrtc::video_coding::EncodedFrame> frame) {
-  if(complete_frame_callback_.isNull()){
+  if (complete_frame_callback_.isNull()) {
     RTC_LOG(LS_WARNING) << "No callback registered for complete frames.";
     return;
   }
-  emscripten::val buffer(emscripten::typed_memory_view(
-      frame->EncodedImage().size(), frame->EncodedImage().data()));
-  complete_frame_callback_(buffer);
+  emscripten_sync_run_in_main_runtime_thread(
+      EM_FUNC_SIG_VIII, RtpVideoReceiver::RunCompleteFrameCallback, this,
+      frame->EncodedImage().data(), frame->EncodedImage().size());
 }
 
 void RtpVideoReceiver::SetCompleteFrameCallback(emscripten::val callback) {
@@ -58,7 +71,6 @@ void RtpVideoReceiver::SendNack(const std::vector<uint16_t>& sequence_numbers,
   // Only buffering_allowed == true is supported. Same as
   // https://source.chromium.org/chromium/chromium/src/+/main:third_party/webrtc/video/video_receive_stream2.h;l=157;bpv=1;bpt=1
   RTC_DCHECK(buffering_allowed);
-  RTC_LOG(LS_INFO)<<"Request packet rtx";
   receiver_->RequestPacketRetransmit(sequence_numbers);
 }
 

talk/owt/sdk/wasm/rtp_video_receiver.h

@@ -14,13 +14,13 @@ class RtpVideoReceiver
     : public webrtc::RtpVideoStreamReceiver2::OnCompleteFrameCallback,
       public webrtc::NackSender {
  public:
-  explicit RtpVideoReceiver(webrtc::TaskQueueBase* current_queue,
+  explicit RtpVideoReceiver(rtc::Thread* worker_thread,
                             webrtc::Transport* transport,
                             const webrtc::VideoReceiveStream::Config* config,
                             webrtc::ReceiveStatistics* rtp_receive_statistics,
                             webrtc::ProcessThread* process_thread);
   virtual ~RtpVideoReceiver() {}
-  virtual bool OnRtpPacket(uintptr_t packet_ptr, size_t packet_size);
+  virtual void OnRtpPacket(uintptr_t packet_ptr, size_t packet_size);
 
   // Overrides webrtc::RtpVideoStreamReceiver2::OnCompleteFrameCallback.
   void OnCompleteFrame(
@@ -33,6 +33,11 @@ class RtpVideoReceiver
                 bool buffering_allowed) override;
 
  private:
+  // Run on main WebAssembly thread.
+  static void RunCompleteFrameCallback(RtpVideoReceiver* receiver,
+                                  uint8_t* buffer,
+                                  size_t size);
+  rtc::Thread* worker_thread_;
   std::unique_ptr<webrtc::RtpVideoStreamReceiver2> receiver_;
   emscripten::val complete_frame_callback_;
 };