[NaCl SDK] Support non blocking TCP/UDP

native_client_sdk/src/libraries/nacl_io/mount_node_tcp.cc

 // Copyright (c) 2013 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 
 #include "nacl_io/ossocket.h"
 #ifdef PROVIDES_SOCKET_API
 
 #include <errno.h>
 #include <string.h>
 #include <algorithm>
 
-#include "nacl_io/mount.h"
-#include "nacl_io/mount_node_socket.h"
 #include "nacl_io/mount_node_tcp.h"
+#include "nacl_io/mount_stream.h"
 #include "nacl_io/pepper_interface.h"
 
+namespace {
+  size_t kMaxPacketSize = 65536;

[binji, 2013/09/15 22:18:58]

const

[noelallen1, 2013/09/17 21:21:54]

Done.

+  size_t kDefaultFifoSize = kMaxPacketSize * 8;
+}
+
 namespace nacl_io {
 
-MountNodeTCP::MountNodeTCP(Mount* mount) : MountNodeSocket(mount) {}
+class TCPWork : public MountStreamWork {
+ public:
+  TCPWork(EventEmitterTCP* emitter)

[binji, 2013/09/15 22:18:58]

explicit

[noelallen1, 2013/09/17 21:21:54]

Done.

+      : MountStreamWork(emitter->stream()->mount_stream()),
+        emitter_(emitter) {}
+
+  TCPSocketInterface* TCPInterface() {
+    return mount()->ppapi()->GetTCPSocketInterface();
+  }
+
+ protected:
+  ScopedEventEmitterTCP emitter_;
+};
 
 
-TCPSocketInterface* MountNodeTCP::TCPSocket() {
-  if (mount_->ppapi() == NULL)
-    return NULL;
+class TCPSendWork : public TCPWork {
+ public:
+  TCPSendWork(EventEmitterTCP* emitter) : TCPWork(emitter) {}

[binji, 2013/09/15 22:18:58]

explicit

[noelallen1, 2013/09/17 21:21:54]

Done.

 
-  return mount_->ppapi()->GetTCPSocketInterface();
+  virtual bool Start(int32_t val) {
+    AUTO_LOCK(emitter_->GetLock());
+    MountNodeTCP* stream = reinterpret_cast<MountNodeTCP*>(emitter_->stream());

[binji, 2013/09/15 22:18:58]

static_cast (though unnecessary if you use covariant returns on
EventEmitterTCP).

[noelallen1, 2013/09/17 21:21:54]

Done.

+    // If the stream is valid, and we are not currently sending
+    if (stream && ((stream->GetStreamFlags() & SSF_SENDING) == 0)) {
+      size_t tx_data_avail = emitter_->tx_fifo()->ReadAvailable();
+      int capped_len =
+          static_cast<int32_t>(std::min(tx_data_avail, kMaxPacketSize));
+      if (capped_len == 0)
+        return false;
+
+      char* data = new char[capped_len];
+      emitter_->ReadTXBytes_Locked(data, capped_len);
+
+      stream->SetStreamFlags(SSF_SENDING);
+      int err = TCPInterface()->Write(stream->socket_resource(),
+                                      data,
+                                      capped_len,
+                                      mount()->GetRunCompletion(this));
+      delete data;

[binji, 2013/09/15 22:18:58]

Deleting the data sent to an async call?

[noelallen1, 2013/09/17 21:21:54]

Consolidated into base class.

+      if (err == PP_OK_COMPLETIONPENDING)

[binji, 2013/09/15 22:18:58]

any way to return this error in case it is not PP_OK_COMPLETIONPENDING?

[noelallen1, 2013/09/17 21:21:54]

Will log once we officially have logging.

+        return true;
+
+      stream->ClearStreamFlags(SSF_SENDING);
+    }
+    return false;
+  }
+
+  virtual void Run(int32_t val) {

[binji, 2013/09/15 22:18:58]

Run is a strange name for this; it isn't meant to be called directly. Maybe
OnCompleted?

[noelallen1, 2013/09/17 21:21:54]

MountStream::GetOnCompletedCompletion?

+    AUTO_LOCK(emitter_->GetLock());
+    MountNodeTCP* stream = reinterpret_cast<MountNodeTCP*>(emitter_->stream());

[binji, 2013/09/15 22:18:58]

static_cast

[noelallen1, 2013/09/17 21:21:54]

Done.

+
+    // If the stream is still there, see if we can queue more output
+    if (stream) {
+      stream->ClearStreamFlags(SSF_SENDING);
+      stream->QueueOutput();
+    }
+  }
+};
+
+class TCPRecvWork : public TCPWork {
+ public:
+  TCPRecvWork(EventEmitterTCP* emitter) : TCPWork(emitter), data_(NULL) {}

[binji, 2013/09/15 22:18:58]

explicit

[noelallen1, 2013/09/17 21:21:54]

Done.

+
+  ~TCPRecvWork() {
+    delete[] data_;
+  }
+
+  virtual bool Start(int32_t val) {
+    AUTO_LOCK(emitter_->GetLock());
+    MountNodeTCP* stream = reinterpret_cast<MountNodeTCP*>(emitter_->stream());

[binji, 2013/09/15 22:18:58]

static_cast

[noelallen1, 2013/09/17 21:21:54]

Done.

+
+    size_t rx_space_avail = emitter_->rx_fifo()->WriteAvailable();
+    int capped_len =
+        static_cast<int32_t>(std::min(rx_space_avail, kMaxPacketSize));
+
+    if (NULL == stream || (stream->GetStreamFlags() & SSF_RECVING) == 0)
+      return false;
+
+    if (capped_len <= 0)
+      return false;
+
+    stream->SetStreamFlags(SSF_RECVING);
+    data_ = new char[capped_len];
+    int err = TCPInterface()->Read(stream->socket_resource(),
+                                    data_,

[binji, 2013/09/15 22:18:58]

nit: align params

[noelallen1, 2013/09/17 21:21:54]

Done.

+                                    capped_len,
+                                    mount()->GetRunCompletion(this));
+    if (err == PP_OK_COMPLETIONPENDING)
+      return true;
+
+    stream->ClearStreamFlags(SSF_RECVING);
+    return false;
+  }
+
+  virtual void Run(int32_t val) {
+    AUTO_LOCK(emitter_->GetLock());
+    MountNodeTCP* stream =
+        reinterpret_cast<MountNodeTCP*>(emitter_->stream());

[binji, 2013/09/15 22:18:58]

static_cast

[noelallen1, 2013/09/17 21:21:54]

Done.

+
+    if (NULL == stream)
+      return;
+
+    // If the stream is still there, see if we can queue more input
+    if (val > 0)

[binji, 2013/09/15 22:18:58]

better name for val? (bytes_read or something?)

[noelallen1, 2013/09/17 21:21:54]

Done.

+      emitter_->WriteRXBytes_Locked(data_, val);
+
+    stream->ClearStreamFlags(SSF_RECVING);
+    stream->QueueInput();
+  }
+
+ private:
+  char* data_;

[binji, 2013/09/15 22:18:58]

std::vector

+};
+
+
+MountNodeTCP::MountNodeTCP(Mount* mount)
+    : MountNodeSocket(mount) {
 }
 
 Error MountNodeTCP::Init(int flags) {
-  if (TCPSocket() == NULL)
+  if (TCPInterface() == NULL)
     return EACCES;
 
-  socket_resource_ = TCPSocket()->Create(mount_->ppapi()->GetInstance());
+  socket_resource_ = TCPInterface()->Create(mount_->ppapi()->GetInstance());
   if (0 == socket_resource_)
     return EACCES;
 
   return 0;
 }
 
+EventEmitter* MountNodeTCP::GetEventEmitter() {
+  return NULL;

[binji, 2013/09/15 22:18:58]

Why no emitter?

[noelallen1, 2013/09/17 21:21:54]

Done.

+}
+

[binji, 2013/09/15 22:18:58]

Looks like TCP is not yet implemented?

[noelallen1, 2013/09/17 21:21:54]

Bind is not allowed on client sockets.  There's no API for it for TCP.
Added comment.

[binji, 2013/09/19 00:48:54]

Sorry, I wasn't talking about Bind, I meant non-blocking TCP is not yet
implemented. Send/Recv below still use BlockUntilComplete.

[noelallen1, 2013/09/19 21:29:27]

Ah I see, fixed.

 Error MountNodeTCP::Bind(const struct sockaddr* addr, socklen_t len) {
   AUTO_LOCK(node_lock_);
 
   if (0 == socket_resource_)
     return EBADF;
 
   /* Only bind once. */
   if (local_addr_ != 0)
     return EINVAL;
 
   /* Lie, we won't know until we connect. */
   return 0;
 }
 
 Error MountNodeTCP::Connect(const struct sockaddr* addr, socklen_t len) {
   AUTO_LOCK(node_lock_);
 
   if (0 == socket_resource_)
     return EBADF;
 
   if (remote_addr_ != 0)
     return EISCONN;
 
   remote_addr_ = SockAddrToResource(addr, len);
   if (0 == remote_addr_)
     return EINVAL;
 
-  int err = TCPSocket()->Connect(socket_resource_,
-                                 remote_addr_,
-                                 PP_BlockUntilComplete());
+  int err = TCPInterface()->Connect(socket_resource_,
+                                    remote_addr_,
+                                    PP_BlockUntilComplete());
 
   // If we fail, release the dest addr resource
   if (err != PP_OK) {
     mount_->ppapi()->ReleaseResource(remote_addr_);
     remote_addr_ = 0;
     return PPErrorToErrno(err);
   }
 
-  local_addr_ = TCPSocket()->GetLocalAddress(socket_resource_);
+  local_addr_ = TCPInterface()->GetLocalAddress(socket_resource_);
   mount_->ppapi()->AddRefResource(local_addr_);
   return 0;
 }
 
 Error MountNodeTCP::Recv(void* buf, size_t len, int flags, int* out_len) {
   AUTO_LOCK(node_lock_);
   if (0 == socket_resource_)
     return EBADF;
 
   int capped_len = static_cast<int32_t>(std::min(len, MAX_SOCK_TRANSFER));
-  int err = TCPSocket()->Read(socket_resource_,
-                              static_cast<char*>(buf),
-                              capped_len,
-                              PP_BlockUntilComplete());
+  int err = TCPInterface()->Read(socket_resource_,
+                                 static_cast<char*>(buf),
+                                 capped_len,
+                                 PP_BlockUntilComplete());
   if (err < 0)
     return PPErrorToErrno(err);
 
   *out_len = err;
   return 0;
 }
 
 Error MountNodeTCP::RecvFrom(void* buf,
                              size_t len,
                              int flags,
@@ skipping 10 matching lines @@
 Error MountNodeTCP::Send(const void* buf, size_t len, int flags, int* out_len) {
   AUTO_LOCK(node_lock_);
 
   if (0 == socket_resource_)
     return EBADF;
 
   if (0 == remote_addr_)
     return ENOTCONN;
 
   int capped_len = static_cast<int32_t>(std::min(len, MAX_SOCK_TRANSFER));
-  int err = TCPSocket()->Write(socket_resource_,
-                               static_cast<const char*>(buf),
-                               capped_len,
-                               PP_BlockUntilComplete());
+  int err = TCPInterface()->Write(socket_resource_,
+                                  static_cast<const char*>(buf),
+                                  capped_len,
+                                  PP_BlockUntilComplete());
   if (err < 0)
     return PPErrorToErrno(err);
 
   *out_len = err;
   return 0;
 }
 
 Error MountNodeTCP::SendTo(const void* buf,
                            size_t len,
                            int flags,
                            const struct sockaddr* dest_addr,
                            socklen_t addrlen,
                            int* out_len) {
   return Send(buf, len, flags, out_len);
 }
 
 }  // namespace nacl_io
 
 #endif  // PROVIDES_SOCKET_API