Replace send packet throttling with a scheme based on number of in-flight bytes in IpcPacketSocket.

content/renderer/p2p/ipc_socket_factory.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 "content/renderer/p2p/ipc_socket_factory.h"
 
+#include <deque>
+
 #include "base/compiler_specific.h"
 #include "base/debug/trace_event.h"
 #include "base/message_loop.h"
 #include "base/message_loop_proxy.h"
 #include "content/renderer/p2p/socket_client.h"
 #include "content/renderer/p2p/socket_dispatcher.h"
 #include "jingle/glue/utils.h"
 #include "third_party/libjingle/source/talk/base/asyncpacketsocket.h"
 
 namespace content {
 
 namespace {
 
-// TODO(hclam): This shouldn't be a pre-defined value. Bug: crbug.com/181321.
-const int kMaxPendingPackets = 32;
-const int kWritableSignalThreshold = 0;
+// TODO(miu): This needs tuning.  http://crbug.com/237960
+const size_t kMaximumInFlightBytes = 64 * 1024;  // 64 KB
 
 // IpcPacketSocket implements talk_base::AsyncPacketSocket interface
 // using P2PSocketClient that works over IPC-channel. It must be used
 // on the thread it was created.
 class IpcPacketSocket : public talk_base::AsyncPacketSocket,
                         public P2PSocketClient::Delegate {
  public:
   IpcPacketSocket();
   virtual ~IpcPacketSocket();
 
@@ skipping 26 matching lines @@
 
  private:
   enum InternalState {
     IS_UNINITIALIZED,
     IS_OPENING,
     IS_OPEN,
     IS_CLOSED,
     IS_ERROR,
   };
 
+  // Update trace of send throttling internal state. This should be called
+  // immediately after any changes to |send_bytes_available_| and/or
+  // |in_flight_packet_sizes_|.
+  void TraceSendThrottlingState() const;
+
   void InitAcceptedTcp(P2PSocketClient* client,
                        const talk_base::SocketAddress& local_address,
                        const talk_base::SocketAddress& remote_address);
 
   P2PSocketType type_;
 
   // Message loop on which this socket was created and being used.
   base::MessageLoop* message_loop_;
 
   // Corresponding P2P socket client.
   scoped_refptr<P2PSocketClient> client_;
 
   // Local address is allocated by the browser process, and the
   // renderer side doesn't know the address until it receives OnOpen()
   // event from the browser.
   talk_base::SocketAddress local_address_;
 
   // Remote address for client TCP connections.
   talk_base::SocketAddress remote_address_;
 
   // Current state of the object.
   InternalState state_;
 
-  // Number which have been sent to the browser, but for which we haven't
-  // received response.
-  int send_packets_pending_;
+  // Track the number of bytes allowed to be sent non-blocking. This is used to
+  // throttle the sending of packets to the browser process. For each packet
+  // sent, the value is decreased. As callbacks to OnSendComplete() (as IPCs
+  // from the browser process) are made, the value is increased back. This
+  // allows short bursts of high-rate sending without dropping packets, but
+  // quickly restricts the client to a sustainable steady-state rate.
+  size_t send_bytes_available_;
+  std::deque<size_t> in_flight_packet_sizes_;
 
   // Set to true once EWOULDBLOCK was returned from Send(). Indicates that the
   // caller expects SignalWritable notification.
   bool writable_signal_expected_;
 
   // Current error code. Valid when state_ == IS_ERROR.
   int error_;
 
   DISALLOW_COPY_AND_ASSIGN(IpcPacketSocket);
 };
 
 IpcPacketSocket::IpcPacketSocket()
     : type_(P2P_SOCKET_UDP),
       message_loop_(base::MessageLoop::current()),
       state_(IS_UNINITIALIZED),
-      send_packets_pending_(0),
+      send_bytes_available_(kMaximumInFlightBytes),
       writable_signal_expected_(false),
-      error_(0) {}
+      error_(0) {
+  COMPILE_ASSERT(kMaximumInFlightBytes > 0, would_send_at_zero_rate);
+}
 
 IpcPacketSocket::~IpcPacketSocket() {
   if (state_ == IS_OPENING || state_ == IS_OPEN ||
       state_ == IS_ERROR) {
     Close();
   }
 }
 
+void IpcPacketSocket::TraceSendThrottlingState() const {
+  TRACE_COUNTER1("p2p", "P2PSendBytesAvailable", send_bytes_available_);
+  TRACE_COUNTER1("p2p", "P2PSendPacketsInFlight",
+                 in_flight_packet_sizes_.size());
+}
+
 bool IpcPacketSocket::Init(P2PSocketType type, P2PSocketClient* client,
                            const talk_base::SocketAddress& local_address,
                            const talk_base::SocketAddress& remote_address) {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
   DCHECK_EQ(state_, IS_UNINITIALIZED);
 
   type_ = type;
   client_ = client;
   local_address_ = local_address;
   remote_address_ = remote_address;
@@ skipping 19 matching lines @@
     P2PSocketClient* client,
     const talk_base::SocketAddress& local_address,
     const talk_base::SocketAddress& remote_address) {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
   DCHECK_EQ(state_, IS_UNINITIALIZED);
 
   client_ = client;
   local_address_ = local_address;
   remote_address_ = remote_address;
   state_ = IS_OPEN;
+  TraceSendThrottlingState();
   client_->set_delegate(this);
 }
 
 // talk_base::AsyncPacketSocket interface.
 talk_base::SocketAddress IpcPacketSocket::GetLocalAddress() const {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
   return local_address_;
 }
 
 talk_base::SocketAddress IpcPacketSocket::GetRemoteAddress() const {
@@ skipping 18 matching lines @@
       return EWOULDBLOCK;
     case IS_CLOSED:
       return ENOTCONN;
     case IS_ERROR:
       return error_;
     case IS_OPEN:
       // Continue sending the packet.
       break;
   }
 
-  if (send_packets_pending_ > kMaxPendingPackets) {
-    TRACE_EVENT_INSTANT1("p2p", "MaxPendingPacketsWouldBlock",
+  if (data_size == 0) {
+    NOTREACHED();
+    return 0;
+  }
+
+  if (data_size > send_bytes_available_) {
+    TRACE_EVENT_INSTANT1("p2p", "MaxPendingBytesWouldBlock",
                          TRACE_EVENT_SCOPE_THREAD, "id", client_->socket_id());
     writable_signal_expected_ = true;
     error_ = EWOULDBLOCK;
     return -1;
   }
 
-  const char* data_char = reinterpret_cast<const char*>(data);
-  std::vector<char> data_vector(data_char, data_char + data_size);
-
   net::IPEndPoint address_chrome;
   if (!jingle_glue::SocketAddressToIPEndPoint(address, &address_chrome)) {
     NOTREACHED();
     return -1;
   }
 
-  ++send_packets_pending_;
+  send_bytes_available_ -= data_size;
+  in_flight_packet_sizes_.push_back(data_size);
+  TraceSendThrottlingState();
+
+  const char* data_char = reinterpret_cast<const char*>(data);
+  std::vector<char> data_vector(data_char, data_char + data_size);
   client_->Send(address_chrome, data_vector);
 
   // Fake successful send. The caller ignores result anyway.
   return data_size;
 }
 
 int IpcPacketSocket::Close() {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 
   client_->Close();
@@ skipping 53 matching lines @@
   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 
   if (!jingle_glue::IPEndPointToSocketAddress(address, &local_address_)) {
     // Always expect correct IPv4 address to be allocated.
     NOTREACHED();
     OnError();
     return;
   }
 
   state_ = IS_OPEN;
+  TraceSendThrottlingState();
 
   SignalAddressReady(this, local_address_);
   if (type_ == P2P_SOCKET_TCP_CLIENT)
     SignalConnect(this);
 }
 
 void IpcPacketSocket::OnIncomingTcpConnection(
     const net::IPEndPoint& address,
     P2PSocketClient* client) {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 
   scoped_ptr<IpcPacketSocket> socket(new IpcPacketSocket());
 
   talk_base::SocketAddress remote_address;
   if (!jingle_glue::IPEndPointToSocketAddress(address, &remote_address)) {
     // Always expect correct IPv4 address to be allocated.
     NOTREACHED();
   }
   socket->InitAcceptedTcp(client, local_address_, remote_address);
   SignalNewConnection(this, socket.release());
 }
 
 void IpcPacketSocket::OnSendComplete() {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 
-  --send_packets_pending_;
-  DCHECK_GE(send_packets_pending_, 0);
+  CHECK(!in_flight_packet_sizes_.empty());
+  send_bytes_available_ += in_flight_packet_sizes_.front();
+  DCHECK_LE(send_bytes_available_, kMaximumInFlightBytes);
+  in_flight_packet_sizes_.pop_front();
+  TraceSendThrottlingState();
 
-  if (writable_signal_expected_ &&
-      send_packets_pending_ <= kWritableSignalThreshold) {
+  if (writable_signal_expected_ && send_bytes_available_ > 0) {
     SignalReadyToSend(this);
     writable_signal_expected_ = false;
   }
 }
 
 void IpcPacketSocket::OnError() {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
   state_ = IS_ERROR;
   error_ = ECONNABORTED;
 }
@@ skipping 66 matching lines @@
   talk_base::SocketAddress crome_address;
   P2PSocketClient* socket_client = new P2PSocketClient(socket_dispatcher_);
   scoped_ptr<IpcPacketSocket> socket(new IpcPacketSocket());
   if (!socket->Init(P2P_SOCKET_TCP_CLIENT, socket_client, local_address,
                     remote_address))
     return NULL;
   return socket.release();
 }
 
 }  // namespace content