Implement write-waits-for-send mode for PseudoTCP and enable it on the host side.

In the new send-confirmation mode the PseudoTCP layer will wait until data
is sent to remote end before completing each write. This significantly 
improves latency for remoting on low bandwidth networks.


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

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

3 files changed, 124 insertions, 1 deletions

diff --git a/jingle/glue/pseudotcp_adapter.cc b/jingle/glue/pseudotcp_adapter.cc
index 79dfa02..7458f89 100644
--- a/jingle/glue/pseudotcp_adapter.cc
+++ b/jingle/glue/pseudotcp_adapter.cc
@@ -52,6 +52,7 @@ class PseudoTcpAdapter::Core : public cricket::IPseudoTcpNotify,
   void SetNoDelay(bool no_delay);
   void SetReceiveBufferSize(int32 size);
   void SetSendBufferSize(int32 size);
+  void SetWriteWaitsForSend(bool write_waits_for_send);
 
   void DeleteSocket();
 
@@ -67,6 +68,10 @@ class PseudoTcpAdapter::Core : public cricket::IPseudoTcpNotify,
   void HandleReadResults(int result);
   void HandleTcpClock();
 
+  // Checks if current write has completed in the write-waits-for-send
+  // mode.
+  void CheckWriteComplete();
+
   // This re-sets |timer| without triggering callbacks.
   void AdjustClock();
 
@@ -82,6 +87,18 @@ class PseudoTcpAdapter::Core : public cricket::IPseudoTcpNotify,
   scoped_refptr<net::IOBuffer> write_buffer_;
   int write_buffer_size_;
 
+  // Whether we need to wait for data to be sent before completing write.
+  bool write_waits_for_send_;
+
+  // Set to true in the write-waits-for-send mode when we've
+  // successfully writtend data to the send buffer and waiting for the
+  // data to be sent to the remote end.
+  bool waiting_write_position_;
+
+  // Number of the bytes written by the last write stored while we wait
+  // for the data to be sent (i.e. when waiting_write_position_ = true).
+  int last_write_result_;
+
   bool socket_write_pending_;
   scoped_refptr<net::IOBuffer> socket_read_buffer_;
 
@@ -94,6 +111,8 @@ class PseudoTcpAdapter::Core : public cricket::IPseudoTcpNotify,
 PseudoTcpAdapter::Core::Core(net::Socket* socket)
     : ALLOW_THIS_IN_INITIALIZER_LIST(pseudo_tcp_(this, 0)),
       socket_(socket),
+      write_waits_for_send_(false),
+      waiting_write_position_(false),
       socket_write_pending_(false) {
   // Doesn't trigger callbacks.
   pseudo_tcp_.NotifyMTU(kDefaultMtu);
@@ -139,13 +158,29 @@ int PseudoTcpAdapter::Core::Write(net::IOBuffer* buffer, int buffer_size,
     DCHECK(result < 0);
   }
 
+  AdjustClock();
+
   if (result == net::ERR_IO_PENDING) {
     write_buffer_ = buffer;
     write_buffer_size_ = buffer_size;
     write_callback_ = callback;
+    return result;
   }
 
-  AdjustClock();
+  if (result < 0)
+    return result;
+
+  // Need to wait until the data is sent to the peer when
+  // send-confirmation mode is enabled.
+  if (write_waits_for_send_ && pseudo_tcp_.GetBytesBufferedNotSent() > 0) {
+    DCHECK(!waiting_write_position_);
+    waiting_write_position_ = true;
+    last_write_result_ = result;
+    write_buffer_ = buffer;
+    write_buffer_size_ = buffer_size;
+    write_callback_ = callback;
+    return net::ERR_IO_PENDING;
+  }
 
   return result;
 }
