Implement video frame acknowledgements in the chromoting protocol.

Added frame_id field in the  VideoPacket message. Client now acknowledges
every frame that has frame_id set by sending VideoAck messages after
the corresponding frame is decoded and rendered.

On the host the VideoAck messages are processed by the new
VideoFeedbackStub, which is implemented in CaptureScheduler.
CaptureScheduler limits number of unacknowledged frames to 4. This
number was chosen experimentally to minimize latency (using
remoting_perftests).

BUG=448838

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

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

18 files changed, 526 insertions, 50 deletions

diff --git a/remoting/host/capture_scheduler.cc b/remoting/host/capture_scheduler.cc
index ccd141d..56a2d28 100644
--- a/remoting/host/capture_scheduler.cc
+++ b/remoting/host/capture_scheduler.cc
@@ -10,6 +10,7 @@
 #include "base/sys_info.h"
 #include "base/time/default_tick_clock.h"
 #include "base/time/time.h"
+#include "remoting/proto/video.pb.h"
 
 namespace {
 
@@ -25,7 +26,22 @@ 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.
+// Maximum number of captured frames in the encoding queue. Currently capturer
+// implementations do not allow to keep more than 2 DesktopFrame objects.
+static const int kMaxFramesInEncodingQueue = 2;
+
+// Maximum number of unacknowledged frames. Ignored if the client doesn't
+// support ACKs. This value was chosen experimentally, using synthetic
+// performance tests (see ProtocolPerfTest), to maximize frame rate, while
+// keeping round-trip latency low.
+static const int kMaxUnacknowledgedFrames = 4;
+
+// Maximum number of frames that can be processed (captured, encoded or sent)
+// simultaneously. It's used only in the case when the client doesn't support
+// ACKs.
+//
+// TODO(sergeyu): Remove this once all current versions support ACKs.
+// crbug.com/460963 .
 static const int kMaxPendingFrames = 2;
 
 }  // namespace
@@ -35,6 +51,7 @@ namespace remoting {
 // We assume that the number of available cores is constant.
 CaptureScheduler::CaptureScheduler(const base::Closure& capture_closure)
     : capture_closure_(capture_closure),
+      acks_supported_(false),
       tick_clock_(new base::DefaultTickClock()),
       capture_timer_(new base::Timer(false, false)),
       minimum_interval_(
@@ -42,9 +59,12 @@ CaptureScheduler::CaptureScheduler(const base::Closure& capture_closure)
       num_of_processors_(base::SysInfo::NumberOfProcessors()),
       capture_time_(kStatisticsWindow),
       encode_time_(kStatisticsWindow),
-      pending_frames_(0),
+      num_encoding_frames_(0),
+      num_sending_frames_(0),
+      num_unacknowledged_frames_(0),
       capture_pending_(false),
-      is_paused_(false) {
+      is_paused_(false),
+      next_frame_id_(0) {
   DCHECK(num_of_processors_);
 }
 
@@ -52,13 +72,13 @@ CaptureScheduler::~CaptureScheduler() {
 }
 
 void CaptureScheduler::Start() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
 
   ScheduleNextCapture();
 }
 
 void CaptureScheduler::Pause(bool pause) {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
 
   if (is_paused_ != pause) {
     is_paused_ = pause;
@@ -72,29 +92,54 @@ void CaptureScheduler::Pause(bool pause) {
 }
 
 void CaptureScheduler::OnCaptureCompleted() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
 
   capture_pending_ = false;
   capture_time_.Record(
       (tick_clock_->NowTicks() - last_capture_started_time_).InMilliseconds());
 
+  ++num_encoding_frames_;
+
+  ScheduleNextCapture();
+}
+
+void CaptureScheduler::OnFrameEncoded(VideoPacket* packet) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // Set packet_id for the outgoing packet.
+  packet->set_frame_id(next_frame_id_);
+  ++next_frame_id_;
+
+  // Update internal stats.
+  encode_time_.Record(packet->encode_time_ms());
+
+  --num_encoding_frames_;
+  ++num_sending_frames_;
+  ++num_unacknowledged_frames_;
+
   ScheduleNextCapture();
 }
 
 void CaptureScheduler::OnFrameSent() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
 
-  // Decrement the pending capture count.
-  pending_frames_--;
-  DCHECK_GE(pending_frames_, 0);
+  --num_sending_frames_;
+  DCHECK_GE(num_sending_frames_, 0);
 
   ScheduleNextCapture();
 }
 
-void CaptureScheduler::OnFrameEncoded(base::TimeDelta encode_time) {
-  DCHECK(CalledOnValidThread());
+void CaptureScheduler::ProcessVideoAck(scoped_ptr<VideoAck> video_ack) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // Host always sets |frame_id| field to indicated that it expects ACK from the
+  // client. It's assumed that the client doesn't support ACKs until the first
+  // ACK message is received.
+  acks_supported_ = true;
+
+  --num_unacknowledged_frames_;
+  DCHECK_GE(num_unacknowledged_frames_, 0);
 
-  encode_time_.Record(encode_time.InMilliseconds());
   ScheduleNextCapture();
 }
 
@@ -102,18 +147,35 @@ 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());
+  DCHECK(thread_checker_.CalledOnValidThread());
 
