native_client_sdk/src/libraries/nacl_io/socket/udp_node.cc

   1 // Copyright 2013 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 "nacl_io/socket/udp_node.h"
   6
   7 #include <errno.h>
   8 #include <string.h>
   9
  10 #include <algorithm>
  11
  12 #include "nacl_io/log.h"
  13 #include "nacl_io/pepper_interface.h"
  14 #include "nacl_io/socket/packet.h"
  15 #include "nacl_io/socket/udp_event_emitter.h"
  16 #include "nacl_io/stream/stream_fs.h"
  17
  18 namespace {
  19 const size_t kMaxPacketSize = 65536;
  20 const size_t kDefaultFifoSize = kMaxPacketSize * 8;
  21 }
  22
  23 namespace nacl_io {
  24
  25 class UdpWork : public StreamFs::Work {
  26  public:
  27   explicit UdpWork(const ScopedUdpEventEmitter& emitter)
  28       : StreamFs::Work(emitter->stream()->stream()),
  29         emitter_(emitter),
  30         packet_(NULL) {}
  31
  32   ~UdpWork() { delete packet_; }
  33
  34   UDPSocketInterface* UDPInterface() {
  35     return filesystem()->ppapi()->GetUDPSocketInterface();
  36   }
  37
  38  protected:
  39   ScopedUdpEventEmitter emitter_;
  40   Packet* packet_;
  41 };
  42
  43 class UdpSendWork : public UdpWork {
  44  public:
  45   explicit UdpSendWork(const ScopedUdpEventEmitter& emitter,
  46                        const ScopedSocketNode& node)
  47       : UdpWork(emitter), node_(node) {}
  48
  49   virtual bool Start(int32_t val) {
  50     AUTO_LOCK(emitter_->GetLock());
  51
  52     // Does the stream exist, and can it send?
  53     if (!node_->TestStreamFlags(SSF_CAN_SEND))
  54       return false;
  55
  56     packet_ = emitter_->ReadTXPacket_Locked();
  57     if (NULL == packet_)
  58       return false;
  59
  60     int err = UDPInterface()->SendTo(node_->socket_resource(),
  61                                      packet_->buffer(),
  62                                      packet_->len(),
  63                                      packet_->addr(),
  64                                      filesystem()->GetRunCompletion(this));
  65     if (err != PP_OK_COMPLETIONPENDING) {
  66       // Anything else, we should assume the socket has gone bad.
  67       node_->SetError_Locked(err);
  68       return false;
  69     }
  70
  71     node_->SetStreamFlags(SSF_SENDING);
  72     return true;
  73   }
  74
  75   virtual void Run(int32_t length_error) {
  76     AUTO_LOCK(emitter_->GetLock());
  77
  78     if (length_error < 0) {
  79       node_->SetError_Locked(length_error);
  80       return;
  81     }
  82
  83     // If we did send, then Q more work.
  84     node_->ClearStreamFlags(SSF_SENDING);
  85     node_->QueueOutput();
  86   }
  87
  88  private:
  89   // We assume that transmits will always complete.  If the upstream
  90   // actually back pressures, enough to prevent the Send callback
  91   // from triggering, this resource may never go away.
  92   ScopedSocketNode node_;
  93 };
  94
  95 class UdpRecvWork : public UdpWork {
  96  public:
  97   explicit UdpRecvWork(const ScopedUdpEventEmitter& emitter)
  98       : UdpWork(emitter) {
  99     data_ = new char[kMaxPacketSize];
 100   }
 101
 102   ~UdpRecvWork() { delete[] data_; }
 103
 104   virtual bool Start(int32_t val) {
 105     AUTO_LOCK(emitter_->GetLock());
 106     UdpNode* stream = static_cast<UdpNode*>(emitter_->stream());
 107
 108     // Does the stream exist, and can it recv?
 109     if (NULL == stream || !stream->TestStreamFlags(SSF_CAN_RECV))
 110       return false;
 111
 112     // Check if we are already receiving.
 113     if (stream->TestStreamFlags(SSF_RECVING))
 114       return false;
 115
 116     stream->SetStreamFlags(SSF_RECVING);
 117     int err = UDPInterface()->RecvFrom(stream->socket_resource(),
 118                                        data_,
 119                                        kMaxPacketSize,
 120                                        &addr_,
 121                                        filesystem()->GetRunCompletion(this));
 122     if (err != PP_OK_COMPLETIONPENDING) {
 123       stream->SetError_Locked(err);
 124       return false;
 125     }
 126
 127     return true;
 128   }
 129
 130   virtual void Run(int32_t length_error) {
 131     AUTO_LOCK(emitter_->GetLock());
 132     UdpNode* stream = static_cast<UdpNode*>(emitter_->stream());
 133     if (NULL == stream)
 134       return;
 135
 136     // On successful receive we queue more input
 137     if (length_error > 0) {
 138       Packet* packet = new Packet(filesystem()->ppapi());
 139       packet->Copy(data_, length_error, addr_);
 140       emitter_->WriteRXPacket_Locked(packet);
 141       stream->ClearStreamFlags(SSF_RECVING);
 142       stream->QueueInput();
 143     } else {
 144       stream->SetError_Locked(length_error);
 145     }
 146   }
 147
 148  private:
 149   char* data_;
 150   PP_Resource addr_;
 151 };
 152
 153 UdpNode::UdpNode(Filesystem* filesystem)
 154     : SocketNode(filesystem),
 155       emitter_(new UdpEventEmitter(kDefaultFifoSize, kDefaultFifoSize)) {
 156   emitter_->AttachStream(this);
 157 }
 158
 159 void UdpNode::Destroy() {
 160   emitter_->DetachStream();
 161   SocketNode::Destroy();
 162 }
 163
 164 UdpEventEmitter* UdpNode::GetEventEmitter() {
 165   return emitter_.get();
 166 }
 167
 168 Error UdpNode::Init(int open_flags) {
 169   Error err = SocketNode::Init(open_flags);
 170   if (err != 0)
 171     return err;
 172
 173   if (UDPInterface() == NULL) {
 174     LOG_ERROR("Got NULL interface: UDP");
 175     return EACCES;
 176   }
 177
 178   socket_resource_ =
 179       UDPInterface()->Create(filesystem_->ppapi()->GetInstance());
 180   if (0 == socket_resource_) {
 181     LOG_ERROR("Unable to create UDP resource.");
 182     return EACCES;
 183   }
 184
 185   return 0;
 186 }
 187
 188 void UdpNode::QueueInput() {
 189   UdpRecvWork* work = new UdpRecvWork(emitter_);
 190   stream()->EnqueueWork(work);
 191 }
 192
 193 void UdpNode::QueueOutput() {
 194   if (!TestStreamFlags(SSF_CAN_SEND))
 195     return;
 196
 197   if (TestStreamFlags(SSF_SENDING))
 198     return;
 199
 200   UdpSendWork* work = new UdpSendWork(emitter_, ScopedSocketNode(this));
 201   stream()->EnqueueWork(work);
 202 }
 203
 204 Error UdpNode::Bind(const struct sockaddr* addr, socklen_t len) {
 205   if (0 == socket_resource_)
 206     return EBADF;
 207
 208   /* Only bind once. */
 209   if (IsBound())
 210     return EINVAL;
 211
 212   PP_Resource out_addr = SockAddrToResource(addr, len);
 213   if (0 == out_addr)
 214     return EINVAL;
 215
 216   int err =
 217       UDPInterface()->Bind(socket_resource_, out_addr, PP_BlockUntilComplete());
 218   filesystem_->ppapi()->ReleaseResource(out_addr);
 219   if (err != 0)
 220     return PPErrorToErrno(err);
 221
 222   // Get the address that was actually bound (in case addr was 0.0.0.0:0).
 223   out_addr = UDPInterface()->GetBoundAddress(socket_resource_);
 224   if (out_addr == 0)
 225     return EINVAL;
 226
 227   // Now that we are bound, we can start sending and receiving.
 228   SetStreamFlags(SSF_CAN_SEND | SSF_CAN_RECV);
 229   QueueInput();
 230
 231   local_addr_ = out_addr;
 232   return 0;
 233 }
 234
 235 Error UdpNode::Connect(const HandleAttr& attr,
 236                        const struct sockaddr* addr,
 237                        socklen_t len) {
 238   if (0 == socket_resource_)
 239     return EBADF;
 240
 241   /* Connect for UDP is the default dest, it's legal to change it. */
 242   if (remote_addr_ != 0) {
 243     filesystem_->ppapi()->ReleaseResource(remote_addr_);
 244     remote_addr_ = 0;
 245   }
 246
 247   remote_addr_ = SockAddrToResource(addr, len);
 248   if (0 == remote_addr_)
 249     return EINVAL;
 250
 251   return 0;
 252 }
 253
 254 Error UdpNode::Recv_Locked(void* buf,
 255                            size_t len,
 256                            PP_Resource* out_addr,
 257                            int* out_len) {
 258   Packet* packet = emitter_->ReadRXPacket_Locked();
 259   *out_len = 0;
 260   *out_addr = 0;
 261
 262   if (packet) {
 263     int capped_len = static_cast<int32_t>(std::min<int>(len, packet->len()));
 264     memcpy(buf, packet->buffer(), capped_len);
 265
 266     if (packet->addr() != 0) {
 267       filesystem_->ppapi()->AddRefResource(packet->addr());
 268       *out_addr = packet->addr();
 269     }
 270
 271     *out_len = capped_len;
 272     delete packet;
 273     return 0;
 274   }
 275
 276   // Should never happen, Recv_Locked should not be called
 277   // unless already in a POLLIN state.
 278   return EBADF;
 279 }
 280
 281 Error UdpNode::Send_Locked(const void* buf,
 282                            size_t len,
 283                            PP_Resource addr,
 284                            int* out_len) {
 285   if (!IsBound()) {
 286     // Pepper requires a socket to be bound before it can send.
 287     sockaddr_in addr;
 288     addr.sin_family = AF_INET;
 289     addr.sin_port = 0;
 290     memset(&addr.sin_addr, 0, sizeof(addr.sin_addr));
 291     Error err =
 292         Bind(reinterpret_cast<const struct sockaddr*>(&addr), sizeof(addr));
 293     if (err != 0)
 294       return err;
 295   }
 296
 297   *out_len = 0;
 298   int capped_len = static_cast<int32_t>(std::min<int>(len, kMaxPacketSize));
 299   Packet* packet = new Packet(filesystem_->ppapi());
 300   packet->Copy(buf, capped_len, addr);
 301
 302   emitter_->WriteTXPacket_Locked(packet);
 303   *out_len = capped_len;
 304   return 0;
 305 }
 306
 307 }  // namespace nacl_io