admin/webrtc_m130
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add FrameBufferProxy in prep for FrameBuffer3

Browse Source 
This is a delegate that is used by video_receive_stream2 to handle frame
buffer tasks like threading, and stats. This will be used in a follow up
to use FrameBuffer3 as a strategy selected by field trial.

Unit-tests will be used in follow-up CLs containing Frame Buffer 3, and
are expected to work with both Frame buffer proxy versions.

Change-Id: I524279343d60a348d044d9085d618f12d7bf3a23
Bug: webrtc:13343
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/241605
Reviewed-by: Ilya Nikolaevskiy <ilnik@webrtc.org>
Reviewed-by: Philip Eliasson <philipel@webrtc.org>
Commit-Queue: Evan Shrubsole <eshr@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#35803}
This commit is contained in:
Evan Shrubsole 2021-12-15 14:33:40 +01:00 committed by WebRTC LUCI CQ
parent 62238097c9
commit 9d29026216
7 changed files with 908 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
modules/video_coding/BUILD.gn
View File

@ -253,6 +253,7 @@ rtc_library("video_coding") {
"../../api:rtp_packet_info",
"../../api:scoped_refptr",
"../../api:sequence_checker",
"../../api/task_queue",
"../../api/units:data_rate",
"../../api/units:time_delta",
"../../api/units:timestamp",

20
video/BUILD.gn
View File

@ -53,6 +53,7 @@ rtc_library("video") {
]
deps = [
":frame_buffer_proxy",
":frame_cadence_adapter",
":frame_dumping_decoder",
":video_stream_encoder_impl",
@ -285,6 +286,22 @@ rtc_library("frame_cadence_adapter") {
]
}
rtc_library("frame_buffer_proxy") {
sources = [
"frame_buffer_proxy.cc",
"frame_buffer_proxy.h",
]
deps = [
"../api/task_queue",
"../api/video:encoded_frame",
"../modules/video_coding",
"../modules/video_coding:video_codec_interface",
"../rtc_base:rtc_task_queue",
"../system_wrappers",
]
absl_deps = [ "//third_party/abseil-cpp/absl/types:optional" ]
}
rtc_library("video_stream_encoder_impl") {
visibility = [ "*" ]
@ -640,6 +657,7 @@ if (rtc_include_tests) {
"end_to_end_tests/ssrc_tests.cc",
"end_to_end_tests/stats_tests.cc",
"end_to_end_tests/transport_feedback_tests.cc",
"frame_buffer_proxy_unittest.cc",
"frame_cadence_adapter_unittest.cc",
"frame_encode_metadata_writer_unittest.cc",
"picture_id_tests.cc",
@ -662,6 +680,7 @@ if (rtc_include_tests) {
"video_stream_encoder_unittest.cc",
]
deps = [
":frame_buffer_proxy",
":frame_cadence_adapter",
":video",
":video_mocks",
@ -690,6 +709,7 @@ if (rtc_include_tests) {
"../api/task_queue:default_task_queue_factory",
"../api/test/video:function_video_factory",
"../api/units:data_rate",
"../api/units:time_delta",
"../api/units:timestamp",
"../api/video:builtin_video_bitrate_allocator_factory",
"../api/video:encoded_image",

132
video/frame_buffer_proxy.cc Normal file
View File

@ -0,0 +1,132 @@
/*
* Copyright (c) 2022 The WebRTC project authors. All Rights Reserved.
*
* Use of this source code is governed by a BSD-style license
* that can be found in the LICENSE file in the root of the source
* tree. An additional intellectual property rights grant can be found
* in the file PATENTS. All contributing project authors may
* be found in the AUTHORS file in the root of the source tree.
*/
#include "video/frame_buffer_proxy.h"
#include <memory>
#include <utility>
#include "modules/video_coding/frame_buffer2.h"
namespace webrtc {
class FrameBuffer2Proxy : public FrameBufferProxy {
public:
FrameBuffer2Proxy(Clock* clock,
VCMTiming* timing,
VCMReceiveStatisticsCallback* stats_proxy,
rtc::TaskQueue* decode_queue,
FrameSchedulingReceiver* receiver,
TimeDelta max_wait_for_keyframe,
TimeDelta max_wait_for_frame)
: max_wait_for_keyframe_(max_wait_for_keyframe),
max_wait_for_frame_(max_wait_for_frame),
frame_buffer_(clock, timing, stats_proxy),
decode_queue_(decode_queue),
stats_proxy_(stats_proxy),
receiver_(receiver) {
RTC_DCHECK(decode_queue_);
RTC_DCHECK(stats_proxy_);
RTC_DCHECK(receiver_);
}
void StopOnWorker() override {
RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
decode_queue_->PostTask([this] {
frame_buffer_.Stop();
decode_safety_->SetNotAlive();
});
}
void SetProtectionMode(VCMVideoProtection protection_mode) override {
RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
frame_buffer_.SetProtectionMode(kProtectionNackFEC);
}
void Clear() override {
RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
frame_buffer_.Clear();
}
absl::optional<int64_t> InsertFrame(
std::unique_ptr<EncodedFrame> frame) override {
RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
int64_t last_continuous_pid = frame_buffer_.InsertFrame(std::move(frame));
if (last_continuous_pid != -1)
return last_continuous_pid;
return absl::nullopt;
}
void UpdateRtt(int64_t max_rtt_ms) override {
RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
frame_buffer_.UpdateRtt(max_rtt_ms);
}
void StartNextDecode(bool keyframe_required) override {
if (!decode_queue_->IsCurrent()) {
decode_queue_->PostTask(ToQueuedTask(
decode_safety_,
[this, keyframe_required] { StartNextDecode(keyframe_required); }));
return;
}
RTC_DCHECK_RUN_ON(decode_queue_);
frame_buffer_.NextFrame(
MaxWait(keyframe_required).ms(), keyframe_required, decode_queue_,
/* encoded frame handler */
[this, keyframe_required](std::unique_ptr<EncodedFrame> frame) {
RTC_DCHECK_RUN_ON(decode_queue_);
if (!decode_safety_->alive())
return;
if (frame) {
receiver_->OnEncodedFrame(std::move(frame));
} else {
receiver_->OnDecodableFrameTimeout(MaxWait(keyframe_required));
}
});
}
int Size() override {
RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
return frame_buffer_.Size();
}
private:
TimeDelta MaxWait(bool keyframe_required) const {
return keyframe_required ? max_wait_for_keyframe_ : max_wait_for_frame_;
}
RTC_NO_UNIQUE_ADDRESS SequenceChecker worker_sequence_checker_;
const TimeDelta max_wait_for_keyframe_;
const TimeDelta max_wait_for_frame_;
video_coding::FrameBuffer frame_buffer_;
rtc::TaskQueue* const decode_queue_;
VCMReceiveStatisticsCallback* const stats_proxy_;
FrameSchedulingReceiver* const receiver_;
rtc::scoped_refptr<PendingTaskSafetyFlag> decode_safety_ =
PendingTaskSafetyFlag::CreateDetached();
};
// TODO(bugs.webrtc.org/13343): Create FrameBuffer3Proxy when complete.
std::unique_ptr<FrameBufferProxy> FrameBufferProxy::CreateFromFieldTrial(
Clock* clock,
TaskQueueBase* worker_queue,
VCMTiming* timing,
VCMReceiveStatisticsCallback* stats_proxy,
rtc::TaskQueue* decode_queue,
FrameSchedulingReceiver* receiver,
TimeDelta max_wait_for_keyframe,
TimeDelta max_wait_for_frame) {
return std::make_unique<FrameBuffer2Proxy>(
clock, timing, stats_proxy, decode_queue, receiver, max_wait_for_keyframe,
max_wait_for_frame);
}
} // namespace webrtc

64
video/frame_buffer_proxy.h Normal file
View File

@ -0,0 +1,64 @@
/*
* Copyright (c) 2022 The WebRTC project authors. All Rights Reserved.
*
* Use of this source code is governed by a BSD-style license
* that can be found in the LICENSE file in the root of the source
* tree. An additional intellectual property rights grant can be found
* in the file PATENTS. All contributing project authors may
* be found in the AUTHORS file in the root of the source tree.
*/
#ifndef VIDEO_FRAME_BUFFER_PROXY_H_
#define VIDEO_FRAME_BUFFER_PROXY_H_
#include <memory>
#include "api/task_queue/task_queue_base.h"
#include "api/video/encoded_frame.h"
#include "modules/video_coding/include/video_coding_defines.h"
#include "modules/video_coding/timing.h"
#include "rtc_base/task_queue.h"
#include "system_wrappers/include/clock.h"
namespace webrtc {
class FrameSchedulingReceiver {
public:
virtual ~FrameSchedulingReceiver() = default;
virtual void OnEncodedFrame(std::unique_ptr<EncodedFrame> frame) = 0;
virtual void OnDecodableFrameTimeout(TimeDelta wait_time) = 0;
};
// Temporary class to enable replacement of frame_buffer2 with frame_buffer3.
// Once frame_buffer3 has shown to work with a field trial, frame_buffer2 will
// be removed and this class should be directly integrated into
// video_receive_stream2. bugs.webrtc.org/13343 tracks this integration.
class FrameBufferProxy {
public:
static std::unique_ptr<FrameBufferProxy> CreateFromFieldTrial(
Clock* clock,
TaskQueueBase* worker_queue,
VCMTiming* timing,
VCMReceiveStatisticsCallback* stats_proxy,
rtc::TaskQueue* decode_queue,
FrameSchedulingReceiver* receiver,
TimeDelta max_wait_for_keyframe,
TimeDelta max_wait_for_frame);
virtual ~FrameBufferProxy() = default;
// Run on the worker thread.
virtual void StopOnWorker() = 0;
virtual void SetProtectionMode(VCMVideoProtection protection_mode) = 0;
virtual void Clear() = 0;
virtual absl::optional<int64_t> InsertFrame(
std::unique_ptr<EncodedFrame> frame) = 0;
virtual void UpdateRtt(int64_t max_rtt_ms) = 0;
virtual int Size() = 0;
// Run on either the worker thread or the decode thread.
virtual void StartNextDecode(bool keyframe_required) = 0;
};
} // namespace webrtc
#endif // VIDEO_FRAME_BUFFER_PROXY_H_

638
video/frame_buffer_proxy_unittest.cc Normal file
View File

@ -0,0 +1,638 @@
/*
* Copyright (c) 2022 The WebRTC project authors. All Rights Reserved.
*
* Use of this source code is governed by a BSD-style license
* that can be found in the LICENSE file in the root of the source
* tree. An additional intellectual property rights grant can be found
* in the file PATENTS. All contributing project authors may
* be found in the AUTHORS file in the root of the source tree.
*/
#include "video/frame_buffer_proxy.h"
#include <stdint.h>
#include <memory>
#include <string>
#include <utility>
#include <vector>
#include "api/units/time_delta.h"
#include "api/units/timestamp.h"
#include "api/video/video_content_type.h"
#include "rtc_base/event.h"
#include "test/field_trial.h"
#include "test/gmock.h"
#include "test/gtest.h"
#include "test/run_loop.h"
#include "test/time_controller/simulated_time_controller.h"
using ::testing::AllOf;
using ::testing::Contains;
using ::testing::Each;
using ::testing::Eq;
using ::testing::IsEmpty;
using ::testing::Matches;
using ::testing::Ne;
using ::testing::Not;
using ::testing::Optional;
using ::testing::Pointee;
using ::testing::SizeIs;
namespace webrtc {
// For test printing.
void PrintTo(const EncodedFrame& frame, std::ostream* os) {
*os << "EncodedFrame with id=" << frame.Id() << " rtp=" << frame.Timestamp()
<< " size=" << frame.size() << " refs=[";
for (size_t ref = 0; ref < frame.num_references; ++ref) {
*os << frame.references[ref] << ",";
}
*os << "]";
}
namespace {
constexpr size_t kFrameSize = 10;
constexpr uint32_t kFps30Rtp = 90000 / 30;
constexpr TimeDelta kFps30Delay = 1 / Frequency::Hertz(30);
const VideoPlayoutDelay kZeroPlayoutDelay = {0, 0};
constexpr Timestamp kClockStart = Timestamp::Millis(1000);
class FakeEncodedFrame : public EncodedFrame {
public:
// Always 10ms delay and on time.
int64_t ReceivedTime() const override {
if (Timestamp() == 0)
return kClockStart.ms();
return TimeDelta::Seconds(Timestamp() / 90000.0).ms() + kClockStart.ms();
}
int64_t RenderTime() const override { return _renderTimeMs; }
};
MATCHER_P(FrameWithId, id, "") {
return Matches(Eq(id))(arg.Id());
}
MATCHER_P(FrameWithSize, id, "") {
return Matches(Eq(id))(arg.size());
}
class Builder {
public:
Builder& Time(uint32_t rtp_timestamp) {
rtp_timestamp_ = rtp_timestamp;
return *this;
}
Builder& Id(int64_t frame_id) {
frame_id_ = frame_id;
return *this;
}
Builder& AsLast() {
last_spatial_layer_ = true;
return *this;
}
Builder& Refs(const std::vector<int64_t>& references) {
references_ = references;
return *this;
}
Builder& PlayoutDelay(VideoPlayoutDelay playout_delay) {
playout_delay_ = playout_delay;
return *this;
}
Builder& SpatialLayer(int spatial_layer) {
spatial_layer_ = spatial_layer;
return *this;
}
std::unique_ptr<FakeEncodedFrame> Build() {
RTC_CHECK_LE(references_.size(), EncodedFrame::kMaxFrameReferences);
RTC_CHECK(rtp_timestamp_);
RTC_CHECK(frame_id_);
auto frame = std::make_unique<FakeEncodedFrame>();
frame->SetTimestamp(*rtp_timestamp_);
frame->SetId(*frame_id_);
frame->is_last_spatial_layer = last_spatial_layer_;
frame->SetEncodedData(EncodedImageBuffer::Create(kFrameSize));
if (playout_delay_)
frame->SetPlayoutDelay(*playout_delay_);
for (int64_t ref : references_) {
frame->references[frame->num_references] = ref;
frame->num_references++;
}
if (spatial_layer_) {
frame->SetSpatialIndex(spatial_layer_);
}
return frame;
}
private:
absl::optional<uint32_t> rtp_timestamp_;
absl::optional<int64_t> frame_id_;
absl::optional<VideoPlayoutDelay> playout_delay_;
absl::optional<int> spatial_layer_;
bool last_spatial_layer_ = false;
std::vector<int64_t> references_;
};
class VCMReceiveStatisticsCallbackMock : public VCMReceiveStatisticsCallback {
public:
MOCK_METHOD(void,
OnCompleteFrame,
(bool is_keyframe,
size_t size_bytes,
VideoContentType content_type),
(override));
MOCK_METHOD(void, OnDroppedFrames, (uint32_t num_dropped), (override));
MOCK_METHOD(void,
OnFrameBufferTimingsUpdated,
(int max_decode_ms,
int current_delay_ms,
int target_delay_ms,
int jitter_buffer_ms,
int min_playout_delay_ms,
int render_delay_ms),
(override));
MOCK_METHOD(void,
OnTimingFrameInfoUpdated,
(const TimingFrameInfo& info),
(override));
};
} // namespace
constexpr auto kMaxWaitForKeyframe = TimeDelta::Millis(500);
constexpr auto kMaxWaitForFrame = TimeDelta::Millis(1500);
class FrameBufferProxyTest : public ::testing::TestWithParam<std::string>,
public FrameSchedulingReceiver {
public:
FrameBufferProxyTest()
: field_trials_(GetParam()),
time_controller_(kClockStart),
clock_(time_controller_.GetClock()),
decode_queue_(time_controller_.GetTaskQueueFactory()->CreateTaskQueue(
"decode_queue",
TaskQueueFactory::Priority::NORMAL)),
timing_(clock_),
proxy_(FrameBufferProxy::CreateFromFieldTrial(clock_,
run_loop_.task_queue(),
&timing_,
&stats_callback_,
&decode_queue_,
this,
kMaxWaitForKeyframe,
kMaxWaitForFrame)) {
// Avoid starting with negative render times.
timing_.set_min_playout_delay(10);
ON_CALL(stats_callback_, OnDroppedFrames)
.WillByDefault(
[this](auto num_dropped) { dropped_frames_ += num_dropped; });
}
~FrameBufferProxyTest() override {
if (proxy_) {
proxy_->StopOnWorker();
}
}
void OnEncodedFrame(std::unique_ptr<EncodedFrame> frame) override {
last_frame_ = std::move(frame);
run_loop_.Quit();
}
void OnDecodableFrameTimeout(TimeDelta wait_time) override {
timeouts_++;
run_loop_.Quit();
}
bool WaitForFrameOrTimeout(TimeDelta wait) {
if (NewFrameOrTimeout()) {
return true;
}
run_loop_.PostTask([&] { time_controller_.AdvanceTime(wait); });
run_loop_.PostTask([&] {
// If run loop posted to a task queue, flush that.
time_controller_.AdvanceTime(TimeDelta::Zero());
run_loop_.PostTask([&] {
time_controller_.AdvanceTime(TimeDelta::Zero());
run_loop_.Quit();
});
});
run_loop_.Run();
return NewFrameOrTimeout();
}
void StartNextDecode() {
ResetLastResult();
proxy_->StartNextDecode(false);
time_controller_.AdvanceTime(TimeDelta::Zero());
}
void StartNextDecodeForceKeyframe() {
ResetLastResult();
proxy_->StartNextDecode(true);
time_controller_.AdvanceTime(TimeDelta::Zero());
}
void ResetLastResult() {
last_frame_.reset();
last_timeouts_ = timeouts_;
}
int timeouts() const { return timeouts_; }
EncodedFrame* last_frame() const { return last_frame_.get(); }
int dropped_frames() const { return dropped_frames_; }
protected:
test::ScopedFieldTrials field_trials_;
GlobalSimulatedTimeController time_controller_;
Clock* const clock_;
test::RunLoop run_loop_;
rtc::TaskQueue decode_queue_;
VCMTiming timing_;
::testing::NiceMock<VCMReceiveStatisticsCallbackMock> stats_callback_;
std::unique_ptr<FrameBufferProxy> proxy_;
private:
bool NewFrameOrTimeout() const {
return last_frame_ || timeouts_ != last_timeouts_;
}
int timeouts_ = 0;
int last_timeouts_ = 0;
std::unique_ptr<EncodedFrame> last_frame_;
uint32_t dropped_frames_ = 0;
};
TEST_P(FrameBufferProxyTest, InitialTimeoutAfterKeyframeTimeoutPeriod) {
StartNextDecodeForceKeyframe();
// No frame insterted. Timeout expected.
EXPECT_TRUE(WaitForFrameOrTimeout(kMaxWaitForKeyframe));
EXPECT_EQ(timeouts(), 1);
// No new timeout set since receiver has not started new decode.
ResetLastResult();
EXPECT_FALSE(WaitForFrameOrTimeout(kMaxWaitForKeyframe));
EXPECT_EQ(timeouts(), 1);
// Now that receiver has asked for new frame, a new timeout can occur.
StartNextDecodeForceKeyframe();
EXPECT_TRUE(WaitForFrameOrTimeout(kMaxWaitForKeyframe));
EXPECT_EQ(timeouts(), 2);
}
TEST_P(FrameBufferProxyTest, KeyFramesAreScheduled) {
StartNextDecodeForceKeyframe();
time_controller_.AdvanceTime(TimeDelta::Millis(50));
auto frame = Builder().Id(0).Time(0).AsLast().Build();
proxy_->InsertFrame(std::move(frame));
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Zero()));
ASSERT_THAT(last_frame(), Pointee(FrameWithId(0)));
EXPECT_EQ(timeouts(), 0);
}
TEST_P(FrameBufferProxyTest, DeltaFrameTimeoutAfterKeyframeExtracted) {
StartNextDecodeForceKeyframe();
time_controller_.AdvanceTime(TimeDelta::Millis(50));
auto frame = Builder().Id(0).Time(0).AsLast().Build();
proxy_->InsertFrame(std::move(frame));
EXPECT_TRUE(WaitForFrameOrTimeout(kMaxWaitForKeyframe));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
StartNextDecode();
time_controller_.AdvanceTime(TimeDelta::Millis(50));
// Timeouts should now happen at the normal frequency.
const int expected_timeouts = 5;
for (int i = 0; i < expected_timeouts; ++i) {
EXPECT_TRUE(WaitForFrameOrTimeout(kMaxWaitForFrame));
StartNextDecode();
}
EXPECT_EQ(timeouts(), expected_timeouts);
}
TEST_P(FrameBufferProxyTest, DependantFramesAreScheduled) {
StartNextDecodeForceKeyframe();
proxy_->InsertFrame(Builder().Id(0).Time(0).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Zero()));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
StartNextDecode();
time_controller_.AdvanceTime(kFps30Delay);
proxy_->InsertFrame(
Builder().Id(1).Time(kFps30Rtp).AsLast().Refs({0}).Build());
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(1)));
EXPECT_EQ(timeouts(), 0);
}
TEST_P(FrameBufferProxyTest, SpatialLayersAreScheduled) {
StartNextDecodeForceKeyframe();
proxy_->InsertFrame(Builder().Id(0).SpatialLayer(0).Time(0).Build());
proxy_->InsertFrame(Builder().Id(1).SpatialLayer(1).Time(0).Build());
proxy_->InsertFrame(Builder().Id(2).SpatialLayer(2).Time(0).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Zero()));
EXPECT_THAT(last_frame(),
Pointee(AllOf(FrameWithId(0), FrameWithSize(3 * kFrameSize))));
proxy_->InsertFrame(Builder().Id(3).Time(kFps30Rtp).SpatialLayer(0).Build());
proxy_->InsertFrame(Builder().Id(4).Time(kFps30Rtp).SpatialLayer(1).Build());
proxy_->InsertFrame(
Builder().Id(5).Time(kFps30Rtp).SpatialLayer(2).AsLast().Build());
StartNextDecode();
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay * 10));
EXPECT_THAT(last_frame(),
Pointee(AllOf(FrameWithId(3), FrameWithSize(3 * kFrameSize))));
EXPECT_EQ(timeouts(), 0);
}
TEST_P(FrameBufferProxyTest, OutstandingFrameTasksAreCancelledAfterDeletion) {
StartNextDecodeForceKeyframe();
proxy_->InsertFrame(Builder().Id(0).Time(0).AsLast().Build());
// Get keyframe. Delta frame should now be scheduled.
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Zero()));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
StartNextDecode();
proxy_->InsertFrame(
Builder().Id(1).Time(kFps30Rtp).AsLast().Refs({0}).Build());
proxy_->StopOnWorker();
// Wait for 2x max wait time. Since we stopped, this should cause no timeouts
// or frame-ready callbacks.
EXPECT_FALSE(WaitForFrameOrTimeout(kMaxWaitForFrame * 2));
EXPECT_EQ(timeouts(), 0);
}
TEST_P(FrameBufferProxyTest, FramesWaitForDecoderToComplete) {
StartNextDecodeForceKeyframe();
// Start with a keyframe.
proxy_->InsertFrame(Builder().Id(0).Time(0).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Zero()));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
ResetLastResult();
// Insert a delta frame.
proxy_->InsertFrame(
Builder().Id(1).Time(kFps30Rtp).AsLast().Refs({0}).Build());
// Advancing time should not result in a frame since the scheduler has not
// been signalled that we are ready.
EXPECT_FALSE(WaitForFrameOrTimeout(kFps30Delay));
// Signal ready.
StartNextDecode();
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(1)));
}
TEST_P(FrameBufferProxyTest, LateFrameDropped) {
StartNextDecodeForceKeyframe();
// F1
// /
// F0 --> F2
proxy_->InsertFrame(Builder().Id(0).Time(0).AsLast().Build());
// Start with a keyframe.
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Zero()));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
StartNextDecode();
// Simulate late F1 which arrives after F2.
time_controller_.AdvanceTime(kFps30Delay * 2);
proxy_->InsertFrame(
Builder().Id(2).Time(2 * kFps30Rtp).AsLast().Refs({0}).Build());
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(2)));
StartNextDecode();
proxy_->InsertFrame(
Builder().Id(1).Time(1 * kFps30Rtp).AsLast().Refs({0}).Build());
EXPECT_TRUE(WaitForFrameOrTimeout(kMaxWaitForFrame));
// Confirm frame 1 is never scheduled by timing out.
EXPECT_EQ(timeouts(), 1);
}
TEST_P(FrameBufferProxyTest, FramesFastForwardOnSystemHalt) {
StartNextDecodeForceKeyframe();
// F1
// /
// F0 --> F2
proxy_->InsertFrame(Builder().Id(0).Time(0).AsLast().Build());
// Start with a keyframe.
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Zero()));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
time_controller_.AdvanceTime(kFps30Delay);
proxy_->InsertFrame(
Builder().Id(1).Time(kFps30Rtp).AsLast().Refs({0}).Build());
time_controller_.AdvanceTime(kFps30Delay);
proxy_->InsertFrame(
Builder().Id(2).Time(2 * kFps30Rtp).AsLast().Refs({0}).Build());
// Halting time should result in F1 being skipped.
time_controller_.AdvanceTime(kFps30Delay * 2);
StartNextDecode();
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(2)));
EXPECT_EQ(dropped_frames(), 1);
}
TEST_P(FrameBufferProxyTest, ForceKeyFrame) {
StartNextDecodeForceKeyframe();
// Initial keyframe.
proxy_->InsertFrame(Builder().Id(0).Time(0).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Zero()));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
StartNextDecodeForceKeyframe();
// F2 is the next keyframe, and should be extracted since a keyframe was
// forced.
proxy_->InsertFrame(
Builder().Id(1).Time(kFps30Rtp).AsLast().Refs({0}).Build());
proxy_->InsertFrame(Builder().Id(2).Time(kFps30Rtp * 2).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(kMaxWaitForFrame));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(2)));
}
TEST_P(FrameBufferProxyTest, SlowDecoderDropsTemporalLayers) {
StartNextDecodeForceKeyframe();
// 2 temporal layers, at 15fps per layer to make 30fps total.
// Decoder is slower than 30fps, so last_frame() will be skipped.
// F1 --> F3 --> F5
// / / /
// F0 --> F2 --> F4
proxy_->InsertFrame(Builder().Id(0).Time(0).AsLast().Build());
// Keyframe received.
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Zero()));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
// Don't start next decode until slow delay.
time_controller_.AdvanceTime(kFps30Delay);
proxy_->InsertFrame(
Builder().Id(1).Time(1 * kFps30Rtp).Refs({0}).AsLast().Build());
time_controller_.AdvanceTime(kFps30Delay);
proxy_->InsertFrame(
Builder().Id(2).Time(2 * kFps30Rtp).Refs({0}).AsLast().Build());
// Simulate decode taking 3x FPS rate.
time_controller_.AdvanceTime(kFps30Delay * 1.5);
StartNextDecode();
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay * 2));
// F2 is the best frame since decoding was so slow that F1 is too old.
EXPECT_THAT(last_frame(), Pointee(FrameWithId(2)));
EXPECT_EQ(dropped_frames(), 1);
time_controller_.AdvanceTime(kFps30Delay / 2);
proxy_->InsertFrame(
Builder().Id(3).Time(3 * kFps30Rtp).Refs({1, 2}).AsLast().Build());
time_controller_.AdvanceTime(kFps30Delay / 2);
proxy_->InsertFrame(
Builder().Id(4).Time(4 * kFps30Rtp).Refs({2}).AsLast().Build());
time_controller_.AdvanceTime(kFps30Delay / 2);
// F4 is the best frame since decoding was so slow that F1 is too old.
time_controller_.AdvanceTime(kFps30Delay);
StartNextDecode();
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(4)));
proxy_->InsertFrame(
Builder().Id(5).Time(5 * kFps30Rtp).Refs({3, 4}).AsLast().Build());
time_controller_.AdvanceTime(kFps30Delay / 2);
// F5 is not decodable since F4 was decoded, so a timeout is expected.
time_controller_.AdvanceTime(TimeDelta::Millis(10));
StartNextDecode();
EXPECT_TRUE(WaitForFrameOrTimeout(kMaxWaitForFrame));
EXPECT_EQ(timeouts(), 1);
// TODO(bugs.webrtc.org/13343): This should be 2 dropped frames since frames 1
// and 3 were dropped. However, frame_buffer2 does not mark frame 3 as dropped
// which is a bug. Uncomment below when that is fixed for frame_buffer2 is
// deleted.
// EXPECT_EQ(dropped_frames(), 2);
}
TEST_P(FrameBufferProxyTest, OldTimestampNotDecodable) {
StartNextDecodeForceKeyframe();
proxy_->InsertFrame(Builder().Id(0).Time(kFps30Rtp).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
// Timestamp is before previous frame's.
proxy_->InsertFrame(Builder().Id(1).Time(0).AsLast().Build());
StartNextDecode();
// F1 should be dropped since its timestamp went backwards.
EXPECT_TRUE(WaitForFrameOrTimeout(kMaxWaitForFrame));
EXPECT_EQ(timeouts(), 1);
}
TEST_P(FrameBufferProxyTest, NewFrameInsertedWhileWaitingToReleaseFrame) {
StartNextDecodeForceKeyframe();
// Initial keyframe.
proxy_->InsertFrame(Builder().Id(0).Time(0).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Zero()));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
time_controller_.AdvanceTime(kFps30Delay);
proxy_->InsertFrame(
Builder().Id(1).Time(kFps30Rtp).Refs({0}).AsLast().Build());
StartNextDecode();
EXPECT_FALSE(WaitForFrameOrTimeout(TimeDelta::Zero()));
// Scheduler is waiting to deliver Frame 1 now. Insert Frame 2. Frame 1 should
// be delivered still.
proxy_->InsertFrame(
Builder().Id(2).Time(kFps30Rtp * 2).Refs({0}).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(1)));
}
TEST_P(FrameBufferProxyTest, SameFrameNotScheduledTwice) {
// A frame could be scheduled twice if last_frame() arrive out-of-order but
// the older frame is old enough to be fast forwarded.
//
// 1. F2 arrives and is scheduled.
// 2. F3 arrives, but scheduling will not change since F2 is next.
// 3. F1 arrives late and scheduling is checked since it is before F2. F1
// fast-forwarded since it is older.
//
// F2 is the best frame, but should only be scheduled once, followed by F3.
StartNextDecodeForceKeyframe();
// First keyframe.
proxy_->InsertFrame(Builder().Id(0).Time(0).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Millis(15)));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
StartNextDecode();
// Warmup VCMTiming for 30fps.
for (int i = 1; i <= 30; ++i) {
proxy_->InsertFrame(Builder().Id(i).Time(i * kFps30Rtp).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(i)));
StartNextDecode();
}
// F2 arrives and is scheduled.
proxy_->InsertFrame(Builder().Id(32).Time(32 * kFps30Rtp).AsLast().Build());
// F3 arrives before F2 is extracted.
time_controller_.AdvanceTime(kFps30Delay);
proxy_->InsertFrame(Builder().Id(33).Time(33 * kFps30Rtp).AsLast().Build());
// F1 arrives and is fast-forwarded since it is too late.
// F2 is already scheduled and should not be rescheduled.
time_controller_.AdvanceTime(kFps30Delay / 2);
proxy_->InsertFrame(Builder().Id(31).Time(31 * kFps30Rtp).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(32)));
StartNextDecode();
EXPECT_TRUE(WaitForFrameOrTimeout(kFps30Delay));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(33)));
EXPECT_EQ(dropped_frames(), 1);
}
TEST_P(FrameBufferProxyTest, TestStatsCallback) {
EXPECT_CALL(stats_callback_,
OnCompleteFrame(true, kFrameSize, VideoContentType::UNSPECIFIED));
EXPECT_CALL(stats_callback_, OnFrameBufferTimingsUpdated);
// Fake timing having received decoded frame.
timing_.StopDecodeTimer(clock_->TimeInMicroseconds() + 1,
clock_->TimeInMilliseconds());
StartNextDecodeForceKeyframe();
proxy_->InsertFrame(Builder().Id(0).Time(0).AsLast().Build());
EXPECT_TRUE(WaitForFrameOrTimeout(TimeDelta::Zero()));
EXPECT_THAT(last_frame(), Pointee(FrameWithId(0)));
// Flush stats posted on the decode queue.
time_controller_.AdvanceTime(TimeDelta::Zero());
}
INSTANTIATE_TEST_SUITE_P(FrameBufferProxy,
FrameBufferProxyTest,
::testing::Values("WebRTC-FrameBuffer3/Disabled/"));
} // namespace webrtc

