Move capture scheduling logic from VideoScheduler to CaptureScheduler.

BUG=455818

Review URL: https://codereview.chromium.org/872433005

Cr-Commit-Position: refs/heads/master@{#314864}

remoting/host/capture_scheduler.cc

@@ -8,6 +8,7 @@
 
 #include "base/logging.h"
 #include "base/sys_info.h"
+#include "base/time/default_tick_clock.h"
 #include "base/time/time.h"
 
 namespace {
@@ -24,46 +25,125 @@ const int64 kDefaultMinimumIntervalMs = 33;
 // available while 1 means using 100% of all CPUs available.
 const double kRecordingCpuConsumption = 0.5;
 
+// Maximum number of frames that can be processed simultaneously.
+static const int kMaxPendingFrames = 2;
+
 }  // namespace
 
 namespace remoting {
 
 // We assume that the number of available cores is constant.
-CaptureScheduler::CaptureScheduler()
-    : minimum_interval_(
+CaptureScheduler::CaptureScheduler(const base::Closure& capture_closure)
+    : capture_closure_(capture_closure),
+      tick_clock_(new base::DefaultTickClock()),
+      capture_timer_(new base::Timer(false, false)),
+      minimum_interval_(
           base::TimeDelta::FromMilliseconds(kDefaultMinimumIntervalMs)),
       num_of_processors_(base::SysInfo::NumberOfProcessors()),
       capture_time_(kStatisticsWindow),
-      encode_time_(kStatisticsWindow) {
+      encode_time_(kStatisticsWindow),
+      pending_frames_(0),
+      capture_pending_(false),
+      is_paused_(false) {
   DCHECK(num_of_processors_);
 }
 
 CaptureScheduler::~CaptureScheduler() {
 }
 
-base::TimeDelta CaptureScheduler::NextCaptureDelay() {
-  // Delay by an amount chosen such that if capture and encode times
-  // continue to follow the averages, then we'll consume the target
-  // fraction of CPU across all cores.
-  base::TimeDelta delay = base::TimeDelta::FromMilliseconds(
-      (capture_time_.Average() + encode_time_.Average()) /
-      (kRecordingCpuConsumption * num_of_processors_));
+void CaptureScheduler::Start() {
+  DCHECK(CalledOnValidThread());
+
+  ScheduleNextCapture();
+}
+
+void CaptureScheduler::Pause(bool pause) {
+  DCHECK(CalledOnValidThread());
 
-  if (delay < minimum_interval_)
-    return minimum_interval_;
-  return delay;
+  if (is_paused_ != pause) {
+    is_paused_ = pause;
+
+    if (is_paused_) {
+      capture_timer_->Stop();
+    } else {
+      ScheduleNextCapture();
+    }
+  }
 }
 
-void CaptureScheduler::RecordCaptureTime(base::TimeDelta capture_time) {
-  capture_time_.Record(capture_time.InMilliseconds());
+void CaptureScheduler::OnCaptureCompleted() {
+  DCHECK(CalledOnValidThread());
+
+  capture_pending_ = false;
+  capture_time_.Record(
+      (tick_clock_->NowTicks() - last_capture_started_time_).InMilliseconds());
+
+  ScheduleNextCapture();
+}
+
+void CaptureScheduler::OnFrameSent() {
+  DCHECK(CalledOnValidThread());
+
+  // Decrement the pending capture count.
+  pending_frames_--;
+  DCHECK_GE(pending_frames_, 0);
+
+  ScheduleNextCapture();
 }
 
-void CaptureScheduler::RecordEncodeTime(base::TimeDelta encode_time) {
+void CaptureScheduler::OnFrameEncoded(base::TimeDelta encode_time) {
+  DCHECK(CalledOnValidThread());
+
   encode_time_.Record(encode_time.InMilliseconds());
+  ScheduleNextCapture();
 }
 
+void CaptureScheduler::SetTickClockForTest(
+    scoped_ptr<base::TickClock> tick_clock) {
+  tick_clock_ = tick_clock.Pass();
+}
+void CaptureScheduler::SetTimerForTest(scoped_ptr<base::Timer> timer) {
+  capture_timer_ = timer.Pass();
+}
 void CaptureScheduler::SetNumOfProcessorsForTest(int num_of_processors) {
   num_of_processors_ = num_of_processors;
 }
 
+void CaptureScheduler::ScheduleNextCapture() {
+  DCHECK(CalledOnValidThread());
+
+  if (is_paused_ || pending_frames_ >= kMaxPendingFrames || capture_pending_)
+    return;
+
+  // Delay by an amount chosen such that if capture and encode times
+  // continue to follow the averages, then we'll consume the target
+  // fraction of CPU across all cores.
+  base::TimeDelta delay =
+      std::max(minimum_interval_,
+               base::TimeDelta::FromMilliseconds(
+                   (capture_time_.Average() + encode_time_.Average()) /
+                   (kRecordingCpuConsumption * num_of_processors_)));
+
+  // Account for the time that has passed since the last capture.
+  delay = std::max(base::TimeDelta(), delay - (tick_clock_->NowTicks() -
+                                               last_capture_started_time_));
+
+  capture_timer_->Start(
+      FROM_HERE, delay,
+      base::Bind(&CaptureScheduler::CaptureNextFrame, base::Unretained(this)));
+}
+
+void CaptureScheduler::CaptureNextFrame() {
+  DCHECK(CalledOnValidThread());
+  DCHECK(!is_paused_);
+  DCHECK(!capture_pending_);
+
+  pending_frames_++;
+  DCHECK_LE(pending_frames_, kMaxPendingFrames);
+
+  capture_pending_ = true;
+  last_capture_started_time_ = tick_clock_->NowTicks();
+  capture_closure_.Run();
+}
+
 }  // namespace remoting

remoting/host/capture_scheduler.h

@@ -9,38 +9,84 @@
 #ifndef REMOTING_HOST_CAPTURE_SCHEDULER_H_
 #define REMOTING_HOST_CAPTURE_SCHEDULER_H_
 
+#include "base/callback.h"
+#include "base/threading/non_thread_safe.h"
+#include "base/time/tick_clock.h"
 #include "base/time/time.h"
+#include "base/timer/timer.h"
 #include "remoting/base/running_average.h"
 
 namespace remoting {
 
-class CaptureScheduler {
+// CaptureScheduler is used by the VideoScheduler to schedule frame capturer,
+// taking into account capture delay, encoder delay, network bandwidth, etc.
+class CaptureScheduler : public base::NonThreadSafe {
  public:
-  CaptureScheduler();
+  // |capture_closure| is called every time a new frame needs to be captured.
+  explicit CaptureScheduler(const base::Closure& capture_closure);
   ~CaptureScheduler();
 
-  // Returns the time to wait after initiating a capture before triggering
-  // the next.
-  base::TimeDelta NextCaptureDelay();
+  // Starts the scheduler.
+  void Start();
 
-  // Records time spent on capturing and encoding.
-  void RecordCaptureTime(base::TimeDelta capture_time);
-  void RecordEncodeTime(base::TimeDelta encode_time);
+  // Pauses or unpauses the stream.
+  void Pause(bool pause);
+
+  // Notifies the scheduler that a capture has been completed.
+  void OnCaptureCompleted();
+
+  // Notifies the scheduler that a frame has been encoded.
+  void OnFrameEncoded(base::TimeDelta encode_time);
+
+  // Notifies the scheduler that a frame has been sent.
+  void OnFrameSent();
 
   // Sets minimum interval between frames.
   void set_minimum_interval(base::TimeDelta minimum_interval) {
     minimum_interval_ = minimum_interval;
   }
 
-  // Overrides the number of processors for testing.
+  // Helper functions for tests.
+  void SetTickClockForTest(scoped_ptr<base::TickClock> tick_clock);
+  void SetTimerForTest(scoped_ptr<base::Timer> timer);
   void SetNumOfProcessorsForTest(int num_of_processors);
 
  private:
+  // Schedules |capture_timer_| to call CaptureNextFrame() at appropriate time.
+  // Doesn't do anything if next frame cannot be captured yet (e.g. because
+  // there are too many frames being processed).
+  void ScheduleNextCapture();
+
+  // Called by |capture_timer_|. Calls |capture_closure_| to start capturing a
+  // new frame.
+  void CaptureNextFrame();
+
+  base::Closure capture_closure_;
+
+  scoped_ptr<base::TickClock> tick_clock_;
+
+  // Timer used to schedule CaptureNextFrame().
+  scoped_ptr<base::Timer> capture_timer_;
+
+  // Minimum interval between frames that determines maximum possible framerate.
   base::TimeDelta minimum_interval_;
+
   int num_of_processors_;
+
   RunningAverage capture_time_;
   RunningAverage encode_time_;
 
+  // Total number of pending frames that are being captured, encoded or sent.
+  int pending_frames_;
+
+  // Set to true when capture is pending.
+  bool capture_pending_;
+
+  // Time at which the last capture started. Used to schedule |capture_timer_|.
+  base::TimeTicks last_capture_started_time_;
+
+  bool is_paused_;
+
   DISALLOW_COPY_AND_ASSIGN(CaptureScheduler);
 };
 

remoting/host/capture_scheduler_unittest.cc

@@ -3,6 +3,10 @@
 // found in the LICENSE file.
 
 #include "remoting/host/capture_scheduler.h"
+
+#include "base/message_loop/message_loop.h"
+#include "base/test/simple_test_tick_clock.h"
+#include "base/timer/mock_timer.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace remoting {
@@ -10,7 +14,61 @@ namespace remoting {
 static const int kTestInputs[] = { 100, 50, 30, 20, 10, 30, 60, 80 };
 static const int kMinumumFrameIntervalMs = 50;
 
-TEST(CaptureSchedulerTest, SingleSampleSameTimes) {
+class CaptureSchedulerTest : public testing::Test {
+ public:
+  CaptureSchedulerTest() : capture_called_(false) {}
+
+  void InitScheduler() {
+    scheduler_.reset(new CaptureScheduler(
+        base::Bind(&CaptureSchedulerTest::DoCapture, base::Unretained(this))));
+    scheduler_->set_minimum_interval(
+        base::TimeDelta::FromMilliseconds(kMinumumFrameIntervalMs));
+    tick_clock_ = new base::SimpleTestTickClock();
+    scheduler_->SetTickClockForTest(make_scoped_ptr(tick_clock_));
+    capture_timer_ = new base::MockTimer(false, false);
+    scheduler_->SetTimerForTest(make_scoped_ptr(capture_timer_));
+    scheduler_->Start();
+  }
+
+  void DoCapture() {
+    capture_called_ = true;
+  }
+
+  void CheckCaptureCalled() {
+    EXPECT_TRUE(capture_called_);
+    capture_called_ = false;
+  }
+
+  void SimulateSingleFrameCapture(
+      base::TimeDelta capture_delay,
+      base::TimeDelta encode_delay,
+      base::TimeDelta expected_delay_between_frames) {
+    capture_timer_->Fire();
+    CheckCaptureCalled();
+    tick_clock_->Advance(capture_delay);
+    scheduler_->OnCaptureCompleted();
+    scheduler_->OnFrameEncoded(encode_delay);
+    scheduler_->OnFrameSent();
+
+    EXPECT_TRUE(capture_timer_->IsRunning());
+    EXPECT_EQ(std::max(base::TimeDelta(),
+                       expected_delay_between_frames - capture_delay),
+              capture_timer_->GetCurrentDelay());
+  }
+
+ protected:
+  base::MessageLoop message_loop_;
+
+  scoped_ptr<CaptureScheduler> scheduler_;
+
+  // Owned by |scheduler_|.
+  base::SimpleTestTickClock* tick_clock_;
+  base::MockTimer* capture_timer_;
+
+  bool capture_called_;
+};
+
+TEST_F(CaptureSchedulerTest, SingleSampleSameTimes) {
   const int kTestResults[][arraysize(kTestInputs)] = {
     { 400, 200, 120, 80, 50, 120, 240, 320 }, // One core.
     { 200, 100, 60, 50, 50, 60, 120, 160 },   // Two cores.
@@ -20,21 +78,18 @@ TEST(CaptureSchedulerTest, SingleSampleSameTimes) {
 
   for (size_t i = 0; i < arraysize(kTestResults); ++i) {
     for (size_t j = 0; j < arraysize(kTestInputs); ++j) {
-      CaptureScheduler scheduler;
-      scheduler.SetNumOfProcessorsForTest(1 << i);
-      scheduler.set_minimum_interval(
-          base::TimeDelta::FromMilliseconds(kMinumumFrameIntervalMs));
-      scheduler.RecordCaptureTime(
-          base::TimeDelta::FromMilliseconds(kTestInputs[j]));
-      scheduler.RecordEncodeTime(
-          base::TimeDelta::FromMilliseconds(kTestInputs[j]));
-      EXPECT_EQ(kTestResults[i][j],
-                scheduler.NextCaptureDelay().InMilliseconds()) << i  << " "<< j;
+      InitScheduler();
+      scheduler_->SetNumOfProcessorsForTest(1 << i);
+
+      SimulateSingleFrameCapture(
+          base::TimeDelta::FromMilliseconds(kTestInputs[j]),
+          base::TimeDelta::FromMilliseconds(kTestInputs[j]),
+          base::TimeDelta::FromMilliseconds(kTestResults[i][j]));
     }
   }
 }
 
-TEST(CaptureSchedulerTest, SingleSampleDifferentTimes) {
+TEST_F(CaptureSchedulerTest, SingleSampleDifferentTimes) {
   const int kTestResults[][arraysize(kTestInputs)] = {
     { 360, 220, 120, 60, 60, 120, 220, 360 }, // One core.
     { 180, 110, 60, 50, 50, 60, 110, 180 },   // Two cores.
@@ -44,22 +99,19 @@ TEST(CaptureSchedulerTest, SingleSampleDifferentTimes) {
 
   for (size_t i = 0; i < arraysize(kTestResults); ++i) {
     for (size_t j = 0; j < arraysize(kTestInputs); ++j) {
-      CaptureScheduler scheduler;
-      scheduler.SetNumOfProcessorsForTest(1 << i);
-      scheduler.set_minimum_interval(
-          base::TimeDelta::FromMilliseconds(kMinumumFrameIntervalMs));
-      scheduler.RecordCaptureTime(
-          base::TimeDelta::FromMilliseconds(kTestInputs[j]));
-      scheduler.RecordEncodeTime(
+      InitScheduler();
+      scheduler_->SetNumOfProcessorsForTest(1 << i);
+
+      SimulateSingleFrameCapture(
+          base::TimeDelta::FromMilliseconds(kTestInputs[j]),
           base::TimeDelta::FromMilliseconds(
-              kTestInputs[arraysize(kTestInputs) - 1 - j]));
-      EXPECT_EQ(kTestResults[i][j],
-                scheduler.NextCaptureDelay().InMilliseconds());
+              kTestInputs[arraysize(kTestInputs) - 1 - j]),
+          base::TimeDelta::FromMilliseconds(kTestResults[i][j]));
     }
   }
 }
 
-TEST(CaptureSchedulerTest, RollingAverageDifferentTimes) {
+TEST_F(CaptureSchedulerTest, RollingAverageDifferentTimes) {
   const int kTestResults[][arraysize(kTestInputs)] = {
     { 360, 290, 233, 133, 80, 80, 133, 233 }, // One core.
     { 180, 145, 116, 66, 50, 50, 66, 116 },   // Two cores.
@@ -68,20 +120,36 @@ TEST(CaptureSchedulerTest, RollingAverageDifferentTimes) {
   };
 
   for (size_t i = 0; i < arraysize(kTestResults); ++i) {
-    CaptureScheduler scheduler;
-    scheduler.SetNumOfProcessorsForTest(1 << i);
-    scheduler.set_minimum_interval(
-        base::TimeDelta::FromMilliseconds(kMinumumFrameIntervalMs));
+    InitScheduler();
+    scheduler_->SetNumOfProcessorsForTest(1 << i);
     for (size_t j = 0; j < arraysize(kTestInputs); ++j) {
-      scheduler.RecordCaptureTime(
-          base::TimeDelta::FromMilliseconds(kTestInputs[j]));
-      scheduler.RecordEncodeTime(
+      SimulateSingleFrameCapture(
+          base::TimeDelta::FromMilliseconds(kTestInputs[j]),
           base::TimeDelta::FromMilliseconds(
-              kTestInputs[arraysize(kTestInputs) - 1 - j]));
-      EXPECT_EQ(kTestResults[i][j],
-                scheduler.NextCaptureDelay().InMilliseconds());
+              kTestInputs[arraysize(kTestInputs) - 1 - j]),
+          base::TimeDelta::FromMilliseconds(kTestResults[i][j]));
     }
   }
 }
 
+// Verify that we never have more than 2 pending frames.
+TEST_F(CaptureSchedulerTest, MaximumPendingFrames) {
+  InitScheduler();
+
+  capture_timer_->Fire();
+  CheckCaptureCalled();
+  scheduler_->OnCaptureCompleted();
+
+  capture_timer_->Fire();
+  CheckCaptureCalled();
+  scheduler_->OnCaptureCompleted();
+
+  EXPECT_FALSE(capture_timer_->IsRunning());
+
+  scheduler_->OnFrameEncoded(base::TimeDelta());
+  scheduler_->OnFrameSent();
+
+  EXPECT_TRUE(capture_timer_->IsRunning());
+}
+
 }  // namespace remoting

remoting/host/client_session.cc

@@ -464,12 +464,14 @@ void ClientSession::ResetVideoPipeline() {
       &mouse_clamping_filter_);
 
   // Apply video-control parameters to the new scheduler.
-  video_scheduler_->Pause(pause_video_);
   video_scheduler_->SetLosslessEncode(lossless_video_encode_);
   video_scheduler_->SetLosslessColor(lossless_video_color_);
 
   // Start capturing the screen.
   video_scheduler_->Start();
+
+  // Pause capturing if necessary.
+  video_scheduler_->Pause(pause_video_);
 }
 
 void ClientSession::SetGnubbyAuthHandlerForTesting(

remoting/host/video_scheduler.h

@@ -13,7 +13,6 @@
 #include "base/time/time.h"
 #include "base/timer/timer.h"
 #include "remoting/codec/video_encoder.h"
-#include "remoting/host/capture_scheduler.h"
 #include "remoting/proto/video.pb.h"
 #include "third_party/webrtc/modules/desktop_capture/desktop_capturer.h"
 #include "third_party/webrtc/modules/desktop_capture/mouse_cursor_monitor.h"
@@ -28,6 +27,7 @@ class DesktopCapturer;
 
 namespace remoting {
 
+class CaptureScheduler;
 class CursorShapeInfo;
 
 namespace protocol {
@@ -72,7 +72,8 @@ class VideoStub;
 // of the capture, encode and network processes.  However, it also needs to
 // rate-limit captures to avoid overloading the host system, either by consuming
 // too much CPU, or hogging the host's graphics subsystem.
-
+//
+// TODO(sergeyu): Rename this class to VideoFramePipe.
 class VideoScheduler : public base::RefCountedThreadSafe<VideoScheduler>,
                        public webrtc::DesktopCapturer::Callback,
                        public webrtc::MouseCursorMonitor::Callback {
@@ -94,15 +95,6 @@ class VideoScheduler : public base::RefCountedThreadSafe<VideoScheduler>,
       protocol::CursorShapeStub* cursor_stub,
       protocol::VideoStub* video_stub);
 
-  // webrtc::DesktopCapturer::Callback implementation.
-  webrtc::SharedMemory* CreateSharedMemory(size_t size) override;
-  void OnCaptureCompleted(webrtc::DesktopFrame* frame) override;
-
-  // webrtc::MouseCursorMonitor::Callback implementation.
-  void OnMouseCursor(webrtc::MouseCursor* mouse_cursor) override;
-  void OnMouseCursorPosition(webrtc::MouseCursorMonitor::CursorState state,
-                             const webrtc::DesktopVector& position) override;
-
   // Starts scheduling frame captures.
   void Start();
 
@@ -129,25 +121,39 @@ class VideoScheduler : public base::RefCountedThreadSafe<VideoScheduler>,
 
   // Capturer thread ----------------------------------------------------------
 
+  // TODO(sergeyu): Move all methods that run on the capture thread to a
+  // separate class and make VideoScheduler not ref-counted.
+
+  // webrtc::DesktopCapturer::Callback implementation.
+  webrtc::SharedMemory* CreateSharedMemory(size_t size) override;
+  void OnCaptureCompleted(webrtc::DesktopFrame* frame) override;
+
+  // webrtc::MouseCursorMonitor::Callback implementation.
+  void OnMouseCursor(webrtc::MouseCursor* mouse_cursor) override;
+  void OnMouseCursorPosition(webrtc::MouseCursorMonitor::CursorState state,
+                             const webrtc::DesktopVector& position) override;
+
   // Starts the capturer on the capture thread.
   void StartOnCaptureThread();
 
   // Stops scheduling frame captures on the capture thread.
   void StopOnCaptureThread();
 
-  // Schedules the next call to CaptureNextFrame.
-  void ScheduleNextCapture();
+  // Captures next frame on the capture thread.
+  void CaptureNextFrameOnCaptureThread();
 
-  // Starts the next frame capture, unless there are already too many pending.
-  void CaptureNextFrame();
+  // Network thread -----------------------------------------------------------
 
-  // Called when a frame capture has been encoded & sent to the client.
-  void FrameCaptureCompleted();
+  // Captures a new frame. Called by CaptureScheduler.
+  void CaptureNextFrame();
 
-  // Network thread -----------------------------------------------------------
+  // Encodes and sends |frame|.
+  void EncodeAndSendFrame(scoped_ptr<webrtc::DesktopFrame> frame);
 
-  // Send |packet| to the client, unless we are in the process of stopping.
-  void SendVideoPacket(scoped_ptr<VideoPacket> packet);
+  // Sends encoded frame
+  void SendEncodedFrame(int64 latest_event_timestamp,
+                        base::TimeTicks timestamp,
+                        scoped_ptr<VideoPacket> packet);
 
   // Callback passed to |video_stub_| for the last packet in each frame, to
   // rate-limit frame captures to network throughput.
@@ -162,16 +168,6 @@ class VideoScheduler : public base::RefCountedThreadSafe<VideoScheduler>,
   // Send updated cursor shape to client.
   void SendCursorShape(scoped_ptr<protocol::CursorShapeInfo> cursor_shape);
 
-  // Encoder thread -----------------------------------------------------------
-
-  // Encode a frame, passing generated VideoPackets to SendVideoPacket().
-  void EncodeFrame(scoped_ptr<webrtc::DesktopFrame> frame,
-                   int64 latest_event_timestamp,
-                   base::TimeTicks timestamp);
-
-  void EncodedDataAvailableCallback(int64 latest_event_timestamp,
-                                    scoped_ptr<VideoPacket> packet);
-
   // Task runners used by this class.
   scoped_refptr<base::SingleThreadTaskRunner> capture_task_runner_;
   scoped_refptr<base::SingleThreadTaskRunner> encode_task_runner_;
@@ -191,32 +187,14 @@ class VideoScheduler : public base::RefCountedThreadSafe<VideoScheduler>,
   protocol::CursorShapeStub* cursor_stub_;
   protocol::VideoStub* video_stub_;
 
-  // Timer used to schedule CaptureNextFrame().
-  scoped_ptr<base::OneShotTimer<VideoScheduler> > capture_timer_;
-
   // Timer used to ensure that we send empty keep-alive frames to the client
   // even when the video stream is paused or encoder is busy.
   scoped_ptr<base::DelayTimer<VideoScheduler> > keep_alive_timer_;
 
-  // The number of frames being processed, i.e. frames that we are currently
-  // capturing, encoding or sending. The value is capped at 2 to minimize
-  // latency.
-  int pending_frames_;
-
-  // Set when the capturer is capturing a frame.
-  bool capture_pending_;
-
-  // True if the previous scheduled capture was skipped.
-  bool did_skip_frame_;
-
-  // True if capture of video frames is paused.
-  bool is_paused_;
-
   // Number updated by the caller to trace performance.
   int64 latest_event_timestamp_;
 
-  // An object to schedule capturing.
-  CaptureScheduler scheduler_;
+  scoped_ptr<CaptureScheduler> capture_scheduler_;
 
   DISALLOW_COPY_AND_ASSIGN(VideoScheduler);
 };