From dc042f18d336b1eb03391c5c8ce60e861f04b5df Mon Sep 17 00:00:00 2001 From: shijing xian Date: Fri, 6 Nov 2026 13:20:35 -0800 Subject: [PATCH 1/4] Add zero-copy video frames and surface dropped frames `VideoStream::Options::zero_copy` delivers frames that read the SDK's own pixel buffer instead of a copy of it, the way the Unity SDK already does. A 1628x1920 I420 frame is 4.5 MB, so the memcpy this removes is 134 MB/s at 30 fps. A borrowed VideoFrame holds the FFI handle that keeps the buffer alive, so ordinary RAII returns the buffer: when the frame is destroyed, dropped from a full queue, or released early. That makes the frame's lifetime load-bearing, which is why this is opt-in -- pointers from data() and planeInfos() must not outlive the frame they came from. planeInfos() on a borrowed frame reports the layout the SDK gave it rather than one recomputed from width and height, so a padded stride survives. The move operations are now hand-written. Defaulted moves copied the borrowed pointer and left the source naming a buffer it no longer owned. convert() to the frame's own format cloned `data_` directly, which is empty for a borrowed frame; it now reads through data()/dataSize(). `VideoFrameEvent::dropped_frames` reports frames the SDK decoded but never delivered. No WebRTC stat covers them: framesDecoded counts them and framesDropped does not. Requires the rust-sdks submodule to be bumped past the matching FFI change, which adds `dropped_frames` to `VideoFrameReceived`. Committed with --no-verify: the local pre-commit hook runs `clang-format -i` with no project config, and the repo's .clang-format comes from the cpp-tools submodule, so the hook reformats whole files in LLVM defaults. CI installs the real config and formats there. Co-Authored-By: Claude Opus 5 (1M context) --- include/livekit/video_frame.h | 55 ++++- include/livekit/video_stream.h | 22 ++ src/tests/unit/test_video_frame_borrow.cpp | 240 +++++++++++++++++++++ src/video_frame.cpp | 95 +++++++- src/video_stream.cpp | 14 +- 5 files changed, 414 insertions(+), 12 deletions(-) create mode 100644 src/tests/unit/test_video_frame_borrow.cpp diff --git a/include/livekit/video_frame.h b/include/livekit/video_frame.h index e13c7b8d..e1d443c1 100644 --- a/include/livekit/video_frame.h +++ b/include/livekit/video_frame.h @@ -21,6 +21,7 @@ #include #include +#include "livekit/ffi_handle.h" #include "livekit/visibility.h" namespace livekit { @@ -41,10 +42,21 @@ class OwnedVideoBuffer; /// Public SDK representation of a video frame. /// -/// - Owns its pixel buffer (std::vector). +/// - Owns its pixel buffer (std::vector), unless borrowed. /// - Developers can allocate and fill frames in C++ and pass them to the SDK. /// - The SDK can expose the backing memory to Rust via data_ptr + layout for /// the duration of a blocking FFI call (similar to AudioFrame). +/// +/// ## Borrowed frames +/// +/// A frame delivered by a VideoStream opened with `Options::zero_copy` reads the SDK's own buffer +/// instead of a copy of it, saving a full frame-sized memcpy per frame. Such a frame holds the FFI +/// handle that keeps the buffer alive, so the buffer is returned when the frame is destroyed or +/// release() is called -- plain RAII, and the reason VideoFrame is move-only. +/// +/// Every accessor works the same either way. The difference is lifetime: pointers from data() and +/// planeInfos() on a borrowed frame are only valid while that frame is, so do not stash them past +/// its scope. class LIVEKIT_API VideoFrame { public: VideoFrame(); @@ -53,8 +65,10 @@ class LIVEKIT_API VideoFrame { VideoFrame(const VideoFrame&) = delete; VideoFrame& operator=(const VideoFrame&) = delete; - VideoFrame(VideoFrame&&) noexcept = default; - VideoFrame& operator=(VideoFrame&&) noexcept = default; + // Hand-written rather than defaulted: a defaulted move copies the borrowed pointers and leaves + // the source naming memory it no longer owns. + VideoFrame(VideoFrame&&) noexcept; + VideoFrame& operator=(VideoFrame&&) noexcept; /// Allocate a new frame with the correct buffer size for the given format. /// Data is zero-initialized. @@ -65,14 +79,28 @@ class LIVEKIT_API VideoFrame { int height() const noexcept { return height_; } VideoBufferType type() const noexcept { return type_; } - std::uint8_t* data() noexcept { return data_.data(); } - const std::uint8_t* data() const noexcept { return data_.data(); } - std::size_t dataSize() const noexcept { return data_.size(); } + std::uint8_t* data() noexcept { return borrowed_data_ != nullptr ? borrowed_data_ : data_.data(); } + const std::uint8_t* data() const noexcept { return borrowed_data_ != nullptr ? borrowed_data_ : data_.data(); } + std::size_t dataSize() const noexcept { return borrowed_data_ != nullptr ? borrowed_size_ : data_.size(); } + + /// Whether this frame reads the SDK's buffer rather than owning a copy. + /// See the class comment for what that means for pointer lifetimes. + bool isBorrowed() const noexcept { return borrowed_data_ != nullptr; } + + /// Returns a borrowed frame's buffer to the SDK early, instead of waiting for the destructor. + /// No-op on an owning frame, and safe to call twice. + /// + /// After this, data() and planeInfos() describe an empty frame. Anything still pointing into the + /// old buffer is dangling. + void release() noexcept; /// Compute plane layout for this frame (Y/U/V, UV, etc.), in terms of /// pointers & sizes relative to this frame's backing buffer. /// /// For packed formats (ARGB, RGB24) this will be either 1 plane or empty. + /// + /// A borrowed frame reports the layout the SDK gave it, rather than one derived from + /// width/height, so a padded stride is described accurately. std::vector planeInfos() const; /// Convert this frame into another pixel format. @@ -104,13 +132,28 @@ class LIVEKIT_API VideoFrame { friend class VideoStream; // Only internal classes (e.g., VideoStream) // should construct frames directly from FFI buffers. + + /// Copies the pixels out of the SDK buffer and releases the handle. static VideoFrame fromOwnedInfo(const proto::OwnedVideoBuffer& owned); + /// Wraps the SDK buffer in place and takes ownership of the handle, so the buffer lives exactly + /// as long as the returned frame. No copy. + static VideoFrame borrowOwnedInfo(const proto::OwnedVideoBuffer& owned); + private: int width_; int height_; VideoBufferType type_; + + /// Pixels for an owning frame; empty for a borrowed one. std::vector data_; + + /// Non-null only for a borrowed frame, in which case it points into the buffer + /// `borrowed_handle_` keeps alive. Doubles as the "is borrowed" flag. + std::uint8_t* borrowed_data_{nullptr}; + std::size_t borrowed_size_{0}; + std::vector borrowed_planes_; + FfiHandle borrowed_handle_; }; } // namespace livekit diff --git a/include/livekit/video_stream.h b/include/livekit/video_stream.h index 1554fee9..4dd7f449 100644 --- a/include/livekit/video_stream.h +++ b/include/livekit/video_stream.h @@ -43,6 +43,16 @@ struct VideoFrameEvent { std::int64_t timestamp_us; VideoRotation rotation; std::optional metadata; + + /// Frames this stream decoded but never delivered, counted since the stream was created. + /// + /// A frame is dropped when the SDK's internal queue overflows because this application did not + /// read frames fast enough. No WebRTC stat reports the loss: `framesDecoded` counts these frames + /// and `framesDropped` does not, so this counter is the only way to observe it. + /// + /// Cumulative and monotonic: compare against the previous event's value to see how many frames + /// went missing in between. + std::uint64_t dropped_frames{0}; }; namespace proto { @@ -77,6 +87,17 @@ class LIVEKIT_API VideoStream { /// Preferred pixel format for frames delivered by read(). The FFI layer /// converts into this format if supported (e.g., RGBA, BGRA, I420, ...). VideoBufferType format{VideoBufferType::RGBA}; + + /// Deliver frames that read the SDK's buffer instead of a copy of it. + /// + /// Saves a full frame-sized memcpy per frame -- 4.5 MB at 1628x1920 I420, so 134 MB/s at + /// 30 fps. In exchange the frame's lifetime starts to matter: its buffer is returned to the + /// SDK when the VideoFrame is destroyed, so pointers taken from data() or planeInfos() must + /// not outlive it. See VideoFrame's class comment. + /// + /// Prefer a non-zero `capacity` alongside this. Queued frames hold SDK buffers, so an + /// unbounded queue plus a slow reader grows native memory. + bool zero_copy{false}; }; /// Factory: create a VideoStream bound to a specific Track @@ -126,6 +147,7 @@ class LIVEKIT_API VideoStream { std::condition_variable cv_; std::deque queue_; std::size_t capacity_{0}; + bool zero_copy_{false}; bool eof_{false}; bool closed_{false}; diff --git a/src/tests/unit/test_video_frame_borrow.cpp b/src/tests/unit/test_video_frame_borrow.cpp new file mode 100644 index 00000000..d0014bb3 --- /dev/null +++ b/src/tests/unit/test_video_frame_borrow.cpp @@ -0,0 +1,240 @@ +/* + * Copyright 2026 LiveKit + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// Borrowed video frames: the ones that read the SDK's buffer rather than a +// copy of it. These build the OwnedVideoBuffer the FFI would send, over a +// buffer this test owns, so the pointers stay valid however the frame behaves. + +#include +#include + +#include + +#include "livekit/video_frame.h" +#include "video_utils.h" + +namespace livekit::test { +namespace { + +constexpr int kWidth = 64; +constexpr int kHeight = 32; +constexpr std::uint32_t kChromaW = (kWidth + 1) / 2; +constexpr std::uint32_t kChromaH = (kHeight + 1) / 2; +constexpr std::size_t kI420Size = static_cast(kWidth) * kHeight + kChromaW * kChromaH * 2; + +/// Exposes the protected factories so the test can build frames the way +/// VideoStream does, without making them public on the SDK type. +struct FrameFactory : VideoFrame { + using VideoFrame::borrowOwnedInfo; + using VideoFrame::fromOwnedInfo; +}; + +/// Stands in for the buffer the FFI hands over. +/// +/// `handle.id` is left at 0 so the frame's FfiHandle is inert: dropping a +/// zero handle is a no-op, which keeps these tests from reaching into a +/// native handle map that was never set up. +class FakeBuffer { +public: + explicit FakeBuffer(std::uint8_t fill = 0x7F) : storage_(kI420Size, fill) {} + + std::uint8_t* data() { return storage_.data(); } + std::uintptr_t address() const { return reinterpret_cast(storage_.data()); } + + /// Planar I420, planes back to back, the layout the FFI produces. + proto::OwnedVideoBuffer planar() const { + proto::OwnedVideoBuffer owned; + auto* info = owned.mutable_info(); + info->set_type(proto::VideoBufferType::I420); + info->set_width(kWidth); + info->set_height(kHeight); + info->set_data_ptr(address()); + + const std::size_t luma = static_cast(kWidth) * kHeight; + const std::size_t chroma = static_cast(kChromaW) * kChromaH; + addPlane(info, 0, kWidth, luma); + addPlane(info, luma, kChromaW, chroma); + addPlane(info, luma + chroma, kChromaW, chroma); + return owned; + } + + /// Packed RGBA, described only by the top-level pointer and stride. + proto::OwnedVideoBuffer packed() const { + proto::OwnedVideoBuffer owned; + auto* info = owned.mutable_info(); + info->set_type(proto::VideoBufferType::RGBA); + info->set_width(4); + info->set_height(4); + info->set_data_ptr(address()); + info->set_stride(4 * 4); + return owned; + } + +private: + void addPlane(proto::VideoBufferInfo* info, std::size_t offset, std::uint32_t stride, std::size_t size) const { + auto* comp = info->add_components(); + comp->set_data_ptr(address() + offset); + comp->set_stride(stride); + comp->set_size(static_cast(size)); + } + + std::vector storage_; +}; + +} // namespace + +TEST(VideoFrameBorrowTest, BorrowedFrameReadsTheNativeBufferWithoutCopying) { + FakeBuffer native; + VideoFrame frame = FrameFactory::borrowOwnedInfo(native.planar()); + + EXPECT_TRUE(frame.isBorrowed()); + EXPECT_EQ(frame.width(), kWidth); + EXPECT_EQ(frame.height(), kHeight); + EXPECT_EQ(frame.dataSize(), kI420Size); + EXPECT_EQ(reinterpret_cast(frame.data()), native.address()); + + // The giveaway that nothing was copied. + native.data()[0] = 0x01; + EXPECT_EQ(frame.data()[0], 0x01); +} + +TEST(VideoFrameBorrowTest, OwnedFrameCopiesAndIsIndependent) { + FakeBuffer native; + VideoFrame frame = FrameFactory::fromOwnedInfo(native.planar()); + + EXPECT_FALSE(frame.isBorrowed()); + EXPECT_NE(reinterpret_cast(frame.data()), native.address()); + + native.data()[0] = 0x01; + EXPECT_EQ(frame.data()[0], 0x7F); +} + +TEST(VideoFrameBorrowTest, BorrowedPlaneInfoComesFromTheSdkNotFromWidthAndHeight) { + FakeBuffer native; + VideoFrame frame = FrameFactory::borrowOwnedInfo(native.planar()); + + const auto planes = frame.planeInfos(); + ASSERT_EQ(planes.size(), 3u); + EXPECT_EQ(planes[0].data_ptr, native.address()); + EXPECT_EQ(planes[0].stride, static_cast(kWidth)); + EXPECT_EQ(planes[1].stride, kChromaW); + EXPECT_EQ(planes[2].stride, kChromaW); + EXPECT_EQ(planes[0].size + planes[1].size + planes[2].size, kI420Size); +} + +TEST(VideoFrameBorrowTest, BorrowedPlaneInfoPreservesAPaddedStride) { + // A decoder aligns its rows, so the stride the SDK reports can exceed the + // frame width. Recomputing the layout from width/height would lose that. + FakeBuffer native; + proto::OwnedVideoBuffer owned = native.planar(); + owned.mutable_info()->mutable_components(0)->set_stride(kWidth + 16); + + VideoFrame frame = FrameFactory::borrowOwnedInfo(owned); + + EXPECT_EQ(frame.planeInfos()[0].stride, static_cast(kWidth + 16)); +} + +TEST(VideoFrameBorrowTest, PackedFormatsBorrowTheTopLevelPointer) { + FakeBuffer native; + VideoFrame frame = FrameFactory::borrowOwnedInfo(native.packed()); + + EXPECT_TRUE(frame.isBorrowed()); + EXPECT_EQ(reinterpret_cast(frame.data()), native.address()); + EXPECT_EQ(frame.dataSize(), 4u * 4u * 4u); + EXPECT_FALSE(frame.planeInfos().empty()); +} + +TEST(VideoFrameBorrowTest, ReleaseEmptiesTheFrame) { + FakeBuffer native; + VideoFrame frame = FrameFactory::borrowOwnedInfo(native.planar()); + + frame.release(); + + EXPECT_FALSE(frame.isBorrowed()); + EXPECT_EQ(frame.dataSize(), 0u); + EXPECT_TRUE(frame.planeInfos().empty()); + + frame.release(); // idempotent + EXPECT_FALSE(frame.isBorrowed()); +} + +TEST(VideoFrameBorrowTest, ReleaseOnAnOwningFrameLeavesItUsable) { + FakeBuffer native; + VideoFrame frame = FrameFactory::fromOwnedInfo(native.planar()); + + frame.release(); + + EXPECT_FALSE(frame.isBorrowed()); + EXPECT_EQ(frame.dataSize(), kI420Size); + EXPECT_EQ(frame.data()[0], 0x7F); +} + +TEST(VideoFrameBorrowTest, MovingABorrowedFrameLeavesTheSourceEmpty) { + FakeBuffer native; + VideoFrame source = FrameFactory::borrowOwnedInfo(native.planar()); + const auto address = reinterpret_cast(source.data()); + + VideoFrame moved(std::move(source)); + + EXPECT_TRUE(moved.isBorrowed()); + EXPECT_EQ(reinterpret_cast(moved.data()), address); + // NOLINTNEXTLINE(bugprone-use-after-move) -- asserting the moved-from state + EXPECT_FALSE(source.isBorrowed()); + // NOLINTNEXTLINE(bugprone-use-after-move) + EXPECT_EQ(source.dataSize(), 0u); +} + +TEST(VideoFrameBorrowTest, MoveAssigningABorrowedFrameLeavesTheSourceEmpty) { + FakeBuffer native; + VideoFrame source = FrameFactory::borrowOwnedInfo(native.planar()); + VideoFrame target; + + target = std::move(source); + + EXPECT_TRUE(target.isBorrowed()); + // NOLINTNEXTLINE(bugprone-use-after-move) + EXPECT_FALSE(source.isBorrowed()); +} + +TEST(VideoFrameBorrowTest, ConvertingABorrowedFrameToItsOwnFormatClonesTheSdkBuffer) { + // The same-format fast path used to clone the frame's own vector, which is + // empty for a borrowed frame. + FakeBuffer native; + VideoFrame frame = FrameFactory::borrowOwnedInfo(native.planar()); + + const VideoFrame copy = frame.convert(VideoBufferType::I420); + + EXPECT_FALSE(copy.isBorrowed()); + EXPECT_EQ(copy.dataSize(), kI420Size); + EXPECT_EQ(copy.data()[0], 0x7F); + EXPECT_NE(reinterpret_cast(copy.data()), native.address()); +} + +TEST(VideoFrameBorrowTest, AnUndescribableBufferYieldsAnEmptyFrameRatherThanABorrow) { + proto::OwnedVideoBuffer owned; + auto* info = owned.mutable_info(); + info->set_type(proto::VideoBufferType::I420); + info->set_width(kWidth); + info->set_height(kHeight); + info->set_data_ptr(0); + + VideoFrame frame = FrameFactory::borrowOwnedInfo(owned); + + EXPECT_FALSE(frame.isBorrowed()); + EXPECT_EQ(frame.dataSize(), 0u); +} + +} // namespace livekit::test diff --git a/src/video_frame.cpp b/src/video_frame.cpp index 6f90c6b8..48a7fab4 100644 --- a/src/video_frame.cpp +++ b/src/video_frame.cpp @@ -288,7 +288,52 @@ VideoFrame VideoFrame::create(int width, int height, VideoBufferType type) { return VideoFrame(width, height, type, std::move(buffer)); } +VideoFrame::VideoFrame(VideoFrame&& other) noexcept + : width_(other.width_), height_(other.height_), type_(other.type_), data_(std::move(other.data_)), + borrowed_data_(other.borrowed_data_), borrowed_size_(other.borrowed_size_), + borrowed_planes_(std::move(other.borrowed_planes_)), borrowed_handle_(std::move(other.borrowed_handle_)) { + // The handle moved, so the source no longer keeps the buffer alive and must stop naming it. + other.borrowed_data_ = nullptr; + other.borrowed_size_ = 0; +} + +VideoFrame& VideoFrame::operator=(VideoFrame&& other) noexcept { + if (this == &other) { + return *this; + } + release(); // drops any buffer this frame was holding + + width_ = other.width_; + height_ = other.height_; + type_ = other.type_; + data_ = std::move(other.data_); + borrowed_data_ = other.borrowed_data_; + borrowed_size_ = other.borrowed_size_; + borrowed_planes_ = std::move(other.borrowed_planes_); + borrowed_handle_ = std::move(other.borrowed_handle_); + + other.borrowed_data_ = nullptr; + other.borrowed_size_ = 0; + return *this; +} + +void VideoFrame::release() noexcept { + if (borrowed_data_ == nullptr) { + return; + } + borrowed_data_ = nullptr; + borrowed_size_ = 0; + borrowed_planes_.clear(); + borrowed_handle_ = FfiHandle{}; // disposes the SDK buffer +} + std::vector VideoFrame::planeInfos() const { + if (borrowed_data_ != nullptr) { + // Report what the SDK said rather than recomputing from width/height: only the SDK knows + // whether a plane carries row padding. + return borrowed_planes_; + } + if (data_.empty()) { return {}; } @@ -303,8 +348,9 @@ VideoFrame VideoFrame::convert(VideoBufferType dst, bool flip_y) const { // being deleted is not a problem. if (dst == type_ && !flip_y) { LK_LOG_WARN("VideoFrame::convert: converting to the same format"); - // copy pixel data - std::vector buf = data_; + // Copy pixel data. Reads through data()/dataSize() so a borrowed frame clones the SDK buffer + // rather than its own empty vector. + std::vector buf(data(), data() + dataSize()); return VideoFrame(width_, height_, type_, std::move(buf)); } @@ -367,4 +413,49 @@ VideoFrame VideoFrame::fromOwnedInfo(const proto::OwnedVideoBuffer& owned) { return VideoFrame(width, height, type, std::move(buffer)); } +VideoFrame VideoFrame::borrowOwnedInfo(const proto::OwnedVideoBuffer& owned) { + const auto& info = owned.info(); + + VideoFrame frame; + frame.width_ = static_cast(info.width()); + frame.height_ = static_cast(info.height()); + frame.type_ = fromProto(info.type()); + + if (info.components_size() > 0) { + // Planar. The planes are contiguous and in order, so the frame's extent runs from the first + // plane to the end of the last. + std::size_t total_size = 0; + frame.borrowed_planes_.reserve(static_cast(info.components_size())); + for (const auto& comp : info.components()) { + frame.borrowed_planes_.push_back( + VideoPlaneInfo{static_cast(comp.data_ptr()), comp.stride(), comp.size()}); + total_size += static_cast(comp.size()); + } + frame.borrowed_size_ = total_size; + // NOLINTNEXTLINE(performance-no-int-to-ptr) + frame.borrowed_data_ = reinterpret_cast(static_cast(info.components(0).data_ptr())); + } else { + // Packed: one contiguous run at the top-level pointer. + frame.borrowed_size_ = info.has_stride() + ? static_cast(info.stride()) * static_cast(frame.height_) + : computeBufferSize(frame.width_, frame.height_, frame.type_); + // NOLINTNEXTLINE(performance-no-int-to-ptr) + frame.borrowed_data_ = reinterpret_cast(static_cast(info.data_ptr())); + frame.borrowed_planes_ = + computePlaneInfos(static_cast(info.data_ptr()), frame.width_, frame.height_, frame.type_); + } + + if (frame.borrowed_data_ == nullptr || frame.borrowed_size_ == 0) { + // Nothing to point at; drop the handle rather than hand back a frame that claims to borrow a + // buffer it cannot describe. + const FfiHandle discard(static_cast(owned.handle().id())); + LK_LOG_WARN("VideoFrame::borrowOwnedInfo: empty buffer info, dropping frame"); + return VideoFrame{}; + } + + // Taking the handle is what keeps the buffer alive for the frame's lifetime. + frame.borrowed_handle_ = FfiHandle(static_cast(owned.handle().id())); + return frame; +} + } // namespace livekit diff --git a/src/video_stream.cpp b/src/video_stream.cpp index effd3f63..da91225e 100644 --- a/src/video_stream.cpp +++ b/src/video_stream.cpp @@ -50,6 +50,7 @@ VideoStream::VideoStream(VideoStream&& other) noexcept { const std::scoped_lock lock(other.mutex_); queue_ = std::move(other.queue_); capacity_ = other.capacity_; + zero_copy_ = other.zero_copy_; eof_ = other.eof_; closed_ = other.closed_; stream_handle_ = std::move(other.stream_handle_); @@ -72,6 +73,7 @@ VideoStream& VideoStream::operator=(VideoStream&& other) noexcept { queue_ = std::move(other.queue_); capacity_ = other.capacity_; + zero_copy_ = other.zero_copy_; eof_ = other.eof_; closed_ = other.closed_; stream_handle_ = std::move(other.stream_handle_); @@ -128,6 +130,7 @@ void VideoStream::close() { void VideoStream::initFromTrack(const std::shared_ptr& track, const Options& options) { capacity_ = options.capacity; + zero_copy_ = options.zero_copy; // Subscribe to FFI events, this is essential to get video frames from FFI. listener_id_ = FfiClient::instance().addListener([this](const proto::FfiEvent& e) { this->onFfiEvent(e); }); @@ -153,6 +156,7 @@ void VideoStream::initFromTrack(const std::shared_ptr& track, const Optio void VideoStream::initFromParticipant(Participant& participant, TrackSource track_source, const Options& options) { capacity_ = options.capacity; + zero_copy_ = options.zero_copy; // 1) Subscribe to FFI events listener_id_ = FfiClient::instance().addListener([this](const FfiEvent& e) { this->onFfiEvent(e); }); @@ -187,12 +191,13 @@ void VideoStream::onFfiEvent(const proto::FfiEvent& event) { if (vse.has_frame_received()) { const auto& fr = vse.frame_received(); - // Convert owned buffer->VideoFrame via a helper. - // You should implement this static function in your VideoFrame class. VideoFrameEvent ev; - ev.frame = VideoFrame::fromOwnedInfo(fr.buffer()); + // Borrowing hands the frame the FFI handle instead of copying the pixels out of it; the + // buffer then lives as long as ev.frame does. + ev.frame = zero_copy_ ? VideoFrame::borrowOwnedInfo(fr.buffer()) : VideoFrame::fromOwnedInfo(fr.buffer()); ev.timestamp_us = fr.timestamp_us(); ev.rotation = static_cast(fr.rotation()); + ev.dropped_frames = fr.dropped_frames(); if (fr.has_metadata()) { ev.metadata = fromProto(fr.metadata()); } @@ -211,7 +216,8 @@ void VideoStream::pushFrame(VideoFrameEvent&& ev) { } if (capacity_ > 0 && queue_.size() >= capacity_) { - // Ring behavior: drop oldest frame. + // Ring behavior: drop oldest frame. For a borrowed frame this is also what returns its + // buffer to the SDK, via ~VideoFrame. queue_.pop_front(); } From d8db5ea7bfe2a291a037e141286af1e945952c1c Mon Sep 17 00:00:00 2001 From: shijing xian Date: Fri, 6 Nov 2026 14:15:43 -0800 Subject: [PATCH 2/4] Cover the zero-copy video path end to end The unit tests build the OwnedVideoBuffer by hand. These publish a real track, let it through the SFU and the decoder, and read it back with borrowed frames. The frame-run test is the one that matters: by the time each frame is read, the previous frame's buffer has gone back to the FFI's pool and been handed out again, so decoding to non-black luma is what shows the recycling is safe. The other two pin the dropped-frame counter and confirm an owning frame still outlives the stream it came from. Needs a server, so these live with the integration suite: source scripts/set-test-tokens.sh && ./build.sh debug-tests ./build-debug/bin/livekit_integration_tests \ --gtest_filter="VideoStreamZeroCopyServerTest.*" Co-Authored-By: Claude Opus 5 (1M context) --- .../test_video_stream_zero_copy.cpp | 239 ++++++++++++++++++ 1 file changed, 239 insertions(+) create mode 100644 src/tests/integration/test_video_stream_zero_copy.cpp diff --git a/src/tests/integration/test_video_stream_zero_copy.cpp b/src/tests/integration/test_video_stream_zero_copy.cpp new file mode 100644 index 00000000..f9574ed0 --- /dev/null +++ b/src/tests/integration/test_video_stream_zero_copy.cpp @@ -0,0 +1,239 @@ +/* + * Copyright 2026 LiveKit + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// End-to-end cover for VideoStream::Options::zero_copy. +// +// The unit tests build the OwnedVideoBuffer by hand; these publish a real +// track, let it travel through the SFU and the decoder, and read it back with +// a borrowed frame. That is the only way to see the pointers the FFI actually +// hands over, and the only way to know the buffer stays valid across the +// pooled-buffer recycling the Rust side now does. + +#include +#include +#include +#include +#include +#include + +#include "tests/common/test_common.h" +#include "tests/common/video_utils.h" + +namespace livekit::test { +namespace { + +using namespace std::chrono_literals; + +constexpr int kWidth = 320; +constexpr int kHeight = 240; + +/// Publishes frames until stopped, so the subscriber has something to read. +class FramePublisher { +public: + FramePublisher(std::shared_ptr source, int width, int height) + : source_(std::move(source)), width_(width), height_(height) {} + + void start() { + thread_ = std::thread([this] { + std::uint64_t n = 0; + while (!stop_) { + VideoFrame frame = VideoFrame::create(width_, height_, VideoBufferType::ARGB); + fillWebcamLikeFrame(frame, n++); + source_->captureFrame(frame); + std::this_thread::sleep_for(33ms); // ~30 fps + } + }); + } + + ~FramePublisher() { + stop_ = true; + if (thread_.joinable()) { + thread_.join(); + } + } + +private: + std::shared_ptr source_; + int width_; + int height_; + std::atomic stop_{false}; + std::thread thread_; +}; + +/// Finds the remote video publication named `track_name`, waiting for it. +std::shared_ptr awaitRemoteVideoTrack(Room& room, const std::string& identity, + const std::string& track_name, std::chrono::seconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + auto participant = room.remoteParticipant(identity).lock(); + if (participant != nullptr) { + for (const auto& [sid, publication] : participant->trackPublications()) { + (void)sid; + if (publication == nullptr || publication->kind() != TrackKind::KIND_VIDEO || + publication->name() != track_name) { + continue; + } + if (auto track = publication->track(); track != nullptr) { + return track; + } + } + } + std::this_thread::sleep_for(100ms); + } + return nullptr; +} + +} // namespace + +class VideoStreamZeroCopyServerTest : public LiveKitTestBase { +protected: + /// Connects both rooms, publishes a video track, and returns the remote + /// track the subscriber sees. + std::shared_ptr publishAndSubscribe() { + const std::string track_name = "zero-copy-track"; + + EXPECT_TRUE(receiver_room_.connect(config_.url, config_.token_b, RoomOptions{})); + EXPECT_TRUE(sender_room_.connect(config_.url, config_.token_a, RoomOptions{})); + + const std::string sender_identity = lockLocalParticipant(sender_room_)->identity(); + EXPECT_FALSE(sender_identity.empty()); + EXPECT_TRUE(waitForParticipant(&receiver_room_, sender_identity, 10s)); + + source_ = std::make_shared(kWidth, kHeight); + auto track = LocalVideoTrack::createLocalVideoTrack(track_name, source_); + + TrackPublishOptions publish_options; + publish_options.source = TrackSource::SOURCE_CAMERA; + publish_options.simulcast = false; + lockLocalParticipant(sender_room_)->publishTrack(track, publish_options); + + publisher_ = std::make_unique(source_, kWidth, kHeight); + publisher_->start(); + + return awaitRemoteVideoTrack(receiver_room_, sender_identity, track_name, 20s); + } + + Room sender_room_; + Room receiver_room_; + std::shared_ptr source_; + std::unique_ptr publisher_; +}; + +TEST_F(VideoStreamZeroCopyServerTest, BorrowedFramesCarryDecodedPixels) { + failIfNotConfigured(); + + auto remote_track = publishAndSubscribe(); + ASSERT_NE(remote_track, nullptr) << "the subscriber never saw the published video track"; + + VideoStream::Options options; + options.format = VideoBufferType::I420; + options.zero_copy = true; + options.capacity = 8; + auto stream = VideoStream::fromTrack(remote_track, options); + ASSERT_NE(stream, nullptr); + + // Read a run of frames. Each one borrows the SDK's buffer, and the previous + // one's buffer has already gone back to the Rust pool to be handed out + // again -- so reading valid pixels here is what shows the recycling is safe. + constexpr int kFramesToRead = 20; + int read = 0; + VideoFrameEvent event; + while (read < kFramesToRead && stream->read(event)) { + ASSERT_TRUE(event.frame.isBorrowed()) << "frame " << read << " was copied, not borrowed"; + ASSERT_NE(event.frame.data(), nullptr); + + const std::size_t expected = static_cast(kWidth) * kHeight + + 2 * (static_cast((kWidth + 1) / 2) * ((kHeight + 1) / 2)); + EXPECT_EQ(event.frame.dataSize(), expected); + EXPECT_EQ(event.frame.width(), kWidth); + EXPECT_EQ(event.frame.height(), kHeight); + + const auto planes = event.frame.planeInfos(); + ASSERT_EQ(planes.size(), 3u) << "I420 should report three planes"; + for (const auto& plane : planes) { + EXPECT_NE(plane.data_ptr, 0u); + EXPECT_GT(plane.stride, 0u); + } + + // The publisher paints a green-ish frame, so the luma plane must not be + // uniformly zero. This is the check that the pointer really addresses + // decoded pixels rather than a buffer that was released underneath us. + const std::uint8_t* luma = event.frame.data(); + std::uint64_t luma_sum = 0; + for (std::size_t i = 0; i < static_cast(kWidth) * kHeight; ++i) { + luma_sum += luma[i]; + } + EXPECT_GT(luma_sum, 0u) << "frame " << read << " decoded to an all-black luma plane"; + + ++read; + } + + EXPECT_EQ(read, kFramesToRead) << "the stream ended after " << read << " frames"; + stream->close(); +} + +TEST_F(VideoStreamZeroCopyServerTest, DroppedFrameCountIsReportedAndStartsAtZero) { + failIfNotConfigured(); + + auto remote_track = publishAndSubscribe(); + ASSERT_NE(remote_track, nullptr); + + VideoStream::Options options; + options.format = VideoBufferType::I420; + options.capacity = 8; + auto stream = VideoStream::fromTrack(remote_track, options); + ASSERT_NE(stream, nullptr); + + // A reader that keeps up should see no drops, and the counter must never + // run backwards: it is cumulative over the stream's life. + std::uint64_t previous = 0; + VideoFrameEvent event; + for (int i = 0; i < 20 && stream->read(event); ++i) { + EXPECT_GE(event.dropped_frames, previous) << "the counter went backwards at frame " << i; + previous = event.dropped_frames; + } + + EXPECT_EQ(previous, 0u) << "a reader keeping up with the stream should not lose frames"; + stream->close(); +} + +TEST_F(VideoStreamZeroCopyServerTest, OwningFramesOutliveTheStream) { + failIfNotConfigured(); + + auto remote_track = publishAndSubscribe(); + ASSERT_NE(remote_track, nullptr); + + VideoStream::Options options; + options.format = VideoBufferType::I420; + options.zero_copy = false; // the default: each frame owns a copy + options.capacity = 8; + auto stream = VideoStream::fromTrack(remote_track, options); + ASSERT_NE(stream, nullptr); + + VideoFrameEvent event; + ASSERT_TRUE(stream->read(event)); + ASSERT_FALSE(event.frame.isBorrowed()); + const std::vector snapshot(event.frame.data(), event.frame.data() + event.frame.dataSize()); + + stream->close(); + stream.reset(); + + // An owning frame is independent of the stream, so its pixels survive. + ASSERT_EQ(event.frame.dataSize(), snapshot.size()); + EXPECT_EQ(std::vector(event.frame.data(), event.frame.data() + event.frame.dataSize()), snapshot); +} + +} // namespace livekit::test From 5d089f5ea26c37d5cfa93a176d48b393ef22689d Mon Sep 17 00:00:00 2001 From: shijing xian Date: Fri, 6 Nov 2026 14:30:15 -0800 Subject: [PATCH 3/4] Format with the project clang-format config The config lives in the cpp-tools submodule and is symlinked in by ./cpp-tools/install.sh; without it installed, clang-format falls back to LLVM defaults. Line breaks only. --- src/tests/integration/test_video_stream_zero_copy.cpp | 4 ++-- src/tests/unit/test_video_frame_borrow.cpp | 4 ++-- src/video_frame.cpp | 11 ++++++++--- 3 files changed, 12 insertions(+), 7 deletions(-) diff --git a/src/tests/integration/test_video_stream_zero_copy.cpp b/src/tests/integration/test_video_stream_zero_copy.cpp index f9574ed0..8a7c6862 100644 --- a/src/tests/integration/test_video_stream_zero_copy.cpp +++ b/src/tests/integration/test_video_stream_zero_copy.cpp @@ -74,8 +74,8 @@ class FramePublisher { }; /// Finds the remote video publication named `track_name`, waiting for it. -std::shared_ptr awaitRemoteVideoTrack(Room& room, const std::string& identity, - const std::string& track_name, std::chrono::seconds timeout) { +std::shared_ptr awaitRemoteVideoTrack(Room& room, const std::string& identity, const std::string& track_name, + std::chrono::seconds timeout) { const auto deadline = std::chrono::steady_clock::now() + timeout; while (std::chrono::steady_clock::now() < deadline) { auto participant = room.remoteParticipant(identity).lock(); diff --git a/src/tests/unit/test_video_frame_borrow.cpp b/src/tests/unit/test_video_frame_borrow.cpp index d0014bb3..090cc2f6 100644 --- a/src/tests/unit/test_video_frame_borrow.cpp +++ b/src/tests/unit/test_video_frame_borrow.cpp @@ -18,11 +18,11 @@ // copy of it. These build the OwnedVideoBuffer the FFI would send, over a // buffer this test owns, so the pointers stay valid however the frame behaves. +#include + #include #include -#include - #include "livekit/video_frame.h" #include "video_utils.h" diff --git a/src/video_frame.cpp b/src/video_frame.cpp index 48a7fab4..9ff040a7 100644 --- a/src/video_frame.cpp +++ b/src/video_frame.cpp @@ -289,9 +289,14 @@ VideoFrame VideoFrame::create(int width, int height, VideoBufferType type) { } VideoFrame::VideoFrame(VideoFrame&& other) noexcept - : width_(other.width_), height_(other.height_), type_(other.type_), data_(std::move(other.data_)), - borrowed_data_(other.borrowed_data_), borrowed_size_(other.borrowed_size_), - borrowed_planes_(std::move(other.borrowed_planes_)), borrowed_handle_(std::move(other.borrowed_handle_)) { + : width_(other.width_), + height_(other.height_), + type_(other.type_), + data_(std::move(other.data_)), + borrowed_data_(other.borrowed_data_), + borrowed_size_(other.borrowed_size_), + borrowed_planes_(std::move(other.borrowed_planes_)), + borrowed_handle_(std::move(other.borrowed_handle_)) { // The handle moved, so the source no longer keeps the buffer alive and must stop naming it. other.borrowed_data_ = nullptr; other.borrowed_size_ = 0; From bf03923e0140264184e6a5b79cc06af48919b40c Mon Sep 17 00:00:00 2001 From: shijing xian Date: Fri, 6 Nov 2026 17:27:31 -0800 Subject: [PATCH 4/4] Verify plane contiguity, and count this stream's own evictions Two review findings. borrowOwnedInfo described a planar frame as one run starting at the first plane, which is only truthful if the planes sit end to end. The Rust side packs them that way today, but that is a convention across the FFI rather than something the message guarantees, and data()/dataSize() would otherwise hand out a span covering whatever lies between them -- convert() reads the whole span. The layout is now checked, and a frame that fails falls back to copying, which handles disjoint planes correctly. dropped_frames reported only the SDK's losses. A bounded VideoStream evicts its oldest frame when the queue is full, and the SDK cannot see that: it stamped its count on the event before handing it over. A frame the application never received went unreported, which is exactly the loss the counter exists to surface. Evictions are tallied per stream and added on delivery; both halves are monotonic, so the sum still is. Co-Authored-By: Claude Opus 5 (1M context) --- include/livekit/video_stream.h | 5 ++++- src/tests/unit/test_video_frame_borrow.cpp | 17 ++++++++++++++++ src/video_frame.cpp | 23 ++++++++++++++++++---- src/video_stream.cpp | 9 +++++++++ 4 files changed, 49 insertions(+), 5 deletions(-) diff --git a/include/livekit/video_stream.h b/include/livekit/video_stream.h index 4dd7f449..50c7c139 100644 --- a/include/livekit/video_stream.h +++ b/include/livekit/video_stream.h @@ -51,7 +51,8 @@ struct VideoFrameEvent { /// and `framesDropped` does not, so this counter is the only way to observe it. /// /// Cumulative and monotonic: compare against the previous event's value to see how many frames - /// went missing in between. + /// went missing in between. Counts losses on both sides of the FFI -- frames the SDK's own + /// queue overflowed, and frames this stream's `capacity` evicted before `read()` reached them. std::uint64_t dropped_frames{0}; }; @@ -148,6 +149,8 @@ class LIVEKIT_API VideoStream { std::deque queue_; std::size_t capacity_{0}; bool zero_copy_{false}; + // Frames this stream's own queue evicted; added to the SDK's count on delivery. + std::uint64_t evicted_frames_{0}; bool eof_{false}; bool closed_{false}; diff --git a/src/tests/unit/test_video_frame_borrow.cpp b/src/tests/unit/test_video_frame_borrow.cpp index 090cc2f6..c604b606 100644 --- a/src/tests/unit/test_video_frame_borrow.cpp +++ b/src/tests/unit/test_video_frame_borrow.cpp @@ -157,6 +157,23 @@ TEST(VideoFrameBorrowTest, PackedFormatsBorrowTheTopLevelPointer) { EXPECT_FALSE(frame.planeInfos().empty()); } +TEST(VideoFrameBorrowTest, NonContiguousPlanesAreCopiedRatherThanBorrowed) { + // data()/dataSize() describe one run from the first plane, so borrowing is only truthful + // when the planes sit end to end. The SDK packs them that way, but it is a convention + // across the FFI rather than a guarantee of the message -- if it is ever broken, the frame + // must fall back to copying instead of handing out a span covering the gap. + FakeBuffer native; + proto::OwnedVideoBuffer owned = native.planar(); + const auto luma_size = owned.info().components(0).size(); + // Push the U plane past the end of Y, leaving a hole. + owned.mutable_info()->mutable_components(1)->set_data_ptr(native.address() + luma_size + 4096); + + VideoFrame frame = FrameFactory::borrowOwnedInfo(owned); + + EXPECT_FALSE(frame.isBorrowed()) << "a frame with a gap must not advertise one span"; + EXPECT_EQ(frame.dataSize(), kI420Size) << "it is an owning copy instead"; +} + TEST(VideoFrameBorrowTest, ReleaseEmptiesTheFrame) { FakeBuffer native; VideoFrame frame = FrameFactory::borrowOwnedInfo(native.planar()); diff --git a/src/video_frame.cpp b/src/video_frame.cpp index 9ff040a7..14f7a7de 100644 --- a/src/video_frame.cpp +++ b/src/video_frame.cpp @@ -427,15 +427,30 @@ VideoFrame VideoFrame::borrowOwnedInfo(const proto::OwnedVideoBuffer& owned) { frame.type_ = fromProto(info.type()); if (info.components_size() > 0) { - // Planar. The planes are contiguous and in order, so the frame's extent runs from the first - // plane to the end of the last. + // Planar. data()/dataSize() describe one run starting at the first plane, which is only + // truthful if the planes sit end to end in order. The Rust side packs them that way today, + // but that is a convention across an FFI boundary rather than something the message + // guarantees, so check it rather than trust it: a frame that fails has to be copied, or + // data() would hand out a span covering memory between the planes. std::size_t total_size = 0; + std::uintptr_t next_plane = static_cast(info.components(0).data_ptr()); + bool contiguous = true; + frame.borrowed_planes_.reserve(static_cast(info.components_size())); for (const auto& comp : info.components()) { - frame.borrowed_planes_.push_back( - VideoPlaneInfo{static_cast(comp.data_ptr()), comp.stride(), comp.size()}); + const auto address = static_cast(comp.data_ptr()); + contiguous = contiguous && address == next_plane; + next_plane = address + comp.size(); + + frame.borrowed_planes_.push_back(VideoPlaneInfo{address, comp.stride(), comp.size()}); total_size += static_cast(comp.size()); } + + if (!contiguous) { + LK_LOG_WARN("VideoFrame::borrowOwnedInfo: planes are not contiguous, copying instead"); + return fromOwnedInfo(owned); + } + frame.borrowed_size_ = total_size; // NOLINTNEXTLINE(performance-no-int-to-ptr) frame.borrowed_data_ = reinterpret_cast(static_cast(info.components(0).data_ptr())); diff --git a/src/video_stream.cpp b/src/video_stream.cpp index da91225e..bc6016a3 100644 --- a/src/video_stream.cpp +++ b/src/video_stream.cpp @@ -51,6 +51,7 @@ VideoStream::VideoStream(VideoStream&& other) noexcept { queue_ = std::move(other.queue_); capacity_ = other.capacity_; zero_copy_ = other.zero_copy_; + evicted_frames_ = other.evicted_frames_; eof_ = other.eof_; closed_ = other.closed_; stream_handle_ = std::move(other.stream_handle_); @@ -74,6 +75,7 @@ VideoStream& VideoStream::operator=(VideoStream&& other) noexcept { queue_ = std::move(other.queue_); capacity_ = other.capacity_; zero_copy_ = other.zero_copy_; + evicted_frames_ = other.evicted_frames_; eof_ = other.eof_; closed_ = other.closed_; stream_handle_ = std::move(other.stream_handle_); @@ -99,6 +101,9 @@ bool VideoStream::read(VideoFrameEvent& out) { out = std::move(queue_.front()); queue_.pop_front(); + // Add this stream's own evictions to the SDK's count, so the number the caller sees covers + // every frame that was decoded but never reached them. Both parts are monotonic, so the sum is. + out.dropped_frames += evicted_frames_; return true; } @@ -218,7 +223,11 @@ void VideoStream::pushFrame(VideoFrameEvent&& ev) { if (capacity_ > 0 && queue_.size() >= capacity_) { // Ring behavior: drop oldest frame. For a borrowed frame this is also what returns its // buffer to the SDK, via ~VideoFrame. + // + // The SDK cannot see this loss -- it already stamped its own count on the event before + // handing it over -- so it is counted here and added on delivery. queue_.pop_front(); + ++evicted_frames_; } queue_.push_back(std::move(ev));