77
video/video_receive_stream2.cc
View File

@ -23,6 +23,9 @@
#include "absl/types/optional.h"
#include "api/array_view.h"
#include "api/crypto/frame_decryptor_interface.h"
#include "api/scoped_refptr.h"
#include "api/sequence_checker.h"
#include "api/units/time_delta.h"
#include "api/video/encoded_image.h"
#include "api/video_codecs/h264_profile_level_id.h"
#include "api/video_codecs/sdp_video_format.h"
@ -32,6 +35,7 @@
#include "call/rtp_stream_receiver_controller_interface.h"
#include "call/rtx_receive_stream.h"
#include "common_video/include/incoming_video_stream.h"
#include "modules/video_coding/frame_buffer2.h"
#include "modules/video_coding/include/video_codec_interface.h"
#include "modules/video_coding/include/video_coding_defines.h"
#include "modules/video_coding/include/video_error_codes.h"
@ -43,6 +47,9 @@
#include "rtc_base/strings/string_builder.h"
#include "rtc_base/synchronization/mutex.h"
#include "rtc_base/system/thread_registry.h"
#include "rtc_base/task_queue.h"
#include "rtc_base/task_utils/pending_task_safety_flag.h"
#include "rtc_base/task_utils/to_queued_task.h"
#include "rtc_base/time_utils.h"
#include "rtc_base/trace_event.h"
#include "system_wrappers/include/clock.h"
@ -251,8 +258,10 @@ VideoReceiveStream2::VideoReceiveStream2(
timing_->set_render_delay(config_.render_delay_ms);
frame_buffer_.reset(
new video_coding::FrameBuffer(clock_, timing_.get(), &stats_proxy_));
frame_buffer_ = FrameBufferProxy::CreateFromFieldTrial(
clock_, call->worker_thread(), timing_.get(), &stats_proxy_,
&decode_queue_, this, TimeDelta::Millis(max_wait_for_keyframe_ms_),
TimeDelta::Millis(max_wait_for_frame_ms_));
if (config_.rtp.rtx_ssrc) {
rtx_receive_stream_ = std::make_unique<RtxReceiveStream>(
@ -419,10 +428,8 @@ void VideoReceiveStream2::Stop() {
stats_proxy_.OnUniqueFramesCounted(
rtp_video_stream_receiver_.GetUniqueFramesSeen());
decode_queue_.PostTask([this] { frame_buffer_->Stop(); });
frame_buffer_->StopOnWorker();
call_stats_->DeregisterStatsObserver(this);
if (decoder_running_) {
rtc::Event done;
decode_queue_.PostTask([this, &done] {
@ -657,12 +664,12 @@ void VideoReceiveStream2::OnCompleteFrame(std::unique_ptr<EncodedFrame> frame) {
UpdatePlayoutDelays();
}
int64_t last_continuous_pid = frame_buffer_->InsertFrame(std::move(frame));
if (last_continuous_pid != -1) {
auto last_continuous_pid = frame_buffer_->InsertFrame(std::move(frame));
if (last_continuous_pid.has_value()) {
{
// TODO(bugs.webrtc.org/11993): Call on the network thread.
RTC_DCHECK_RUN_ON(&packet_sequence_checker_);
rtp_video_stream_receiver_.FrameContinuous(last_continuous_pid);
rtp_video_stream_receiver_.FrameContinuous(*last_continuous_pid);
}
}
}
@ -715,29 +722,39 @@ int64_t VideoReceiveStream2::GetMaxWaitMs() const {
: max_wait_for_frame_ms_;
}
void VideoReceiveStream2::OnEncodedFrame(std::unique_ptr<EncodedFrame> frame) {
if (!decode_queue_.IsCurrent()) {
decode_queue_.PostTask([this, frame = std::move(frame)]() mutable {
OnEncodedFrame(std::move(frame));
});
return;
}
RTC_DCHECK_RUN_ON(&decode_queue_);
HandleEncodedFrame(std::move(frame));
StartNextDecode();
}
void VideoReceiveStream2::OnDecodableFrameTimeout(TimeDelta wait_time) {
if (!call_->worker_thread()->IsCurrent()) {
call_->worker_thread()->PostTask(ToQueuedTask(
task_safety_,
[this, wait_time] { OnDecodableFrameTimeout(wait_time); }));
return;
}
// TODO(bugs.webrtc.org/11993): PostTask to the network thread.
RTC_DCHECK_RUN_ON(&packet_sequence_checker_);
int64_t now_ms = clock_->TimeInMilliseconds();
HandleFrameBufferTimeout(now_ms, wait_time.ms());
decode_queue_.PostTask([this] {
RTC_DCHECK_RUN_ON(&decode_queue_);
StartNextDecode();
});
}
void VideoReceiveStream2::StartNextDecode() {
// Running on the decode thread.
TRACE_EVENT0("webrtc", "VideoReceiveStream2::StartNextDecode");
frame_buffer_->NextFrame(
GetMaxWaitMs(), keyframe_required_, &decode_queue_,
/* encoded frame handler */
[this](std::unique_ptr<EncodedFrame> frame) {
RTC_DCHECK_RUN_ON(&decode_queue_);
if (decoder_stopped_)
return;
if (frame) {
HandleEncodedFrame(std::move(frame));
} else {
int64_t now_ms = clock_->TimeInMilliseconds();
// TODO(bugs.webrtc.org/11993): PostTask to the network thread.
call_->worker_thread()->PostTask(ToQueuedTask(
task_safety_, [this, now_ms, wait_ms = GetMaxWaitMs()]() {
RTC_DCHECK_RUN_ON(&packet_sequence_checker_);
HandleFrameBufferTimeout(now_ms, wait_ms);
}));
}
StartNextDecode();
});
frame_buffer_->StartNextDecode(keyframe_required_);
}
void VideoReceiveStream2::HandleEncodedFrame(

10
video/video_receive_stream2.h
View File

@ -25,7 +25,6 @@
#include "call/video_receive_stream.h"
#include "modules/rtp_rtcp/include/flexfec_receiver.h"
#include "modules/rtp_rtcp/source/source_tracker.h"
#include "modules/video_coding/frame_buffer2.h"
#include "modules/video_coding/nack_requester.h"
#include "modules/video_coding/video_receiver2.h"
#include "rtc_base/system/no_unique_address.h"
@ -33,6 +32,7 @@
#include "rtc_base/task_utils/pending_task_safety_flag.h"
#include "rtc_base/thread_annotations.h"
#include "system_wrappers/include/clock.h"
#include "video/frame_buffer_proxy.h"
#include "video/receive_statistics_proxy2.h"
#include "video/rtp_streams_synchronizer2.h"
#include "video/rtp_video_stream_receiver2.h"
@ -83,7 +83,8 @@ class VideoReceiveStream2
public NackSender,
public RtpVideoStreamReceiver2::OnCompleteFrameCallback,
public Syncable,
public CallStatsObserver {
public CallStatsObserver,
public FrameSchedulingReceiver {
public:
// The default number of milliseconds to pass before re-requesting a key frame
// to be sent.
@ -184,6 +185,8 @@ class VideoReceiveStream2
void GenerateKeyFrame() override;
private:
void OnEncodedFrame(std::unique_ptr<EncodedFrame> frame) override;
void OnDecodableFrameTimeout(TimeDelta wait_time) override;
void CreateAndRegisterExternalDecoder(const Decoder& decoder);
int64_t GetMaxWaitMs() const RTC_RUN_ON(decode_queue_);
void StartNextDecode() RTC_RUN_ON(decode_queue_);
@ -247,8 +250,7 @@ class VideoReceiveStream2
// moved to the new VideoStreamDecoder.
std::vector<std::unique_ptr<VideoDecoder>> video_decoders_;
// Members for the new jitter buffer experiment.
std::unique_ptr<video_coding::FrameBuffer> frame_buffer_;
std::unique_ptr<FrameBufferProxy> frame_buffer_;
std::unique_ptr<RtpStreamReceiverInterface> media_receiver_
RTC_GUARDED_BY(packet_sequence_checker_);