Moved video stub implementation to RectangleUpdateDecoder.

remoting/client/rectangle_update_decoder.cc

 // Copyright (c) 2012 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/client/rectangle_update_decoder.h"
 
 #include "base/bind.h"
 #include "base/bind_helpers.h"
-#include "base/callback.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/message_loop_proxy.h"
 #include "ppapi/cpp/image_data.h"
 #include "remoting/base/util.h"
 #include "remoting/codec/video_decoder.h"
 #include "remoting/codec/video_decoder_row_based.h"
 #include "remoting/codec/video_decoder_vp8.h"
 #include "remoting/client/frame_consumer.h"
 #include "remoting/protocol/session_config.h"
 
 using base::Passed;
 using remoting::protocol::ChannelConfig;
 using remoting::protocol::SessionConfig;
 
 namespace remoting {
 
+RectangleUpdateDecoder::QueuedVideoPacket::QueuedVideoPacket(
+    scoped_ptr<VideoPacket> packet, const base::Closure& done)

[Sergey Ulanov, 2012/08/23 22:22:09]

nit: one argument per line please when arguments are wrapped.

[kxing, 2012/08/24 17:05:22]

Done.

+    : packet(packet.release()), done(done) {

[Sergey Ulanov, 2012/08/23 22:22:09]

nit: move done initializer to a separate line

[kxing, 2012/08/24 17:05:22]

Done.

+}
+
+RectangleUpdateDecoder::QueuedVideoPacket::~QueuedVideoPacket() {
+}
+
 RectangleUpdateDecoder::RectangleUpdateDecoder(
-    scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+    scoped_refptr<base::SingleThreadTaskRunner> main_task_runner,
+    scoped_refptr<base::SingleThreadTaskRunner> decode_task_runner,
     scoped_refptr<FrameConsumerProxy> consumer)
-    : task_runner_(task_runner),
+    : main_task_runner_(main_task_runner),
+      decode_task_runner_(decode_task_runner),
       consumer_(consumer),
       source_size_(SkISize::Make(0, 0)),
       source_dpi_(SkIPoint::Make(0, 0)),
       view_size_(SkISize::Make(0, 0)),
       clip_area_(SkIRect::MakeEmpty()),
-      paint_scheduled_(false) {
+      paint_scheduled_(false),
+      packet_being_processed_(false),
+      last_sequence_number_(0) {
 }
 
 RectangleUpdateDecoder::~RectangleUpdateDecoder() {
 }
 
 void RectangleUpdateDecoder::Initialize(const SessionConfig& config) {
   // Initialize decoder based on the selected codec.
   ChannelConfig::Codec codec = config.video_config().codec;
   if (codec == ChannelConfig::CODEC_VERBATIM) {
     decoder_.reset(DecoderRowBased::CreateVerbatimDecoder());
   } else if (codec == ChannelConfig::CODEC_ZIP) {
     decoder_.reset(DecoderRowBased::CreateZlibDecoder());
   } else if (codec == ChannelConfig::CODEC_VP8) {
     decoder_.reset(new DecoderVp8());
   } else {
     NOTREACHED() << "Invalid Encoding found: " << codec;
   }
 }
 
 void RectangleUpdateDecoder::DecodePacket(scoped_ptr<VideoPacket> packet,
                                           const base::Closure& done) {
-  if (!task_runner_->BelongsToCurrentThread()) {
-    task_runner_->PostTask(
+  if (!decode_task_runner_->BelongsToCurrentThread()) {

[Sergey Ulanov, 2012/08/23 22:22:09]

Better to PostTask() in DispatchPacket() and here just DCHECK that you are on
the right thread.

[kxing, 2012/08/24 17:05:22]

Done.

+    decode_task_runner_->PostTask(
         FROM_HERE, base::Bind(&RectangleUpdateDecoder::DecodePacket,
                               this, base::Passed(&packet), done));
     return;
   }
 
   base::ScopedClosureRunner done_runner(done);
   bool decoder_needs_reset = false;
   bool notify_size_or_dpi_change = false;
 
   // If the packet includes screen size or DPI information, store them.
@@ skipping 32 matching lines @@
   }
 
   if (decoder_->DecodePacket(packet.get()) == Decoder::DECODE_DONE)
     SchedulePaint();
 }
 
 void RectangleUpdateDecoder::SchedulePaint() {
   if (paint_scheduled_)
     return;
   paint_scheduled_ = true;
-  task_runner_->PostTask(
+  decode_task_runner_->PostTask(
       FROM_HERE, base::Bind(&RectangleUpdateDecoder::DoPaint, this));
 }
 
 void RectangleUpdateDecoder::DoPaint() {
   DCHECK(paint_scheduled_);
   paint_scheduled_ = false;
 
   // If the view size is empty or we have no output buffers ready, return.
   if (buffers_.empty() || view_size_.isEmpty())
     return;
@@ skipping 11 matching lines @@
                         &output_region);
 
   // Notify the consumer that painting is done.
   if (!output_region.isEmpty()) {
     buffers_.pop_front();
     consumer_->ApplyBuffer(view_size_, clip_area_, buffer, output_region);
   }
 }
 
 void RectangleUpdateDecoder::RequestReturnBuffers(const base::Closure& done) {
-  if (!task_runner_->BelongsToCurrentThread()) {
-    task_runner_->PostTask(
+  if (!decode_task_runner_->BelongsToCurrentThread()) {
+    decode_task_runner_->PostTask(
         FROM_HERE, base::Bind(&RectangleUpdateDecoder::RequestReturnBuffers,
         this, done));
     return;
   }
 
   while (!buffers_.empty()) {
     consumer_->ReturnBuffer(buffers_.front());
     buffers_.pop_front();
   }
 
   if (!done.is_null())
     done.Run();
 }
 
 void RectangleUpdateDecoder::DrawBuffer(pp::ImageData* buffer) {
-  if (!task_runner_->BelongsToCurrentThread()) {
-    task_runner_->PostTask(
+  if (!decode_task_runner_->BelongsToCurrentThread()) {
+    decode_task_runner_->PostTask(
         FROM_HERE, base::Bind(&RectangleUpdateDecoder::DrawBuffer,
                               this, buffer));
     return;
   }
 
   DCHECK(clip_area_.width() <= buffer->size().width() &&
          clip_area_.height() <= buffer->size().height());
 
   buffers_.push_back(buffer);
   SchedulePaint();
 }
 
 void RectangleUpdateDecoder::InvalidateRegion(const SkRegion& region) {
-  if (!task_runner_->BelongsToCurrentThread()) {
-    task_runner_->PostTask(
+  if (!decode_task_runner_->BelongsToCurrentThread()) {
+    decode_task_runner_->PostTask(
         FROM_HERE, base::Bind(&RectangleUpdateDecoder::InvalidateRegion,
                               this, region));
     return;
   }
 
   if (decoder_.get()) {
     decoder_->Invalidate(view_size_, region);
     SchedulePaint();
   }
 }
 
 void RectangleUpdateDecoder::SetOutputSizeAndClip(const SkISize& view_size,
                                                   const SkIRect& clip_area) {
-  if (!task_runner_->BelongsToCurrentThread()) {
-    task_runner_->PostTask(
+  if (!decode_task_runner_->BelongsToCurrentThread()) {
+    decode_task_runner_->PostTask(
         FROM_HERE, base::Bind(&RectangleUpdateDecoder::SetOutputSizeAndClip,
                               this, view_size, clip_area));
     return;
   }
 
   // The whole frame needs to be repainted if the scaling factor has changed.
   if (view_size_ != view_size && decoder_.get()) {
     SkRegion region;
     region.op(SkIRect::MakeSize(view_size), SkRegion::kUnion_Op);
     decoder_->Invalidate(view_size, region);
@@ skipping 15 matching lines @@
         i = buffers_.erase(i);
       } else {
         ++i;
       }
     }
 
     SchedulePaint();
   }
 }
 
+void RectangleUpdateDecoder::ProcessVideoPacket(scoped_ptr<VideoPacket> packet,
+                                          const base::Closure& done) {

[Sergey Ulanov, 2012/08/23 22:22:09]

nit: indentation

[kxing, 2012/08/24 17:05:22]

Done.

+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
+  // If the video packet is empty then drop it. Empty packets are used to
+  // maintain activity on the network.
+  if (!packet->has_data() || packet->data().size() == 0) {
+    done.Run();
+    return;
+  }
+
+  // Add one frame to the counter.
+  stats_.video_frame_rate()->Record(1);
+
+  // Record other statistics received from host.
+  stats_.video_bandwidth()->Record(packet->data().size());
+  if (packet->has_capture_time_ms())
+    stats_.video_capture_ms()->Record(packet->capture_time_ms());
+  if (packet->has_encode_time_ms())
+    stats_.video_encode_ms()->Record(packet->encode_time_ms());
+  if (packet->has_client_sequence_number() &&
+      packet->client_sequence_number() > last_sequence_number_) {
+    last_sequence_number_ = packet->client_sequence_number();
+    base::TimeDelta round_trip_latency =
+        base::Time::Now() -
+        base::Time::FromInternalValue(packet->client_sequence_number());
+    stats_.round_trip_ms()->Record(round_trip_latency.InMilliseconds());
+  }
+
+  received_packets_.push_back(QueuedVideoPacket(packet.Pass(), done));
+  if (!packet_being_processed_)
+    DispatchPacket();
+}
+
+int RectangleUpdateDecoder::GetPendingVideoPackets() {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+  return received_packets_.size();
+}
+
+void RectangleUpdateDecoder::DropAllPackets() {

[Sergey Ulanov, 2012/08/23 22:22:09]

I think we no longer need this method. (we needed it previously because the done
task was destroying the packets, but that's not the case anymore).

If you remove this method, then you no longer need received_packets_ list - you
can just queue pending packets in the task queue of the decoder thread. So
instead of this method you will just need StopDecoding() method that just sets a
flag and then DecodePacket() will do nothing once that flag is set.

[kxing, 2012/08/24 17:05:22]

I'll let someone do this in a separate CL.

+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+
+  while(!received_packets_.empty()) {
+    delete received_packets_.front().packet;
+    received_packets_.front().done.Run();
+    received_packets_.pop_front();
+  }
+}
+
+void RectangleUpdateDecoder::DispatchPacket() {

[Wez, 2012/08/23 22:10:48]

nit: Rename to ProcessNextPacket?

[kxing, 2012/08/24 17:05:22]

Done.

+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+  CHECK(!packet_being_processed_);
+
+  if (received_packets_.empty()) {
+    // Nothing to do!
+    return;
+  }
+
+  scoped_ptr<VideoPacket> packet(received_packets_.front().packet);
+  received_packets_.front().packet = NULL;
+  packet_being_processed_ = true;
+
+  // Measure the latency between the last packet being received and presented.
+  bool last_packet = (packet->flags() & VideoPacket::LAST_PACKET) != 0;
+  base::Time decode_start;
+  if (last_packet)
+    decode_start = base::Time::Now();
+
+  DecodePacket(
+      packet.Pass(),
+      base::Bind(&RectangleUpdateDecoder::OnPacketDone, base::Unretained(this),

[Sergey Ulanov, 2012/08/23 22:22:09]

Why do you need Unretained() here? This object is ref-counted isn't it. Not sure
if it's safe here

[kxing, 2012/08/24 17:05:22]

Done.

+                 last_packet, decode_start));
+}
+
+void RectangleUpdateDecoder::OnPacketDone(bool last_packet,
+                                    base::Time decode_start) {

[Sergey Ulanov, 2012/08/23 22:22:09]

nit: indentation

[kxing, 2012/08/24 17:05:22]

Done.

+  if (!main_task_runner_->BelongsToCurrentThread()) {
+    main_task_runner_->PostTask(FROM_HERE, base::Bind(
+        &RectangleUpdateDecoder::OnPacketDone, base::Unretained(this),
+        last_packet, decode_start));
+    return;
+  }
+
+  // Record the latency between the final packet being received and
+  // presented.
+  if (last_packet) {
+    stats_.video_decode_ms()->Record(
+        (base::Time::Now() - decode_start).InMilliseconds());
+  }
+
+  received_packets_.front().done.Run();
+  received_packets_.pop_front();
+
+  packet_being_processed_ = false;
+
+  // Process the next video packet.
+  DispatchPacket();
+}
+
+ChromotingStats* RectangleUpdateDecoder::GetStats() {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+  return &stats_;
+}
+
 }  // namespace remoting