Move capture scheduling logic from VideoScheduler to CaptureScheduler.

BUG=455818

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

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

6 files changed, 405 insertions, 328 deletions

diff --git a/remoting/host/capture_scheduler.cc b/remoting/host/capture_scheduler.cc
index c6b3be0..ccd141d 100644
--- a/remoting/host/capture_scheduler.cc
+++ b/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
diff --git a/remoting/host/capture_scheduler.h b/remoting/host/capture_scheduler.h
index 9dbecda..5f978da 100644
--- a/remoting/host/capture_scheduler.h
+++ b/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);
 };
 
diff --git a/remoting/host/capture_scheduler_unittest.cc b/remoting/host/capture_scheduler_unittest.cc
index a6ca0e6..b07fc6da 100644
--- a/remoting/host/capture_scheduler_unittest.cc
+++ b/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
diff --git a/remoting/host/client_session.cc b/remoting/host/client_session.cc
index 1b580dc..6fae325 100644
--- a/remoting/host/client_session.cc
+++ b/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(
diff --git a/remoting/host/video_scheduler.cc b/remoting/host/video_scheduler.cc
index e2a3f94..ea57e8a 100644
--- a/remoting/host/video_scheduler.cc
+++ b/remoting/host/video_scheduler.cc
@@ -11,14 +11,13 @@
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/message_loop/message_loop_proxy.h"
-#include "base/stl_util.h"
-#include "base/sys_info.h"
+#include "base/task_runner_util.h"
 #include "base/time/time.h"
+#include "remoting/host/capture_scheduler.h"
 #include "remoting/proto/control.pb.h"
 #include "remoting/proto/internal.pb.h"
 #include "remoting/proto/video.pb.h"
 #include "remoting/protocol/cursor_shape_stub.h"
-#include "remoting/protocol/message_decoder.h"
 #include "remoting/protocol/video_stub.h"
 #include "third_party/webrtc/modules/desktop_capture/desktop_capturer.h"
 #include "third_party/webrtc/modules/desktop_capture/desktop_frame.h"
@@ -26,9 +25,23 @@
 
 namespace remoting {
 
-// Maximum number of frames that can be processed simultaneously.
-// TODO(hclam): Move this value to CaptureScheduler.
-static const int kMaxPendingFrames = 2;
+namespace {
+
+// Helper used to encode frames on the encode thread.
+//
+// TODO(sergeyu): This functions doesn't do much beside calling
+// VideoEncoder::Encode(). It's only needed to handle empty frames properly and
+// that logic can be moved to VideoEncoder implementations.
+scoped_ptr<VideoPacket> EncodeFrame(VideoEncoder* encoder,
+                                    scoped_ptr<webrtc::DesktopFrame> frame) {
+  // If there is nothing to encode then send an empty packet.
+  if (!frame || frame->updated_region().is_empty())
+    return make_scoped_ptr(new VideoPacket());
+
+  return encoder->Encode(*frame);
+}
+
+}  // namespace
 
 // Interval between empty keep-alive frames. These frames are sent only when the
 // stream is paused or inactive for some other reason (e.g. when blocked on
@@ -60,10 +73,6 @@ VideoScheduler::VideoScheduler(
       encoder_(encoder.Pass()),
       cursor_stub_(cursor_stub),
       video_stub_(video_stub),
-      pending_frames_(0),
-      capture_pending_(false),
-      did_skip_frame_(false),
-      is_paused_(false),
       latest_event_timestamp_(0) {
   DCHECK(network_task_runner_->BelongsToCurrentThread());
   DCHECK(capturer_);
@@ -75,78 +84,17 @@ VideoScheduler::VideoScheduler(
 
 // Public methods --------------------------------------------------------------
 
-webrtc::SharedMemory* VideoScheduler::CreateSharedMemory(size_t size) {
-  return nullptr;
-}
-
-void VideoScheduler::OnCaptureCompleted(webrtc::DesktopFrame* frame) {
-  DCHECK(capture_task_runner_->BelongsToCurrentThread());
-
-  capture_pending_ = false;
-
-  scoped_ptr<webrtc::DesktopFrame> owned_frame(frame);
-
-  if (owned_frame) {
-    scheduler_.RecordCaptureTime(
-        base::TimeDelta::FromMilliseconds(owned_frame->capture_time_ms()));
-  }
-
-  // Even when |frame| is nullptr we still need to post it to the encode thread
-  // to make sure frames are freed in the same order they are received and
-  // that we don't start capturing frame n+2 before frame n is freed.
-  encode_task_runner_->PostTask(
-      FROM_HERE, base::Bind(&VideoScheduler::EncodeFrame, this,
-                            base::Passed(&owned_frame), latest_event_timestamp_,
-                            base::TimeTicks::Now()));
-
-  // If a frame was skipped, try to capture it again.
-  if (did_skip_frame_) {
-    capture_task_runner_->PostTask(
-        FROM_HERE, base::Bind(&VideoScheduler::CaptureNextFrame, this));
-  }
-}
-
-void VideoScheduler::OnMouseCursor(webrtc::MouseCursor* cursor) {
-  DCHECK(capture_task_runner_->BelongsToCurrentThread());
-
-  scoped_ptr<webrtc::MouseCursor> owned_cursor(cursor);
-
-  // Do nothing if the scheduler is being stopped.
-  if (!capturer_)
-    return;
-
-  scoped_ptr<protocol::CursorShapeInfo> cursor_proto(
-      new protocol::CursorShapeInfo());
-  cursor_proto->set_width(cursor->image()->size().width());
-  cursor_proto->set_height(cursor->image()->size().height());
-  cursor_proto->set_hotspot_x(cursor->hotspot().x());
-  cursor_proto->set_hotspot_y(cursor->hotspot().y());
-
-  cursor_proto->set_data(std::string());
-  uint8_t* current_row = cursor->image()->data();
-  for (int y = 0; y < cursor->image()->size().height(); ++y) {
-    cursor_proto->mutable_data()->append(
-        current_row,
-        current_row + cursor->image()->size().width() *
-            webrtc::DesktopFrame::kBytesPerPixel);
-    current_row += cursor->image()->stride();
-  }
-
-  network_task_runner_->PostTask(
-      FROM_HERE, base::Bind(&VideoScheduler::SendCursorShape, this,
-                            base::Passed(&cursor_proto)));
-}
-
-void VideoScheduler::OnMouseCursorPosition(
-    webrtc::MouseCursorMonitor::CursorState state,
-    const webrtc::DesktopVector& position) {
-  // We're not subscribing to mouse position changes.
-  NOTREACHED();
-}
-
 void VideoScheduler::Start() {
   DCHECK(network_task_runner_->BelongsToCurrentThread());
 
+  keep_alive_timer_.reset(new base::DelayTimer<VideoScheduler>(
+      FROM_HERE, base::TimeDelta::FromMilliseconds(kKeepAlivePacketIntervalMs),
+      this, &VideoScheduler::SendKeepAlivePacket));
+
+  capture_scheduler_.reset(new CaptureScheduler(
+      base::Bind(&VideoScheduler::CaptureNextFrame, this)));
+  capture_scheduler_->Start();
+
   capture_task_runner_->PostTask(
       FROM_HERE, base::Bind(&VideoScheduler::StartOnCaptureThread, this));
 }
@@ -158,6 +106,7 @@ void VideoScheduler::Stop() {
   cursor_stub_ = nullptr;
   video_stub_ = nullptr;
 
+  capture_scheduler_.reset();
   keep_alive_timer_.reset();
 
   capture_task_runner_->PostTask(
@@ -165,56 +114,31 @@ void VideoScheduler::Stop() {
 }
 
 void VideoScheduler::Pause(bool pause) {
-  if (!capture_task_runner_->BelongsToCurrentThread()) {
-    DCHECK(network_task_runner_->BelongsToCurrentThread());
-    capture_task_runner_->PostTask(
-        FROM_HERE, base::Bind(&VideoScheduler::Pause, this, pause));
-    return;
-  }
-
-  if (is_paused_ != pause) {
-    is_paused_ = pause;
+  DCHECK(network_task_runner_->BelongsToCurrentThread());
 
-    // Restart captures if we're resuming and there are none scheduled.
-    if (!is_paused_ && capture_timer_ && !capture_timer_->IsRunning())
-      CaptureNextFrame();
-  }
+  capture_scheduler_->Pause(pause);
 }
 
 void VideoScheduler::SetLatestEventTimestamp(int64 latest_event_timestamp) {
-  if (!capture_task_runner_->BelongsToCurrentThread()) {
-    DCHECK(network_task_runner_->BelongsToCurrentThread());
-    capture_task_runner_->PostTask(
-        FROM_HERE, base::Bind(&VideoScheduler::SetLatestEventTimestamp,
-                              this, latest_event_timestamp));
-    return;
-  }
+  DCHECK(network_task_runner_->BelongsToCurrentThread());
 
   latest_event_timestamp_ = latest_event_timestamp;
 }
 
 void VideoScheduler::SetLosslessEncode(bool want_lossless) {
-  if (!encode_task_runner_->BelongsToCurrentThread()) {
-    DCHECK(network_task_runner_->BelongsToCurrentThread());
-    encode_task_runner_->PostTask(
-        FROM_HERE, base::Bind(&VideoScheduler::SetLosslessEncode,
-                              this, want_lossless));
-    return;
-  }
+  DCHECK(network_task_runner_->BelongsToCurrentThread());
 
-  encoder_->SetLosslessEncode(want_lossless);
+  encode_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&VideoEncoder::SetLosslessEncode,
+                            base::Unretained(encoder_.get()), want_lossless));
 }
 
 void VideoScheduler::SetLosslessColor(bool want_lossless) {
-  if (!encode_task_runner_->BelongsToCurrentThread()) {
-    DCHECK(network_task_runner_->BelongsToCurrentThread());
-    encode_task_runner_->PostTask(
-        FROM_HERE, base::Bind(&VideoScheduler::SetLosslessColor,
-                              this, want_lossless));
-    return;
-  }
+  DCHECK(network_task_runner_->BelongsToCurrentThread());
 
-  encoder_->SetLosslessColor(want_lossless);
+  encode_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&VideoEncoder::SetLosslessColor,
+                            base::Unretained(encoder_.get()), want_lossless));
 }
 
 // Private methods -----------------------------------------------------------
@@ -228,99 +152,122 @@ VideoScheduler::~VideoScheduler() {
 
 // Capturer thread -------------------------------------------------------------
 
-void VideoScheduler::StartOnCaptureThread() {
+webrtc::SharedMemory* VideoScheduler::CreateSharedMemory(size_t size) {
+  return nullptr;
+}
+
+void VideoScheduler::OnCaptureCompleted(webrtc::DesktopFrame* frame) {
   DCHECK(capture_task_runner_->BelongsToCurrentThread());
-  DCHECK(!capture_timer_);
 
-  // Start mouse cursor monitor.
-  mouse_cursor_monitor_->Init(this, webrtc::MouseCursorMonitor::SHAPE_ONLY);
+  network_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&VideoScheduler::EncodeAndSendFrame, this,
+                            base::Passed(make_scoped_ptr(frame))));
+}
 
-  // Start the capturer.
-  capturer_->Start(this);
+void VideoScheduler::OnMouseCursor(webrtc::MouseCursor* cursor) {
+  DCHECK(capture_task_runner_->BelongsToCurrentThread());
 
-  capture_timer_.reset(new base::OneShotTimer<VideoScheduler>());
-  keep_alive_timer_.reset(new base::DelayTimer<VideoScheduler>(
-      FROM_HERE, base::TimeDelta::FromMilliseconds(kKeepAlivePacketIntervalMs),
-      this, &VideoScheduler::SendKeepAlivePacket));
+  scoped_ptr<webrtc::MouseCursor> owned_cursor(cursor);
 
-  // Capture first frame immediately.
-  CaptureNextFrame();
-}
+  scoped_ptr<protocol::CursorShapeInfo> cursor_proto(
+      new protocol::CursorShapeInfo());
+  cursor_proto->set_width(cursor->image()->size().width());
+  cursor_proto->set_height(cursor->image()->size().height());
+  cursor_proto->set_hotspot_x(cursor->hotspot().x());
+  cursor_proto->set_hotspot_y(cursor->hotspot().y());
 
-void VideoScheduler::StopOnCaptureThread() {
-  DCHECK(capture_task_runner_->BelongsToCurrentThread());
+  cursor_proto->set_data(std::string());
+  uint8_t* current_row = cursor->image()->data();
+  for (int y = 0; y < cursor->image()->size().height(); ++y) {
+    cursor_proto->mutable_data()->append(
+        current_row,
+        current_row + cursor->image()->size().width() *
+            webrtc::DesktopFrame::kBytesPerPixel);
+    current_row += cursor->image()->stride();
+  }
 
-  // This doesn't deleted already captured frames, so encoder can keep using the
-  // frames that were captured previously.
-  capturer_.reset();
+  network_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&VideoScheduler::SendCursorShape, this,
+                            base::Passed(&cursor_proto)));
+}
 
-  // |capture_timer_| must be destroyed on the thread on which it is used.
-  capture_timer_.reset();
+void VideoScheduler::OnMouseCursorPosition(
+    webrtc::MouseCursorMonitor::CursorState state,
+    const webrtc::DesktopVector& position) {
+  // We're not subscribing to mouse position changes.
+  NOTREACHED();
 }
 
-void VideoScheduler::ScheduleNextCapture() {
+void VideoScheduler::StartOnCaptureThread() {
   DCHECK(capture_task_runner_->BelongsToCurrentThread());
 
-  capture_timer_->Start(FROM_HERE,
-                        scheduler_.NextCaptureDelay(),
-                        this,
-                        &VideoScheduler::CaptureNextFrame);
+  mouse_cursor_monitor_->Init(this, webrtc::MouseCursorMonitor::SHAPE_ONLY);
+  capturer_->Start(this);
 }
 
-void VideoScheduler::CaptureNextFrame() {
+void VideoScheduler::StopOnCaptureThread() {
   DCHECK(capture_task_runner_->BelongsToCurrentThread());
 
-  // If we are stopping (|capturer_| is nullptr), or paused, then don't capture.
-  if (!capturer_ || is_paused_)
-    return;
-
-  // Make sure we have at most two outstanding recordings. We can simply return
-  // if we can't make a capture now, the next capture will be started by the
-  // end of an encode operation.
-  if (pending_frames_ >= kMaxPendingFrames || capture_pending_) {
-    did_skip_frame_ = true;
-    return;
-  }
-
-  did_skip_frame_ = false;
-
-  // At this point we are going to perform one capture so save the current time.
-  pending_frames_++;
-  DCHECK_LE(pending_frames_, kMaxPendingFrames);
+  // This doesn't deleted already captured frames, so encoder can keep using the
+  // frames that were captured previously.
+  capturer_.reset();
 
-  // Before doing a capture schedule for the next one.
-  ScheduleNextCapture();
+  mouse_cursor_monitor_.reset();
+}
 
-  capture_pending_ = true;
+void VideoScheduler::CaptureNextFrameOnCaptureThread() {
+  DCHECK(capture_task_runner_->BelongsToCurrentThread());
 
-  // Capture the mouse shape.
+  // Capture mouse shape first and then screen content.
   mouse_cursor_monitor_->Capture();
-
-  // And finally perform one capture.
   capturer_->Capture(webrtc::DesktopRegion());
 }
 
-void VideoScheduler::FrameCaptureCompleted() {
-  DCHECK(capture_task_runner_->BelongsToCurrentThread());
+// Network thread --------------------------------------------------------------
 
-  // Decrement the pending capture count.
-  pending_frames_--;
-  DCHECK_GE(pending_frames_, 0);
+void VideoScheduler::CaptureNextFrame() {
+  DCHECK(network_task_runner_->BelongsToCurrentThread());
 
-  // If we've skipped a frame capture because too we had too many captures
-  // pending then schedule one now.
-  if (did_skip_frame_)
-    CaptureNextFrame();
+  capture_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&VideoScheduler::CaptureNextFrameOnCaptureThread, this));
 }
 
-// Network thread --------------------------------------------------------------
+void VideoScheduler::EncodeAndSendFrame(
+    scoped_ptr<webrtc::DesktopFrame> frame) {
+  DCHECK(network_task_runner_->BelongsToCurrentThread());
 
-void VideoScheduler::SendVideoPacket(scoped_ptr<VideoPacket> packet) {
+  if (!video_stub_)
+    return;
+
+  capture_scheduler_->OnCaptureCompleted();
+
+  // Even when |frame| is nullptr we still need to post it to the encode thread
+  // to make sure frames are freed in the same order they are received and
+  // that we don't start capturing frame n+2 before frame n is freed.
+  base::PostTaskAndReplyWithResult(
+      encode_task_runner_.get(), FROM_HERE,
+      base::Bind(&EncodeFrame, encoder_.get(), base::Passed(&frame)),
+      base::Bind(&VideoScheduler::SendEncodedFrame, this,
+                 latest_event_timestamp_, base::TimeTicks::Now()));
+}
+
+void VideoScheduler::SendEncodedFrame(int64 latest_event_timestamp,
+                                      base::TimeTicks timestamp,
+                                      scoped_ptr<VideoPacket> packet) {
   DCHECK(network_task_runner_->BelongsToCurrentThread());
 
   if (!video_stub_)
     return;
 
+  if (g_enable_timestamps)
+    packet->set_timestamp(timestamp.ToInternalValue());
+
+  packet->set_latest_event_timestamp(latest_event_timestamp);
+
+  capture_scheduler_->OnFrameEncoded(
+      base::TimeDelta::FromMilliseconds(packet->encode_time_ms()));
+
   video_stub_->ProcessVideoPacket(
       packet.Pass(), base::Bind(&VideoScheduler::OnVideoPacketSent, this));
 }
@@ -331,18 +278,13 @@ void VideoScheduler::OnVideoPacketSent() {
   if (!video_stub_)
     return;
 
+  capture_scheduler_->OnFrameSent();
   keep_alive_timer_->Reset();
-
-  capture_task_runner_->PostTask(
-      FROM_HERE, base::Bind(&VideoScheduler::FrameCaptureCompleted, this));
 }
 
 void VideoScheduler::SendKeepAlivePacket() {
   DCHECK(network_task_runner_->BelongsToCurrentThread());
 
-  if (!video_stub_)
-    return;
-
   video_stub_->ProcessVideoPacket(
       make_scoped_ptr(new VideoPacket()),
       base::Bind(&VideoScheduler::OnKeepAlivePacketSent, this));
@@ -365,43 +307,4 @@ void VideoScheduler::SendCursorShape(
   cursor_stub_->SetCursorShape(*cursor_shape);
 }
 
-// Encoder thread --------------------------------------------------------------
-
-void VideoScheduler::EncodeFrame(
-    scoped_ptr<webrtc::DesktopFrame> frame,
-    int64 latest_event_timestamp,
-    base::TimeTicks timestamp) {
-  DCHECK(encode_task_runner_->BelongsToCurrentThread());
-
-  // If there is nothing to encode then send an empty packet.
-  if (!frame || frame->updated_region().is_empty()) {
-    capture_task_runner_->DeleteSoon(FROM_HERE, frame.release());
-    scoped_ptr<VideoPacket> packet(new VideoPacket());
-    packet->set_latest_event_timestamp(latest_event_timestamp);
-    network_task_runner_->PostTask(
-        FROM_HERE,
-        base::Bind(
-            &VideoScheduler::SendVideoPacket, this, base::Passed(&packet)));
-    return;
-  }
-
-  scoped_ptr<VideoPacket> packet = encoder_->Encode(*frame);
-  packet->set_latest_event_timestamp(latest_event_timestamp);
-
-  if (g_enable_timestamps) {
-    packet->set_timestamp(timestamp.ToInternalValue());
-  }
-
-  // Destroy the frame before sending |packet| because SendVideoPacket() may
-  // trigger another frame to be captured, and the screen capturer expects the
-  // old frame to be freed by then.
-  frame.reset();
-
-  scheduler_.RecordEncodeTime(
-      base::TimeDelta::FromMilliseconds(packet->encode_time_ms()));
-  network_task_runner_->PostTask(
-      FROM_HERE, base::Bind(&VideoScheduler::SendVideoPacket, this,
-                            base::Passed(&packet)));
-}
-
 }  // namespace remoting
diff --git a/remoting/host/video_scheduler.h b/remoting/host/video_scheduler.h
index 3bcf277..6adab18 100644
--- a/remoting/host/video_scheduler.h
+++ b/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);
 };