Destroy encoded frames where they're dropped, not on the Dispatcher

python-webrtc/cpp/NATIVE.md

@@ -41,11 +41,15 @@ attached when it posts (`utils/libwebrtc_thread.h`). `tests/test_native_rules.py
 - Create wrappers with `NativeObject<T>::Create` or a `Registry`. Construct them outside locks.
 - Don't wait in a constructor: post the setup to its thread instead.
 - Destructors run on the Dispatcher, never on a libwebrtc or Python thread, so they may wait for libwebrtc threads.
+  Leaves are the exception.
+- A leaf (`kLeaf = true`, such as `RTCEncodedFrame`) is destroyed wherever its last reference drops, so its members
+  only free memory: no waits, posts, wrappers or task queues (`tests/test_native_rules.py` lists the allowed types).
 
 A `Registry` constructs its wrappers on the signaling thread, where their observers are registered, except a data
 channel's: libwebrtc calls it on the network thread, where it reads the channel's state and registers.
-Sanitizer builds abort on a destructor that runs off the Dispatcher; a constructor that throws is the exception: the
-half-made object is destroyed where it was made. The factory and the ICE, DTLS and SCTP transports still wait in their
+Sanitizer builds abort on a non-leaf destructor off the Dispatcher and on a leaf's that blocks or releases a wrapper;
+a constructor that throws is the exception: the half-made object is destroyed where it was made. The factory and the
+ICE, DTLS and SCTP transports still wait in their
 constructors.
 
 ### The Dispatcher and mailboxes

python-webrtc/cpp/src/media/encoded_frame.h

@@ -23,6 +23,7 @@ namespace python_webrtc {
   class EncodedFrame : public NativeObject<EncodedFrame> {
   public:
     static constexpr const char *kName = "RTCEncodedFrame";
+    static constexpr bool kLeaf = true;
 
     EncodedFrame(std::unique_ptr<webrtc::TransformableFrameInterface> frame, uint64_t source);
 

python-webrtc/cpp/src/testing.cpp

@@ -35,6 +35,7 @@
 #include <api/make_ref_counted.h>
 #include <api/ref_count.h>
 #include <api/scoped_refptr.h>
+#include <rtc_base/thread.h>
 
 #include <pybind11/stl.h>
 
@@ -59,6 +60,52 @@ namespace python_webrtc {
       }
     };
 
+    class Leaf : public NativeObject<Leaf> {
+    public:
+      static constexpr const char *kName = "Leaf";
+      static constexpr bool kLeaf = true;
+
+      Leaf() = default;
+    };
+
+    // breaks the leaf rules: its wrapper is destroyed on the Dispatcher
+    class LeafOwningCounted : public NativeObject<LeafOwningCounted> {
+    public:
+      static constexpr const char *kName = "LeafOwningCounted";
+      static constexpr bool kLeaf = true;
+
+      LeafOwningCounted() = default;
+
+    private:
+      const std::shared_ptr<Counted> _counted = Counted::Create();
+    };
+
+    // waits for its own libwebrtc thread when destroyed
+    struct BlockingOnDestruction {
+      const std::unique_ptr<webrtc::Thread> thread = webrtc::Thread::Create();
+
+      BlockingOnDestruction() { thread->Start(); }
+
+      BlockingOnDestruction(const BlockingOnDestruction &) = delete;
+      BlockingOnDestruction &operator=(const BlockingOnDestruction &) = delete;
+
+      ~BlockingOnDestruction() {
+        BlockingCallOn(thread, []() {});
+      }
+    };
+
+    // breaks the leaf rules: it waits for a libwebrtc thread
+    class LeafBlocking : public NativeObject<LeafBlocking> {
+    public:
+      static constexpr const char *kName = "LeafBlocking";
+      static constexpr bool kLeaf = true;
+
+      LeafBlocking() = default;
+
+    private:
+      const BlockingOnDestruction _blocking;
+    };
+
     class Throwing : public NativeObject<Throwing> {
     public:
       static constexpr const char *kName = "Throwing";
@@ -214,6 +261,22 @@ namespace python_webrtc {
           BlockingCallOn(factory->signalingThread(), []() {});
         },
         "Blocks on a libwebrtc thread with the GIL held, which sanitizer builds abort on.");
+    pybind11::class_<Leaf, std::shared_ptr<Leaf>>(testing, "Leaf",
+                                                  "A leaf NativeObject counted under 'Leaf', made by Create.")
+        .def(pybind11::init([]() { return Leaf::Create(); }));
+    testing.def(
+        "destroy_leaf_on_native_thread",
+        []() {
+          auto leaf = Leaf::Create();
+          std::thread([&leaf]() { leaf = nullptr; }).join();
+        },
+        nogil(), "Makes a Leaf and drops it on a native thread.");
+    testing.def(
+        "check_leaf_releasing_a_wrapper", []() { (void)LeafOwningCounted::Create(); },
+        "Drops a leaf that owns a wrapper, which sanitizer builds abort on.");
+    testing.def(
+        "check_leaf_blocking", []() { (void)LeafBlocking::Create(); }, nogil(),
+        "Drops a leaf that blocks on a libwebrtc thread, which sanitizer builds abort on.");
     pybind11::class_<Throwing, std::shared_ptr<Throwing>>(testing, "Throwing",
                                                           "A NativeObject counted under 'Throwing', made by Create.")
         .def(pybind11::init(nogil_factory(+[](bool fail) { return Throwing::Create(fail); })), pybind11::arg("fail"),

python-webrtc/cpp/src/utils/dispatcher.cpp

@@ -12,6 +12,8 @@
 #include "parking.h"
 
 #include <chrono>
+#include <cstdio>
+#include <cstdlib>
 #include <thread>
 #include <utility>
 
@@ -68,6 +70,12 @@ namespace python_webrtc {
   }
 
   void Dispatcher::Destroy(std::function<void()> destroy, int generation) {
+#ifdef WRTC_SANITIZED
+    if (InLeafDestruction() > 0) {
+      (void)std::fputs("python-webrtc: a leaf's destructor released a wrapper or task queue\n", stderr);
+      std::abort();
+    }
+#endif
     auto &dispatcher = Instance();
     {
       const auto lock = dispatcher.Locked();

python-webrtc/cpp/src/utils/libwebrtc_thread.h

@@ -67,6 +67,21 @@ namespace python_webrtc {
     OnNativeThread() = true;
   }
 
+  inline int &InLeafDestruction() {
+    static thread_local int destroying = 0;
+    return destroying;
+  }
+
+  class LeafDestruction {
+  public:
+    LeafDestruction() { InLeafDestruction()++; }
+
+    ~LeafDestruction() { InLeafDestruction()--; }
+
+    LeafDestruction(const LeafDestruction &) = delete;
+    LeafDestruction &operator=(const LeafDestruction &) = delete;
+  };
+
   inline void CheckNativeThreadDetached(const char *where) {
     if (OnNativeThread() && AttachedToPython()) {
       (void)std::fprintf(stderr, "python-webrtc: a native thread is attached to Python at %s\n", where);
@@ -86,6 +101,10 @@ namespace python_webrtc {
       std::abort();
     }
 #ifdef WRTC_SANITIZED
+    if (InLeafDestruction() > 0) {
+      (void)std::fputs("python-webrtc: blocking on a libwebrtc thread in a leaf's destructor\n", stderr);
+      std::abort();
+    }
     if (AttachedToPython()) {
       (void)std::fputs("python-webrtc: blocking on a libwebrtc thread while attached to Python\n", stderr);
       std::abort();

python-webrtc/cpp/src/utils/native_object.h

@@ -17,6 +17,7 @@
 #include <memory>
 #include <mutex>
 #include <string>
+#include <type_traits>
 #include <utility>
 
 #include "dispatcher.h"
@@ -65,6 +66,12 @@ namespace python_webrtc {
     return AliveCounter::Counts();
   }
 
+  class Binding;
+
+  // a leaf is destroyed where its last reference drops
+  template <typename T>
+  constexpr bool kIsLeaf = requires { requires T::kLeaf; };
+
   inline std::atomic<uint64_t> &NextNativeId() {
     static std::atomic<uint64_t> next{1};
     return next;
@@ -91,17 +98,29 @@ namespace python_webrtc {
 
     template <typename... Args>
     static std::shared_ptr<T> Create(Args &&...args) {
-      return std::shared_ptr<T>(new T(std::forward<Args>(args)...), [generation = Forks().load()](T *dying) {
-        Dispatcher::Destroy([dying]() { delete dying; }, generation);
-      });
+      if constexpr (kIsLeaf<T>) {
+        static_assert(!std::is_base_of_v<Binding, T>, "a leaf doesn't emit");
+        return std::shared_ptr<T>(new T(std::forward<Args>(args)...), [generation = Forks().load()](T *dying) {
+          if (generation == Forks().load()) {
+#ifdef WRTC_SANITIZED
+            const LeafDestruction destruction;
+#endif
+            delete dying;
+          }
+        });
+      } else {
+        return std::shared_ptr<T>(new T(std::forward<Args>(args)...), [generation = Forks().load()](T *dying) {
+          Dispatcher::Destroy([dying]() { delete dying; }, generation);
+        });
+      }
     }
 
   protected:
     // runs after T's members: the count holds while T is destroyed
     ~NativeObject() {
 #ifdef WRTC_SANITIZED
       // a throwing constructor destroys the half-made object where it was made
-      if (std::uncaught_exceptions() == 0 && !Dispatcher::IsCurrent()) {
+      if (!kIsLeaf<T> && std::uncaught_exceptions() == 0 && !Dispatcher::IsCurrent()) {
         (void)std::fprintf(stderr, "python-webrtc: %s destroyed off the Dispatcher, see NativeObject::Create\n",
                            T::kName);
         std::abort();

stubs/wrtc/_testing.pyi

@@ -35,6 +35,9 @@ class Emitting:
 class Identity:
     def __init__(self) -> None:
         ...
+class Leaf:
+    def __init__(self) -> None:
+        ...
 class Registered:
     @property
     def id(self) -> int:
@@ -57,10 +60,16 @@ def check_blocking_while_attached() -> None:
     ...
 def check_destroyed_off_dispatcher() -> None:
     ...
+def check_leaf_blocking() -> None:
+    ...
+def check_leaf_releasing_a_wrapper() -> None:
+    ...
 def complete(mailbox: Mailbox, token: typing.SupportsInt | typing.SupportsIndex, value: typing.SupportsInt | typing.SupportsIndex) -> None:
     ...
 def delete_queue_from_its_task(by_captures: bool) -> bool:
     ...
+def destroy_leaf_on_native_thread() -> None:
+    ...
 def dispatcher_idle() -> None:
     ...
 def dispatcher_threads() -> int:

tests/test_encoded_transform.py

@@ -11,6 +11,7 @@
 
 import asyncio
 import gc
+import threading
 import time
 from typing import TYPE_CHECKING, Callable, Union
 
@@ -25,6 +26,7 @@
     exchange_offer,
     mistyped,
     settled_alive,
+    undelivered,
     wait_progressing,
     wait_until,
     wait_until_unmuted,
@@ -36,16 +38,18 @@
 
 TIMEOUT = seconds(15)
 MARK = 0xAB
+KEPT = 3
 
 Frame = Union[webrtc.RTCEncodedVideoFrame, webrtc.RTCEncodedAudioFrame]
 
 
 class Recorder:
-    """A worker that records the frames it reads, writing them back after ``change``."""
+    """A worker that records the frames it reads, writing back all but the first ``keep``, after ``change``."""
 
-    def __init__(self, change: Callable[[Frame], None] | None = None, *, write: bool = True) -> None:
+    def __init__(self, change: Callable[[Frame], None] | None = None, *, write: bool = True, keep: int = 0) -> None:
         self.change = change
         self.write = write
+        self.keep = keep
         self.frames: list[Frame] = []
         self.transformer: webrtc.RTCRtpScriptTransformer | None = None
         self.started = asyncio.get_running_loop().create_future()
@@ -65,7 +69,7 @@ async def __call__(self, event: webrtc.RTCTransformEvent) -> None:
             self.frames.append(frame)
             if self.change is not None:
                 self.change(frame)
-            if writer is not None:
+            if writer is not None and len(self.frames) > self.keep:
                 _ = writer.write(frame)
         self.done.set_result(None)
 
@@ -677,6 +681,36 @@ async def session() -> None:
         raise TimeoutError(msg) from None
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize('kind', ['video', 'audio'])
+async def test_frames_are_released_where_dropped(kind: str) -> None:
+    """Frames kept past their connection are destroyed on the thread dropping them, while the Dispatcher is held."""
+    caller, callee = webrtc.RTCPeerConnection(), webrtc.RTCPeerConnection()
+    sending, receiving = Recorder(keep=KEPT), Recorder(keep=KEPT)
+    await transformed_call(caller, callee, kind, sender_worker=sending, receiver_worker=receiving)
+    await receiving.wait_frames(KEPT)
+    caller.close()
+    callee.close()
+    await asyncio.wait_for(asyncio.gather(sending.done, receiving.done), TIMEOUT)
+
+    # the frames each side kept
+    kept = [frame for frame in sending.frames + receiving.frames if frame._native is not None]
+    assert len(kept) == 2 * KEPT
+    sending.frames.clear()
+    receiving.frames.clear()
+    alive = settled_alive()[0]['RTCEncodedFrame']
+
+    def drop() -> None:
+        kept.clear()
+        gc.collect()
+
+    with undelivered():
+        dropping = threading.Thread(target=drop)
+        dropping.start()
+        dropping.join()
+        assert wrtc._testing.alive_counts()['RTCEncodedFrame'] == alive - 2 * KEPT
+
+
 @pytest.mark.asyncio
 async def test_async_worker_exception_goes_to_the_loop() -> None:
     """An async worker's exception reaches the loop's exception handler."""

tests/test_mailbox.py

@@ -21,7 +21,7 @@
 
 import webrtc
 import wrtc
-from tests.helpers import ROOT, exit_code_within
+from tests.helpers import ROOT, exit_code_within, undelivered
 from tests.isolation import isolated
 from tests.timing import seconds
 
@@ -267,6 +267,39 @@ def test_destroyed_off_the_dispatcher_aborts() -> None:
     assert 'destroyed off the Dispatcher' in result.stderr, result.stderr
 
 
+def test_a_leaf_is_destroyed_where_released() -> None:
+    """A leaf is destroyed on the thread dropping it, Python or native, while the Dispatcher is held."""
+    with undelivered():
+        leaf = wrtc._testing.Leaf()
+        assert wrtc._testing.alive_counts()['Leaf'] == 1
+        del leaf
+        assert wrtc._testing.alive_counts()['Leaf'] == 0
+        wrtc._testing.destroy_leaf_on_native_thread()
+        assert wrtc._testing.alive_counts()['Leaf'] == 0
+
+
+@pytest.mark.skipif(not wrtc._sanitized, reason='checked in sanitizer builds only')
+@pytest.mark.parametrize(
+    ('check', 'message'),
+    [
+        ('check_leaf_releasing_a_wrapper', "a leaf's destructor released a wrapper or task queue"),
+        ('check_leaf_blocking', "blocking on a libwebrtc thread in a leaf's destructor"),
+    ],
+)
+def test_a_leaf_releasing_a_wrapper_or_blocking_aborts(check: str, message: str) -> None:
+    """A leaf's destructor that would need the Dispatcher aborts."""
+    result = subprocess.run(
+        [sys.executable, '-c', f'import wrtc; wrtc._testing.{check}()'],
+        capture_output=True,
+        text=True,
+        timeout=seconds(60),
+        cwd=ROOT,
+        check=False,
+    )
+    assert result.returncode != 0
+    assert message in result.stderr, result.stderr
+
+
 @pytest.mark.skipif(not wrtc._sanitized, reason='checked in sanitizer builds only')
 def test_blocking_on_a_libwebrtc_thread_while_attached_aborts() -> None:
     """BlockingCallOn with the GIL attached aborts."""