From 29de82d2c51a8b0df9960e8700dc15b4722c3525 Mon Sep 17 00:00:00 2001 From: "Ilya (Marshal)" Date: Fri, 16 Oct 2026 19:06:04 +0200 Subject: [PATCH] Destroy encoded frames where they're dropped, not on the Dispatcher --- python-webrtc/cpp/NATIVE.md | 8 +- python-webrtc/cpp/src/media/encoded_frame.h | 1 + python-webrtc/cpp/src/testing.cpp | 63 ++++++++++ python-webrtc/cpp/src/utils/dispatcher.cpp | 8 ++ .../cpp/src/utils/libwebrtc_thread.h | 19 +++ python-webrtc/cpp/src/utils/native_object.h | 27 ++++- stubs/wrtc/_testing.pyi | 9 ++ tests/test_encoded_transform.py | 40 ++++++- tests/test_mailbox.py | 35 +++++- tests/test_native_rules.py | 113 ++++++++++++++++++ tests/test_sframe.py | 49 ++++++++ 11 files changed, 362 insertions(+), 10 deletions(-) diff --git a/python-webrtc/cpp/NATIVE.md b/python-webrtc/cpp/NATIVE.md index 50768c0..047646a 100644 --- a/python-webrtc/cpp/NATIVE.md +++ b/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::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 diff --git a/python-webrtc/cpp/src/media/encoded_frame.h b/python-webrtc/cpp/src/media/encoded_frame.h index f058c38..88ad32e 100644 --- a/python-webrtc/cpp/src/media/encoded_frame.h +++ b/python-webrtc/cpp/src/media/encoded_frame.h @@ -23,6 +23,7 @@ namespace python_webrtc { class EncodedFrame : public NativeObject { public: static constexpr const char *kName = "RTCEncodedFrame"; + static constexpr bool kLeaf = true; EncodedFrame(std::unique_ptr frame, uint64_t source); diff --git a/python-webrtc/cpp/src/testing.cpp b/python-webrtc/cpp/src/testing.cpp index 04a4c78..d151074 100644 --- a/python-webrtc/cpp/src/testing.cpp +++ b/python-webrtc/cpp/src/testing.cpp @@ -35,6 +35,7 @@ #include #include #include +#include #include @@ -59,6 +60,52 @@ namespace python_webrtc { } }; + class Leaf : public NativeObject { + 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 { + public: + static constexpr const char *kName = "LeafOwningCounted"; + static constexpr bool kLeaf = true; + + LeafOwningCounted() = default; + + private: + const std::shared_ptr _counted = Counted::Create(); + }; + + // waits for its own libwebrtc thread when destroyed + struct BlockingOnDestruction { + const std::unique_ptr 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 { + public: + static constexpr const char *kName = "LeafBlocking"; + static constexpr bool kLeaf = true; + + LeafBlocking() = default; + + private: + const BlockingOnDestruction _blocking; + }; + class Throwing : public NativeObject { 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_>(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_>(testing, "Throwing", "A NativeObject counted under 'Throwing', made by Create.") .def(pybind11::init(nogil_factory(+[](bool fail) { return Throwing::Create(fail); })), pybind11::arg("fail"), diff --git a/python-webrtc/cpp/src/utils/dispatcher.cpp b/python-webrtc/cpp/src/utils/dispatcher.cpp index fe2f9dc..506ebdc 100644 --- a/python-webrtc/cpp/src/utils/dispatcher.cpp +++ b/python-webrtc/cpp/src/utils/dispatcher.cpp @@ -12,6 +12,8 @@ #include "parking.h" #include +#include +#include #include #include @@ -68,6 +70,12 @@ namespace python_webrtc { } void Dispatcher::Destroy(std::function 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(); diff --git a/python-webrtc/cpp/src/utils/libwebrtc_thread.h b/python-webrtc/cpp/src/utils/libwebrtc_thread.h index d2b51d5..58c2ec3 100644 --- a/python-webrtc/cpp/src/utils/libwebrtc_thread.h +++ b/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(); diff --git a/python-webrtc/cpp/src/utils/native_object.h b/python-webrtc/cpp/src/utils/native_object.h index 5029440..b845197 100644 --- a/python-webrtc/cpp/src/utils/native_object.h +++ b/python-webrtc/cpp/src/utils/native_object.h @@ -17,6 +17,7 @@ #include #include #include +#include #include #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 + constexpr bool kIsLeaf = requires { requires T::kLeaf; }; + inline std::atomic &NextNativeId() { static std::atomic next{1}; return next; @@ -91,9 +98,21 @@ namespace python_webrtc { template static std::shared_ptr Create(Args &&...args) { - return std::shared_ptr(new T(std::forward(args)...), [generation = Forks().load()](T *dying) { - Dispatcher::Destroy([dying]() { delete dying; }, generation); - }); + if constexpr (kIsLeaf) { + static_assert(!std::is_base_of_v, "a leaf doesn't emit"); + return std::shared_ptr(new T(std::forward(args)...), [generation = Forks().load()](T *dying) { + if (generation == Forks().load()) { +#ifdef WRTC_SANITIZED + const LeafDestruction destruction; +#endif + delete dying; + } + }); + } else { + return std::shared_ptr(new T(std::forward(args)...), [generation = Forks().load()](T *dying) { + Dispatcher::Destroy([dying]() { delete dying; }, generation); + }); + } } protected: @@ -101,7 +120,7 @@ namespace python_webrtc { ~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 && 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(); diff --git a/stubs/wrtc/_testing.pyi b/stubs/wrtc/_testing.pyi index ac64497..636449e 100644 --- a/stubs/wrtc/_testing.pyi +++ b/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: diff --git a/tests/test_encoded_transform.py b/tests/test_encoded_transform.py index da1bf9b..2199bd7 100644 --- a/tests/test_encoded_transform.py +++ b/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.""" diff --git a/tests/test_mailbox.py b/tests/test_mailbox.py index 4dc4f96..739d6bf 100644 --- a/tests/test_mailbox.py +++ b/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.""" diff --git a/tests/test_native_rules.py b/tests/test_native_rules.py index 778fb85..1eb8e00 100644 --- a/tests/test_native_rules.py +++ b/tests/test_native_rules.py @@ -10,6 +10,7 @@ from __future__ import annotations import re +from typing import NamedTuple from tests.helpers import ROOT @@ -89,6 +90,118 @@ def test_wrappers_are_created_by_create_only() -> None: assert offending == [], 'a NativeObject is created outside NativeObject::Create:\n' + '\n'.join(offending) +LEAVES = frozenset({'EncodedFrame'}) + +# member types a leaf may have, each freeing only memory or an atomic count +LEAF_MEMBERS = { + 'const bool': 'a value', + 'const uint64_t': 'a value', + 'std::mutex': 'unlocked once nothing references the leaf', + 'std::unique_ptr': "libwebrtc's frames have implicit destructors over buffers", +} + +LEAF_FORBIDDEN = r'BlockingCall|PostTask|Guard\(|Disarm\(|Dispatcher::|Emit\(|Registry<' + +FIXTURES = frozenset({'testing.cpp'}) + +LEAF_MARKER = 'static constexpr bool kLeaf = true;' + + +class Leaf(NamedTuple): + path: str + bases: str + body: str + + +def _class_body(text: str, start: int) -> str: + """The text between a class's braces, from just past its opening brace.""" + depth, end = 1, start + while depth > 0: + depth += {'{': 1, '}': -1}.get(text[end], 0) + end += 1 + + return text[start : end - 1] + + +def _leaves() -> dict[str, Leaf]: + """Leaf classes by name.""" + leaves: dict[str, Leaf] = {} + for relative, text in _sources(): + if relative in FIXTURES: + continue + + for marker in re.finditer(re.escape(LEAF_MARKER), text): + head = list(re.finditer(r'\bclass (\w+)\b([^;{]*)\{', text[: marker.start()]))[-1] + leaves[head.group(1)] = Leaf(relative, ' '.join(head.group(2).split()), _class_body(text, head.end())) + + return leaves + + +def _declarations(body: str) -> list[str]: + """A class body's declarations, without comments, inline bodies, initializers or access specifiers.""" + code = re.sub(r'//[^\n]*|/\*.*?\*/', '', body, flags=re.DOTALL) + while (inner := re.sub(r'\{[^{}]*\}', ';', code)) != code: + code = inner + + statements = (re.sub(r'\b(public|protected|private)\s*:', '', statement) for statement in code.split(';')) + # up to an initializer: = default, = delete or a member's value + declarations = (re.split(r'(? list[str]: + """The types of a class body's instance data members, whatever their names.""" + members: list[str] = [] + for declaration in _declarations(body): + if re.match(r'(static|using|friend|typedef|template)\b', declaration) is not None: + continue + + # a function: its parameters close the declarator, then qualifiers + if re.search(r'\)( ?(const|noexcept|override|final|&&?))*$', declaration) is not None: + continue + + member = re.fullmatch(r'(.+?[\s*&])(\w+)(\[[^\]]*\])?', declaration) + assert member is not None, f'not parsed as a member: {declaration}' + members.append(' '.join(member.group(1).split())) + + return members + + +def test_leaves_are_known() -> None: + """Every kLeaf class is listed, marked the one way the tests find.""" + markers = _lines_matching(r'\bkLeaf\b', allowed=FIXTURES | {'utils/native_object.h'}) + offending = [line for line in markers if not line.endswith(f': {LEAF_MARKER}')] + assert offending == [], f'kLeaf set otherwise than `{LEAF_MARKER}`:\n' + '\n'.join(offending) + assert set(_leaves()) == LEAVES + + +def test_leaves_hold_only_values() -> None: + """A leaf's implicit destructor frees memory only.""" + used: set[str] = set() + for name, leaf in _leaves().items(): + assert leaf.bases == f': public NativeObject<{name}>', f'{leaf.path}: the leaf {name} derives from {leaf.bases}' + assert re.search(rf'~{name}\s*\(', leaf.body) is None, f'{leaf.path}: the leaf {name} declares a destructor' + members = _member_types(leaf.body) + unknown = [member for member in members if member not in LEAF_MEMBERS] + assert unknown == [], f'{leaf.path}: the leaf {name} holds members not known to only free memory: {unknown}' + used.update(members) + + # a parse that finds nothing can't pass + assert used == set(LEAF_MEMBERS) + + +def test_leaves_dont_wait_or_post() -> None: + """A leaf's sources don't wait, post, emit or register.""" + sources = _sources() + for name, leaf in _leaves().items(): + stem = leaf.path.rsplit('.', 1)[0] + for path, text in sources: + if path.rsplit('.', 1)[0] == stem: + matching = re.findall(LEAF_FORBIDDEN, text) + assert matching == [], f'{path}: the leaf {name} waits, posts or registers: {matching}' + + def test_waits_on_libwebrtc_threads_go_through_blocking_call_on() -> None: """Waits go through BlockingCallOn.""" offending = _lines_matching(r'\bBlockingCall\(', allowed=frozenset({'utils/libwebrtc_thread.h'})) diff --git a/tests/test_sframe.py b/tests/test_sframe.py index 50edbac..9e35fc1 100644 --- a/tests/test_sframe.py +++ b/tests/test_sframe.py @@ -13,6 +13,7 @@ import contextlib import gc import re +import time from typing import TYPE_CHECKING, NamedTuple, Union import pytest @@ -26,10 +27,12 @@ exchange_offer_answer, mistyped, settled_alive, + undelivered, wait_progressing, wait_until, ) from tests.timing import seconds +from webrtc.utils.loops import LoopState if TYPE_CHECKING: from collections.abc import Iterator @@ -40,6 +43,7 @@ SUITES = list(webrtc.SFrameCipherSuite) KEY = bytes(range(16)) OTHER_KEY = bytes(range(1, 17)) +QUEUED_ERRORS = 3 DEPENDENCY_DESCRIPTOR = 'https://aomediacodec.github.io/av1-rtp-spec/#dependency-descriptor-rtp-header-extension' Frame = Union[webrtc.RTCEncodedVideoFrame, webrtc.RTCEncodedAudioFrame] @@ -650,6 +654,51 @@ async def test_decryption_errors(pair: tuple[webrtc.RTCPeerConnection, webrtc.RT assert sender.transform == encryptor +def queued(mailbox: wrtc.Mailbox, count: int) -> bool: + """Whether a mailbox holds count records within the timeout, waiting without a loop.""" + deadline = time.monotonic() + TIMEOUT + while len(mailbox) < count and time.monotonic() < deadline: + time.sleep(0.05) + return len(mailbox) >= count + + +def test_error_frames_of_a_closed_loop_are_released_with_its_mailbox() -> None: + """Errors queued for a closed loop drop their frames on the thread releasing it, while the Dispatcher is held.""" + baseline = settled_alive()[0]['RTCEncodedFrame'] + caller, callee = webrtc.RTCPeerConnection(), webrtc.RTCPeerConnection() + loop = asyncio.new_event_loop() + + async def failing_call() -> wrtc.Mailbox: + encryptor, decryptor = sframe_pair() + await encryptor.set_encryption_key(KEY, 1) + await decryptor.add_decryption_key(OTHER_KEY, 1) + failed = asyncio.get_running_loop().create_future() + # keeps no event: only the queued records hold frames + decryptor.on('error', lambda _event: failed.done() or failed.set_result(None)) + await encrypted_call(caller, callee, 'video', encryptor=encryptor, receiver_transform=decryptor) + await asyncio.wait_for(failed, TIMEOUT) + return LoopState.of(asyncio.get_running_loop()).mailbox + + try: + try: + mailbox = loop.run_until_complete(failing_call()) + # the loop doesn't run any more: errors queue in its mailbox + assert queued(mailbox, QUEUED_ERRORS) + finally: + caller.close() + callee.close() + assert settled_alive()[0]['RTCEncodedFrame'] > baseline + + with undelivered(): + loop.close() + # a collection releases the closed loop's state, which closes its mailbox + gc.collect() + assert mailbox.closed + assert wrtc._testing.alive_counts()['RTCEncodedFrame'] == baseline + finally: + loop.close() + + def counter_of(frame: Frame) -> int: header = wrtc._sframeParseHeader(frame.data) assert header is not None