Never lock the Pepper proxy lock on the IO thread

ppapi/proxy/udp_socket_resource_base.cc

Index: ppapi/proxy/udp_socket_resource_base.cc
diff --git a/ppapi/proxy/udp_socket_resource_base.cc b/ppapi/proxy/udp_socket_resource_base.cc
index e61e16be3675e69f0500a4b8da4828115d67a698..a2f7ce7c67c44e3724fae6187101b2f53b9516c7 100644
--- a/ppapi/proxy/udp_socket_resource_base.cc
+++ b/ppapi/proxy/udp_socket_resource_base.cc
@@ -4,32 +4,40 @@
 
 #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 = 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;
 const size_t UDPSocketResourceBase::kPluginSendBufferSlots = 8u;
 
+namespace {
+
+void RunCallback(scoped_refptr<TrackedCallback> callback,
+                 int32_t pp_result,
+                 bool private_api) {
+  callback->Run(ConvertNetworkAPIErrorForCompatibility(pp_result, private_api));
+}
+
+void PostAbortIfNecessary(const scoped_refptr<TrackedCallback>& callback) {
+  if (TrackedCallback::IsPending(callback))
+    callback->PostAbort();
+}
+
+}  // namespace
+
 UDPSocketResourceBase::UDPSocketResourceBase(Connection connection,
                                              PP_Instance instance,
                                              bool private_api)
