Cleanup RectangleUpdateDecoder and VideoStub

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 "base/single_thread_task_runner.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)
-    : packet(packet.release()),
-      done(done) {
-}
-
-RectangleUpdateDecoder::QueuedVideoPacket::~QueuedVideoPacket() {
-}
-
 RectangleUpdateDecoder::RectangleUpdateDecoder(
     scoped_refptr<base::SingleThreadTaskRunner> main_task_runner,
     scoped_refptr<base::SingleThreadTaskRunner> decode_task_runner,
     scoped_refptr<FrameConsumerProxy> consumer)
     : 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),
-      packet_being_processed_(false),
+      stopped_(false),
       latest_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(VideoDecoderRowBased::CreateVerbatimDecoder());
   } else if (codec == ChannelConfig::CODEC_ZIP) {
     decoder_.reset(VideoDecoderRowBased::CreateZlibDecoder());
   } else if (codec == ChannelConfig::CODEC_VP8) {
     decoder_.reset(new VideoDecoderVp8());
   } else {
     NOTREACHED() << "Invalid Encoding found: " << codec;
   }
 }
 
+void RectangleUpdateDecoder::Stop() {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());

[Wez, 2012/08/28 16:42:27]

nit: Blank line after DCHECK.

[Sergey Ulanov, 2012/08/28 18:30:27]

Done.

+  // |stopped_| is used on the decode thread, but changed only here on the main
+  // thread. Don't need any synchronization here because it's not an issue
+  // if the decode thread sees false instead of true.

[Wez, 2012/08/28 16:42:27]

Why not just re-post the Stop() to set this on the decode thread, so you know
that it'll be seen as soon as that gets processed?

[Sergey Ulanov, 2012/08/28 18:30:27]

Posting a task on the decoder thread will not do what we want here. We want to
stop decoding of the frames that are already in the task queue for the decode
thread. If I post a task to the decode thread it will only be executed after all
pending decoder tasks.

[Wez, 2012/08/29 23:18:05]

OK, but you've stated that we don't need to lock here because the it's not an
issue if the thread sees false instead of true; if the optimization of stopping
processing on the other thread is valuable then it's worth putting a lock around
this to create a memory barrier and ensure that the other thread sees the
change.

[Sergey Ulanov, 2012/08/29 23:31:10]

Actually after thinking more about it I think we don't need this method at all.
Normally MessageReader class doesn't read from the socket if there are any
pending messages. The only case when it may allow several messages to be
processed concurrently is when they were read in a single Read() call - in that
case they are likely to be empty packets, so we don't win much by canceling them
anyway.
Removed this method.

+  stopped_ = true;
+}
+
 void RectangleUpdateDecoder::DecodePacket(scoped_ptr<VideoPacket> packet,
                                           const base::Closure& done) {
   DCHECK(decode_task_runner_->BelongsToCurrentThread());
 
   base::ScopedClosureRunner done_runner(done);
+
+  if (stopped_)
+    return;
+
   bool decoder_needs_reset = false;
   bool notify_size_or_dpi_change = false;
 
   // If the packet includes screen size or DPI information, store them.
   if (packet->format().has_screen_width() &&
       packet->format().has_screen_height()) {
     SkISize source_size = SkISize::Make(packet->format().screen_width(),
                                         packet->format().screen_height());
     if (source_size_ != source_size) {
       source_size_ = source_size;
@@ skipping 171 matching lines @@
     stats_.video_encode_ms()->Record(packet->encode_time_ms());
   if (packet->has_client_sequence_number() &&
       packet->client_sequence_number() > latest_sequence_number_) {
     latest_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_)
-    ProcessNextPacket();
-}
-
-int RectangleUpdateDecoder::GetPendingVideoPackets() {
-  DCHECK(main_task_runner_->BelongsToCurrentThread());
-  return received_packets_.size();
-}
-
-void RectangleUpdateDecoder::DropAllPackets() {
-  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::ProcessNextPacket() {
-  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();
 
-  base::Closure callback = base::Bind(&RectangleUpdateDecoder::OnPacketDone,
-                                      this,
-                                      last_packet,
-                                      decode_start);
+  base::Closure callback =

[Wez, 2012/08/28 16:42:27]

nit: callback -> decode_done? (More consistent with |done|).

[Sergey Ulanov, 2012/08/28 18:30:27]

Done.

+      base::Bind(&RectangleUpdateDecoder::OnPacketDone, this,
+                 last_packet, decode_start, done);
 
   decode_task_runner_->PostTask(FROM_HERE, base::Bind(
       &RectangleUpdateDecoder::DecodePacket, this,
       base::Passed(&packet), callback));
 }
 
 void RectangleUpdateDecoder::OnPacketDone(bool last_packet,
-                                          base::Time decode_start) {
+                                          base::Time decode_start,
+                                          const base::Closure& done) {
   if (!main_task_runner_->BelongsToCurrentThread()) {
     main_task_runner_->PostTask(FROM_HERE, base::Bind(
         &RectangleUpdateDecoder::OnPacketDone, this,
-        last_packet, decode_start));
+        last_packet, decode_start, done));
     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.
-  ProcessNextPacket();
+  done.Run();
 }
 
 ChromotingStats* RectangleUpdateDecoder::GetStats() {
   DCHECK(main_task_runner_->BelongsToCurrentThread());
   return &stats_;
 }
 
 }  // namespace remoting