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

ppapi/proxy/udp_socket_resource_base.cc

 // Copyright 2013 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 "ppapi/proxy/udp_socket_resource_base.h"
 
 #include <algorithm>
 #include <cstring>
 
 #include "base/logging.h"
 #include "ppapi/c/pp_bool.h"
 #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"
 #include "ppapi/thunk/resource_creation_api.h"
 
 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,
                                              bool private_api)
     : PluginResource(connection, instance),
       private_api_(private_api),
       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));
   bound_addr_.size = 0;
   memset(bound_addr_.data, 0,
          arraysize(bound_addr_.data) * sizeof(*bound_addr_.data));
 
   if (private_api)
     SendCreate(BROWSER, PpapiHostMsg_UDPSocket_CreatePrivate());
   else
     SendCreate(BROWSER, PpapiHostMsg_UDPSocket_Create());
+
+  PluginGlobals::Get()->resource_reply_thread_registrar()->HandleOnIOThread(
+      pp_resource(),
+      PpapiPluginMsg_UDPSocket_PushRecvResult::ID);
 }
 
 UDPSocketResourceBase::~UDPSocketResourceBase() {
 }
 
 int32_t UDPSocketResourceBase::SetOptionImpl(
     PP_UDPSocket_Option name,
     const PP_Var& value,
     scoped_refptr<TrackedCallback> callback) {
   if (closed_)
@@ skipping 71 matching lines @@
     int32_t num_bytes,
     PP_Resource* addr,
     scoped_refptr<TrackedCallback> callback) {
   if (!buffer || num_bytes <= 0)
     return PP_ERROR_BADARGUMENT;
   if (!bound_)
     return PP_ERROR_FAILED;
   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(
     PP_NetAddress_Private* addr) {
   if (!addr)
     return PP_FALSE;
   *addr = recvfrom_addr_;
   return PP_TRUE;
 }
 
@@ skipping 34 matching lines @@
   Post(BROWSER, PpapiHostMsg_UDPSocket_Close());
 
   PostAbortIfNecessary(&bind_callback_);
   PostAbortIfNecessary(&recvfrom_callback_);
   PostAbortIfNecessary(&sendto_callback_);
 
   read_buffer_ = NULL;
   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))
     (*callback)->PostAbort();
 }
 
 void UDPSocketResourceBase::OnPluginMsgSetOptionReply(
     scoped_refptr<TrackedCallback> callback,
     const ResourceMessageReplyParams& params) {
   if (TrackedCallback::IsPending(callback))
     RunCallback(callback, params.result());
 }
 
 void UDPSocketResourceBase::OnPluginMsgBindReply(
     const ResourceMessageReplyParams& params,
     const PP_NetAddress_Private& bound_addr) {
   // It is possible that |bind_callback_| is pending while |closed_| is true:
   // CloseImpl() has been called, but a BindReply came earlier than the task to
   // abort |bind_callback_|. We don't want to update |bound_| or |bound_addr_|
   // in that case.
   if (!TrackedCallback::IsPending(bind_callback_) || closed_)
     return;
 
   if (params.result() == PP_OK)
     bound_ = true;
   bound_addr_ = bound_addr;
   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_)
+  // TODO(yzshen): Support passing in a non-const string ref, so that we can
+  // eliminate one copy when storing the data in the buffer.
+
+  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;
-
-  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;
-    }
   }
 
-  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(
     const ResourceMessageReplyParams& params,
     int32_t bytes_written) {
   if (!TrackedCallback::IsPending(sendto_callback_))
     return;
 
   if (params.result() == PP_OK)
     RunCallback(sendto_callback_, bytes_written);
   else
     RunCallback(sendto_callback_, params.result());
 }
 
 void UDPSocketResourceBase::RunCallback(scoped_refptr<TrackedCallback> callback,
                                         int32_t pp_result) {
   callback->Run(ConvertNetworkAPIErrorForCompatibility(pp_result,
                                                        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());

[Tom Sepez, 2014/09/12 01:28:14]

The DCHECK that output_buffer is big enough to handle data.size() should also be
present in release builds to prevent this ever overflowing.

[yzshen1, 2014/09/12 16:49:17]

This is a private method of the class, and all callers have ensured that
num_bytes >= data.size(). Therefore, I think it is okay to just have a DCHECK().
I added a comment above the method declaration. Do you think that is okay?

Thanks!

+
+  recvfrom_addr_ = addr;
+
+  return result == PP_OK ? static_cast<int32_t>(data.size()) : result;
+}
+
 }  // namespace proxy
 }  // namespace ppapi