-  if (is_paused_ || pending_frames_ >= kMaxPendingFrames || capture_pending_)
+  if (is_paused_ || capture_pending_ ||
+      num_encoding_frames_ >= kMaxFramesInEncodingQueue) {
     return;
+  }
+
+  if (acks_supported_) {
+    if (num_encoding_frames_ + num_unacknowledged_frames_ >=
+        kMaxUnacknowledgedFrames) {
+      return;
+    }
+  } else {
+    // TODO(sergeyu): Remove this once all current versions support ACKs.
+    // crbug.com/460963 .
+    if (num_encoding_frames_ + num_sending_frames_ >= kMaxPendingFrames) {
+      return;
+    }
+  }
 
   // Delay by an amount chosen such that if capture and encode times
   // continue to follow the averages, then we'll consume the target
@@ -134,13 +196,10 @@ void CaptureScheduler::ScheduleNextCapture() {
 }
 
 void CaptureScheduler::CaptureNextFrame() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.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();
diff --git a/remoting/host/capture_scheduler.h b/remoting/host/capture_scheduler.h
index b3136e7..42a1657 100644
--- a/remoting/host/capture_scheduler.h
+++ b/remoting/host/capture_scheduler.h
@@ -2,29 +2,35 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-// This class chooses a capture interval so as to limit CPU usage to not exceed
-// a specified %age. It bases this on the CPU usage of recent capture and encode
-// operations, and on the number of available CPUs.
-
 #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/threading/thread_checker.h"
 #include "base/time/tick_clock.h"
 #include "base/time/time.h"
 #include "base/timer/timer.h"
 #include "remoting/base/running_average.h"
