Simplified frame rate control in the chromoting host.

Insted of keeping semi-fixed frame rate, now capturing rate is controlled
by how fast we can send data to the client. Capturing of frame n is
started only after frame n-2 is sent (while n-1 is being encoded). This
guarantees that we don't clog the video channel buffers, and that we start
capturing only if we know that the frame will not need to wait for too long
in the buffer.

TEST=None
BUG=None

Review URL: http://codereview.chromium.org/5634002

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@68688 0039d316-1c4b-4281-b951-d872f2087c98

7 files changed, 64 insertions, 32 deletions

diff --git a/remoting/protocol/buffered_socket_writer.cc b/remoting/protocol/buffered_socket_writer.cc
index 0cf39bc..11b8395 100644
--- a/remoting/protocol/buffered_socket_writer.cc
+++ b/remoting/protocol/buffered_socket_writer.cc
@@ -5,11 +5,34 @@
 #include "remoting/protocol/buffered_socket_writer.h"
 
 #include "base/message_loop.h"
+#include "base/stl_util-inl.h"
 #include "net/base/net_errors.h"
 
 namespace remoting {
 namespace protocol {
 
+class BufferedSocketWriterBase::PendingPacket {
+ public:
+  PendingPacket(scoped_refptr<net::IOBufferWithSize> data, Task* done_task)
+      : data_(data),
+        done_task_(done_task) {
+  }
+  ~PendingPacket() {
+    if (done_task_.get())
+      done_task_->Run();
+  }
+
+  net::IOBufferWithSize* data() {
+    return data_;
+  }
+
+ private:
+  scoped_refptr<net::IOBufferWithSize> data_;
+  scoped_ptr<Task> done_task_;
+
+  DISALLOW_COPY_AND_ASSIGN(PendingPacket);
+};
+
 BufferedSocketWriterBase::BufferedSocketWriterBase()
     : buffer_size_(0),
       socket_(NULL),
@@ -32,11 +55,11 @@ void BufferedSocketWriterBase::Init(net::Socket* socket,
 }
 
 bool BufferedSocketWriterBase::Write(
-    scoped_refptr<net::IOBufferWithSize> data) {
+    scoped_refptr<net::IOBufferWithSize> data, Task* done_task) {
   AutoLock auto_lock(lock_);
   if (!socket_)
     return false;
-  queue_.push(data);
+  queue_.push_back(new PendingPacket(data, done_task));
   buffer_size_ += data->size();
   message_loop_->PostTask(
       FROM_HERE, NewRunnableMethod(this, &BufferedSocketWriterBase::DoWrite));
@@ -112,9 +135,7 @@ void BufferedSocketWriterBase::OnWritten(int result) {
 void BufferedSocketWriterBase::HandleError(int result) {
   AutoLock auto_lock(lock_);
   closed_ = true;
-  while (!queue_.empty()) {
-    queue_.pop();
-  }
+  STLDeleteElements(&queue_);
 
   // Notify subclass that an error is received.
   OnError_Locked(result);
@@ -135,19 +156,27 @@ void BufferedSocketWriterBase::Close() {
   closed_ = true;
 }
 
+void BufferedSocketWriterBase::PopQueue() {
+  // This also calls |done_task|.
+  delete queue_.front();
+  queue_.pop_front();
+}
+
 BufferedSocketWriter::BufferedSocketWriter() { }
-BufferedSocketWriter::~BufferedSocketWriter() { }
+
+BufferedSocketWriter::~BufferedSocketWriter() {
+  STLDeleteElements(&queue_);
+}
 
 void BufferedSocketWriter::GetNextPacket_Locked(
     net::IOBuffer** buffer, int* size) {
-  while (!current_buf_ || current_buf_->BytesRemaining() == 0) {
+  if (!current_buf_) {
     if (queue_.empty()) {
       *buffer = NULL;
       return;  // Nothing to write.
     }
-    current_buf_ =
-        new net::DrainableIOBuffer(queue_.front(), queue_.front()->size());
-    queue_.pop();
+    current_buf_ = new net::DrainableIOBuffer(
+        queue_.front()->data(), queue_.front()->data()->size());
   }
 
   *buffer = current_buf_;
@@ -157,6 +186,11 @@ void BufferedSocketWriter::GetNextPacket_Locked(
 void BufferedSocketWriter::AdvanceBufferPosition_Locked(int written) {
   buffer_size_ -= written;
   current_buf_->DidConsume(written);
+
+  if (current_buf_->BytesRemaining() == 0) {
+    PopQueue();
+    current_buf_ = NULL;
+  }
 }
 
 void BufferedSocketWriter::OnError_Locked(int result) {
@@ -172,14 +206,14 @@ void BufferedDatagramWriter::GetNextPacket_Locked(
     *buffer = NULL;
     return;  // Nothing to write.
   }
-  *buffer = queue_.front();
-  *size = queue_.front()->size();
+  *buffer = queue_.front()->data();
+  *size = queue_.front()->data()->size();
 }
 
 void BufferedDatagramWriter::AdvanceBufferPosition_Locked(int written) {
-  DCHECK_EQ(written, queue_.front()->size());
-  buffer_size_ -= queue_.front()->size();
-  queue_.pop();
+  DCHECK_EQ(written, queue_.front()->data()->size());
+  buffer_size_ -= queue_.front()->data()->size();
+  PopQueue();
 }
 
 void BufferedDatagramWriter::OnError_Locked(int result) {
diff --git a/remoting/protocol/buffered_socket_writer.h b/remoting/protocol/buffered_socket_writer.h
index c786c44..3ea127e 100644
--- a/remoting/protocol/buffered_socket_writer.h
+++ b/remoting/protocol/buffered_socket_writer.h
@@ -5,7 +5,7 @@
 #ifndef REMOTING_PROTOCOL_BUFFERED_SOCKET_WRITER_H_
 #define REMOTING_PROTOCOL_BUFFERED_SOCKET_WRITER_H_
 
-#include <queue>
+#include <list>
 
 #include "base/lock.h"
 #include "base/ref_counted.h"
@@ -13,6 +13,7 @@
 #include "net/socket/socket.h"
 
 class MessageLoop;
+class Task;
 
 namespace net {
 class Socket;
@@ -45,7 +46,7 @@ class BufferedSocketWriterBase
 
   // Puts a new data chunk in the buffer. Returns false and doesn't enqueue
   // the data if called before Init(). Can be called on any thread.
-  bool Write(scoped_refptr<net::IOBufferWithSize> buffer);
+  bool Write(scoped_refptr<net::IOBufferWithSize> buffer, Task* done_task);
 
   // Returns current size of the buffer. Can be called on any thread.
   int GetBufferSize();
@@ -58,11 +59,16 @@ class BufferedSocketWriterBase
   void Close();
 
  protected:
-  typedef std::queue<scoped_refptr<net::IOBufferWithSize> > DataQueue;
+  class PendingPacket;
+  typedef std::list<PendingPacket*> DataQueue;
 
   DataQueue queue_;
   int buffer_size_;
 
+  // Removes element from the front of the queue and calls |done_task|
+  // for that element.
+  void PopQueue();
+
   // Following three methods must be implemented in child classes.
   // GetNextPacket() returns next packet that needs to be written to the
   // socket. |buffer| must be set to NULL if there is nothing left in the queue.
diff --git a/remoting/protocol/client_control_sender.cc b/remoting/protocol/client_control_sender.cc
index d1fd2f5..c16d235 100644
--- a/remoting/protocol/client_control_sender.cc
+++ b/remoting/protocol/client_control_sender.cc
@@ -28,9 +28,7 @@ void ClientControlSender::NotifyResolution(
     const NotifyResolutionRequest* msg, Task* done) {
   protocol::ControlMessage message;
   message.mutable_notify_resolution()->CopyFrom(*msg);
-  buffered_writer_->Write(SerializeAndFrameMessage(message));
-  done->Run();
-  delete done;
+  buffered_writer_->Write(SerializeAndFrameMessage(message), done);
 }
 
 }  // namespace protocol
diff --git a/remoting/protocol/input_sender.cc b/remoting/protocol/input_sender.cc
index e45149e..8831649 100644
--- a/remoting/protocol/input_sender.cc
+++ b/remoting/protocol/input_sender.cc
@@ -31,9 +31,7 @@ void InputSender::InjectKeyEvent(const KeyEvent* event, Task* done) {
   // TODO(hclam): Provide timestamp.
   evt->set_timestamp(0);
   evt->mutable_key()->CopyFrom(*event);
-  buffered_writer_->Write(SerializeAndFrameMessage(message));
-  done->Run();
-  delete done;
+  buffered_writer_->Write(SerializeAndFrameMessage(message), done);
 }
 
 void InputSender::InjectMouseEvent(const MouseEvent* event, Task* done) {
@@ -42,9 +40,7 @@ void InputSender::InjectMouseEvent(const MouseEvent* event, Task* done) {
   // TODO(hclam): Provide timestamp.
   evt->set_timestamp(0);
   evt->mutable_mouse()->CopyFrom(*event);
-  buffered_writer_->Write(SerializeAndFrameMessage(message));
-  done->Run();
-  delete done;
+  buffered_writer_->Write(SerializeAndFrameMessage(message), done);
 }
 
 }  // namespace protocol
diff --git a/remoting/protocol/protobuf_video_writer.cc b/remoting/protocol/protobuf_video_writer.cc
index 73a9b0d..025445d 100644
--- a/remoting/protocol/protobuf_video_writer.cc
+++ b/remoting/protocol/protobuf_video_writer.cc
@@ -25,9 +25,7 @@ void ProtobufVideoWriter::Init(protocol::Session* session) {
 
 void ProtobufVideoWriter::ProcessVideoPacket(const VideoPacket* packet,
                                              Task* done) {
-  buffered_writer_->Write(SerializeAndFrameMessage(*packet));
-  done->Run();
-  delete done;
+  buffered_writer_->Write(SerializeAndFrameMessage(*packet), done);
 }
 
 int ProtobufVideoWriter::GetPendingPackets() {
diff --git a/remoting/protocol/rtcp_writer.cc b/remoting/protocol/rtcp_writer.cc
index c7d53f0..1996665 100644
--- a/remoting/protocol/rtcp_writer.cc
+++ b/remoting/protocol/rtcp_writer.cc
@@ -32,7 +32,7 @@ void RtcpWriter::SendReport(const RtcpReceiverReport& report) {
   PackRtcpReceiverReport(report, reinterpret_cast<uint8*>(buffer->data()),
                          size);
 
-  buffered_rtcp_writer_->Write(buffer);
+  buffered_rtcp_writer_->Write(buffer, NULL);
 }
 
 }  // namespace protocol
diff --git a/remoting/protocol/rtp_writer.cc b/remoting/protocol/rtp_writer.cc
index 7341741..e26bdb7 100644
--- a/remoting/protocol/rtp_writer.cc
+++ b/remoting/protocol/rtp_writer.cc
@@ -69,7 +69,7 @@ void RtpWriter::SendPacket(uint32 timestamp, bool marker,
                  payload_size);
 
   // And write the packet.
-  buffered_rtp_writer_->Write(buffer);
+  buffered_rtp_writer_->Write(buffer, NULL);
 }
 
 int RtpWriter::GetPendingPackets() {