perf: replace extract_dup+from_buffer_copy with buffer.map+memmove zero-copy

Instead of extract_dup (GLib alloc+memcpy → Python bytes) followed by
from_buffer_copy (Python bytes → ctypes array) — two 3MB copies per frame —
use Gst.Buffer.map(READ) to get a zero-allocation pointer to the decoded
frame memory, then memmove directly into a pre-allocated reusable ctypes
array (_raw_arr).

This reduces the per-frame copy path from 2 copies (6MB) to 1 memmove
(3MB), with no Python bytes object allocation at all.  The memmove happens
under _frame_lock so render() on the main thread never reads a partial frame.
_raw_arr is allocated once on the first frame (or on resolution change) and
reused for every subsequent frame.

_Frame no longer carries a pixels field.  Tests updated accordingly.
Benchmark updated to use the same buffer.map+memmove path as the app.

tests/benchmark_nv12_decode.py

@@ -153,6 +153,8 @@ class Stats:
     lock:              threading.Lock = field(default_factory=threading.Lock)
 
 stats = Stats()
+stats._raw_arr = None
+stats._raw_arr_size = 0
 
 # ── Callback ────────────────────────────────────────────────────────────────
 
@@ -183,23 +185,22 @@ def _on_sample(sink) -> Gst.FlowReturn:
         except Exception:
             pass
 
-    # Measure extract_dup (GStreamer buf → Python bytes) + from_buffer_copy
-    # (Python bytes → ctypes array for SDL upload).  del objects immediately
-    # after timing so CPython's ref-counting frees the 3+ MB allocations at
-    # once rather than letting them accumulate across frames (OOM on 1 GB device).
+    # Measure buffer.map(READ) + memmove into a pre-allocated ctypes array
+    # (same path as the app).  Reuse a single ctypes array across frames to
+    # avoid per-frame allocation.  del is not needed — ctypes array is reused.
     t0 = time.monotonic()
-    raw = buf.extract_dup(0, buf.get_size())
-    extract_us = (time.monotonic() - t0) * 1e6
-
-    if fmt_str == "NV12":
-        y_size = int(info.stride[0]) * int(info.height)
-        t1 = time.monotonic()
-        arr = (ctypes.c_ubyte * len(raw)).from_buffer_copy(raw)
-        copy_us = extract_us + (time.monotonic() - t1) * 1e6
-        del arr   # free 3 MB ctypes array immediately
-    else:
-        copy_us = extract_us
-    del raw       # free 3 MB bytes object immediately
+    ok, map_info = buf.map(Gst.MapFlags.READ)
+    if not ok:
+        return Gst.FlowReturn.OK
+    try:
+        src_size = map_info.size
+        if not hasattr(stats, '_raw_arr') or stats._raw_arr_size < src_size:
+            stats._raw_arr = (ctypes.c_ubyte * src_size)()
+            stats._raw_arr_size = src_size
+        ctypes.memmove(stats._raw_arr, map_info.data, src_size)
+        copy_us = (time.monotonic() - t0) * 1e6
+    finally:
+        buf.unmap(map_info)
 
     with stats.lock:
         stats.total_frames += 1
@@ -341,7 +342,7 @@ else:
 if copy_us:
     mean_copy = statistics.mean(copy_us)
     max_copy  = max(copy_us)
-    print(f"\n  --- CPU copy cost (from_buffer_copy) ---")
+    print(f"\n  --- CPU copy cost (buffer.map + memmove) ---")
     print(f"  Mean copy time       : {mean_copy:.0f} µs")
     print(f"  Max  copy time       : {max_copy:.0f} µs")
     budget_us = 1_000_000 / (actual_fps if len(wall_times) >= 2 and actual_fps > 0 else 30)

tests/test_player.py

@@ -13,6 +13,7 @@ class FakeMessageType:
     EOS = 2
     BUFFERING = 4
     STATE_CHANGED = 8
+    WARNING = 16
 
 
 class FakeFlowReturn:
@@ -64,6 +65,12 @@ class FakeStructure:
         return self._width if name == "width" else self._height
 
 
+class FakeMapInfo:
+    def __init__(self, data: bytes):
+        self.data = data
+        self.size = len(data)
+
+
 class FakeBuffer:
     def __init__(self, payload: bytes):
         self._payload = payload
@@ -74,6 +81,12 @@ class FakeBuffer:
     def extract_dup(self, _offset, _size):
         return self._payload
 
+    def map(self, _flags):
+        return True, FakeMapInfo(self._payload)
+
+    def unmap(self, _map_info):
+        pass
+
 
 class FakeSample:
     def __init__(self, width=1280, height=720, payload: bytes | None = None):
@@ -176,6 +189,10 @@ class FakeMessage:
         return SimpleNamespace(get_name=lambda: self._structure_name)
 
 
+class FakeMapFlags:
+    READ = 1
+
+
 class FakeGst:
     State = FakeState
     StateChangeReturn = FakeStateChangeReturn
@@ -183,6 +200,7 @@ class FakeGst:
     SeekFlags = FakeSeekFlags
     MessageType = FakeMessageType
     FlowReturn = FakeFlowReturn
+    MapFlags = FakeMapFlags
     SECOND = 1_000_000_000
     MSECOND = 1_000_000
     Caps = SimpleNamespace(from_string=lambda value: value)
@@ -352,7 +370,10 @@ class TestGStreamerBackend:
 
     def test_render_uploads_latest_frame_and_clears_dirty_flag(self, monkeypatch):
         backend, _pipeline, _sink = self._make_backend()
-        backend._latest_frame = SimpleNamespace(width=320, height=180, pitch=1280, pixels=b"\x00" * (320 * 180 * 4), pixel_format="BGRA", uv_pixels=None, uv_pitch=0)
+        raw_data = b"\x00" * (320 * 180 * 4)
+        backend._raw_arr = (ctypes.c_ubyte * len(raw_data)).from_buffer_copy(raw_data)
+        backend._raw_arr_size = len(raw_data)
+        backend._latest_frame = SimpleNamespace(width=320, height=180, pitch=1280, pixel_format="BGRA", y_size=0, uv_pitch=0, buf_size=len(raw_data))
         backend._frame_dirty = True
 
         calls = []