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

content/renderer/p2p/ipc_socket_factory.cc

Index: content/renderer/p2p/ipc_socket_factory.cc
diff --git a/content/renderer/p2p/ipc_socket_factory.cc b/content/renderer/p2p/ipc_socket_factory.cc
index 3fd0556a1881370425fd466b1c7dc43411b9f4dd..6aaa12b9ab00e155ab201d0323a39edbebcbf8f0 100644
--- a/content/renderer/p2p/ipc_socket_factory.cc
+++ b/content/renderer/p2p/ipc_socket_factory.cc
@@ -4,6 +4,8 @@
 
 #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"
@@ -17,9 +19,8 @@ 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
@@ -66,6 +67,15 @@ class IpcPacketSocket : public talk_base::AsyncPacketSocket,
     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);
@@ -89,9 +99,13 @@ class IpcPacketSocket : public talk_base::AsyncPacketSocket,
   // 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.
@@ -107,8 +121,6 @@ IpcPacketSocket::IpcPacketSocket()
     : type_(P2P_SOCKET_UDP),
       message_loop_(MessageLoop::current()),
       state_(IS_UNINITIALIZED),
-      send_packets_pending_(0),
-      writable_signal_expected_(false),
       error_(0) {
 }
 
@@ -119,6 +131,20 @@ IpcPacketSocket::~IpcPacketSocket() {
   }
 }
 
+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) {
@@ -158,6 +184,7 @@ void IpcPacketSocket::InitAcceptedTcp(
   local_address_ = local_address;
   remote_address_ = remote_address;
   state_ = IS_OPEN;
+  ResetSendThrottling();
   client_->set_delegate(this);
 }
 
@@ -196,24 +223,29 @@ int IpcPacketSocket::SendTo(const void *data, size_t data_size,
       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.
@@ -287,6 +319,7 @@ void IpcPacketSocket::OnOpen(const net::IPEndPoint& address) {
   }
 
   state_ = IS_OPEN;
+  ResetSendThrottling();
 
   SignalAddressReady(this, local_address_);
   if (type_ == P2P_SOCKET_TCP_CLIENT)
@@ -312,11 +345,21 @@ void IpcPacketSocket::OnIncomingTcpConnection(
 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;
   }