src/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   }
 100
 101   virtual bool Start(int32_t val) {
 102     AUTO_LOCK(emitter_->GetLock());
 103     UdpNode* stream = static_cast<UdpNode*>(emitter_->stream());
 104
 105     // Does the stream exist, and can it recv?
 106     if (NULL == stream || !stream->TestStreamFlags(SSF_CAN_RECV))
 107       return false;
 108
 109     // Check if we are already receiving.
 110     if (stream->TestStreamFlags(SSF_RECVING))
 111       return false;
 112
 113     stream->SetStreamFlags(SSF_RECVING);
 114     int err = UDPInterface()->RecvFrom(stream->socket_resource(),
 115                                        data_,
 116                                        kMaxPacketSize,
 117                                        &addr_,
 118                                        filesystem()->GetRunCompletion(this));
 119     if (err != PP_OK_COMPLETIONPENDING) {
 120       stream->SetError_Locked(err);
 121       return false;
 122     }
 123
 124     return true;
 125   }
 126
 127   virtual void Run(int32_t length_error) {
 128     AUTO_LOCK(emitter_->GetLock());
 129     UdpNode* stream = static_cast<UdpNode*>(emitter_->stream());
 130     if (NULL == stream)
 131       return;
 132
 133     // On successful receive we queue more input
 134     if (length_error > 0) {
 135       Packet* packet = new Packet(filesystem()->ppapi());
 136       packet->Copy(data_, length_error, addr_);
 137       filesystem()->ppapi()->ReleaseResource(addr_);
 138       emitter_->WriteRXPacket_Locked(packet);
 139       stream->ClearStreamFlags(SSF_RECVING);
 140       stream->QueueInput();
 141     } else {
 142       stream->SetError_Locked(length_error);
 143     }
 144   }
 145
 146  private:
 147   char data_[kMaxPacketSize];
 148   PP_Resource addr_;
 149 };
 150
 151 UdpNode::UdpNode(Filesystem* filesystem)
 152     : SocketNode(filesystem),
 153       emitter_(new UdpEventEmitter(kDefaultFifoSize, kDefaultFifoSize)) {
 154   emitter_->AttachStream(this);
 155 }
 156
 157 void UdpNode::Destroy() {
 158   emitter_->DetachStream();
 159   SocketNode::Destroy();
 160 }
 161
 162 UdpEventEmitter* UdpNode::GetEventEmitter() {
 163   return emitter_.get();
 164 }
 165
 166 Error UdpNode::Init(int open_flags) {
 167   Error err = SocketNode::Init(open_flags);
 168   if (err != 0)
 169     return err;
 170
 171   if (UDPInterface() == NULL) {
 172     LOG_ERROR("Got NULL interface: UDP");
 173     return EACCES;
 174   }
 175
 176   socket_resource_ =
 177       UDPInterface()->Create(filesystem_->ppapi()->GetInstance());
 178   if (0 == socket_resource_) {
 179     LOG_ERROR("Unable to create UDP resource.");
 180     return EACCES;
 181   }
 182
 183   return 0;
 184 }
 185
 186 void UdpNode::QueueInput() {
 187   UdpRecvWork* work = new UdpRecvWork(emitter_);
 188   stream()->EnqueueWork(work);
 189 }
 190
 191 void UdpNode::QueueOutput() {
 192   if (!TestStreamFlags(SSF_CAN_SEND))
 193     return;
 194
 195   if (TestStreamFlags(SSF_SENDING))
 196     return;
 197
 198   UdpSendWork* work = new UdpSendWork(emitter_, ScopedSocketNode(this));
 199   stream()->EnqueueWork(work);
 200 }
 201
 202 Error UdpNode::Bind(const struct sockaddr* addr, socklen_t len) {
 203   if (0 == socket_resource_)
 204     return EBADF;
 205
 206   /* Only bind once. */
 207   if (IsBound())
 208     return EINVAL;
 209
 210   PP_Resource out_addr = SockAddrToResource(addr, len);
 211   if (0 == out_addr)
 212     return EINVAL;
 213
 214   int err =
 215       UDPInterface()->Bind(socket_resource_, out_addr, PP_BlockUntilComplete());
 216   filesystem_->ppapi()->ReleaseResource(out_addr);
 217   if (err != 0)
 218     return PPErrorToErrno(err);
 219
 220   // Get the address that was actually bound (in case addr was 0.0.0.0:0).
 221   out_addr = UDPInterface()->GetBoundAddress(socket_resource_);
 222   if (out_addr == 0)
 223     return EINVAL;
 224
 225   // Now that we are bound, we can start sending and receiving.
 226   SetStreamFlags(SSF_CAN_SEND | SSF_CAN_RECV);
 227   QueueInput();
 228
 229   local_addr_ = out_addr;
 230   return 0;
 231 }
 232
 233 Error UdpNode::Connect(const HandleAttr& attr,
 234                        const struct sockaddr* addr,
 235                        socklen_t len) {
 236   if (0 == socket_resource_)
 237     return EBADF;
 238
 239   /* Connect for UDP is the default dest, it's legal to change it. */
 240   if (remote_addr_ != 0) {
 241     filesystem_->ppapi()->ReleaseResource(remote_addr_);
 242     remote_addr_ = 0;
 243   }
 244
 245   remote_addr_ = SockAddrToResource(addr, len);
 246   if (0 == remote_addr_)
 247     return EINVAL;
 248
 249   return 0;
 250 }
 251
 252 Error UdpNode::Recv_Locked(void* buf,
 253                            size_t len,
 254                            PP_Resource* out_addr,
 255                            int* out_len) {
 256   Packet* packet = emitter_->ReadRXPacket_Locked();
 257   *out_len = 0;
 258   *out_addr = 0;
 259
 260   if (packet) {
 261     int capped_len = static_cast<int32_t>(std::min<int>(len, packet->len()));
 262     memcpy(buf, packet->buffer(), capped_len);
 263
 264     if (packet->addr() != 0) {
 265       filesystem_->ppapi()->AddRefResource(packet->addr());
 266       *out_addr = packet->addr();
 267     }
 268
 269     *out_len = capped_len;
 270     delete packet;
 271     return 0;
 272   }
 273
 274   // Should never happen, Recv_Locked should not be called
 275   // unless already in a POLLIN state.
 276   return EBADF;
 277 }
 278
 279 Error UdpNode::Send_Locked(const void* buf,
 280                            size_t len,
 281                            PP_Resource addr,
 282                            int* out_len) {
 283   if (!IsBound()) {
 284     // Pepper requires a socket to be bound before it can send.
 285     sockaddr_in addr;
 286     addr.sin_family = AF_INET;
 287     addr.sin_port = 0;
 288     memset(&addr.sin_addr, 0, sizeof(addr.sin_addr));
 289     Error err =
 290         Bind(reinterpret_cast<const struct sockaddr*>(&addr), sizeof(addr));
 291     if (err != 0)
 292       return err;
 293   }
 294
 295   *out_len = 0;
 296   int capped_len = static_cast<int32_t>(std::min<int>(len, kMaxPacketSize));
 297   Packet* packet = new Packet(filesystem_->ppapi());
 298   packet->Copy(buf, capped_len, addr);
 299
 300   emitter_->WriteTXPacket_Locked(packet);
 301   *out_len = capped_len;
 302   return 0;
 303 }
 304
 305 }  // namespace nacl_io