おためし

Author: melpon

.vscode/launch.json

@@ -10,7 +10,7 @@
             "type": "lldb-dap",
             "request": "launch",
             "program": "uv",
-            "args": ["run", "pytest", "tests/test_openh264.py"],
+            "args": ["run", "pytest", "-s", "tests/test_openh264.py::test_openh264_simulcast"],
             "cwd": "${workspaceFolder}",
             "sourceMap": [
               ["../../../../_source/ubuntu-24.04_x86_64/webrtc/src/", "/home/melpon/dev/webrtc-build/_source/ubuntu-24.04_x86_64/webrtc/src/"],

src/sora.cpp

@@ -18,6 +18,7 @@ Sora::Sora(std::optional<std::string> openh264,
 
 Sora::~Sora() {
   std::cerr << __FUNCTION__ << std::endl;
+  factory_.reset();
   if (thread_) {
     std::cerr << __FUNCTION__ << " 1" << std::endl;
     ioc_->stop();

src/sora_connection.cpp

@@ -45,9 +45,11 @@ void SoraConnection::Disposed() {
   DisposePublisher::Disposed();
   if (video_source_) {
     video_source_->RemoveSubscriber(this);
+    video_source_ = nullptr;
   }
   if (audio_source_) {
     audio_source_->RemoveSubscriber(this);
+    audio_source_ = nullptr;
   }
 }
 
@@ -72,7 +74,14 @@ void SoraConnection::Connect() {
 void SoraConnection::Disconnect() {
   std::cerr << __FUNCTION__ << std::endl;
   if (conn_) {
+    Disposed();
     conn_->Disconnect();
+    // OnDisconnect が来るまで待つ
+    {
+      GILLock lock;
+      on_disconnect_cv_.wait(lock,
+                             [this]() -> bool { return on_disconnected_; });
+    }
     // Connection から生成したものは、ここで消す
     audio_sender_ = nullptr;
     video_sender_ = nullptr;
@@ -189,6 +198,8 @@ void SoraConnection::OnDisconnect(sora::SoraSignalingErrorCode ec,
   if (on_disconnect_) {
     call_python(on_disconnect_, ec, message);
   }
+  on_disconnected_ = true;
+  on_disconnect_cv_.notify_all();
 }
 
 void SoraConnection::OnNotify(std::string text) {

src/sora_connection.h

@@ -1,6 +1,7 @@
 #ifndef SORA_CONNECTION_H_
 #define SORA_CONNECTION_H_
 
+#include <condition_variable>
 #include <memory>
 #include <thread>
 
@@ -164,6 +165,8 @@ class SoraConnection : public DisposePublisher,
       audio_sender_frame_transformer_;
   rtc::scoped_refptr<SoraFrameTransformerInterface>
       video_sender_frame_transformer_;
+  bool on_disconnected_ = false;
+  std::condition_variable_any on_disconnect_cv_;
 };
 
 class SoraSignalingObserver : public sora::SoraSignalingObserver {

src/sora_log.cpp

@@ -5,3 +5,7 @@ void EnableLibwebrtcLog(rtc::LoggingSeverity severity) {
   rtc::LogMessage::LogTimestamps();
   rtc::LogMessage::LogThreads();
 }
+
+void RtcLog(rtc::LoggingSeverity severity, const std::string& message) {
+  RTC_LOG_V(severity) << message;
+}

src/sora_log.h

@@ -4,5 +4,6 @@
 #include "sora.h"
 
 void EnableLibwebrtcLog(rtc::LoggingSeverity severity);
+void RtcLog(rtc::LoggingSeverity severity, const std::string& message);
 
 #endif

src/sora_sdk_ext.cpp

@@ -361,6 +361,7 @@ NB_MODULE(sora_sdk_ext, m) {
       .value("NONE", rtc::LoggingSeverity::LS_NONE);
 
   m.def("enable_libwebrtc_log", &EnableLibwebrtcLog);
+  m.def("rtc_log", &RtcLog);
 
   nb::class_<SoraTrackInterface>(
       m, "SoraTrackInterface",

src/sora_video_source.cpp

@@ -15,21 +15,21 @@ SoraVideoSource::SoraVideoSource(
     rtc::scoped_refptr<webrtc::MediaStreamTrackInterface> track)
     : SoraTrackInterface(publisher, track), source_(source), finished_(false) {
   publisher_->AddSubscriber(this);
-  thread_.reset();
+  thread_.reset(new std::thread([this]() {
+    gil_scoped_acquire acq;
+    while (SendFrameProcess()) {
+    }
+  }));
 }
 
 SoraVideoSource::~SoraVideoSource() {
   std::cerr << __FUNCTION__ << std::endl;
   if (!finished_) {
-    {
-      std::lock_guard<std::recursive_mutex> lock(queue_mtx_);
-      finished_ = true;
-    }
+    finished_ = true;
     queue_cond_.notify_all();
-    if (thread_ && thread_->joinable()) {
-      thread_->join();
-      thread_ = nullptr;
-    }
+    gil_scoped_release release;
+    thread_->join();
+    thread_ = nullptr;
   }
 }
 
@@ -58,18 +58,15 @@ void SoraVideoSource::OnCaptured(
   if (finished_) {
     return;
   }
-  {
-    std::lock_guard<std::recursive_mutex> lock(queue_mtx_);
-    queue_.push(
-        std::make_unique<Frame>(std::move(data), width, height, timestamp_us));
-  }
+  queue_.push(
+      std::make_unique<Frame>(std::move(data), width, height, timestamp_us));
   queue_cond_.notify_all();
 }
 
 bool SoraVideoSource::SendFrameProcess() {
   std::unique_ptr<Frame> frame;
   {
-    std::unique_lock<std::recursive_mutex> lock(queue_mtx_);
+    GILLock lock;
     queue_cond_.wait(lock, [&] { return !queue_.empty() || finished_; });
     if (finished_) {
       return false;

tests/client.py

@@ -210,6 +210,10 @@ def __enter__(self) -> "SoraClient":
     def __exit__(self, exc_type, exc_value, traceback) -> None:
         self.disconnect()
 
+    def __del__(self) -> None:
+        sora_sdk.rtc_log(sora_sdk.SoraLoggingSeverity.INFO, "SoraClient.__del__")
+        self.disconnect()
+
     def connect(self, fake_audio=False, fake_video=False) -> None:
         # スレッドは connect 前に起動する
         if fake_audio:

tests/conftest.py

@@ -4,14 +4,19 @@
 import pytest
 from dotenv import load_dotenv
 
-from sora_sdk import Sora
+from sora_sdk import Sora, SoraLoggingSeverity, rtc_log
 
 
 @pytest.fixture
 def sora():
     sora = Sora()
+    rtc_log(SoraLoggingSeverity.ERROR, "sora created")
     yield sora
+    rtc_log(SoraLoggingSeverity.ERROR, "sora destroy started")
     del sora
+    rtc_log(SoraLoggingSeverity.ERROR, "sora destroy completed")
+    gc.collect()
+    rtc_log(SoraLoggingSeverity.ERROR, "sora destroy completed2")
 
 
 class SoraDeferred: