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

[Alpha Left Google, 2013/05/04 07:14:14]

This limit should be small. M27 limit is 40KB. We acutally don't want this to be
too large so I think this value shouldn't be greater than 64KB which is what
talk uses.

The reason this should be small is because we want to react to network events or
congestion due to IPC traffic quicker, such that we can react by pausing the
encoder.

[miu, 2013/05/06 19:21:26]

I agree with you.  One problem, though: When sending a 150KB key frame, if we
start error'ing with EWOULDBLOCK after 64 KB worth of packets, will the encoder
drop frames?  That's the problem we're trying to solve here, correct?

 
 // 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.
   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
+  // 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_;
+  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_(MessageLoop::current()),
       state_(IS_UNINITIALIZED),
-      send_packets_pending_(0),
-      writable_signal_expected_(false),
       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(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(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(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)
+    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;

[Alpha Left Google, 2013/05/04 04:11:20]

I'm not sure how this helps.

data_size is always going to be smaller than 1200 because it's packetized by
WebRTC. When we get a burst of packets this behaviors like limiting by bytes
isn't it?

[Alpha Left Google, 2013/05/04 07:14:14]

An alternative is to do this decrement with a delayed task, this way we allow a
sudden burst of packets in time but keep the cap over time. Being able to burst
out a series of packets is critical to sending out unexpected large frames.

[miu, 2013/05/06 19:21:26]

Exactly.  That's what I thought we had discussed over VC on Friday.  We wanted
to stop doing the packet-counting/limiting approach, and limit based on number
of bytes instead.  These are the goals (from your e-mail):
The cap is set by the kMaximumInFlightBytes constant.  I need guidance from the
testing on your end on how we should tune this.
This is a behavior we're getting from the token bucket approach.  Packets can be
sent at *infinite* rate up to the kMaximumInFlightBytes limit.  Thereafter, the
ACKs from the browser must be received in order to re-fill the token bucket. 
This means that, after the initial burst, the renderer will only be allowed to
send at the rate the current network conditions allow for.
This occurs once the token bucket has insufficient value.

[miu, 2013/05/06 19:21:26]

The two are identical approaches, except that the token bucket doesn't rely on
the system clock (nor the additional code complexity of a callback-by-timer). 
Here's a summary of the two approaches:

Delayed task scheme: Renderer can send at infinite rate for N seconds, and is
then capped.
Token bucket scheme: Renderer can send at infinite rate until byte limit, K, is
reached.

The two are related by a simple equation (assuming 20 Mbps is the bursty send
rate):

  K = N * (20 Mbps / 8)

So, if you set up a delayed task to run 100 ms after the burst begins, then K
(the kMaximumInFlightBytes constant) should be 250 KB.

+  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(MessageLoop::current(), message_loop_);
 
   client_->Close();
@@ skipping 53 matching lines @@
   DCHECK_EQ(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(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(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 production builds, auto-recover by resetting the throttling state.

[Alpha Left Google, 2013/05/04 04:11:20]

How is it related to production builds?

[miu, 2013/05/06 19:21:26]

Wrong wording on my part.  I meant: s/production/release/  Will fix.

+    const bool signal_expected = writable_signal_expected_;
+    ResetSendThrottling();
+    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(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