Fix IpcPacketSocketFactory to return EWOULDBLOCK when network interface is congested.

PeerConnection implementation in libjingle will need to handle the case when 
a network interface is congested, so it needs to be notified when send buffers
are full. This CL adds send result notification message that is sent to the
renderer process after each message is sent. Sockets created using 
IpcPacketSocketFactory will be returning EWOULDBLOCK when Send() buffers are 
full instead of dropping the packet silently.

BUG=226158

Review URL: https://chromiumcodereview.appspot.com/13584008

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

5 files changed, 68 insertions, 6 deletions

diff --git a/content/renderer/p2p/ipc_socket_factory.cc b/content/renderer/p2p/ipc_socket_factory.cc
index ea53678..defe31f 100644
--- a/content/renderer/p2p/ipc_socket_factory.cc
+++ b/content/renderer/p2p/ipc_socket_factory.cc
@@ -16,6 +16,10 @@ namespace content {
 
 namespace {
 
+// TODO(sergeyu): Try adjusting these parameters to achieve optimal performance.
+const int kMaxPendingPackets = 8;
+const int kWritableSignalThreshold = 0;
+
 // IpcPacketSocket implements talk_base::AsyncPacketSocket interface
 // using P2PSocketClient that works over IPC-channel. It must be used
 // on the thread it was created.
@@ -47,6 +51,7 @@ class IpcPacketSocket : public talk_base::AsyncPacketSocket,
   virtual void OnOpen(const net::IPEndPoint& address) OVERRIDE;
   virtual void OnIncomingTcpConnection(const net::IPEndPoint& address,
                                        P2PSocketClient* client) OVERRIDE;
+  virtual void OnSendComplete() OVERRIDE;
   virtual void OnError() OVERRIDE;
   virtual void OnDataReceived(const net::IPEndPoint& address,
                               const std::vector<char>& data) OVERRIDE;
@@ -83,6 +88,14 @@ 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_;
+
+  // 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_;
 
@@ -93,6 +106,8 @@ IpcPacketSocket::IpcPacketSocket()
     : type_(P2P_SOCKET_UDP),
       message_loop_(MessageLoop::current()),
       state_(IS_UNINITIALIZED),
+      send_packets_pending_(0),
+      writable_signal_expected_(false),
       error_(0) {
 }
 
@@ -180,15 +195,22 @@ int IpcPacketSocket::SendTo(const void *data, size_t data_size,
       break;
   }
 
+  if (send_packets_pending_ > kMaxPendingPackets) {
+    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)) {
-    // Just drop the packet if we failed to convert the address.
-    return 0;
+    NOTREACHED();
+    return -1;
   }
 
+  ++send_packets_pending_;
   client_->Send(address_chrome, data_vector);
 
   // Fake successful send. The caller ignores result anyway.
@@ -238,9 +260,6 @@ int IpcPacketSocket::GetOption(talk_base::Socket::Option opt, int* value) {
 
 int IpcPacketSocket::SetOption(talk_base::Socket::Option opt, int value) {
   // We don't support socket options for IPC sockets.
-  //
-  // TODO(sergeyu): Make sure we set proper socket options on the
-  // browser side.
   return -1;
 }
 
@@ -287,6 +306,22 @@ void IpcPacketSocket::OnIncomingTcpConnection(
   SignalNewConnection(this, socket.release());
 }
 
+void IpcPacketSocket::OnSendComplete() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+
+  --send_packets_pending_;
+  DCHECK_GE(send_packets_pending_, 0);
+
+  if (writable_signal_expected_ &&
+      send_packets_pending_ <= kWritableSignalThreshold) {
+    // TODO(sergeyu): Uncomment this line once SignalWritable is added in
+    // talk_base::AsyncPacketSocket.
+    //
+    // SignalWritable(this);
+    writable_signal_expected_ = false;
+  }
+}
+
 void IpcPacketSocket::OnError() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   state_ = IS_ERROR;
diff --git a/content/renderer/p2p/socket_client.cc b/content/renderer/p2p/socket_client.cc
index e6f5a0f..ed3f211 100644
--- a/content/renderer/p2p/socket_client.cc
+++ b/content/renderer/p2p/socket_client.cc
@@ -134,6 +134,19 @@ void P2PSocketClient::DeliverOnIncomingTcpConnection(
   }
 }
 
