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