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 probably needs tuning.  http://crbug.com/237960
+const size_t kMaximumInFlightBytes = 256 << 10;  // 256 KB

[Sergey Ulanov, 2013/05/06 22:18:02]

nit: *1024 is more readable than <<10.

[miu, 2013/05/06 23:36:13]

Done.

 
 // 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,
   };
 
+  // Reset send throttling mechanism to initial (i.e., no packets in-flight)
+  // state.
+  void ResetSendThrottling();
+
+  // Update trace of send throttling internal state. This should be called
+  // immediately after any changes to |token_bucket_level_| 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_;
+  // A token bucket of bytes is used to throttle the sending of packets to the

[Sergey Ulanov, 2013/05/06 22:18:02]

It doesn't look right to me that this is called "Token bucket". Traditional
token bucket algorithm adds tokens on regular time intervals, but it's not the
way this code works. The way it's implemented this variable just indicates
available buffer size.

[miu, 2013/05/06 23:36:13]

Ah, yes.  This did start out as a token bucket, where we were going to use 20
Mbps as the rate of adding tokens.  Then, it collapsed into a simple "in-flight
bytes" counter.

Changed to send_bytes_available_.

+  // browser process. As calls to OnSendComplete() return, the bucket is
+  // refilled. This allows short bursts of high-rate sending without dropping
+  // packets, but quickly restricts the client to a sustainable steady-state
+  // rate.
+  size_t token_bucket_level_;

[Sergey Ulanov, 2013/05/06 22:18:02]

maybe int instead of size_t, because it's not size?

[miu, 2013/05/06 23:36:13]

But it *is* a size: It's always equal to kMaximumInFlightBytes -
SUM(in_flight_packet_sizes_).  It can never become negative.

+  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),
-      writable_signal_expected_(false),
-      error_(0) {}
+      error_(0) {
+}
 
 IpcPacketSocket::~IpcPacketSocket() {
   if (state_ == IS_OPENING || state_ == IS_OPEN ||
       state_ == IS_ERROR) {
     Close();
   }
 }
 
+void IpcPacketSocket::ResetSendThrottling() {
+  COMPILE_ASSERT(kMaximumInFlightBytes > 0, would_send_at_zero_rate);
+  token_bucket_level_ = kMaximumInFlightBytes;
+  in_flight_packet_sizes_.clear();
+  TraceSendThrottlingState();
+  writable_signal_expected_ = false;
+}
+
+void IpcPacketSocket::TraceSendThrottlingState() const {
+  TRACE_COUNTER1("p2p", "P2PSendBytesAvailable", token_bucket_level_);
+  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;
+  ResetSendThrottling();
   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)

[Sergey Ulanov, 2013/05/06 22:18:02]

I think this can be DCHECK_GT(data_size, 0).

[miu, 2013/05/06 23:36:13]

After thinking this over, I believe a NOTREACHED() would be most appropriate
here.  In debug builds, we want to crash so devs can make sure their code isn't
trying to send empty packets (probably indicative of a bug in the calling code).
 However, in release builds, we should take an efficient no-op path.

+    return 0;  // No-op.
+
+  if (data_size > token_bucket_level_) {
+    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_;
+  token_bucket_level_ -= 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;
+  ResetSendThrottling();
 
   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);
+  if (in_flight_packet_sizes_.empty()) {
+    NOTREACHED() << "Received SendComplete() with no known in-flight packets.";
+    // In Release builds, auto-recover by resetting the throttling state.
+    const bool signal_expected = writable_signal_expected_;
+    ResetSendThrottling();

[Sergey Ulanov, 2013/05/06 22:18:02]

Why do you need this given it's marked as NOTREACHED()? If we get into that
state it means browser is broken, so i don't think it's worth trying to recover
from this situation. Just crashing or calling OnError() would be better, IMO.

[miu, 2013/05/06 23:36:13]

Done.  It's a CHECK() now.

+    if (signal_expected)
+      SignalReadyToSend(this);
+    return;
+  }
+  token_bucket_level_ += in_flight_packet_sizes_.front();
+  DCHECK_LE(token_bucket_level_, kMaximumInFlightBytes);
+  in_flight_packet_sizes_.pop_front();
+  TraceSendThrottlingState();
 
-  if (writable_signal_expected_ &&
-      send_packets_pending_ <= kWritableSignalThreshold) {
+  if (writable_signal_expected_ && token_bucket_level_ > 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