+#include "remoting/protocol/video_feedback_stub.h"
 
 namespace remoting {
 
+class VideoPacket;
+
 // CaptureScheduler is used by the VideoFramePump to schedule frame capturer,
 // taking into account capture delay, encoder delay, network bandwidth, etc.
-class CaptureScheduler : public base::NonThreadSafe {
+// It implements VideoFeedbackStub to receive frame acknowledgments from the
+// client.
+//
+// It attempts to achieve the following goals when scheduling frames:
+//  - Keep round-trip latency as low a possible.
+//  - Parallelize capture, encode and transmission, to achieve frame rate as
+//    close to the target of 30fps as possible.
+//  - Limit CPU usage to 50%.
+class CaptureScheduler : public protocol::VideoFeedbackStub {
  public:
-  // |capture_closure| is called every time a new frame needs to be captured.
   explicit CaptureScheduler(const base::Closure& capture_closure);
-  ~CaptureScheduler();
+  ~CaptureScheduler() override;
 
   // Starts the scheduler.
   void Start();
@@ -35,12 +41,16 @@ class CaptureScheduler : public base::NonThreadSafe {
   // 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 encoded. The scheduler can
+  // change |packet| if necessary, e.g. set |frame_id|.
+  void OnFrameEncoded(VideoPacket* packet);
 
   // Notifies the scheduler that a frame has been sent.
   void OnFrameSent();
 
+  // VideoFeedbackStub interface.
+  void ProcessVideoAck(scoped_ptr<VideoAck> video_ack) override;
+
   // Sets minimum interval between frames.
   void set_minimum_interval(base::TimeDelta minimum_interval) {
     minimum_interval_ = minimum_interval;
@@ -63,6 +73,9 @@ class CaptureScheduler : public base::NonThreadSafe {
 
   base::Closure capture_closure_;
 
+  // Set to true if the connection supports video frame acknowledgments.
+  bool acks_supported_;
+
   scoped_ptr<base::TickClock> tick_clock_;
 
   // Timer used to schedule CaptureNextFrame().
@@ -76,8 +89,14 @@ class CaptureScheduler : public base::NonThreadSafe {
   RunningAverage capture_time_;
   RunningAverage encode_time_;
 
-  // Total number of pending frames that are being captured, encoded or sent.
-  int pending_frames_;
+  // Number of frames pending encoding.
+  int num_encoding_frames_;
+
+  // Number of frames in the sending queue.
+  int num_sending_frames_;
+
+  // Number of outgoing frames for which we haven't received an acknowledgment.
+  int num_unacknowledged_frames_;
 
   // Set to true when capture is pending.
   bool capture_pending_;
@@ -87,6 +106,11 @@ class CaptureScheduler : public base::NonThreadSafe {
 
   bool is_paused_;
 
+  // Frame ID to be assigned to the next outgoing video frame.
+  uint32_t next_frame_id_;
+
+  base::ThreadChecker thread_checker_;
+
   DISALLOW_COPY_AND_ASSIGN(CaptureScheduler);
 };
 
diff --git a/remoting/host/capture_scheduler_unittest.cc b/remoting/host/capture_scheduler_unittest.cc
index b07fc6da..95ddf6b 100644
--- a/remoting/host/capture_scheduler_unittest.cc
+++ b/remoting/host/capture_scheduler_unittest.cc
@@ -7,6 +7,7 @@
 #include "base/message_loop/message_loop.h"
 #include "base/test/simple_test_tick_clock.h"
 #include "base/timer/mock_timer.h"
+#include "remoting/proto/video.pb.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace remoting {
@@ -47,9 +48,17 @@ class CaptureSchedulerTest : public testing::Test {
     CheckCaptureCalled();
     tick_clock_->Advance(capture_delay);
     scheduler_->OnCaptureCompleted();
-    scheduler_->OnFrameEncoded(encode_delay);
+
+    VideoPacket packet;
+    packet.set_encode_time_ms(encode_delay.InMilliseconds());
+    scheduler_->OnFrameEncoded(&packet);
+
     scheduler_->OnFrameSent();
 
+    scoped_ptr<VideoAck> ack(new VideoAck());
+    ack->set_frame_id(packet.frame_id());
+    scheduler_->ProcessVideoAck(ack.Pass());
+
     EXPECT_TRUE(capture_timer_->IsRunning());
     EXPECT_EQ(std::max(base::TimeDelta(),
                        expected_delay_between_frames - capture_delay),
@@ -132,10 +141,16 @@ TEST_F(CaptureSchedulerTest, RollingAverageDifferentTimes) {
   }
 }
 
-// Verify that we never have more than 2 pending frames.
-TEST_F(CaptureSchedulerTest, MaximumPendingFrames) {
+// Verify that we never have more than 2 encoding frames.
+TEST_F(CaptureSchedulerTest, MaximumEncodingFrames) {
   InitScheduler();
 
+  // Process the first frame to let the scheduler know that receiver supports
+  // ACKs.
+  SimulateSingleFrameCapture(
+      base::TimeDelta(), base::TimeDelta(),
+      base::TimeDelta::FromMilliseconds(kMinumumFrameIntervalMs));
+
   capture_timer_->Fire();
   CheckCaptureCalled();
   scheduler_->OnCaptureCompleted();
@@ -145,11 +160,55 @@ TEST_F(CaptureSchedulerTest, MaximumPendingFrames) {
   scheduler_->OnCaptureCompleted();
 
   EXPECT_FALSE(capture_timer_->IsRunning());
+  VideoPacket packet;
+  scheduler_->OnFrameEncoded(&packet);
+  EXPECT_TRUE(capture_timer_->IsRunning());
+}
 
-  scheduler_->OnFrameEncoded(base::TimeDelta());
-  scheduler_->OnFrameSent();
+// Verify that the scheduler doesn't exceed maximum number of pending frames.
+TEST_F(CaptureSchedulerTest, MaximumPendingFrames) {
+  InitScheduler();
 
+  // Process the first frame to let the scheduler know that receiver supports
+  // ACKs.
+  SimulateSingleFrameCapture(
+      base::TimeDelta(), base::TimeDelta(),
+      base::TimeDelta::FromMilliseconds(kMinumumFrameIntervalMs));
+
+  // Queue some frames until the sender is blocked.
+  while (capture_timer_->IsRunning()) {
+    capture_timer_->Fire();
+    CheckCaptureCalled();
+    scheduler_->OnCaptureCompleted();
+    VideoPacket packet;
+    scheduler_->OnFrameEncoded(&packet);
+    scheduler_->OnFrameSent();
+  }
+
+  // Next frame should be scheduled, once one of the queued frames is
+  // acknowledged.
+  EXPECT_FALSE(capture_timer_->IsRunning());
+  scheduler_->ProcessVideoAck(make_scoped_ptr(new VideoAck()));
   EXPECT_TRUE(capture_timer_->IsRunning());
 }
 
+// Verify that the scheduler doesn't exceed maximum number of pending frames
+// when acks are not supported.
+TEST_F(CaptureSchedulerTest, MaximumPendingFramesNoAcks) {
+  InitScheduler();
+
+  // Queue some frames until the sender is blocked.
+  while (capture_timer_->IsRunning()) {
+    capture_timer_->Fire();
+    CheckCaptureCalled();
+    scheduler_->OnCaptureCompleted();
+    VideoPacket packet;
+    scheduler_->OnFrameEncoded(&packet);
+  }
+
+  // Next frame should be scheduled, once one of the queued frames is sent.
+  EXPECT_FALSE(capture_timer_->IsRunning());
+  scheduler_->OnFrameSent();
+  EXPECT_TRUE(capture_timer_->IsRunning());
+}
 }  // namespace remoting
diff --git a/remoting/host/client_session.cc b/remoting/host/client_session.cc
index ef6f5a6..905539b 100644
--- a/remoting/host/client_session.cc
+++ b/remoting/host/client_session.cc
@@ -447,6 +447,7 @@ void ClientSession::ResetVideoPipeline() {
   DCHECK(CalledOnValidThread());
 
   mouse_shape_pump_.reset();
+  connection_->set_video_feedback_stub(nullptr);
   video_frame_pump_.reset();
 
   // Create VideoEncoder and DesktopCapturer to match the session's video
@@ -484,6 +485,9 @@ void ClientSession::ResetVideoPipeline() {
 
   // Pause capturing if necessary.
   video_frame_pump_->Pause(pause_video_);
+
+  connection_->set_video_feedback_stub(
+      video_frame_pump_->video_feedback_stub());
 }
 
 void ClientSession::SetGnubbyAuthHandlerForTesting(
diff --git a/remoting/host/video_frame_pump.cc b/remoting/host/video_frame_pump.cc
index 6451313..526615c 100644
--- a/remoting/host/video_frame_pump.cc
+++ b/remoting/host/video_frame_pump.cc
@@ -146,8 +146,7 @@ void VideoFramePump::SendEncodedFrame(int64 latest_event_timestamp,
 
   packet->set_latest_event_timestamp(latest_event_timestamp);
 
-  capture_scheduler_.OnFrameEncoded(
-      base::TimeDelta::FromMilliseconds(packet->encode_time_ms()));
+  capture_scheduler_.OnFrameEncoded(packet.get());
 
   video_stub_->ProcessVideoPacket(packet.Pass(),
                                   base::Bind(&VideoFramePump::OnVideoPacketSent,
diff --git a/remoting/host/video_frame_pump.h b/remoting/host/video_frame_pump.h
index 2f36d0d..5bafa3e 100644
--- a/remoting/host/video_frame_pump.h
+++ b/remoting/host/video_frame_pump.h
@@ -23,6 +23,7 @@ class SingleThreadTaskRunner;
 namespace remoting {
 
 namespace protocol {
+class VideoFeedbackStub;
 class VideoStub;
 }  // namespace protocol
 
@@ -90,6 +91,10 @@ class VideoFramePump : public webrtc::DesktopCapturer::Callback {
   void SetLosslessEncode(bool want_lossless);
   void SetLosslessColor(bool want_lossless);
 
+  protocol::VideoFeedbackStub* video_feedback_stub() {
+    return &capture_scheduler_;
+  }
+
  private:
   // webrtc::DesktopCapturer::Callback interface.
   webrtc::SharedMemory* CreateSharedMemory(size_t size) override;
@@ -103,8 +108,7 @@ class VideoFramePump : public webrtc::DesktopCapturer::Callback {
                         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.
+  // Callback passed to |video_stub_|.
   void OnVideoPacketSent();
 
   // Called by |keep_alive_timer_|.
@@ -113,8 +117,6 @@ class VideoFramePump : public webrtc::DesktopCapturer::Callback {
   // Callback for |video_stub_| called after a keep-alive packet is sent.
   void OnKeepAlivePacketSent();
 
-  base::ThreadChecker thread_checker_;
-
   // Task runner used to run |encoder_|.
   scoped_refptr<base::SingleThreadTaskRunner> encode_task_runner_;
 
@@ -138,6 +140,8 @@ class VideoFramePump : public webrtc::DesktopCapturer::Callback {
   // Number updated by the caller to trace performance.
   int64 latest_event_timestamp_;
 
+  base::ThreadChecker thread_checker_;
+
   base::WeakPtrFactory<VideoFramePump> weak_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(VideoFramePump);
diff --git a/remoting/proto/video.proto b/remoting/proto/video.proto
index 0ee29c9..211b023 100644
--- a/remoting/proto/video.proto
+++ b/remoting/proto/video.proto
@@ -66,4 +66,16 @@ message VideoPacket {
 
   // Optional frame timestamp. Used in tests to estimate frame latency.
   optional int64 timestamp = 12;
+
+  // Frame identifier used to match VideoFrame and VideoAck.
+  optional int32 frame_id = 13;
+}
+
+// VideoAck acknowledges that the frame in the VideoPacket with the same
+// frame_id has been rendered. VideoAck messages must be sent only for frames
+// that have frame_id field set. They must be sent the same order in which
+// the corresponding VideoPackets were received.
+message VideoAck {
+  // Frame ID.
+  optional int32 frame_id = 1;
 }
diff --git a/remoting/protocol/client_video_dispatcher.cc b/remoting/protocol/client_video_dispatcher.cc
index 287bba3..d9918fb 100644
--- a/remoting/protocol/client_video_dispatcher.cc
+++ b/remoting/protocol/client_video_dispatcher.cc
@@ -5,23 +5,71 @@
 #include "remoting/protocol/client_video_dispatcher.h"
 
 #include "base/bind.h"
+#include "base/callback_helpers.h"
 #include "net/socket/stream_socket.h"
 #include "remoting/base/constants.h"
 #include "remoting/proto/video.pb.h"
+#include "remoting/protocol/message_serialization.h"
 #include "remoting/protocol/video_stub.h"
 
 namespace remoting {
 namespace protocol {
 
+struct ClientVideoDispatcher::PendingFrame {
+  PendingFrame(int frame_id)
+      : frame_id(frame_id),
+        done(false) {}
+  int frame_id;
+  bool done;
+};
+
 ClientVideoDispatcher::ClientVideoDispatcher(VideoStub* video_stub)
     : ChannelDispatcherBase(kVideoChannelName),
-      parser_(base::Bind(&VideoStub::ProcessVideoPacket,
-                         base::Unretained(video_stub)),
+      video_stub_(video_stub),
+      parser_(base::Bind(&ClientVideoDispatcher::ProcessVideoPacket,
+                         base::Unretained(this)),
               reader()) {
 }
 
 ClientVideoDispatcher::~ClientVideoDispatcher() {
 }
 
+void ClientVideoDispatcher::ProcessVideoPacket(
+    scoped_ptr<VideoPacket> video_packet,
+    const base::Closure& done) {
+  base::ScopedClosureRunner done_runner(done);
+
+  int frame_id = video_packet->frame_id();
+
+  if (!video_packet->has_frame_id()) {
+    video_stub_->ProcessVideoPacket(video_packet.Pass(), done_runner.Release());
+    return;
+  }
+
+  PendingFramesList::iterator pending_frame =
+      pending_frames_.insert(pending_frames_.end(), PendingFrame(frame_id));
+
+  video_stub_->ProcessVideoPacket(
+      video_packet.Pass(), base::Bind(&ClientVideoDispatcher::OnPacketDone,
+                                      base::Unretained(this), pending_frame));
+
+}
+
+void ClientVideoDispatcher::OnPacketDone(
+    PendingFramesList::iterator pending_frame) {
+  // Mark the frame as done.
+  DCHECK(!pending_frame->done);
+  pending_frame->done = true;
+
+  // Send VideoAck for all packets in the head of the queue that have finished
+  // rendering.
+  while (!pending_frames_.empty() && pending_frames_.front().done) {
+    VideoAck ack_message;
+    ack_message.set_frame_id(pending_frames_.front().frame_id);
+    writer()->Write(SerializeAndFrameMessage(ack_message), base::Closure());
+    pending_frames_.pop_front();
+  }
+}
+
 }  // namespace protocol
 }  // namespace remoting
diff --git a/remoting/protocol/client_video_dispatcher.h b/remoting/protocol/client_video_dispatcher.h
index b2b1e31..a86142f 100644
--- a/remoting/protocol/client_video_dispatcher.h
+++ b/remoting/protocol/client_video_dispatcher.h
@@ -21,6 +21,18 @@ class ClientVideoDispatcher : public ChannelDispatcherBase {
   ~ClientVideoDispatcher() override;
 
  private:
+  struct PendingFrame;
+  typedef std::list<PendingFrame> PendingFramesList;
+
+  void ProcessVideoPacket(scoped_ptr<VideoPacket> video_packet,
+                          const base::Closure& done);
+
+  // Callback for VideoStub::ProcessVideoPacket().
+  void OnPacketDone(PendingFramesList::iterator pending_frame);
+
+  PendingFramesList pending_frames_;
+
+  VideoStub* video_stub_;
   ProtobufMessageParser<VideoPacket> parser_;
 
   DISALLOW_COPY_AND_ASSIGN(ClientVideoDispatcher);
diff --git a/remoting/protocol/client_video_dispatcher_unittest.cc b/remoting/protocol/client_video_dispatcher_unittest.cc
new file mode 100644
index 0000000..92087b3
--- /dev/null
+++ b/remoting/protocol/client_video_dispatcher_unittest.cc
@@ -0,0 +1,180 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "remoting/protocol/client_video_dispatcher.h"
+
+#include "base/bind.h"
+#include "base/memory/scoped_vector.h"
+#include "base/message_loop/message_loop.h"
+#include "base/run_loop.h"
+#include "remoting/base/constants.h"
+#include "remoting/proto/video.pb.h"
+#include "remoting/protocol/fake_session.h"
+#include "remoting/protocol/fake_stream_socket.h"
+#include "remoting/protocol/message_serialization.h"
+#include "remoting/protocol/video_stub.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace remoting {
+namespace protocol {
+
+class ClientVideoDispatcherTest : public testing::Test,
+                                  public VideoStub,
+                                  public ChannelDispatcherBase::EventHandler {
+ public:
+  ClientVideoDispatcherTest();
+
+  // VideoStub interface.
+  void ProcessVideoPacket(scoped_ptr<VideoPacket> video_packet,
+                          const base::Closure& done) override;
+
+  // ChannelDispatcherBase::EventHandler interface.
+  void OnChannelInitialized(ChannelDispatcherBase* channel_dispatcher) override;
+  void OnChannelError(ChannelDispatcherBase* channel_dispatcher,
+                      ErrorCode error) override;
+
+ protected:
+  void OnVideoAck(scoped_ptr<VideoAck> ack, const base::Closure& done);
+
+  base::MessageLoop message_loop_;
+
+  // Set to true in OnChannelInitialized().
+  bool initialized_;
+
+  // Client side.
+  ClientVideoDispatcher dispatcher_;
+  FakeSession session_;
+
+  // Host side.
+  FakeStreamSocket host_socket_;
+  MessageReader reader_;
+  ProtobufMessageParser<VideoAck> parser_;
+  BufferedSocketWriter writer_;
+
+  ScopedVector<VideoPacket> video_packets_;
+  std::vector<base::Closure> packet_done_callbacks_;
+
+  ScopedVector<VideoAck> ack_messages_;
+};
+
+ClientVideoDispatcherTest::ClientVideoDispatcherTest()
+    : initialized_(false),
+      dispatcher_(this),
+      parser_(base::Bind(&ClientVideoDispatcherTest::OnVideoAck,
+                         base::Unretained(this)),
+              &reader_) {
+  dispatcher_.Init(&session_, ChannelConfig(ChannelConfig::TRANSPORT_MUX_STREAM,
+                                            kDefaultStreamVersion,
+                                            ChannelConfig::CODEC_UNDEFINED),
+                   this);
+  base::RunLoop().RunUntilIdle();
+  DCHECK(initialized_);
+  host_socket_.PairWith(
+      session_.fake_channel_factory().GetFakeChannel(kVideoChannelName));
+  reader_.StartReading(&host_socket_);
+  writer_.Init(&host_socket_, BufferedSocketWriter::WriteFailedCallback());
+}
+
+void ClientVideoDispatcherTest::ProcessVideoPacket(
+    scoped_ptr<VideoPacket> video_packet,
+    const base::Closure& done) {
+  video_packets_.push_back(video_packet.release());
+  packet_done_callbacks_.push_back(done);
+}
+
+void ClientVideoDispatcherTest::OnChannelInitialized(
+    ChannelDispatcherBase* channel_dispatcher) {
+  initialized_ = true;
+}
+
+void ClientVideoDispatcherTest::OnChannelError(
+    ChannelDispatcherBase* channel_dispatcher,
+    ErrorCode error) {
+  // Don't expect channel creation to fail.
+  FAIL();
+}
+
+void ClientVideoDispatcherTest::OnVideoAck(scoped_ptr<VideoAck> ack,
+                                           const base::Closure& done) {
+  ack_messages_.push_back(ack.release());
+  done.Run();
+}
+
+// Verify that the client can receive video packets and acks are not sent for
+// VideoPackets that don't have frame_id field set.
+TEST_F(ClientVideoDispatcherTest, WithoutAcks) {
+  VideoPacket packet;
+  packet.set_data(std::string());
+
+  // Send a VideoPacket and verify that the client receives it.
+  writer_.Write(SerializeAndFrameMessage(packet), base::Closure());
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(1U, video_packets_.size());
+
+  packet_done_callbacks_.front().Run();
+  base::RunLoop().RunUntilIdle();
+
+  // Ack should never be sent for the packet without frame_id.
+  EXPECT_TRUE(ack_messages_.empty());
+}
+
+// Verifies that the dispatcher sends Ack message with correct rendering delay.
+TEST_F(ClientVideoDispatcherTest, WithAcks) {
+  int kTestFrameId = 3;
+
+  VideoPacket packet;
+  packet.set_data(std::string());
+  packet.set_frame_id(kTestFrameId);
+
+  // Send a VideoPacket and verify that the client receives it.
+  writer_.Write(SerializeAndFrameMessage(packet), base::Closure());
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(1U, video_packets_.size());
+
+  // Ack should only be sent after the packet is processed.
+  EXPECT_TRUE(ack_messages_.empty());
+  base::RunLoop().RunUntilIdle();
+
+  // Fake completion of video packet decoding, to trigger the Ack.
+  packet_done_callbacks_.front().Run();
+  base::RunLoop().RunUntilIdle();
+
+  // Verify that the Ack message has been received.
+  ASSERT_EQ(1U, ack_messages_.size());
+  EXPECT_EQ(kTestFrameId, ack_messages_[0]->frame_id());
+}
+
+// Verify that Ack messages are sent in correct order.
+TEST_F(ClientVideoDispatcherTest, AcksOrder) {
+  int kTestFrameId = 3;
+
+  VideoPacket packet;
+  packet.set_data(std::string());
+  packet.set_frame_id(kTestFrameId);
+
+  // Send two VideoPackets.
+  writer_.Write(SerializeAndFrameMessage(packet), base::Closure());
+  base::RunLoop().RunUntilIdle();
+
+  packet.set_frame_id(kTestFrameId + 1);
+  writer_.Write(SerializeAndFrameMessage(packet), base::Closure());
+  base::RunLoop().RunUntilIdle();
+
+  EXPECT_EQ(2U, video_packets_.size());
+  EXPECT_TRUE(ack_messages_.empty());
+
+  // Call completion callbacks in revers order.
+  packet_done_callbacks_[1].Run();
+  packet_done_callbacks_[0].Run();
+
+  base::RunLoop().RunUntilIdle();
+
+  // Verify order of Ack messages.
+  ASSERT_EQ(2U, ack_messages_.size());
+  EXPECT_EQ(kTestFrameId, ack_messages_[0]->frame_id());
+  EXPECT_EQ(kTestFrameId + 1, ack_messages_[1]->frame_id());
+}
+
+}  // namespace protocol
+}  // namespace remoting
diff --git a/remoting/protocol/connection_to_client.cc b/remoting/protocol/connection_to_client.cc
index 6fd811e..b026ee3 100644
--- a/remoting/protocol/connection_to_client.cc
+++ b/remoting/protocol/connection_to_client.cc
@@ -84,6 +84,12 @@ void ConnectionToClient::set_input_stub(protocol::InputStub* input_stub) {
   event_dispatcher_->set_input_stub(input_stub);
 }
 
+void ConnectionToClient::set_video_feedback_stub(
+    VideoFeedbackStub* video_feedback_stub) {
+  DCHECK(CalledOnValidThread());
+  video_dispatcher_->set_video_feedback_stub(video_feedback_stub);
+}
+
 void ConnectionToClient::OnSessionStateChange(Session::State state) {
   DCHECK(CalledOnValidThread());
 
diff --git a/remoting/protocol/connection_to_client.h b/remoting/protocol/connection_to_client.h
index 2cdd60b..43c0b1a 100644
--- a/remoting/protocol/connection_to_client.h
+++ b/remoting/protocol/connection_to_client.h
@@ -25,6 +25,7 @@ class HostEventDispatcher;
 class HostStub;
 class HostVideoDispatcher;
 class InputStub;
+class VideoFeedbackStub;
 class VideoStub;
 
 // This class represents a remote viewer connection to the chromoting
@@ -98,6 +99,10 @@ class ConnectionToClient : public base::NonThreadSafe,
   virtual void set_host_stub(HostStub* host_stub);
   virtual void set_input_stub(InputStub* input_stub);
 
+  // Sets video feedback stub. Can be called at any time after connection is
+  // authenticated.
+  virtual void set_video_feedback_stub(VideoFeedbackStub* video_feedback_stub);
+
   // Session::EventHandler interface.
   void OnSessionStateChange(Session::State state) override;
   void OnSessionRouteChange(const std::string& channel_name,
diff --git a/remoting/protocol/host_video_dispatcher.cc b/remoting/protocol/host_video_dispatcher.cc
index b9a12c0..4a752fb 100644
--- a/remoting/protocol/host_video_dispatcher.cc
+++ b/remoting/protocol/host_video_dispatcher.cc
@@ -7,14 +7,18 @@
 #include "base/bind.h"
 #include "net/socket/stream_socket.h"
 #include "remoting/base/constants.h"
-#include "remoting/proto/video.pb.h"
 #include "remoting/protocol/message_serialization.h"
+#include "remoting/protocol/video_feedback_stub.h"
 
 namespace remoting {
 namespace protocol {
 
 HostVideoDispatcher::HostVideoDispatcher()
-    : ChannelDispatcherBase(kVideoChannelName) {
+    : ChannelDispatcherBase(kVideoChannelName),
+      parser_(
+          base::Bind(&HostVideoDispatcher::OnVideoAck, base::Unretained(this)),
+          reader()),
+      video_feedback_stub_(nullptr) {
 }
 
 HostVideoDispatcher::~HostVideoDispatcher() {
@@ -25,5 +29,13 @@ void HostVideoDispatcher::ProcessVideoPacket(scoped_ptr<VideoPacket> packet,
   writer()->Write(SerializeAndFrameMessage(*packet), done);
 }
 
+void HostVideoDispatcher::OnVideoAck(scoped_ptr<VideoAck> ack,
+                                     const base::Closure& done) {
+  if (video_feedback_stub_)
+    video_feedback_stub_->ProcessVideoAck(ack.Pass());
+
+  done.Run();
+}
+
 }  // namespace protocol
 }  // namespace remoting
diff --git a/remoting/protocol/host_video_dispatcher.h b/remoting/protocol/host_video_dispatcher.h
index c1a9573..79d1585 100644
--- a/remoting/protocol/host_video_dispatcher.h
+++ b/remoting/protocol/host_video_dispatcher.h
@@ -5,26 +5,37 @@
 #ifndef REMOTING_PROTOCOL_HOST_VIDEO_DISPATCHER_H_
 #define REMOTING_PROTOCOL_HOST_VIDEO_DISPATCHER_H_
 
-#include <string>
-
 #include "base/compiler_specific.h"
-#include "remoting/protocol/buffered_socket_writer.h"
+#include "remoting/proto/video.pb.h"
 #include "remoting/protocol/channel_dispatcher_base.h"
+#include "remoting/protocol/protobuf_message_parser.h"
 #include "remoting/protocol/video_stub.h"
 
 namespace remoting {
 namespace protocol {
 
+class VideoFeedbackStub;
+
 class HostVideoDispatcher : public ChannelDispatcherBase, public VideoStub {
  public:
   HostVideoDispatcher();
   ~HostVideoDispatcher() override;
 
+  void set_video_feedback_stub(VideoFeedbackStub* video_feedback_stub) {
+    video_feedback_stub_ = video_feedback_stub;
+  }
+
   // VideoStub interface.
   void ProcessVideoPacket(scoped_ptr<VideoPacket> packet,
                           const base::Closure& done) override;
 
  private:
+  void OnVideoAck(scoped_ptr<VideoAck> ack, const base::Closure& done);
+
+  ProtobufMessageParser<VideoAck> parser_;
+
+  VideoFeedbackStub* video_feedback_stub_;
+
   DISALLOW_COPY_AND_ASSIGN(HostVideoDispatcher);
 };
 
diff --git a/remoting/protocol/protocol_mock_objects.cc b/remoting/protocol/protocol_mock_objects.cc
index 7b51fe2..9087636 100644
--- a/remoting/protocol/protocol_mock_objects.cc
+++ b/remoting/protocol/protocol_mock_objects.cc
@@ -15,7 +15,8 @@ MockConnectionToClient::MockConnectionToClient(Session* session,
     : ConnectionToClient(session),
       clipboard_stub_(nullptr),
       host_stub_(host_stub),
-      input_stub_(nullptr) {
+      input_stub_(nullptr),
+      video_feedback_stub_(nullptr) {
 }
 
 MockConnectionToClient::~MockConnectionToClient() {}
diff --git a/remoting/protocol/protocol_mock_objects.h b/remoting/protocol/protocol_mock_objects.h
index b0ff4f2..3ee6a1a 100644
--- a/remoting/protocol/protocol_mock_objects.h
+++ b/remoting/protocol/protocol_mock_objects.h
@@ -50,14 +50,21 @@ class MockConnectionToClient : public ConnectionToClient {
     input_stub_ = input_stub;
   }
 
+  void set_video_feedback_stub(
+      VideoFeedbackStub* video_feedback_stub) override {
+    video_feedback_stub_ = video_feedback_stub;
+  }
+
   ClipboardStub* clipboard_stub() { return clipboard_stub_; }
   HostStub* host_stub() { return host_stub_; }
   InputStub* input_stub() { return input_stub_; }
+  VideoFeedbackStub* video_feedback_stub() { return video_feedback_stub_; }
 
  private:
   ClipboardStub* clipboard_stub_;
   HostStub* host_stub_;
   InputStub* input_stub_;
+  VideoFeedbackStub* video_feedback_stub_;
 
   DISALLOW_COPY_AND_ASSIGN(MockConnectionToClient);
 };
diff --git a/remoting/protocol/video_feedback_stub.h b/remoting/protocol/video_feedback_stub.h
new file mode 100644
index 0000000..0bb5557
--- /dev/null
+++ b/remoting/protocol/video_feedback_stub.h
@@ -0,0 +1,32 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef REMOTING_PROTOCOL_VIDEO_FEEDBACK_STUB_H_
+#define REMOTING_PROTOCOL_VIDEO_FEEDBACK_STUB_H_
+
+#include "base/callback_forward.h"
+#include "base/memory/scoped_ptr.h"
+
+namespace remoting {
+
+class VideoAck;
+
+namespace protocol {
+
+class VideoFeedbackStub {
+ public:
+  virtual void ProcessVideoAck(scoped_ptr<VideoAck> video_ack) = 0;
+
+ protected:
+  VideoFeedbackStub() {}
+  virtual ~VideoFeedbackStub() {}
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(VideoFeedbackStub);
+};
+
+}  // namespace protocol
+}  // namespace remoting
+
+#endif  // REMOTING_PROTOCOL_VIDEO_FEEDBACK_STUB_H_
diff --git a/remoting/remoting_test.gypi b/remoting/remoting_test.gypi
index ad1d371..7fb8fc9 100644
--- a/remoting/remoting_test.gypi
+++ b/remoting/remoting_test.gypi
@@ -193,6 +193,7 @@
         'protocol/buffered_socket_writer_unittest.cc',
         'protocol/channel_multiplexer_unittest.cc',
         'protocol/chromium_socket_factory_unittest.cc',
+        'protocol/client_video_dispatcher_unittest.cc',
         'protocol/clipboard_echo_filter_unittest.cc',
         'protocol/clipboard_filter_unittest.cc',
         'protocol/connection_tester.cc',