From d5bc912aa1e9970012e7b5def8f59fcd658372f0 Mon Sep 17 00:00:00 2001 From: "Ilya (Marshal)" Date: Fri, 16 Oct 2026 18:31:22 +0200 Subject: [PATCH] Copy encoded frames less, cache RTP sources, shorten the transform queue --- python-webrtc/cpp/src/media/encoded_frame.cpp | 14 +++----- python-webrtc/cpp/src/media/encoded_frame.h | 2 +- .../src/media/rtc_rtp_script_transform.cpp | 17 ++++----- .../cpp/src/media/rtc_rtp_script_transform.h | 2 +- python-webrtc/cpp/src/utils/buffer.h | 5 +++ .../webrtc/interfaces/rtc_rtp_receiver.py | 36 +++++++++++++++++-- .../interfaces/rtc_rtp_script_transform.py | 2 +- .../webrtc/interfaces/rtc_rtp_sender.py | 2 ++ .../python/webrtc/models/rtc_encoded_frame.py | 2 +- stubs/wrtc/__init__.pyi | 2 +- tests/test_encoded_transform.py | 26 ++++++++++++-- tests/test_rtp_sender_receiver.py | 24 +++++++++++++ 12 files changed, 105 insertions(+), 29 deletions(-) diff --git a/python-webrtc/cpp/src/media/encoded_frame.cpp b/python-webrtc/cpp/src/media/encoded_frame.cpp index 18e61ed..96b0d4d 100644 --- a/python-webrtc/cpp/src/media/encoded_frame.cpp +++ b/python-webrtc/cpp/src/media/encoded_frame.cpp @@ -112,16 +112,10 @@ namespace python_webrtc { .def("getMetadata", &EncodedFrame::GetMetadata); } - pybind11::bytes EncodedFrame::GetData() { - std::vector data; - { - const std::scoped_lock lock(_mutex); - if (_frame) { - auto span = _frame->GetData(); - data.assign(span.begin(), span.end()); - } - } - return Bytes(data.data(), data.size()); + pybind11::bytearray EncodedFrame::GetData() { + // copied once, under _mutex taken with the GIL held (GIL, then _mutex: its holders never wait for the GIL) + const std::scoped_lock lock(_mutex); + return _frame ? ByteArray(_frame->GetData()) : pybind11::bytearray(); } pybind11::dict EncodedFrame::GetMetadata() { diff --git a/python-webrtc/cpp/src/media/encoded_frame.h b/python-webrtc/cpp/src/media/encoded_frame.h index 32e33fe..f058c38 100644 --- a/python-webrtc/cpp/src/media/encoded_frame.h +++ b/python-webrtc/cpp/src/media/encoded_frame.h @@ -32,7 +32,7 @@ namespace python_webrtc { [[nodiscard]] uint64_t Source() const { return _source; } - pybind11::bytes GetData(); + pybind11::bytearray GetData(); pybind11::dict GetMetadata(); diff --git a/python-webrtc/cpp/src/media/rtc_rtp_script_transform.cpp b/python-webrtc/cpp/src/media/rtc_rtp_script_transform.cpp index 6eaa7dc..42e6bd1 100644 --- a/python-webrtc/cpp/src/media/rtc_rtp_script_transform.cpp +++ b/python-webrtc/cpp/src/media/rtc_rtp_script_transform.cpp @@ -7,7 +7,7 @@ #include "rtc_rtp_script_transform.h" -#include +#include #include #include @@ -121,14 +121,11 @@ namespace python_webrtc { } bool RTCRtpScriptTransform::Write(EncodedFrame &frame, std::optional data) { - std::optional> payload; + // the view keeps the octets in place without the GIL (a bytearray can't resize), released once it's back + std::optional info; + std::span payload; if (data) { - auto info = ContiguousBuffer(*data); - const auto size = static_cast(info.size * info.itemsize); - payload.emplace(size); - if (size > 0) { - std::memcpy(payload->data(), info.ptr, size); - } + payload = BufferSpan(*data, info.emplace()); } const gil_release release; webrtc::scoped_refptr bridge; @@ -147,8 +144,8 @@ namespace python_webrtc { if (!transformable) { return false; } - if (payload) { - transformable->SetData(*payload); + if (info) { + transformable->SetData(payload); } bridge->Output(std::move(transformable)); return true; diff --git a/python-webrtc/cpp/src/media/rtc_rtp_script_transform.h b/python-webrtc/cpp/src/media/rtc_rtp_script_transform.h index 9002f1e..c5c83d1 100644 --- a/python-webrtc/cpp/src/media/rtc_rtp_script_transform.h +++ b/python-webrtc/cpp/src/media/rtc_rtp_script_transform.h @@ -32,7 +32,7 @@ namespace python_webrtc { public Emitter { public: static constexpr const char *kName = "RTCRtpScriptTransform"; - static constexpr size_t kMaxQueuedFrames = 120; + static constexpr size_t kMaxQueuedFrames = 60; enum class State : uint8_t { kNew, kAssociated, kDisassociated }; diff --git a/python-webrtc/cpp/src/utils/buffer.h b/python-webrtc/cpp/src/utils/buffer.h index 7de6482..b3c7814 100644 --- a/python-webrtc/cpp/src/utils/buffer.h +++ b/python-webrtc/cpp/src/utils/buffer.h @@ -37,6 +37,11 @@ namespace python_webrtc { return {reinterpret_cast(data), size}; } + inline pybind11::bytearray ByteArray(std::span data) { + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast): Python bytes are chars, octets alias them + return {reinterpret_cast(data.data()), data.size()}; + } + // Whether rows of rowBytes, stride apart and starting at offset, fit in size bytes, without overflowing inline bool RowsFit(size_t offset, size_t stride, size_t rows, size_t rowBytes, size_t size) { if (rows == 0 || rowBytes == 0) { diff --git a/python-webrtc/python/webrtc/interfaces/rtc_rtp_receiver.py b/python-webrtc/python/webrtc/interfaces/rtc_rtp_receiver.py index f73dfaf..a1b5395 100644 --- a/python-webrtc/python/webrtc/interfaces/rtc_rtp_receiver.py +++ b/python-webrtc/python/webrtc/interfaces/rtc_rtp_receiver.py @@ -9,9 +9,12 @@ from __future__ import annotations +import asyncio import math from typing import TypeVar, cast +from typing_extensions import override + import webrtc import wrtc from webrtc.base import WebRTCObject @@ -36,14 +39,40 @@ class RTCRtpReceiver(WebRTCObject[wrtc.RTCRtpReceiver]): See :mdn:`RTCRtpReceiver`. """ - __slots__ = () + __slots__ = ('_cached_sources',) _class = wrtc.RTCRtpReceiver + _cached_sources: tuple[asyncio.AbstractEventLoop, list[tuple[bool, int, float, int, int | None]]] | None + + @override + def _init_native(self, native_obj: wrtc.RTCRtpReceiver | None) -> None: + super()._init_native(native_obj) + self._cached_sources = None + + def _native_sources(self) -> list[tuple[bool, int, float, int, int | None]]: + # kept for the loop iteration (the spec's task): both getters in it cost one hop to the worker thread + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return self._native_obj._getSources() + + cached = self._cached_sources + if cached is not None and cached[0] is loop: + return cached[1] + + sources = self._native_obj._getSources() + self._cached_sources = (loop, sources) + _ = loop.call_soon(self._clear_sources) + + return sources + + def _clear_sources(self) -> None: + self._cached_sources = None def _sources(self, cls: type[_SourceT]) -> list[_SourceT]: synchronization = cls is RTCRtpSynchronizationSource # each native source starts with whether it's an SSRC - return [cls._from_native(source) for source in self._native_obj._getSources() if source[0] == synchronization] + return [cls._from_native(source) for source in self._native_sources() if source[0] == synchronization] @property def track(self) -> webrtc.MediaStreamTrack: @@ -155,6 +184,8 @@ async def get_stats(self) -> webrtc.RTCStatsReport: def get_synchronization_sources(self) -> list[webrtc.RTCRtpSynchronizationSource]: """Returns the synchronization sources (SSRCs) of the media received in the last 10 seconds. + Calls in the same loop iteration, of this method and :meth:`get_contributing_sources`, see the same sources. + See :mdn:`RTCRtpReceiver/getSynchronizationSources`. Returns: @@ -166,6 +197,7 @@ def get_contributing_sources(self) -> list[webrtc.RTCRtpContributingSource]: """Returns the contributing sources (CSRCs) of the media received in the last 10 seconds. These are the original streams a mixer, like a conference server, combined into the received one. + Calls in the same loop iteration, of this method and :meth:`get_synchronization_sources`, see the same sources. See :mdn:`RTCRtpReceiver/getContributingSources`. diff --git a/python-webrtc/python/webrtc/interfaces/rtc_rtp_script_transform.py b/python-webrtc/python/webrtc/interfaces/rtc_rtp_script_transform.py index 82879f1..eca88da 100644 --- a/python-webrtc/python/webrtc/interfaces/rtc_rtp_script_transform.py +++ b/python-webrtc/python/webrtc/interfaces/rtc_rtp_script_transform.py @@ -109,7 +109,7 @@ class RTCRtpScriptTransformer(UniformEventTarget[Literal['keyframerequest'], Key The worker gets it as the ``transformer`` of the ``rtctransform`` event. Frames read from :attr:`readable` and written to :attr:`writable` go on to the packetizer or the decoder, whether they were changed or not. Frames - that aren't written are dropped. Up to 120 unread frames are queued, and after that the oldest is dropped. Both + that aren't written are dropped. Up to 60 unread frames are queued, and after that the oldest is dropped. Both streams end once the transform is removed from its sender or receiver. See :mdn:`RTCRtpScriptTransformer`. diff --git a/python-webrtc/python/webrtc/interfaces/rtc_rtp_sender.py b/python-webrtc/python/webrtc/interfaces/rtc_rtp_sender.py index 85ca9ab..e202322 100644 --- a/python-webrtc/python/webrtc/interfaces/rtc_rtp_sender.py +++ b/python-webrtc/python/webrtc/interfaces/rtc_rtp_sender.py @@ -81,6 +81,8 @@ def transform(self) -> webrtc.RTCRtpScriptTransform | webrtc.RTCRtpSFrameEncrypt It changes the encoded frames before they're sent. With :obj:`None`, frames are sent as encoded. A transform serves one sender or receiver for its lifetime, so once attached it can't be set elsewhere. + Setting the first transform on a video sender that is already sending restarts its encoder (a short glitch and + a key frame); set it before negotiation to avoid that. See :mdn:`RTCRtpSender/transform`. diff --git a/python-webrtc/python/webrtc/models/rtc_encoded_frame.py b/python-webrtc/python/webrtc/models/rtc_encoded_frame.py index c25144e..14f335c 100644 --- a/python-webrtc/python/webrtc/models/rtc_encoded_frame.py +++ b/python-webrtc/python/webrtc/models/rtc_encoded_frame.py @@ -229,7 +229,7 @@ def data(self) -> bytearray: TypeError: If the value set isn't a contiguous buffer. """ if self._payload is None: - self._payload = bytearray(self._native.getData()) if self._native is not None else bytearray() + self._payload = self._native.getData() if self._native is not None else bytearray() return self._payload @data.setter diff --git a/stubs/wrtc/__init__.pyi b/stubs/wrtc/__init__.pyi index 1b0c1ea..17bbb7f 100644 --- a/stubs/wrtc/__init__.pyi +++ b/stubs/wrtc/__init__.pyi @@ -832,7 +832,7 @@ class RTCEncodedFrame: @property def _id(self) -> int: ... - def getData(self) -> bytes: + def getData(self) -> bytearray: ... def getMetadata(self) -> dict[str, object]: ... diff --git a/tests/test_encoded_transform.py b/tests/test_encoded_transform.py index 962dd0f..da1bf9b 100644 --- a/tests/test_encoded_transform.py +++ b/tests/test_encoded_transform.py @@ -307,6 +307,28 @@ async def test_frame_data(pair: tuple[webrtc.RTCPeerConnection, webrtc.RTCPeerCo frame.data = memoryview(b'abcd')[::2] +@pytest.mark.asyncio +async def test_changed_data_is_received(pair: tuple[webrtc.RTCPeerConnection, webrtc.RTCPeerConnection]) -> None: + """Data changed in place, or replaced by bytes or a memoryview, is what the receiver's transform reads.""" + caller, callee = pair + written = 0 + + def change(frame: Frame) -> None: + nonlocal written + written += 1 + if written % 3 == 0: + frame.data[:] = b'in place' + elif written % 3 == 1: + frame.data = b'bytes' + else: + frame.data = memoryview(b'a memoryview') + + payloads: set[bytes] = set() + receiving = Recorder(lambda frame: payloads.add(bytes(frame.data))) + await transformed_call(caller, callee, 'audio', sender_worker=Recorder(change), receiver_worker=receiving) + await wait_until(lambda: {b'in place', b'bytes', b'a memoryview'} <= payloads, 'each changed payload', TIMEOUT) + + @pytest.mark.asyncio async def test_written_frame_is_detached(pair: tuple[webrtc.RTCPeerConnection, webrtc.RTCPeerConnection]) -> None: caller, callee = pair @@ -601,8 +623,8 @@ async def frames_encoded() -> int: encoded = max((frames for frames in outbound if frames is not None), default=0) return encoded - # past the 120 frames a transformer queues, the oldest are dropped; slow cores encode slower, not never - await wait_progressing(frames_encoded, lambda: encoded > 130, 'more frames than the queue keeps') + # past the 60 frames a transformer queues, the oldest are dropped; slow cores encode slower, not never + await wait_progressing(frames_encoded, lambda: encoded > 70, 'more frames than the queue keeps') assert len(started) == 1 caller.close() callee.close() diff --git a/tests/test_rtp_sender_receiver.py b/tests/test_rtp_sender_receiver.py index 266d08d..c0de645 100644 --- a/tests/test_rtp_sender_receiver.py +++ b/tests/test_rtp_sender_receiver.py @@ -16,6 +16,7 @@ import pytest import webrtc +import wrtc from tests.helpers import ( connect, exchange_offer, @@ -589,6 +590,29 @@ async def test_synchronization_sources_of_audio_have_its_level( assert 0 <= source.audio_level <= 1 +@pytest.mark.asyncio +async def test_sources_are_read_once_per_task(pc: webrtc.RTCPeerConnection, monkeypatch: pytest.MonkeyPatch) -> None: + """Both getters in one task share one read of the native sources, a later task reads them again.""" + receiver = pc.add_transceiver(webrtc.MediaType.audio).receiver + reads = 0 + + def counted(_native: wrtc.RTCRtpReceiver) -> list[tuple[bool, int, float, int, int | None]]: + nonlocal reads + reads += 1 + return [(True, reads, 0.0, 0, None), (False, reads, 0.0, 0, None)] + + monkeypatch.setattr(wrtc.RTCRtpReceiver, '_getSources', counted) + + [synchronization] = receiver.get_synchronization_sources() + [contributing] = receiver.get_contributing_sources() + assert reads == 1 + assert synchronization.source == contributing.source == 1 + await next_task() + [synchronization] = receiver.get_synchronization_sources() + assert reads == 2 + assert synchronization.source == 2 + + @pytest.mark.parametrize( 'encoding', [