@@ -38,26 +46,19 @@ UDPSocketResourceBase::UDPSocketResourceBase(Connection connection,
       bind_called_(false),
       bound_(false),
       closed_(false),
-      read_buffer_(NULL),
-      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));
-
+      recv_filter_(PluginGlobals::Get()->udp_socket_filter()),
+      bound_addr_() {
+  recv_filter_->AddUDPResource(
+      pp_instance(), pp_resource(), private_api,
+      base::Bind(&UDPSocketResourceBase::SlotBecameAvailable, pp_resource()));
   if (private_api)
     SendCreate(BROWSER, PpapiHostMsg_UDPSocket_CreatePrivate());
   else
     SendCreate(BROWSER, PpapiHostMsg_UDPSocket_Create());
-
-  PluginGlobals::Get()->resource_reply_thread_registrar()->HandleOnIOThread(
-      PpapiPluginMsg_UDPSocket_PushRecvResult::ID);
 }
 
 UDPSocketResourceBase::~UDPSocketResourceBase() {
+  CloseImpl();
 }
 
 int32_t UDPSocketResourceBase::SetOptionImpl(
@@ -165,45 +166,19 @@ PP_Bool UDPSocketResourceBase::GetBoundAddressImpl(
 }
 
 int32_t UDPSocketResourceBase::RecvFromImpl(
-    char* buffer,
+    char* buffer_out,
     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;
-
-  if (recv_buffers_.empty()) {
-    read_buffer_ = buffer;
-    bytes_to_read_ = std::min(num_bytes, kMaxReadSize);
-    recvfrom_addr_resource_ = addr;
-    recvfrom_callback_ = callback;
-
-    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;
-  }
+  return recv_filter_->RequestData(pp_resource(), num_bytes, buffer_out, addr,
+                                   callback);
 }
 
 PP_Bool UDPSocketResourceBase::GetRecvFromAddressImpl(
     PP_NetAddress_Private* addr) {
   if (!addr)
     return PP_FALSE;
-  *addr = recvfrom_addr_;
+  *addr = recv_filter_->GetLastAddrPrivate(pp_resource());
   return PP_TRUE;
 }
 
@@ -243,16 +218,13 @@ void UDPSocketResourceBase::CloseImpl() {
 
   Post(BROWSER, PpapiHostMsg_UDPSocket_Close());
 
-  PostAbortIfNecessary(&bind_callback_);
-  PostAbortIfNecessary(&recvfrom_callback_);
+  PostAbortIfNecessary(bind_callback_);
   while (!sendto_callbacks_.empty()) {
     scoped_refptr<TrackedCallback> callback = sendto_callbacks_.front();
     sendto_callbacks_.pop();
-    PostAbortIfNecessary(&callback);
+    PostAbortIfNecessary(callback);
   }
-
-  read_buffer_ = NULL;
-  bytes_to_read_ = -1;
+  recv_filter_->RemoveUDPResource(pp_resource());
 }
 
 int32_t UDPSocketResourceBase::JoinGroupImpl(
@@ -285,29 +257,11 @@ int32_t UDPSocketResourceBase::LeaveGroupImpl(
   return PP_OK_COMPLETIONPENDING;
 }
 
-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::OnPluginMsgGeneralReply(
     scoped_refptr<TrackedCallback> callback,
     const ResourceMessageReplyParams& params) {
   if (TrackedCallback::IsPending(callback))
-    RunCallback(callback, params.result());
+    RunCallback(callback, params.result(), private_api_);
 }
 
 void UDPSocketResourceBase::OnPluginMsgBindReply(
@@ -323,50 +277,7 @@ void UDPSocketResourceBase::OnPluginMsgBindReply(
   if (params.result() == PP_OK)
     bound_ = true;
   bound_addr_ = bound_addr;
-  RunCallback(bind_callback_, params.result());
-}
-
-void UDPSocketResourceBase::OnPluginMsgPushRecvResult(
-    const ResourceMessageReplyParams& params,
-    int32_t result,
-    const std::string& data,
-    const PP_NetAddress_Private& addr) {
-  // 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;
-  }
-
-  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_resource_ = NULL;
-
-  RunCallback(recvfrom_callback_, result);
+  RunCallback(bind_callback_, params.result(), private_api_);
 }
 
 void UDPSocketResourceBase::OnPluginMsgSendToReply(
@@ -383,43 +294,45 @@ void UDPSocketResourceBase::OnPluginMsgSendToReply(
     return;
 
   if (params.result() == PP_OK)
-    RunCallback(callback, bytes_written);
+    RunCallback(callback, bytes_written, private_api_);
   else
-    RunCallback(callback, params.result());
-}
-
-void UDPSocketResourceBase::RunCallback(scoped_refptr<TrackedCallback> callback,
-                                        int32_t pp_result) {
-  callback->Run(ConvertNetworkAPIErrorForCompatibility(pp_result,
-                                                       private_api_));
+    RunCallback(callback, params.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;
+// static
+void UDPSocketResourceBase::SlotBecameAvailable(PP_Resource resource) {
+  PluginGlobals* globals = PluginGlobals::Get();
+  if (globals->ipc_task_runner()->RunsTasksOnCurrentThread()) {
+    // We were invoked on the IO thread, so we do *not* have the ProxyLock and
+    // we have to post over to the main thread in order to get the resource and
+    // send the message.

[dmichael (off chromium), 2015/03/24 20:10:26]

This is kind of weird and a bummer... the callback might be invoked:
1) On the IO thread *without* the lock
2) On some other thread *with* the lock
Pretty confusing. I think I'll move this logic in to the filter instead so that
it just always posts to the main thread using RunWhileLocked. It will mean that
in case #2, we do an unnecessary PostTask to the main thread for updating the
slot info, but it will simplify things a bit.
WDYT?

[raymes, 2015/03/26 06:30:51]

Do you still think this is worth doing? If it leads to simpler code here I'm in
favour :)

[dmichael (off chromium), 2015/03/26 18:19:42]

Done. I did it slightly different than I suggested initially...  from the IO
thread, I post to the main thread with the lock to invoke the callback. As
before, the callback can still be invoked from within the call to RequestData.
It's slightly akward that it's called in a "nested" sort of way, but it has the
proxy lock and can call Post immediately without a thread-hop, so it seemed
worthwhile. This bit of UDPSocketResourceBase is nicely simplified.

+    // TODO(dmichael): This is kind of stupid; we ought to be able to build up
+    // the right message and send it directly from the IO thread. Unfortunately,
+    // we need to get a sequence number, which right now requires the ProxyLock.
+    if (globals->GetMainThreadMessageLoop()) {
+      globals->GetMainThreadMessageLoop()->PostTask(FROM_HERE,
+          RunWhileLocked(base::Bind(
+              &UDPSocketResourceBase::SlotBecameAvailableWithLock, resource)));
     }
+  } else {
+    SlotBecameAvailableWithLock(resource);
   }
+}
 
-  if (result == PP_OK && !data.empty())
-    memcpy(output_buffer, data.c_str(), data.size());
-
-  recvfrom_addr_ = addr;
+// static
+void UDPSocketResourceBase::SlotBecameAvailableWithLock(PP_Resource resource) {
+  ProxyLock::AssertAcquired();
+  // Note, we don't have the ProxyLock yet, so we use EnterResource to acquire
+  // it. (We couldn't use RunWhileLocked in SlotBecameAvailableOnIOThread
+  // because it requires the ProxyLock, which we can't acquire on the IO
+  // thread).
+  thunk::EnterResourceNoLock<thunk::PPB_UDPSocket_API> enter(resource, false);
+  if (enter.failed())
+    return;
+  auto thiz(static_cast<UDPSocketResourceBase*>(enter.resource()));
 
-  return result == PP_OK ? static_cast<int32_t>(data.size()) : result;
+  if (!thiz->closed_)
+    thiz->Post(BROWSER, PpapiHostMsg_UDPSocket_RecvSlotAvailable());
 }
 
 }  // namespace proxy