VideoReceiveStream: eliminate task post in decode path.
VideoReceiveStream2 unnecessarily posts a decode complete call to
its own queue while already being executed on it. A popular use
case in downstream project has a large amount of decoders
in use so the cost of this is multiplied by the number of active
decoders.
Fix this by removing the unnecessary task post. To allow for this,
this change also changes the frame buffer to call out to it's
handler without any locks held.
Bug: webrtc:12297
Change-Id: Ib2e26445458228a44c53dad9d585d4025f2f2945
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/197420
Reviewed-by: Philip Eliasson <philipel@webrtc.org>
Reviewed-by: Erik Språng <sprang@webrtc.org>
Commit-Queue: Mirko Bonadei <mbonadei@webrtc.org>
Commit-Queue: Markus Handell <handellm@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#32845}
diff --git a/modules/video_coding/frame_buffer2.cc b/modules/video_coding/frame_buffer2.cc
index 5195087..c085557 100644
--- a/modules/video_coding/frame_buffer2.cc
+++ b/modules/video_coding/frame_buffer2.cc
@@ -102,24 +102,28 @@
RTC_DCHECK_RUN_ON(&callback_checker_);
// If this task has not been cancelled, we did not get any new frames
// while waiting. Continue with frame delivery.
- MutexLock lock(&mutex_);
- if (!frames_to_decode_.empty()) {
- // We have frames, deliver!
- frame_handler_(absl::WrapUnique(GetNextFrame()), kFrameFound);
+ std::unique_ptr<EncodedFrame> frame;
+ std::function<void(std::unique_ptr<EncodedFrame>, ReturnReason)>
+ frame_handler;
+ {
+ MutexLock lock(&mutex_);
+ if (!frames_to_decode_.empty()) {
+ // We have frames, deliver!
+ frame = absl::WrapUnique(GetNextFrame());
+ } else if (clock_->TimeInMilliseconds() < latest_return_time_ms_) {
+ // If there's no frames to decode and there is still time left, it
+ // means that the frame buffer was cleared between creation and
+ // execution of this task. Continue waiting for the remaining time.
+ int64_t wait_ms = FindNextFrame(clock_->TimeInMilliseconds());
+ return TimeDelta::Millis(wait_ms);
+ }
+ frame_handler = std::move(frame_handler_);
CancelCallback();
- return TimeDelta::Zero(); // Ignored.
- } else if (clock_->TimeInMilliseconds() >= latest_return_time_ms_) {
- // We have timed out, signal this and stop repeating.
- frame_handler_(nullptr, kTimeout);
- CancelCallback();
- return TimeDelta::Zero(); // Ignored.
- } else {
- // If there's no frames to decode and there is still time left, it
- // means that the frame buffer was cleared between creation and
- // execution of this task. Continue waiting for the remaining time.
- int64_t wait_ms = FindNextFrame(clock_->TimeInMilliseconds());
- return TimeDelta::Millis(wait_ms);
}
+ // Deliver frame, if any. Otherwise signal timeout.
+ ReturnReason reason = frame ? kFrameFound : kTimeout;
+ frame_handler(std::move(frame), reason);
+ return TimeDelta::Zero(); // Ignored.
});
}
diff --git a/video/video_receive_stream.cc b/video/video_receive_stream.cc
index fbd32e4..fbe994f 100644
--- a/video/video_receive_stream.cc
+++ b/video/video_receive_stream.cc
@@ -634,17 +634,15 @@
[this](std::unique_ptr<EncodedFrame> frame, ReturnReason res) {
RTC_DCHECK_EQ(frame == nullptr, res == ReturnReason::kTimeout);
RTC_DCHECK_EQ(frame != nullptr, res == ReturnReason::kFrameFound);
- decode_queue_.PostTask([this, frame = std::move(frame)]() mutable {
- RTC_DCHECK_RUN_ON(&decode_queue_);
- if (decoder_stopped_)
- return;
- if (frame) {
- HandleEncodedFrame(std::move(frame));
- } else {
- HandleFrameBufferTimeout();
- }
- StartNextDecode();
- });
+ RTC_DCHECK_RUN_ON(&decode_queue_);
+ if (decoder_stopped_)
+ return;
+ if (frame) {
+ HandleEncodedFrame(std::move(frame));
+ } else {
+ HandleFrameBufferTimeout();
+ }
+ StartNextDecode();
});
}
diff --git a/video/video_receive_stream2.cc b/video/video_receive_stream2.cc
index 8cc14e5..5431ae8 100644
--- a/video/video_receive_stream2.cc
+++ b/video/video_receive_stream2.cc
@@ -633,22 +633,20 @@
[this](std::unique_ptr<EncodedFrame> frame, ReturnReason res) {
RTC_DCHECK_EQ(frame == nullptr, res == ReturnReason::kTimeout);
RTC_DCHECK_EQ(frame != nullptr, res == ReturnReason::kFrameFound);
- decode_queue_.PostTask([this, frame = std::move(frame)]() mutable {
- RTC_DCHECK_RUN_ON(&decode_queue_);
- if (decoder_stopped_)
- return;
- if (frame) {
- HandleEncodedFrame(std::move(frame));
- } else {
- int64_t now_ms = clock_->TimeInMilliseconds();
- worker_thread_->PostTask(ToQueuedTask(
- task_safety_, [this, now_ms, wait_ms = GetMaxWaitMs()]() {
- RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
- HandleFrameBufferTimeout(now_ms, wait_ms);
- }));
- }
- StartNextDecode();
- });
+ RTC_DCHECK_RUN_ON(&decode_queue_);
+ if (decoder_stopped_)
+ return;
+ if (frame) {
+ HandleEncodedFrame(std::move(frame));
+ } else {
+ int64_t now_ms = clock_->TimeInMilliseconds();
+ worker_thread_->PostTask(ToQueuedTask(
+ task_safety_, [this, now_ms, wait_ms = GetMaxWaitMs()]() {
+ RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
+ HandleFrameBufferTimeout(now_ms, wait_ms);
+ }));
+ }
+ StartNextDecode();
});
}
