Copy encoded frames less, cache RTP sources, shorten the transform queue

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<uint8_t> 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() {

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();
 

python-webrtc/cpp/src/media/rtc_rtp_script_transform.cpp

@@ -7,7 +7,7 @@
 
 #include "rtc_rtp_script_transform.h"
 
-#include <cstring>
+#include <span>
 #include <utility>
 #include <vector>
 
@@ -121,14 +121,11 @@ namespace python_webrtc {
   }
 
   bool RTCRtpScriptTransform::Write(EncodedFrame &frame, std::optional<pybind11::buffer> data) {
-    std::optional<std::vector<uint8_t>> payload;
+    // the view keeps the octets in place without the GIL (a bytearray can't resize), released once it's back
+    std::optional<pybind11::buffer_info> info;
+    std::span<const uint8_t> payload;
     if (data) {
-      auto info = ContiguousBuffer(*data);
-      const auto size = static_cast<size_t>(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<FrameTransformerBridge> 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;

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

@@ -32,7 +32,7 @@ namespace python_webrtc {
                                 public Emitter<RTCRtpScriptTransform> {
   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 };
 

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

@@ -37,6 +37,11 @@ namespace python_webrtc {
     return {reinterpret_cast<const char *>(data), size};
   }
 
+  inline pybind11::bytearray ByteArray(std::span<const uint8_t> data) {
+    // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast): Python bytes are chars, octets alias them
+    return {reinterpret_cast<const char *>(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) {

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`.
 

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`.

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`.
 

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

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]:
         ...

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()

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',
     [