ppapi/proxy/udp_socket_filter.cc

   1 // Copyright 2015 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "ppapi/proxy/udp_socket_filter.h"
   6
   7 #include <algorithm>
   8 #include <cstring>
   9
  10 #include "base/logging.h"
  11 #include "ppapi/c/pp_errors.h"
  12 #include "ppapi/proxy/error_conversion.h"
  13 #include "ppapi/proxy/plugin_globals.h"
  14 #include "ppapi/proxy/ppapi_messages.h"
  15 #include "ppapi/thunk/enter.h"
  16 #include "ppapi/thunk/resource_creation_api.h"
  17
  18 namespace ppapi {
  19 namespace proxy {
  20
  21 const int32_t UDPSocketFilter::kMaxReadSize = 128 * 1024;
  22 const int32_t UDPSocketFilter::kMaxReceiveBufferSize =
  23     1024 * UDPSocketFilter::kMaxReadSize;
  24 const size_t UDPSocketFilter::kPluginReceiveBufferSlots = 32u;
  25
  26 namespace {
  27
  28 int32_t SetRecvFromOutput(PP_Instance pp_instance,
  29                           const scoped_ptr<std::string>& data,
  30                           const PP_NetAddress_Private& addr,
  31                           char* output_buffer,
  32                           int32_t num_bytes,
  33                           PP_Resource* output_addr,
  34                           int32_t browser_result) {
  35   ProxyLock::AssertAcquired();
  36   DCHECK_GE(num_bytes, static_cast<int32_t>(data->size()));
  37
  38   int32_t result = browser_result;
  39   if (result == PP_OK && output_addr) {
  40     thunk::EnterResourceCreationNoLock enter(pp_instance);
  41     if (enter.succeeded()) {
  42       *output_addr = enter.functions()->CreateNetAddressFromNetAddressPrivate(
  43           pp_instance, addr);
  44     } else {
  45       result = PP_ERROR_FAILED;
  46     }
  47   }
  48
  49   if (result == PP_OK && !data->empty())
  50     memcpy(output_buffer, data->c_str(), data->size());
  51
  52   return result == PP_OK ? static_cast<int32_t>(data->size()) : result;
  53 }
  54
  55 }  // namespace
  56
  57 UDPSocketFilter::UDPSocketFilter() {
  58 }
  59
  60 UDPSocketFilter::~UDPSocketFilter() {
  61 }
  62
  63 void UDPSocketFilter::AddUDPResource(
  64     PP_Instance instance,
  65     PP_Resource resource,
  66     bool private_api,
  67     const base::Closure& slot_available_callback) {
  68   ProxyLock::AssertAcquired();
  69   base::AutoLock acquire(lock_);
  70   DCHECK(!queues_.contains(resource));
  71   queues_.add(resource, scoped_ptr<RecvQueue>(new RecvQueue(
  72                             instance, private_api, slot_available_callback)));
  73 }
  74
  75 void UDPSocketFilter::RemoveUDPResource(PP_Resource resource) {
  76   ProxyLock::AssertAcquired();
  77   base::AutoLock acquire(lock_);
  78   DCHECK(queues_.contains(resource));
  79   queues_.erase(resource);
  80 }
  81
  82 int32_t UDPSocketFilter::RequestData(
  83     PP_Resource resource,
  84     int32_t num_bytes,
  85     char* buffer,
  86     PP_Resource* addr,
  87     const scoped_refptr<TrackedCallback>& callback) {
  88   ProxyLock::AssertAcquired();
  89   base::AutoLock acquire(lock_);
  90   RecvQueue* queue_ptr = queues_.get(resource);
  91   if (!queue_ptr) {
  92     NOTREACHED();
  93     return PP_ERROR_FAILED;
  94   }
  95   return queue_ptr->RequestData(num_bytes, buffer, addr, callback);
  96 }
  97
  98 bool UDPSocketFilter::OnResourceReplyReceived(
  99     const ResourceMessageReplyParams& params,
 100     const IPC::Message& nested_msg) {
 101   bool handled = true;
 102   PPAPI_BEGIN_MESSAGE_MAP(UDPSocketFilter, nested_msg)
 103     PPAPI_DISPATCH_PLUGIN_RESOURCE_CALL(PpapiPluginMsg_UDPSocket_PushRecvResult,
 104                                         OnPluginMsgPushRecvResult)
 105     PPAPI_DISPATCH_PLUGIN_RESOURCE_CALL_UNHANDLED(handled = false)
 106   PPAPI_END_MESSAGE_MAP()
 107   return handled;
 108 }
 109
 110 PP_NetAddress_Private UDPSocketFilter::GetLastAddrPrivate(
 111     PP_Resource resource) const {
 112   base::AutoLock acquire(lock_);
 113   return queues_.get(resource)->GetLastAddrPrivate();
 114 }
 115
 116 void UDPSocketFilter::OnPluginMsgPushRecvResult(
 117     const ResourceMessageReplyParams& params,
 118     int32_t result,
 119     const std::string& data,
 120     const PP_NetAddress_Private& addr) {
 121   DCHECK(PluginGlobals::Get()->ipc_task_runner()->RunsTasksOnCurrentThread());
 122   base::AutoLock acquire(lock_);
 123   RecvQueue* queue_ptr = queues_.get(params.pp_resource());
 124   // The RecvQueue might be gone if there were messages in-flight for a
 125   // resource that has been destroyed.
 126   if (queue_ptr) {
 127     // TODO(yzshen): Support passing in a non-const string ref, so that we can
 128     // eliminate one copy when storing the data in the buffer.
 129     queue_ptr->DataReceivedOnIOThread(result, data, addr);
 130   }
 131 }
 132
 133 UDPSocketFilter::RecvQueue::RecvQueue(
 134     PP_Instance pp_instance,
 135     bool private_api,
 136     const base::Closure& slot_available_callback)
 137     : pp_instance_(pp_instance),
 138       read_buffer_(nullptr),
 139       bytes_to_read_(0),
 140       recvfrom_addr_resource_(nullptr),
 141       last_recvfrom_addr_(),
 142       private_api_(private_api),
 143       slot_available_callback_(slot_available_callback) {
 144 }
 145
 146 UDPSocketFilter::RecvQueue::~RecvQueue() {
 147   if (TrackedCallback::IsPending(recvfrom_callback_))
 148     recvfrom_callback_->PostAbort();
 149 }
 150
 151 void UDPSocketFilter::RecvQueue::DataReceivedOnIOThread(
 152     int32_t result,
 153     const std::string& data,
 154     const PP_NetAddress_Private& addr) {
 155   DCHECK(PluginGlobals::Get()->ipc_task_runner()->RunsTasksOnCurrentThread());
 156   DCHECK_LT(recv_buffers_.size(), UDPSocketFilter::kPluginReceiveBufferSlots);
 157
 158   if (!TrackedCallback::IsPending(recvfrom_callback_) || !read_buffer_) {
 159     recv_buffers_.push(RecvBuffer());
 160     RecvBuffer& back = recv_buffers_.back();
 161     back.result = result;
 162     back.data = data;
 163     back.addr = addr;
 164     return;
 165   }
 166   DCHECK_EQ(recv_buffers_.size(), 0u);
 167
 168   if (bytes_to_read_ < static_cast<int32_t>(data.size())) {
 169     recv_buffers_.push(RecvBuffer());
 170     RecvBuffer& back = recv_buffers_.back();
 171     back.result = result;
 172     back.data = data;
 173     back.addr = addr;
 174
 175     result = PP_ERROR_MESSAGE_TOO_BIG;
 176   } else {
 177     // Instead of calling SetRecvFromOutput directly, post it as a completion
 178     // task, so that:
 179     // 1) It can run with the ProxyLock (we can't lock it on the IO thread.)
 180     // 2) So that we only write to the output params in the case of success.
 181     //    (Since the callback will complete on another thread, it's possible
 182     //     that the resource will be deleted and abort the callback before it
 183     //     is actually run.)
 184     scoped_ptr<std::string> data_to_pass(new std::string(data));
 185     recvfrom_callback_->set_completion_task(base::Bind(
 186         &SetRecvFromOutput, pp_instance_, base::Passed(data_to_pass.Pass()),
 187         addr, base::Unretained(read_buffer_), bytes_to_read_,
 188         base::Unretained(recvfrom_addr_resource_)));
 189     last_recvfrom_addr_ = addr;
 190     PpapiGlobals::Get()->GetMainThreadMessageLoop()->PostTask(
 191         FROM_HERE,
 192         RunWhileLocked(slot_available_callback_));
 193   }
 194
 195   read_buffer_ = NULL;
 196   bytes_to_read_ = -1;
 197   recvfrom_addr_resource_ = NULL;
 198
 199   recvfrom_callback_->Run(
 200       ConvertNetworkAPIErrorForCompatibility(result, private_api_));
 201 }
 202
 203 int32_t UDPSocketFilter::RecvQueue::RequestData(
 204     int32_t num_bytes,
 205     char* buffer_out,
 206     PP_Resource* addr_out,
 207     const scoped_refptr<TrackedCallback>& callback) {
 208   ProxyLock::AssertAcquired();
 209   if (!buffer_out || num_bytes <= 0)
 210     return PP_ERROR_BADARGUMENT;
 211   if (TrackedCallback::IsPending(recvfrom_callback_))
 212     return PP_ERROR_INPROGRESS;
 213
 214   if (recv_buffers_.empty()) {
 215     read_buffer_ = buffer_out;
 216     bytes_to_read_ = std::min(num_bytes, UDPSocketFilter::kMaxReadSize);
 217     recvfrom_addr_resource_ = addr_out;
 218     recvfrom_callback_ = callback;
 219     return PP_OK_COMPLETIONPENDING;
 220   } else {
 221     RecvBuffer& front = recv_buffers_.front();
 222
 223     if (static_cast<size_t>(num_bytes) < front.data.size())
 224       return PP_ERROR_MESSAGE_TOO_BIG;
 225
 226     int32_t result = static_cast<int32_t>(front.data.size());
 227     scoped_ptr<std::string> data_to_pass(new std::string);
 228     data_to_pass->swap(front.data);
 229     SetRecvFromOutput(pp_instance_, data_to_pass.Pass(), front.addr, buffer_out,
 230                       num_bytes, addr_out, PP_OK);
 231     last_recvfrom_addr_ = front.addr;
 232     recv_buffers_.pop();
 233     slot_available_callback_.Run();
 234
 235     return result;
 236   }
 237 }
 238
 239 PP_NetAddress_Private UDPSocketFilter::RecvQueue::GetLastAddrPrivate() const {
 240   CHECK(private_api_);
 241   return last_recvfrom_addr_;
 242 }
 243
 244 }  // namespace proxy
 245 }  // namespace ppapi