Send notification about outgoing p2p packets from browser to renderer.

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 "base/compiler_specific.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(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.
 class IpcPacketSocket : public talk_base::AsyncPacketSocket,
                         public P2PSocketClient::Delegate {
  public:
   IpcPacketSocket();
   virtual ~IpcPacketSocket();
 
   // Always takes ownership of client even if initialization fails.
@@ skipping 11 matching lines @@
   virtual State GetState() const OVERRIDE;
   virtual int GetOption(talk_base::Socket::Option opt, int* value) OVERRIDE;
   virtual int SetOption(talk_base::Socket::Option opt, int value) OVERRIDE;
   virtual int GetError() const OVERRIDE;
   virtual void SetError(int error) OVERRIDE;
 
   // P2PSocketClient::Delegate implementation.
   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;
 
  private:
   enum InternalState {
     IS_UNINITIALIZED,
     IS_OPENING,
     IS_OPEN,
     IS_CLOSED,
@@ skipping 16 matching lines @@
   // 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_;
+
+  // 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();
   }
 }
 
@@ skipping 67 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) {
+    writable_signal_expected_ = true;
+    return EWOULDBLOCK;

[Ronghua Wu (Left Chromium), 2013/04/04 23:53:02]

Justin, you mentioned in my cl that you want something like
SignalReadyToSend(bool) for both down and resume. Or are you ok with EWOULDBLOCK
+ SignalWritable?

[Ronghua Wu (Left Chromium), 2013/04/05 17:56:04]

Should we set error_ to EWOULDBLOCK? IOW, should the upper layer get EWOULDBLOCK
from return value of SendTo or from the GetError()?

[Sergey Ulanov, 2013/04/05 21:04:03]

That's a good point. I think we should return -1 here and EWOULDBLOCK in
GetError() for consistency with AsyncSocket and also because EWOULDBLOCK is
positive.

+  }
+
   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;
   }
 
+  ++send_packets_pending_;
   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 29 matching lines @@
   return STATE_CLOSED;
 }
 
 int IpcPacketSocket::GetOption(talk_base::Socket::Option opt, int* value) {
   // We don't support socket options for IPC sockets.
   return -1;
 }
 
 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;
 }
 
 int IpcPacketSocket::GetError() const {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   return error_;
 }
 
 void IpcPacketSocket::SetError(int error) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
@@ skipping 26 matching lines @@
 
   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 (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;
   error_ = ECONNABORTED;
 }
 
 void IpcPacketSocket::OnDataReceived(const net::IPEndPoint& address,
                                      const std::vector<char>& data) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
@@ skipping 61 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