@@ -230,6 +265,11 @@ void PseudoTcpAdapter::Core::OnTcpWriteable(PseudoTcp* tcp) {
   if (write_callback_.is_null())
     return;
 
+  if (waiting_write_position_) {
+    CheckWriteComplete();
+    return;
+  }
+
   int result = pseudo_tcp_.Send(write_buffer_->data(), write_buffer_size_);
   if (result < 0) {
     result = net::MapSystemError(pseudo_tcp_.GetError());
@@ -240,6 +280,13 @@ void PseudoTcpAdapter::Core::OnTcpWriteable(PseudoTcp* tcp) {
 
   AdjustClock();
 
+  if (write_waits_for_send_ && pseudo_tcp_.GetBytesBufferedNotSent() > 0) {
+    DCHECK(!waiting_write_position_);
+    waiting_write_position_ = true;
+    last_write_result_ = result;
+    return;
+  }
+
   net::CompletionCallback callback = write_callback_;
   write_callback_.Reset();
   write_buffer_ = NULL;
@@ -284,6 +331,10 @@ void PseudoTcpAdapter::Core::SetSendBufferSize(int32 size) {
   pseudo_tcp_.SetOption(cricket::PseudoTcp::OPT_SNDBUF, size);
 }
 
+void PseudoTcpAdapter::Core::SetWriteWaitsForSend(bool write_waits_for_send) {
+  write_waits_for_send_ = write_waits_for_send;
+}
+
 void PseudoTcpAdapter::Core::DeleteSocket() {
   socket_.reset();
 }
@@ -348,6 +399,8 @@ void PseudoTcpAdapter::Core::HandleReadResults(int result) {
   // TODO(wez): Disconnect on failure of NotifyPacket?
   pseudo_tcp_.NotifyPacket(socket_read_buffer_->data(), result);
   AdjustClock();
+
+  CheckWriteComplete();
 }
 
 void PseudoTcpAdapter::Core::OnRead(int result) {
@@ -385,6 +438,21 @@ void PseudoTcpAdapter::Core::HandleTcpClock() {
 
   pseudo_tcp_.NotifyClock(PseudoTcp::Now());
   AdjustClock();
+
+  CheckWriteComplete();
+}
+
+void PseudoTcpAdapter::Core::CheckWriteComplete() {
+  if (!write_callback_.is_null() && waiting_write_position_) {
+    if (pseudo_tcp_.GetBytesBufferedNotSent() == 0) {
+      waiting_write_position_ = false;
+
+      net::CompletionCallback callback = write_callback_;
+      write_callback_.Reset();
+      write_buffer_ = NULL;
+      callback.Run(last_write_result_);
+    }
+  }
 }
 
 // Public interface implemention.
@@ -518,4 +586,9 @@ void PseudoTcpAdapter::SetNoDelay(bool no_delay) {
   core_->SetNoDelay(no_delay);
 }
 
+void PseudoTcpAdapter::SetWriteWaitsForSend(bool write_waits_for_send) {
+  DCHECK(CalledOnValidThread());
+  core_->SetWriteWaitsForSend(write_waits_for_send);
+}
+
 }  // namespace jingle_glue
diff --git a/jingle/glue/pseudotcp_adapter.h b/jingle/glue/pseudotcp_adapter.h
index b112fec..eed2504 100644
--- a/jingle/glue/pseudotcp_adapter.h
+++ b/jingle/glue/pseudotcp_adapter.h
@@ -59,6 +59,24 @@ class PseudoTcpAdapter : public net::StreamSocket, base::NonThreadSafe {
   // Set whether Nagle's algorithm is enabled.
   void SetNoDelay(bool no_delay);
 
+  // When write_waits_for_send flag is set to true the Write() method
+  // will wait until the data is sent to the remote end before the
+  // write completes (it still doesn't wait until the data is received
+  // and acknowledged by the remote end). Otherwise write completes
+  // after the data has been copied to the send buffer.
+  //
+  // This flag is useful in cases when the sender needs to get
+  // feedback from the connection when it is congested. E.g. remoting
+  // host uses this feature to adjust screen capturing rate according
+  // to the available bandwidth. In the same time it may negatively
+  // impact performance in some cases. E.g. when the sender writes one
+  // byte at a time then each byte will always be sent in a separate
+  // packet.
+  //
+  // TODO(sergeyu): Remove this flag once remoting has a better
+  // flow-control solution.
+  void SetWriteWaitsForSend(bool write_waits_for_send);
+
  private:
   class Core;
 
diff --git a/jingle/glue/pseudotcp_adapter_unittest.cc b/jingle/glue/pseudotcp_adapter_unittest.cc
index fd070e6..4b4431e 100644
--- a/jingle/glue/pseudotcp_adapter_unittest.cc
+++ b/jingle/glue/pseudotcp_adapter_unittest.cc
@@ -407,6 +407,38 @@ TEST_F(PseudoTcpAdapterTest, DeleteOnConnected) {
   ASSERT_EQ(NULL, host_pseudotcp_.get());
 }
 
+// Verify that we can send/receive data with the write-waits-for-send
+// flag set.
+TEST_F(PseudoTcpAdapterTest, WriteWaitsForSendLetsDataThrough) {
+  net::TestCompletionCallback host_connect_cb;
+  net::TestCompletionCallback client_connect_cb;
+
+  host_pseudotcp_->SetWriteWaitsForSend(true);
+  client_pseudotcp_->SetWriteWaitsForSend(true);
+
+  // Disable Nagle's algorithm because the test is slow when it is
+  // enabled.
+  host_pseudotcp_->SetNoDelay(true);
+
+  int rv1 = host_pseudotcp_->Connect(host_connect_cb.callback());
+  int rv2 = client_pseudotcp_->Connect(client_connect_cb.callback());
+
+  if (rv1 == net::ERR_IO_PENDING)
+    rv1 = host_connect_cb.WaitForResult();
+  if (rv2 == net::ERR_IO_PENDING)
+    rv2 = client_connect_cb.WaitForResult();
+  ASSERT_EQ(net::OK, rv1);
+  ASSERT_EQ(net::OK, rv2);
+
+  scoped_refptr<TCPChannelTester> tester =
+      new TCPChannelTester(&message_loop_, host_pseudotcp_.get(),
+                           client_pseudotcp_.get());
+
+  tester->Start();
+  message_loop_.Run();
+  tester->CheckResults();
+}
+
 }  // namespace
 
 }  // namespace jingle_glue