Reland of Use TaskQueue in IncomingVideoStream (patchset #1 id:1 of https://codereview.webrtc.org/2714393003/ )

Use TaskQueue in IncomingVideoStream instead of the PlatformThread + event timer approach.

TBR=mflodman@webrtc.org
# Skipping CQ checks because original CL landed less than 1 days ago.
NOPRESUBMIT=true
NOTREECHECKS=true
NOTRY=true
BUG=webrtc:7219, webrtc:7253

Reland of
686aa37382 (revert)
e2d1d64295 (original)

Review-Url: https://codereview.webrtc.org/2720773002
Cr-Commit-Position: refs/heads/master@{#16872}

webrtc/base/task_queue_unittest.cc

@@ -80,6 +80,15 @@ TEST(TaskQueueTest, PostLambda) {
   EXPECT_TRUE(event.Wait(1000));
 }
 
+TEST(TaskQueueTest, PostDelayedZero) {
+  static const char kQueueName[] = "PostDelayedZero";
+  Event event(false, false);
+  TaskQueue queue(kQueueName);
+
+  queue.PostDelayedTask([&event]() { event.Set(); }, 0);
+  EXPECT_TRUE(event.Wait(1000));
+}
+
 TEST(TaskQueueTest, PostFromQueue) {
   static const char kQueueName[] = "PostFromQueue";
   Event event(false, false);

webrtc/common_video/include/incoming_video_stream.h

@@ -11,18 +11,12 @@
 #ifndef WEBRTC_COMMON_VIDEO_INCLUDE_INCOMING_VIDEO_STREAM_H_
 #define WEBRTC_COMMON_VIDEO_INCLUDE_INCOMING_VIDEO_STREAM_H_
 
-#include <memory>
-
-#include "webrtc/base/criticalsection.h"
-#include "webrtc/base/platform_thread.h"
 #include "webrtc/base/race_checker.h"
-#include "webrtc/base/thread_annotations.h"
+#include "webrtc/base/task_queue.h"
-#include "webrtc/base/thread_checker.h"
 #include "webrtc/common_video/video_render_frames.h"
 #include "webrtc/media/base/videosinkinterface.h"
 
 namespace webrtc {
-class EventTimerWrapper;
 
 class IncomingVideoStream : public rtc::VideoSinkInterface<VideoFrame> {
  public:
@@ -30,24 +24,19 @@ class IncomingVideoStream : public rtc::VideoSinkInterface<VideoFrame> {
                       rtc::VideoSinkInterface<VideoFrame>* callback);
   ~IncomingVideoStream() override;
 
- protected:
-  static void IncomingVideoStreamThreadFun(void* obj);
-  void IncomingVideoStreamProcess();
-
  private:
   void OnFrame(const VideoFrame& video_frame) override;
+  void Dequeue();
+
+  // Fwd decl of a QueuedTask implementation for carrying frames over to the TQ.
+  class NewFrameTask;
 
   rtc::ThreadChecker main_thread_checker_;
-  rtc::ThreadChecker render_thread_checker_;
   rtc::RaceChecker decoder_race_checker_;
 
-  rtc::CriticalSection buffer_critsect_;
+  VideoRenderFrames render_buffers_;  // Only touched on the TaskQueue.
-  rtc::PlatformThread incoming_render_thread_;
+  rtc::VideoSinkInterface<VideoFrame>* const callback_;
-  std::unique_ptr<EventTimerWrapper> deliver_buffer_event_;
+  rtc::TaskQueue incoming_render_queue_;
-
-  rtc::VideoSinkInterface<VideoFrame>* const external_callback_;
-  std::unique_ptr<VideoRenderFrames> render_buffers_
-      GUARDED_BY(buffer_critsect_);
 };
 
 }  // namespace webrtc

webrtc/common_video/incoming_video_stream.cc

@@ -10,6 +10,8 @@
 
 #include "webrtc/common_video/include/incoming_video_stream.h"
 
+#include <memory>
+
 #include "webrtc/base/timeutils.h"
 #include "webrtc/common_video/video_render_frames.h"
 #include "webrtc/system_wrappers/include/critical_section_wrapper.h"
@@ -17,87 +19,62 @@
 
 namespace webrtc {
 namespace {
-const int kEventStartupTimeMs = 10;
+const char kIncomingQueueName[] = "IncomingVideoStream";
-const int kEventMaxWaitTimeMs = 100;
+}
-}  // namespace
+
+// Capture by moving (std::move) into a lambda isn't possible in C++11
+// (supported in C++14). This class provides the functionality of what would be
+// something like (inside OnFrame):
+// VideoFrame frame(video_frame);
+// incoming_render_queue_.PostTask([this, frame = std::move(frame)](){
+//   if (render_buffers_.AddFrame(std::move(frame)) == 1)
+//     Dequeue();
+// });
+class IncomingVideoStream::NewFrameTask : public rtc::QueuedTask {
+ public:
+  NewFrameTask(IncomingVideoStream* stream, VideoFrame frame)
+      : stream_(stream), frame_(std::move(frame)) {}
+
+ private:
+  bool Run() override {
+    RTC_DCHECK(rtc::TaskQueue::IsCurrent(kIncomingQueueName));
+    if (stream_->render_buffers_.AddFrame(std::move(frame_)) == 1)
+      stream_->Dequeue();
+    return true;
+  }
+
+  IncomingVideoStream* stream_;
+  VideoFrame frame_;
+};
 
 IncomingVideoStream::IncomingVideoStream(
     int32_t delay_ms,
     rtc::VideoSinkInterface<VideoFrame>* callback)
-    : incoming_render_thread_(&IncomingVideoStreamThreadFun,
+    : render_buffers_(delay_ms),
-                              this,
+      callback_(callback),
-                              "IncomingVideoStreamThread",
+      incoming_render_queue_(kIncomingQueueName,
-                              rtc::kRealtimePriority),
+                             rtc::TaskQueue::Priority::HIGH) {}
-      deliver_buffer_event_(EventTimerWrapper::Create()),
-      external_callback_(callback),
-      render_buffers_(new VideoRenderFrames(delay_ms)) {
-  RTC_DCHECK(external_callback_);
-
-  render_thread_checker_.DetachFromThread();
-
-  deliver_buffer_event_->StartTimer(false, kEventStartupTimeMs);
-  incoming_render_thread_.Start();
-}
 
 IncomingVideoStream::~IncomingVideoStream() {
   RTC_DCHECK(main_thread_checker_.CalledOnValidThread());
-
-  {
-    rtc::CritScope cs(&buffer_critsect_);
-    render_buffers_.reset();
-  }
-
-  deliver_buffer_event_->Set();
-  incoming_render_thread_.Stop();
-  deliver_buffer_event_->StopTimer();
 }
 
 void IncomingVideoStream::OnFrame(const VideoFrame& video_frame) {
   RTC_CHECK_RUNS_SERIALIZED(&decoder_race_checker_);
-  // Hand over or insert frame.
+  RTC_DCHECK(!incoming_render_queue_.IsCurrent());
-  rtc::CritScope csB(&buffer_critsect_);
+  incoming_render_queue_.PostTask(
-  if (render_buffers_->AddFrame(video_frame) == 1) {
+      std::unique_ptr<rtc::QueuedTask>(new NewFrameTask(this, video_frame)));
-    deliver_buffer_event_->Set();
-  }
 }
 
-// static
+void IncomingVideoStream::Dequeue() {
-void IncomingVideoStream::IncomingVideoStreamThreadFun(void* obj) {
+  RTC_DCHECK(incoming_render_queue_.IsCurrent());
-  static_cast<IncomingVideoStream*>(obj)->IncomingVideoStreamProcess();
+  rtc::Optional<VideoFrame> frame_to_render = render_buffers_.FrameToRender();
-}
+  if (frame_to_render)
+    callback_->OnFrame(*frame_to_render);
 
-void IncomingVideoStream::IncomingVideoStreamProcess() {
+  if (render_buffers_.HasPendingFrames()) {
-  RTC_DCHECK_RUN_ON(&render_thread_checker_);
+    uint32_t wait_time = render_buffers_.TimeToNextFrameRelease();
-
+    incoming_render_queue_.PostDelayedTask([this]() { Dequeue(); }, wait_time);
-  while (true) {
-    if (kEventError != deliver_buffer_event_->Wait(kEventMaxWaitTimeMs)) {
-      // Get a new frame to render and the time for the frame after this one.
-      rtc::Optional<VideoFrame> frame_to_render;
-      uint32_t wait_time;
-      {
-        rtc::CritScope cs(&buffer_critsect_);
-        if (!render_buffers_.get()) {
-          // Terminating
-          return;
-        }
-
-        frame_to_render = render_buffers_->FrameToRender();
-        wait_time = render_buffers_->TimeToNextFrameRelease();
-      }
-
-      // Set timer for next frame to render.
-      if (wait_time > kEventMaxWaitTimeMs) {
-        wait_time = kEventMaxWaitTimeMs;
-      }
-
-      deliver_buffer_event_->StartTimer(false, wait_time);
-
-      if (frame_to_render) {
-        external_callback_->OnFrame(*frame_to_render);
-      }
-    } else {
-      RTC_NOTREACHED();
-    }
   }
 }
 

webrtc/common_video/video_render_frames.cc

@@ -10,6 +10,8 @@
 
 #include "webrtc/common_video/video_render_frames.h"
 
+#include <utility>
+
 #include "webrtc/base/logging.h"
 #include "webrtc/base/timeutils.h"
 #include "webrtc/modules/include/module_common_types.h"
@@ -17,6 +19,10 @@
 
 namespace webrtc {
 namespace {
+// Don't render frames with timestamp older than 500ms from now.
+const int kOldRenderTimestampMS = 500;
+// Don't render frames with timestamp more than 10s into the future.
+const int kFutureRenderTimestampMS = 10000;
 
 const uint32_t kEventMaxWaitTimeMs = 200;
 const uint32_t kMinRenderDelayMs = 10;
@@ -33,13 +39,13 @@ uint32_t EnsureValidRenderDelay(uint32_t render_delay) {
 VideoRenderFrames::VideoRenderFrames(uint32_t render_delay_ms)
     : render_delay_ms_(EnsureValidRenderDelay(render_delay_ms)) {}
 
-int32_t VideoRenderFrames::AddFrame(const VideoFrame& new_frame) {
+int32_t VideoRenderFrames::AddFrame(VideoFrame&& new_frame) {
   const int64_t time_now = rtc::TimeMillis();
 
   // Drop old frames only when there are other frames in the queue, otherwise, a
   // really slow system never renders any frames.
   if (!incoming_frames_.empty() &&
-      new_frame.render_time_ms() + KOldRenderTimestampMS < time_now) {
+      new_frame.render_time_ms() + kOldRenderTimestampMS < time_now) {
     WEBRTC_TRACE(kTraceWarning,
                  kTraceVideoRenderer,
                  -1,
@@ -49,14 +55,25 @@ int32_t VideoRenderFrames::AddFrame(const VideoFrame& new_frame) {
     return -1;
   }
 
-  if (new_frame.render_time_ms() > time_now + KFutureRenderTimestampMS) {
+  if (new_frame.render_time_ms() > time_now + kFutureRenderTimestampMS) {
     WEBRTC_TRACE(kTraceWarning, kTraceVideoRenderer, -1,
                  "%s: frame too long into the future, timestamp=%u.",
                  __FUNCTION__, new_frame.timestamp());
     return -1;
   }
 
-  incoming_frames_.push_back(new_frame);
+  if (new_frame.render_time_ms() < last_render_time_ms_) {
+    WEBRTC_TRACE(kTraceWarning, kTraceVideoRenderer, -1,
+                 "%s: frame scheduled out of order, render_time=%u, latest=%u.",
+                 __FUNCTION__, new_frame.render_time_ms(),
+                 last_render_time_ms_);
+    // TODO(mflodman): Decide what to do when this happens.
+    // See bug: https://bugs.chromium.org/p/webrtc/issues/detail?id=7253
+  }
+
+  last_render_time_ms_ = new_frame.render_time_ms();
+  incoming_frames_.emplace_back(std::move(new_frame));
+
   if (incoming_frames_.size() > kMaxIncomingFramesBeforeLogged)
     LOG(LS_WARNING) << "Stored incoming frames: " << incoming_frames_.size();
   return static_cast<int32_t>(incoming_frames_.size());
@@ -66,7 +83,8 @@ rtc::Optional<VideoFrame> VideoRenderFrames::FrameToRender() {
   rtc::Optional<VideoFrame> render_frame;
   // Get the newest frame that can be released for rendering.
   while (!incoming_frames_.empty() && TimeToNextFrameRelease() <= 0) {
-    render_frame = rtc::Optional<VideoFrame>(incoming_frames_.front());
+    render_frame =
+        rtc::Optional<VideoFrame>(std::move(incoming_frames_.front()));
     incoming_frames_.pop_front();
   }
   return render_frame;
@@ -82,4 +100,8 @@ uint32_t VideoRenderFrames::TimeToNextFrameRelease() {
   return time_to_release < 0 ? 0u : static_cast<uint32_t>(time_to_release);
 }
 
+bool VideoRenderFrames::HasPendingFrames() const {
+  return !incoming_frames_.empty();
+}
+
 }  // namespace webrtc

webrtc/common_video/video_render_frames.h

@@ -27,7 +27,7 @@ class VideoRenderFrames {
   VideoRenderFrames(const VideoRenderFrames&) = delete;
 
   // Add a frame to the render queue
-  int32_t AddFrame(const VideoFrame& new_frame);
+  int32_t AddFrame(VideoFrame&& new_frame);
 
   // Get a frame for rendering, or false if it's not time to render.
   rtc::Optional<VideoFrame> FrameToRender();
@@ -35,19 +35,16 @@ class VideoRenderFrames {
   // Returns the number of ms to next frame to render
   uint32_t TimeToNextFrameRelease();
 
- private:
+  bool HasPendingFrames() const;
-  // 10 seconds for 30 fps.
-  enum { KMaxNumberOfFrames = 300 };
-  // Don't render frames with timestamp older than 500ms from now.
-  enum { KOldRenderTimestampMS = 500 };
-  // Don't render frames with timestamp more than 10s into the future.
-  enum { KFutureRenderTimestampMS = 10000 };
 
+ private:
   // Sorted list with framed to be rendered, oldest first.
   std::list<VideoFrame> incoming_frames_;
 
   // Estimated delay from a frame is released until it's rendered.
   const uint32_t render_delay_ms_;
+
+  int64_t last_render_time_ms_ = 0;
 };
 
 }  // namespace webrtc