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..50c7c139 100644 --- a/include/livekit/video_stream.h +++ b/include/livekit/video_stream.h @@ -43,6 +43,17 @@ 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. 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}; }; namespace proto { @@ -77,6 +88,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 +148,9 @@ class LIVEKIT_API VideoStream { std::condition_variable cv_; 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/integration/test_video_stream_zero_copy.cpp b/src/tests/integration/test_video_stream_zero_copy.cpp new file mode 100644 index 00000000..8a7c6862 --- /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 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..c604b606 --- /dev/null +++ b/src/tests/unit/test_video_frame_borrow.cpp @@ -0,0 +1,257 @@ +/* + * 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, 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()); + + 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..14f7a7de 100644 --- a/src/video_frame.cpp +++ b/src/video_frame.cpp @@ -288,7 +288,57 @@ 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 +353,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 +418,64 @@ 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. 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()) { + 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())); + } 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..bc6016a3 100644 --- a/src/video_stream.cpp +++ b/src/video_stream.cpp @@ -50,6 +50,8 @@ 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_; + evicted_frames_ = other.evicted_frames_; eof_ = other.eof_; closed_ = other.closed_; stream_handle_ = std::move(other.stream_handle_); @@ -72,6 +74,8 @@ 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_); @@ -97,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; } @@ -128,6 +135,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 +161,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 +196,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,8 +221,13 @@ 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. + // + // 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));