Use TaskQueue in IncomingVideoStream

webrtc/common_video/incoming_video_stream.cc

Index: webrtc/common_video/incoming_video_stream.cc
diff --git a/webrtc/common_video/incoming_video_stream.cc b/webrtc/common_video/incoming_video_stream.cc
index c1f61d1cfb87c94b55f57246a4ceaf684f98526d..a9d6983277900d650fe1333f2edbffb0e648d9fe 100644
--- a/webrtc/common_video/incoming_video_stream.cc
+++ b/webrtc/common_video/incoming_video_stream.cc
@@ -10,94 +10,66 @@
 
 #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"
 #include "webrtc/system_wrappers/include/event_wrapper.h"
 
 namespace webrtc {
-namespace {
-const int kEventStartupTimeMs = 10;
-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 {
+    if (stream_->render_buffers_.AddFrame(std::move(frame_)) == 1)

[mflodman, 2017/02/23 15:23:25]

There is no check that this happens on the right thread, 'render_buffers_' have
no thread/queue checks, should we add something here? Dequeue is covered below.

[tommi, 2017/02/23 19:00:08]

Check added.  Dequeue would have caught it, but one more check doesn't hurt.

[mflodman, 2017/02/24 15:29:58]

If render_buffers_ were empty yes, but not otherwise. But I agree that would
probably be enough anyway.

+      stream_->Dequeue();
+    return true;
+  }
+
+  IncomingVideoStream* stream_;
+  VideoFrame frame_;
+};
 
 IncomingVideoStream::IncomingVideoStream(
     int32_t delay_ms,
     rtc::VideoSinkInterface<VideoFrame>* callback)
-    : incoming_render_thread_(&IncomingVideoStreamThreadFun,
-                              this,
-                              "IncomingVideoStreamThread",
-                              rtc::kRealtimePriority),
-      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();
-}
+    : render_buffers_(delay_ms),
+      callback_(callback),
+      incoming_render_queue_("IncomingVideoStream", rtc::TaskQueue::HIGH) {}
 
 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::CritScope csB(&buffer_critsect_);
-  if (render_buffers_->AddFrame(video_frame) == 1) {
-    deliver_buffer_event_->Set();
-  }
-}
-
-// static
-void IncomingVideoStream::IncomingVideoStreamThreadFun(void* obj) {
-  static_cast<IncomingVideoStream*>(obj)->IncomingVideoStreamProcess();
+  RTC_DCHECK(!incoming_render_queue_.IsCurrent());
+  incoming_render_queue_.PostTask(
+      std::unique_ptr<rtc::QueuedTask>(new NewFrameTask(this, video_frame)));
 }
 
-void IncomingVideoStream::IncomingVideoStreamProcess() {
-  RTC_DCHECK_RUN_ON(&render_thread_checker_);
-
-  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);
+void IncomingVideoStream::Dequeue() {
+  RTC_DCHECK(incoming_render_queue_.IsCurrent());
+  rtc::Optional<VideoFrame> frame_to_render = render_buffers_.FrameToRender();
+  if (frame_to_render)
+    callback_->OnFrame(*frame_to_render);
 
-      if (frame_to_render) {
-        external_callback_->OnFrame(*frame_to_render);
-      }
-    } else {
-      RTC_NOTREACHED();
-    }
+  if (render_buffers_.HasPendingFrames()) {
+    uint32_t wait_time = render_buffers_.TimeToNextFrameRelease();
+    incoming_render_queue_.PostDelayedTask([this]() { Dequeue(); }, wait_time);

[mflodman, 2017/02/23 15:23:25]

'wait_time' can in theory be '0, e.g. if 'OnFrame' above takes a long time, but
I assume that is ok. Right? There is no unittest for that case in TaskQueue.

[tommi, 2017/02/23 19:00:08]

That's ok.  Also added test for TaskQueue.

[mflodman, 2017/02/24 15:29:58]

Thanks

   }
 }