+void P2PSocketClient::OnSendComplete() {
+  DCHECK(ipc_message_loop_->BelongsToCurrentThread());
+
+  delegate_message_loop_->PostTask(
+      FROM_HERE, base::Bind(&P2PSocketClient::DeliverOnSendComplete, this));
+}
+
+void P2PSocketClient::DeliverOnSendComplete() {
+  DCHECK(delegate_message_loop_->BelongsToCurrentThread());
+  if (delegate_)
+    delegate_->OnSendComplete();
+}
+
 void P2PSocketClient::OnError() {
   DCHECK(ipc_message_loop_->BelongsToCurrentThread());
   state_ = STATE_ERROR;
diff --git a/content/renderer/p2p/socket_client.h b/content/renderer/p2p/socket_client.h
index d74107f..fd8537e 100644
--- a/content/renderer/p2p/socket_client.h
+++ b/content/renderer/p2p/socket_client.h
@@ -27,7 +27,8 @@ class P2PSocketDispatcher;
 // thread which is specified in Init().
 class P2PSocketClient : public base::RefCountedThreadSafe<P2PSocketClient> {
  public:
-  // Delegate is called on the the same thread on the delegate thread.
+  // Delegate is called on the the same thread on which P2PSocketCLient is
+  // created.
   class Delegate {
    public:
     virtual ~Delegate() { }
@@ -35,6 +36,7 @@ class P2PSocketClient : public base::RefCountedThreadSafe<P2PSocketClient> {
     virtual void OnOpen(const net::IPEndPoint& address) = 0;
     virtual void OnIncomingTcpConnection(const net::IPEndPoint& address,
                                          P2PSocketClient* client) = 0;
+    virtual void OnSendComplete() = 0;
     virtual void OnError() = 0;
     virtual void OnDataReceived(const net::IPEndPoint& address,
                                 const std::vector<char>& data) = 0;
@@ -80,6 +82,8 @@ class P2PSocketClient : public base::RefCountedThreadSafe<P2PSocketClient> {
   // Message handlers that run on IPC thread.
   void OnSocketCreated(const net::IPEndPoint& address);
   void OnIncomingTcpConnection(const net::IPEndPoint& address);
+  void OnSendComplete(int packet_id);
+  void OnSendComplete();
   void OnError();
   void OnDataReceived(const net::IPEndPoint& address,
                       const std::vector<char>& data);
@@ -89,6 +93,7 @@ class P2PSocketClient : public base::RefCountedThreadSafe<P2PSocketClient> {
   void DeliverOnIncomingTcpConnection(
       const net::IPEndPoint& address,
       scoped_refptr<P2PSocketClient> new_client);
+  void DeliverOnSendComplete();
   void DeliverOnError();
   void DeliverOnDataReceived(const net::IPEndPoint& address,
                              const std::vector<char>& data);
diff --git a/content/renderer/p2p/socket_dispatcher.cc b/content/renderer/p2p/socket_dispatcher.cc
index c999e62..7fe71ee 100644
--- a/content/renderer/p2p/socket_dispatcher.cc
+++ b/content/renderer/p2p/socket_dispatcher.cc
@@ -63,6 +63,7 @@ bool P2PSocketDispatcher::OnMessageReceived(const IPC::Message& message) {
     IPC_MESSAGE_HANDLER(P2PMsg_GetHostAddressResult, OnGetHostAddressResult)
     IPC_MESSAGE_HANDLER(P2PMsg_OnSocketCreated, OnSocketCreated)
     IPC_MESSAGE_HANDLER(P2PMsg_OnIncomingTcpConnection, OnIncomingTcpConnection)
+    IPC_MESSAGE_HANDLER(P2PMsg_OnSendComplete, OnSendComplete)
     IPC_MESSAGE_HANDLER(P2PMsg_OnError, OnError)
     IPC_MESSAGE_HANDLER(P2PMsg_OnDataReceived, OnDataReceived)
     IPC_MESSAGE_UNHANDLED(handled = false)
@@ -152,6 +153,13 @@ void P2PSocketDispatcher::OnIncomingTcpConnection(
   }
 }
 
+void P2PSocketDispatcher::OnSendComplete(int socket_id) {
+  P2PSocketClient* client = GetClient(socket_id);
+  if (client) {
+    client->OnSendComplete();
+  }
+}
+
 void P2PSocketDispatcher::OnError(int socket_id) {
   P2PSocketClient* client = GetClient(socket_id);
   if (client) {
diff --git a/content/renderer/p2p/socket_dispatcher.h b/content/renderer/p2p/socket_dispatcher.h
index e17a8c1..cbd8f2a 100644
--- a/content/renderer/p2p/socket_dispatcher.h
+++ b/content/renderer/p2p/socket_dispatcher.h
@@ -102,6 +102,7 @@ class CONTENT_EXPORT P2PSocketDispatcher
                               const net::IPAddressNumber& address);
   void OnSocketCreated(int socket_id, const net::IPEndPoint& address);
   void OnIncomingTcpConnection(int socket_id, const net::IPEndPoint& address);
+  void OnSendComplete(int socket_id);
   void OnError(int socket_id);
   void OnDataReceived(int socket_id, const net::IPEndPoint& address,
                       const std::vector<char>& data);