Pepper UDP socket: buffer received packets in the plugin process to improve performance.

For each bound UDP socket, the Pepper layer in the plugin process maintains a
buffer to store received packets, and informs the browser process of available
buffer slots. The browser process does recvfrom and pushes the results to the
plugin process when there are available slots.

BUG=None
TEST=None

Review URL: https://codereview.chromium.org/563073002

Cr-Commit-Position: refs/heads/master@{#295184}

7 files changed, 200 insertions, 83 deletions

diff --git a/ppapi/host/resource_message_filter.h b/ppapi/host/resource_message_filter.h
index b72a7df..67e9b79 100644
--- a/ppapi/host/resource_message_filter.h
+++ b/ppapi/host/resource_message_filter.h
@@ -106,6 +106,10 @@ class PPAPI_HOST_EXPORT ResourceMessageFilter
  protected:
   virtual ~ResourceMessageFilter();
 
+  // Please see the comments of |resource_host_| for on which thread it can be
+  // used and when it is NULL.
+  ResourceHost* resource_host() const { return resource_host_; }
+
   // If you want the message to be handled on another thread, return a non-null
   // task runner which will target tasks accordingly.
   virtual scoped_refptr<base::TaskRunner> OverrideTaskRunnerForMessage(
diff --git a/ppapi/proxy/plugin_message_filter.cc b/ppapi/proxy/plugin_message_filter.cc
index 1846957..f2f1cbe 100644
--- a/ppapi/proxy/plugin_message_filter.cc
+++ b/ppapi/proxy/plugin_message_filter.cc
@@ -83,12 +83,16 @@ void PluginMessageFilter::OnMsgResourceReply(
     const ResourceMessageReplyParams& reply_params,
     const IPC::Message& nested_msg) {
   scoped_refptr<base::MessageLoopProxy> target =
-      resource_reply_thread_registrar_->GetTargetThreadAndUnregister(
-          reply_params.pp_resource(), reply_params.sequence());
-
-  target->PostTask(
-      FROM_HERE,
-      base::Bind(&DispatchResourceReply, reply_params, nested_msg));
+      resource_reply_thread_registrar_->GetTargetThread(reply_params,
+                                                        nested_msg);
+
+  if (!target.get()) {
+    DispatchResourceReply(reply_params, nested_msg);
+  } else {
+    target->PostTask(
+        FROM_HERE,
+        base::Bind(&DispatchResourceReply, reply_params, nested_msg));
+  }
 }
 
 // static
diff --git a/ppapi/proxy/ppapi_messages.h b/ppapi/proxy/ppapi_messages.h
index 506f476..de7caf3 100644
--- a/ppapi/proxy/ppapi_messages.h
+++ b/ppapi/proxy/ppapi_messages.h
@@ -1728,11 +1728,11 @@ IPC_MESSAGE_CONTROL1(PpapiHostMsg_UDPSocket_Bind,
                      PP_NetAddress_Private /* net_addr */)
 IPC_MESSAGE_CONTROL1(PpapiPluginMsg_UDPSocket_BindReply,
                      PP_NetAddress_Private /* bound_addr */)
-IPC_MESSAGE_CONTROL1(PpapiHostMsg_UDPSocket_RecvFrom,
-                     int32_t /* num_bytes */)
-IPC_MESSAGE_CONTROL2(PpapiPluginMsg_UDPSocket_RecvFromReply,
+IPC_MESSAGE_CONTROL3(PpapiPluginMsg_UDPSocket_PushRecvResult,
+                     int32_t /* result */,
                      std::string /* data */,
                      PP_NetAddress_Private /* remote_addr */)
+IPC_MESSAGE_CONTROL0(PpapiHostMsg_UDPSocket_RecvSlotAvailable)
 IPC_MESSAGE_CONTROL2(PpapiHostMsg_UDPSocket_SendTo,
                      std::string /* data */,
                      PP_NetAddress_Private /* net_addr */)
diff --git a/ppapi/proxy/resource_reply_thread_registrar.cc b/ppapi/proxy/resource_reply_thread_registrar.cc
index 1332987..d0c697d 100644
--- a/ppapi/proxy/resource_reply_thread_registrar.cc
+++ b/ppapi/proxy/resource_reply_thread_registrar.cc
@@ -6,6 +6,8 @@
 
 #include "base/logging.h"
 #include "base/message_loop/message_loop_proxy.h"
+#include "ipc/ipc_message.h"
+#include "ppapi/proxy/resource_message_params.h"
 #include "ppapi/shared_impl/proxy_lock.h"
 #include "ppapi/shared_impl/tracked_callback.h"
 
@@ -13,8 +15,8 @@ namespace ppapi {
 namespace proxy {
 
 ResourceReplyThreadRegistrar::ResourceReplyThreadRegistrar(
-    scoped_refptr<base::MessageLoopProxy> default_thread)
-    : default_thread_(default_thread) {
+    scoped_refptr<base::MessageLoopProxy> main_thread)
+    : main_thread_(main_thread) {
 }
 
 ResourceReplyThreadRegistrar::~ResourceReplyThreadRegistrar() {
@@ -26,7 +28,7 @@ void ResourceReplyThreadRegistrar::Register(
     scoped_refptr<TrackedCallback> reply_thread_hint) {
   ProxyLock::AssertAcquiredDebugOnly();
 
-  // Use the default thread if |reply_thread_hint| is NULL or blocking.
+  // Use the main thread if |reply_thread_hint| is NULL or blocking.
   if (!reply_thread_hint.get() || reply_thread_hint->is_blocking())
     return;
 
@@ -36,7 +38,7 @@ void ResourceReplyThreadRegistrar::Register(
   {
     base::AutoLock auto_lock(lock_);
 
-    if (reply_thread.get() == default_thread_.get())
+    if (reply_thread.get() == main_thread_.get())
       return;
 
     map_[resource][sequence_number] = reply_thread;
@@ -48,23 +50,32 @@ void ResourceReplyThreadRegistrar::Unregister(PP_Resource resource) {
   map_.erase(resource);
 }
 
+void ResourceReplyThreadRegistrar::HandleOnIOThread(uint32 nested_msg_type) {
+  base::AutoLock auto_lock(lock_);
+  io_thread_message_types_.insert(nested_msg_type);
+}
+
 scoped_refptr<base::MessageLoopProxy>
-ResourceReplyThreadRegistrar::GetTargetThreadAndUnregister(
-    PP_Resource resource,
-    int32_t sequence_number) {
+ResourceReplyThreadRegistrar::GetTargetThread(
+    const ResourceMessageReplyParams& reply_params,
+    const IPC::Message& nested_msg) {
   base::AutoLock auto_lock(lock_);
-  ResourceMap::iterator resource_iter = map_.find(resource);
-  if (resource_iter == map_.end())
-    return default_thread_;
-
-  SequenceNumberMap::iterator sequence_number_iter =
-      resource_iter->second.find(sequence_number);
-  if (sequence_number_iter == resource_iter->second.end())
-    return default_thread_;
-
-  scoped_refptr<base::MessageLoopProxy> target = sequence_number_iter->second;
-  resource_iter->second.erase(sequence_number_iter);
-  return target;
+  ResourceMap::iterator resource_iter = map_.find(reply_params.pp_resource());
+  if (resource_iter != map_.end()) {
+    SequenceThreadMap::iterator sequence_thread_iter =
+        resource_iter->second.find(reply_params.sequence());
+    if (sequence_thread_iter != resource_iter->second.end()) {
+      scoped_refptr<base::MessageLoopProxy> target =
+          sequence_thread_iter->second;
+      resource_iter->second.erase(sequence_thread_iter);
+      return target;
+    }
+  }
+
+  if (io_thread_message_types_.count(nested_msg.type()) != 0)
+    return scoped_refptr<base::MessageLoopProxy>();
+
+  return main_thread_;
 }
 
 }  // namespace proxy
diff --git a/ppapi/proxy/resource_reply_thread_registrar.h b/ppapi/proxy/resource_reply_thread_registrar.h
index eef6d61..ded54e9 100644
--- a/ppapi/proxy/resource_reply_thread_registrar.h
+++ b/ppapi/proxy/resource_reply_thread_registrar.h
@@ -6,6 +6,7 @@
 #define PPAPI_PROXY_RESOURCE_REPLY_THREAD_REGISTRAR_H_
 
 #include <map>
+#include <set>
 
 #include "base/basictypes.h"
 #include "base/memory/ref_counted.h"
@@ -18,12 +19,18 @@ namespace base {
 class MessageLoopProxy;
 }
 
+namespace IPC {
+class Message;
+}
+
 namespace ppapi {
 
 class TrackedCallback;
 
 namespace proxy {
 
+class ResourceMessageReplyParams;
+
 // ResourceReplyThreadRegistrar records the handling thread for
 // PpapiPluginMsg_ResourceReply messages.
 // This class is thread safe.
@@ -31,26 +38,32 @@ class PPAPI_PROXY_EXPORT ResourceReplyThreadRegistrar
     : public base::RefCountedThreadSafe<ResourceReplyThreadRegistrar> {
  public:
   explicit ResourceReplyThreadRegistrar(
-      scoped_refptr<base::MessageLoopProxy> default_thread);
+      scoped_refptr<base::MessageLoopProxy> main_thread);
 
   // This method can only be called while holding the Pepper proxy lock; the
   // other methods can be called with/without the Pepper proxy lock.
   void Register(PP_Resource resource,
                 int32_t sequence_number,
                 scoped_refptr<TrackedCallback> reply_thread_hint);
-
   void Unregister(PP_Resource resource);
 
-  scoped_refptr<base::MessageLoopProxy> GetTargetThreadAndUnregister(
-      PP_Resource resource,
-      int32_t sequence_number);
+  // This results in Resource::OnReplyReceived() for the specified message type
+  // to be called on the IO thread directly, while holding the Pepper proxy
+  // lock.
+  void HandleOnIOThread(uint32 nested_msg_type);
+
+  // This method returns NULL if the target thread is the IO thread (because
+  // that is the thread on which this method is supposed to be called).
+  scoped_refptr<base::MessageLoopProxy> GetTargetThread(
+      const ResourceMessageReplyParams& reply_params,
+      const IPC::Message& nested_msg);
 
  private:
   friend class base::RefCountedThreadSafe<ResourceReplyThreadRegistrar>;
 
   typedef std::map<int32_t, scoped_refptr<base::MessageLoopProxy> >
-      SequenceNumberMap;
-  typedef std::map<PP_Resource, SequenceNumberMap> ResourceMap;
+      SequenceThreadMap;
+  typedef std::map<PP_Resource, SequenceThreadMap> ResourceMap;
 
   ~ResourceReplyThreadRegistrar();
 
@@ -59,7 +72,8 @@ class PPAPI_PROXY_EXPORT ResourceReplyThreadRegistrar
   // holding |lock_|, otherwise we will cause deadlock.
   base::Lock lock_;
   ResourceMap map_;
-  scoped_refptr<base::MessageLoopProxy> default_thread_;
+  std::set<uint32> io_thread_message_types_;
+  scoped_refptr<base::MessageLoopProxy> main_thread_;
 
   DISALLOW_COPY_AND_ASSIGN(ResourceReplyThreadRegistrar);
 };
diff --git a/ppapi/proxy/udp_socket_resource_base.cc b/ppapi/proxy/udp_socket_resource_base.cc
index 79ac348..521a6e2 100644
--- a/ppapi/proxy/udp_socket_resource_base.cc
+++ b/ppapi/proxy/udp_socket_resource_base.cc
@@ -12,6 +12,7 @@
 #include "ppapi/c/pp_completion_callback.h"
 #include "ppapi/c/pp_errors.h"
 #include "ppapi/proxy/error_conversion.h"
+#include "ppapi/proxy/plugin_globals.h"
 #include "ppapi/proxy/ppapi_messages.h"
 #include "ppapi/shared_impl/socket_option_data.h"
 #include "ppapi/thunk/enter.h"
@@ -20,13 +21,13 @@
 namespace ppapi {
 namespace proxy {
 
-const int32_t UDPSocketResourceBase::kMaxReadSize = 1024 * 1024;
-const int32_t UDPSocketResourceBase::kMaxWriteSize = 1024 * 1024;
+const int32_t UDPSocketResourceBase::kMaxReadSize = 128 * 1024;
+const int32_t UDPSocketResourceBase::kMaxWriteSize = 128 * 1024;
 const int32_t UDPSocketResourceBase::kMaxSendBufferSize =
     1024 * UDPSocketResourceBase::kMaxWriteSize;
 const int32_t UDPSocketResourceBase::kMaxReceiveBufferSize =
     1024 * UDPSocketResourceBase::kMaxReadSize;
-
+const size_t UDPSocketResourceBase::kPluginReceiveBufferSlots = 32u;
 
 UDPSocketResourceBase::UDPSocketResourceBase(Connection connection,
                                              PP_Instance instance,
@@ -36,7 +37,8 @@ UDPSocketResourceBase::UDPSocketResourceBase(Connection connection,
       bound_(false),
       closed_(false),
       read_buffer_(NULL),
-      bytes_to_read_(-1) {
+      bytes_to_read_(-1),
+      recvfrom_addr_resource_(NULL) {
   recvfrom_addr_.size = 0;
   memset(recvfrom_addr_.data, 0,
          arraysize(recvfrom_addr_.data) * sizeof(*recvfrom_addr_.data));
@@ -48,6 +50,9 @@ UDPSocketResourceBase::UDPSocketResourceBase(Connection connection,
     SendCreate(BROWSER, PpapiHostMsg_UDPSocket_CreatePrivate());
   else
     SendCreate(BROWSER, PpapiHostMsg_UDPSocket_Create());
+
+  PluginGlobals::Get()->resource_reply_thread_registrar()->HandleOnIOThread(
+      PpapiPluginMsg_UDPSocket_PushRecvResult::ID);
 }
 
 UDPSocketResourceBase::~UDPSocketResourceBase() {
@@ -139,18 +144,27 @@ int32_t UDPSocketResourceBase::RecvFromImpl(
   if (TrackedCallback::IsPending(recvfrom_callback_))
     return PP_ERROR_INPROGRESS;
 
-  read_buffer_ = buffer;
-  bytes_to_read_ = std::min(num_bytes, kMaxReadSize);
-  recvfrom_callback_ = callback;
+  if (recv_buffers_.empty()) {
+    read_buffer_ = buffer;
+    bytes_to_read_ = std::min(num_bytes, kMaxReadSize);
+    recvfrom_addr_resource_ = addr;
+    recvfrom_callback_ = callback;
 
-  // Send the request, the browser will call us back via RecvFromReply.
-  Call<PpapiPluginMsg_UDPSocket_RecvFromReply>(
-      BROWSER,
-      PpapiHostMsg_UDPSocket_RecvFrom(bytes_to_read_),
-      base::Bind(&UDPSocketResourceBase::OnPluginMsgRecvFromReply,
-                 base::Unretained(this), addr),
-      callback);
-  return PP_OK_COMPLETIONPENDING;
+    return PP_OK_COMPLETIONPENDING;
+  } else {
+    RecvBuffer& front = recv_buffers_.front();
+
+    if (num_bytes < static_cast<int32_t>(front.data.size()))
+      return PP_ERROR_MESSAGE_TOO_BIG;
+
+    int32_t result = SetRecvFromOutput(front.result, front.data, front.addr,
+                                       buffer, num_bytes, addr);
+
+    recv_buffers_.pop();
+    Post(BROWSER, PpapiHostMsg_UDPSocket_RecvSlotAvailable());
+
+    return result;
+  }
 }
 
 PP_Bool UDPSocketResourceBase::GetRecvFromAddressImpl(
@@ -205,6 +219,18 @@ void UDPSocketResourceBase::CloseImpl() {
   bytes_to_read_ = -1;
 }
 
+void UDPSocketResourceBase::OnReplyReceived(
+    const ResourceMessageReplyParams& params,
+    const IPC::Message& msg) {
+  PPAPI_BEGIN_MESSAGE_MAP(UDPSocketResourceBase, msg)
+    PPAPI_DISPATCH_PLUGIN_RESOURCE_CALL(
+        PpapiPluginMsg_UDPSocket_PushRecvResult,
+        OnPluginMsgPushRecvResult)
+    PPAPI_DISPATCH_PLUGIN_RESOURCE_CALL_UNHANDLED(
+        PluginResource::OnReplyReceived(params, msg))
+  PPAPI_END_MESSAGE_MAP()
+}
+
 void UDPSocketResourceBase::PostAbortIfNecessary(
     scoped_refptr<TrackedCallback>* callback) {
   if (TrackedCallback::IsPending(*callback))
@@ -234,43 +260,47 @@ void UDPSocketResourceBase::OnPluginMsgBindReply(
   RunCallback(bind_callback_, params.result());
 }
 
-void UDPSocketResourceBase::OnPluginMsgRecvFromReply(
-    PP_Resource* output_addr,
+void UDPSocketResourceBase::OnPluginMsgPushRecvResult(
     const ResourceMessageReplyParams& params,
+    int32_t result,
     const std::string& data,
     const PP_NetAddress_Private& addr) {
-  // It is possible that |recvfrom_callback_| is pending while |read_buffer_| is
-  // NULL: CloseImpl() has been called, but a RecvFromReply came earlier than
-  // the task to abort |recvfrom_callback_|. We shouldn't access the buffer in
-  // that case. The user may have released it.
-  if (!TrackedCallback::IsPending(recvfrom_callback_) || !read_buffer_)
-    return;
+  // TODO(yzshen): Support passing in a non-const string ref, so that we can
+  // eliminate one copy when storing the data in the buffer.
 
-  int32_t result = params.result();
-  if (result == PP_OK && output_addr) {
-    thunk::EnterResourceCreationNoLock enter(pp_instance());
-    if (enter.succeeded()) {
-      *output_addr = enter.functions()->CreateNetAddressFromNetAddressPrivate(
-          pp_instance(), addr);
-    } else {
-      result = PP_ERROR_FAILED;
-    }
+  DCHECK_LT(recv_buffers_.size(), kPluginReceiveBufferSlots);
+
+  if (!TrackedCallback::IsPending(recvfrom_callback_) || !read_buffer_) {
+    recv_buffers_.push(RecvBuffer());
+    RecvBuffer& back = recv_buffers_.back();
+    back.result = result;
+    back.data = data;
+    back.addr = addr;
+
+    return;
   }
 
-  if (result == PP_OK) {
-    CHECK_LE(static_cast<int32_t>(data.size()), bytes_to_read_);
-    if (!data.empty())
-      memcpy(read_buffer_, data.c_str(), data.size());
+  DCHECK_EQ(recv_buffers_.size(), 0u);
+
+  if (bytes_to_read_ < static_cast<int32_t>(data.size())) {
+    recv_buffers_.push(RecvBuffer());
+    RecvBuffer& back = recv_buffers_.back();
+    back.result = result;
+    back.data = data;
+    back.addr = addr;
+
+    result = PP_ERROR_MESSAGE_TOO_BIG;
+  } else {
+    result = SetRecvFromOutput(result, data, addr, read_buffer_, bytes_to_read_,
+                               recvfrom_addr_resource_);
+    Post(BROWSER, PpapiHostMsg_UDPSocket_RecvSlotAvailable());
   }
 
   read_buffer_ = NULL;
   bytes_to_read_ = -1;
-  recvfrom_addr_ = addr;
+  recvfrom_addr_resource_ = NULL;
 
-  if (result == PP_OK)
-    RunCallback(recvfrom_callback_, static_cast<int32_t>(data.size()));
-  else
-    RunCallback(recvfrom_callback_, result);
+  RunCallback(recvfrom_callback_, result);
 }
 
 void UDPSocketResourceBase::OnPluginMsgSendToReply(
@@ -291,5 +321,33 @@ void UDPSocketResourceBase::RunCallback(scoped_refptr<TrackedCallback> callback,
                                                        private_api_));
 }
 
+int32_t UDPSocketResourceBase::SetRecvFromOutput(
+    int32_t browser_result,
+    const std::string& data,
+    const PP_NetAddress_Private& addr,
+    char* output_buffer,
+    int32_t num_bytes,
+    PP_Resource* output_addr) {
+  DCHECK_GE(num_bytes, static_cast<int32_t>(data.size()));
+
+  int32_t result = browser_result;
+  if (result == PP_OK && output_addr) {
+    thunk::EnterResourceCreationNoLock enter(pp_instance());
+    if (enter.succeeded()) {
+      *output_addr = enter.functions()->CreateNetAddressFromNetAddressPrivate(
+          pp_instance(), addr);
+    } else {
+      result = PP_ERROR_FAILED;
+    }
+  }
+
+  if (result == PP_OK && !data.empty())
+    memcpy(output_buffer, data.c_str(), data.size());
+
+  recvfrom_addr_ = addr;
+
+  return result == PP_OK ? static_cast<int32_t>(data.size()) : result;
+}
+
 }  // namespace proxy
 }  // namespace ppapi
diff --git a/ppapi/proxy/udp_socket_resource_base.h b/ppapi/proxy/udp_socket_resource_base.h
index 978e774..1ed9bbf 100644
--- a/ppapi/proxy/udp_socket_resource_base.h
+++ b/ppapi/proxy/udp_socket_resource_base.h
@@ -5,6 +5,7 @@
 #ifndef PPAPI_PROXY_UDP_SOCKET_RESOURCE_BASE_H_
 #define PPAPI_PROXY_UDP_SOCKET_RESOURCE_BASE_H_
 
+#include <queue>
 #include <string>
 
 #include "base/basictypes.h"
@@ -23,8 +24,8 @@ class ResourceMessageReplyParams;
 
 class PPAPI_PROXY_EXPORT UDPSocketResourceBase: public PluginResource {
  public:
-  // The maximum number of bytes that each PpapiHostMsg_PPBUDPSocket_RecvFrom
-  // message is allowed to request.
+  // The maximum number of bytes that each
+  // PpapiPluginMsg_PPBUDPSocket_PushRecvResult message is allowed to carry.
   static const int32_t kMaxReadSize;
   // The maximum number of bytes that each PpapiHostMsg_PPBUDPSocket_SendTo
   // message is allowed to carry.
@@ -41,6 +42,10 @@ class PPAPI_PROXY_EXPORT UDPSocketResourceBase: public PluginResource {
   // such a buffer size.
   static const int32_t kMaxReceiveBufferSize;
 
+  // The maximum number of received packets that we allow instances of this
+  // class to buffer.
+  static const size_t kPluginReceiveBufferSlots;
+
  protected:
   UDPSocketResourceBase(Connection connection,
                         PP_Instance instance,
@@ -66,6 +71,16 @@ class PPAPI_PROXY_EXPORT UDPSocketResourceBase: public PluginResource {
   void CloseImpl();
 
  private:
+  struct RecvBuffer {
+    int32_t result;
+    std::string data;
+    PP_NetAddress_Private addr;
+  };
+
+  // Resource overrides.
+  virtual void OnReplyReceived(const ResourceMessageReplyParams& params,
+                               const IPC::Message& msg) OVERRIDE;
+
   void PostAbortIfNecessary(scoped_refptr<TrackedCallback>* callback);
 
   // IPC message handlers.
@@ -73,15 +88,23 @@ class PPAPI_PROXY_EXPORT UDPSocketResourceBase: public PluginResource {
                                  const ResourceMessageReplyParams& params);
   void OnPluginMsgBindReply(const ResourceMessageReplyParams& params,
                             const PP_NetAddress_Private& bound_addr);
-  void OnPluginMsgRecvFromReply(PP_Resource* output_addr,
-                                const ResourceMessageReplyParams& params,
-                                const std::string& data,
-                                const PP_NetAddress_Private& addr);
+  void OnPluginMsgPushRecvResult(const ResourceMessageReplyParams& params,
+                                 int32_t result,
+                                 const std::string& data,
+                                 const PP_NetAddress_Private& addr);
   void OnPluginMsgSendToReply(const ResourceMessageReplyParams& params,
                               int32_t bytes_written);
 
   void RunCallback(scoped_refptr<TrackedCallback> callback, int32_t pp_result);
 
+  // Callers must ensure that |output_buffer| is big enough to store |data|.
+  int32_t SetRecvFromOutput(int32_t browser_result,
+                            const std::string& data,
+                            const PP_NetAddress_Private& addr,
+                            char* output_buffer,
+                            int32_t num_bytes,
+                            PP_Resource* output_addr);
+
   bool private_api_;
   bool bound_;
   bool closed_;
@@ -92,10 +115,13 @@ class PPAPI_PROXY_EXPORT UDPSocketResourceBase: public PluginResource {
 
   char* read_buffer_;
   int32_t bytes_to_read_;
+  PP_Resource* recvfrom_addr_resource_;
 
   PP_NetAddress_Private recvfrom_addr_;
   PP_NetAddress_Private bound_addr_;
 
+  std::queue<RecvBuffer> recv_buffers_;
+
   DISALLOW_COPY_AND_ASSIGN(UDPSocketResourceBase);
 };