[NaCl SDK] Support non blocking TCP/UDP

native_client_sdk/src/libraries/nacl_io/kernel_proxy.cc

 // Copyright (c) 2012 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/kernel_proxy.h"
 
 
 #include <assert.h>
 #include <errno.h>
 #include <fcntl.h>
@@ skipping 11 matching lines @@
 #include "nacl_io/dbgprint.h"
 #include "nacl_io/host_resolver.h"
 #include "nacl_io/kernel_handle.h"
 #include "nacl_io/kernel_wrap_real.h"
 #include "nacl_io/mount.h"
 #include "nacl_io/mount_dev.h"
 #include "nacl_io/mount_html5fs.h"
 #include "nacl_io/mount_http.h"
 #include "nacl_io/mount_mem.h"
 #include "nacl_io/mount_node.h"
+#include "nacl_io/mount_node_pipe.h"
 #include "nacl_io/mount_node_tcp.h"
 #include "nacl_io/mount_node_udp.h"
 #include "nacl_io/mount_passthrough.h"
+#include "nacl_io/mount_stream.h"
 #include "nacl_io/osmman.h"
 #include "nacl_io/ossocket.h"
 #include "nacl_io/osstat.h"
 #include "nacl_io/path.h"
 #include "nacl_io/pepper_interface.h"
 #include "nacl_io/typed_mount_factory.h"
 #include "sdk_util/auto_lock.h"
 #include "sdk_util/ref_object.h"
 #include "sdk_util/string_util.h"
 
 #ifndef MAXPATHLEN
 #define MAXPATHLEN 256
 #endif
 
 namespace nacl_io {
 
-class SignalEmitter : public EventEmitter {
- public:
-  // From EventEmitter.  The SignalEmitter exists in order
-  // to inturrupt anything waiting in select()/poll() when kill()
-  // is called.  It is an edge trigger only and therefore has no
-  // persistent readable/wriable/error state.
-  uint32_t GetEventStatus() {
-     return 0;
-  }
-
-  int GetType() {
-    // For lack of a better type, report socket to signify it can be in an
-    // used to signal.
-    return S_IFSOCK;
-  }
-
-  void SignalOccurred() {
-    RaiseEvent(POLLERR);
-  }
-};
 
 KernelProxy::KernelProxy() : dev_(0), ppapi_(NULL),
                              sigwinch_handler_(SIG_IGN),
-                             signal_emitter_(new SignalEmitter) {
+                             signal_emitter_(new EventEmitterSignal) {
 
 }
 
 KernelProxy::~KernelProxy() {
   // Clean up the MountFactories.
   for (MountFactoryMap_t::iterator i = factories_.begin();
        i != factories_.end();
        ++i) {
     delete i->second;
   }
@@ skipping 40 matching lines @@
   fd = open("/dev/stderr", O_WRONLY);
   assert(fd == 2);
   if (fd < 0)
     rtn = errno;
 
 #ifdef PROVIDES_SOCKET_API
   host_resolver_.Init(ppapi_);
 #endif
 
   StringMap_t args;
-  socket_mount_.reset(new MountSocket());
-  result = socket_mount_->Init(0, args, ppapi);
+  stream_mount_.reset(new MountStream());
+  result = stream_mount_->Init(0, args, ppapi);
   if (result != 0) {
     assert(false);
     rtn = result;
   }
 
   return rtn;
 }
 
 int KernelProxy::open_resource(const char* path) {
   ScopedMount mnt;
@@ skipping 39 matching lines @@
   ScopedKernelHandle handle(new KernelHandle(mnt, node));
   error = handle->Init(oflags);
   if (error) {
     errno = error;
     return -1;
   }
 
   return AllocateFD(handle);
 }
 
+int KernelProxy::pipe(int pipefds[2]) {
+  MountNodePipe* pipe = new MountNodePipe(stream_mount_.get());
+  ScopedMountNode node(pipe);
+
+  if (pipe->Init(S_IREAD | S_IWRITE) == 0) {
+    ScopedKernelHandle handle0(new KernelHandle(stream_mount_, node));
+    ScopedKernelHandle handle1(new KernelHandle(stream_mount_, node));
+
+    handle0->Init(S_IREAD);

[binji, 2013/09/12 01:47:57]

check return from Init?

[noelallen1, 2013/09/12 23:19:03]

Done.

+    handle1->Init(S_IWRITE);
+
+    pipefds[0] = AllocateFD(handle0);
+    if (pipefds[0] == 0)

[binji, 2013/09/12 01:47:57]

AllocateFD can never return a zero FD.

[noelallen1, 2013/09/12 23:19:03]

Done.

+      return -1;
+
+    pipefds[1] = AllocateFD(handle1);
+    if (pipefds[1] == 0) {
+      close(pipefds[0]);
+      return -1;
+    }
+
+    return 0;
+  }
+
+  return -1;

[binji, 2013/09/12 01:47:57]

set errno?

[noelallen1, 2013/09/12 23:19:03]

Done.

+}
+
 int KernelProxy::close(int fd) {
   ScopedKernelHandle handle;
   Error error = AcquireHandle(fd, &handle);
   if (error) {
     errno = error;
     return -1;
   }
 
   // Remove the FD from the process open file descriptor map
   FreeFD(fd);
@@ skipping 560 matching lines @@
   // Currently we don't even pretend that other processes exist
   // so we can only send a signal to outselves.  For kill(2)
   // pid 0 means the current process group and -1 means all the
   // processes we have permission to send signals to.
   if (pid != getpid() && pid != -1 && pid != 0) {
     errno = ESRCH;
     return -1;
   }
 
   // Raise an event so that select/poll get interrupted.
-  signal_emitter_->SignalOccurred();
+  signal_emitter_->Raise(POLLERR);
   switch (sig) {
     case SIGWINCH:
       if (sigwinch_handler_ != SIG_IGN)
         sigwinch_handler_(SIGWINCH);
       break;
 
     case SIGUSR1:
     case SIGUSR2:
       break;
 
     default:
       errno = EINVAL;
       return -1;
   }
-
   return 0;
 }
 
 sighandler_t KernelProxy::sigset(int signum, sighandler_t handler) {
   switch (signum) {
     // Handled signals.
     case SIGWINCH: {
       sighandler_t old_value = sigwinch_handler_;
       if (handler == SIG_DFL)
         handler = SIG_IGN;
@@ skipping 22 matching lines @@
   }
 
   errno = EINVAL;
   return SIG_ERR;
 }
 
 #ifdef PROVIDES_SOCKET_API
 
 int KernelProxy::select(int nfds, fd_set* readfds, fd_set* writefds,
                         fd_set* exceptfds, struct timeval* timeout) {
-  ScopedEventListener listener(new EventListener);
+  std::vector<pollfd> pollfds;
 
-  std::vector<struct pollfd> fds;
-
-  fd_set readout, writeout, exceptout;
-
-  FD_ZERO(&readout);
-  FD_ZERO(&writeout);
-  FD_ZERO(&exceptout);
-
-  int fd;
-  size_t event_cnt = 0;
-  int event_track = 0;
-  for (fd = 0; fd < nfds; fd++) {
+  for (int fd = 0; fd < nfds; fd++) {
     int events = 0;
-
     if (readfds != NULL && FD_ISSET(fd, readfds))
       events |= POLLIN;
 
     if (writefds != NULL && FD_ISSET(fd, writefds))
       events |= POLLOUT;
 
     if (exceptfds != NULL && FD_ISSET(fd, exceptfds))
       events |= POLLERR | POLLHUP;
 
-    // If we are not interested in this FD, skip it
-    if (0 == events) continue;
+    if (events) {
+      pollfd info;
+      info.fd = fd;
+      info.events = events;
+      pollfds.push_back(info);
+    }
+  }
 
-    ScopedKernelHandle handle;
-    Error err = AcquireHandle(fd, &handle);
+  // NULL timeout signals wait forever.
+  int ms_timeout = -1;
+  if (timeout != NULL) {
+    int64_t ms = timeout->tv_sec * 1000 + ((timeout->tv_usec + 500) / 1000);
 
-    // Select will return immediately if there are bad FDs.
-    if (err != 0) {
-      errno = EBADF;
+    // If the timeout is invalid or too long (larger than signed 32 bit).
+    if ((timeout->tv_sec < 0) || (timeout->tv_sec >= (INT_MAX / 1000)) ||
+        (timeout->tv_usec < 0) || (timeout->tv_usec >= 1000000) ||
+        (ms < 0) || (ms >= INT_MAX)) {
+      errno = EINVAL;
       return -1;
     }
 
-    int status = handle->node()->GetEventStatus() & events;
-    if (status & POLLIN) {
-      FD_SET(fd, &readout);
+    ms_timeout = static_cast<int>(ms);
+  }
+
+  int result = poll(&pollfds[0], pollfds.size(), ms_timeout);
+  if (result == -1)
+    return -1;
+
+  FD_ZERO(readfds);
+  FD_ZERO(writefds);
+  FD_ZERO(exceptfds);
+
+  int event_cnt = 0;
+  for (size_t index = 0; index < pollfds.size(); index++) {
+    pollfd* info = &pollfds[index];
+    if (info->revents & POLLIN) {
+      FD_SET(info->fd, readfds);
       event_cnt++;
     }
-
-    if (status & POLLOUT) {
-      FD_SET(fd, &writeout);
+    if (info->revents & POLLOUT) {
+      FD_SET(info->fd, writefds);
       event_cnt++;
     }
-
-    if (status & (POLLERR | POLLHUP)) {
-      FD_SET(fd, &exceptout);
+    if (info->revents & POLLHUP) {

[binji, 2013/09/12 01:47:57]

not POLLERR? You request it above.

[noelallen1, 2013/09/12 23:19:03]

Done.

+      FD_SET(info->fd, exceptfds);
       event_cnt++;
     }
-
-    // Otherwise track it.
-    if (0 == status) {
-      err = listener->Track(fd, handle->node(), events, fd);
-      if (err != 0) {
-        errno = EBADF;
-        return -1;
-      }
-      event_track++;
-    }
   }
 
-  // If nothing is signaled, then we must wait.
-  if (event_cnt == 0) {
-    std::vector<EventData> events;
-    int ready_cnt;
-    int ms_timeout;
-
-    // NULL timeout signals wait forever.
-    if (timeout == NULL) {
-      ms_timeout = -1;
-    } else {
-      int64_t ms = timeout->tv_sec * 1000 + ((timeout->tv_usec + 500) / 1000);
-
-      // If the timeout is invalid or too long (larger than signed 32 bit).
-      if ((timeout->tv_sec < 0) || (timeout->tv_sec >= (INT_MAX / 1000)) ||
-          (timeout->tv_usec < 0) || (timeout->tv_usec >= 1000000) ||
-          (ms < 0) || (ms >= INT_MAX)) {
-        errno = EINVAL;
-        return -1;
-      }
-
-      ms_timeout = static_cast<int>(ms);
-    }
-
-    // Add a special node to listen for events
-    // coming from the KernelProxy itself (kill will
-    // generated a SIGERR event).
-    listener->Track(-1, signal_emitter_, POLLERR, -1);
-    event_track += 1;
-
-    events.resize(event_track);
-
-    bool interrupted = false;
-    listener->Wait(events.data(), event_track, ms_timeout, &ready_cnt);
-    for (fd = 0; static_cast<int>(fd) < ready_cnt; fd++) {
-      if (events[fd].user_data == static_cast<uint64_t>(-1)) {
-        if (events[fd].events & POLLERR) {
-          interrupted = true;
-        }
-        continue;
-      }
-
-      if (events[fd].events & POLLIN) {
-        FD_SET(events[fd].user_data, &readout);
-        event_cnt++;
-      }
-
-      if (events[fd].events & POLLOUT) {
-        FD_SET(events[fd].user_data, &writeout);
-        event_cnt++;
-      }
-
-      if (events[fd].events & (POLLERR | POLLHUP)) {
-        FD_SET(events[fd].user_data, &exceptout);
-        event_cnt++;
-      }
-    }
-
-    if (0 == event_cnt && interrupted) {
-      errno = EINTR;
-      return -1;
-    }
-  }
-
-  // Copy out the results
-  if (readfds != NULL)
-    *readfds = readout;
-
-  if (writefds != NULL)
-    *writefds = writeout;
-
-  if (exceptfds != NULL)
-    *exceptfds = exceptout;
-
   return event_cnt;
 }
 
+struct PollInfo {
+  PollInfo() : index(-1) {};
+
+  std::vector<struct pollfd*> fds;
+  int index;
+};
+
+typedef std::map<EventEmitter*, PollInfo> EventPollMap_t;
+
 int KernelProxy::poll(struct pollfd *fds, nfds_t nfds, int timeout) {
-  ScopedEventListener listener(new EventListener);
-  listener->Track(-1, signal_emitter_, POLLERR, 0);
+  EventPollMap_t event_map;
 
-  int index;
+  std::vector<EventRequest> requests;
   size_t event_cnt = 0;
-  size_t event_track = 1;
-  for (index = 0; static_cast<nfds_t>(index) < nfds; index++) {
+
+  for (int index = 0; static_cast<nfds_t>(index) < nfds; index++) {
     ScopedKernelHandle handle;
-    struct pollfd* info = &fds[index];
-    Error err = AcquireHandle(info->fd, &handle);
+    struct pollfd* fd_info = &fds[index];
+    Error err = AcquireHandle(fd_info->fd, &handle);
 
     // If the node isn't open, or somehow invalid, mark it so.
     if (err != 0) {
-      info->revents = POLLNVAL;
+      fd_info->revents = POLLNVAL;
       event_cnt++;
       continue;
     }
 
     // If it's already signaled, then just capture the event
-    if (handle->node()->GetEventStatus() & info->events) {
-      info->revents = info->events & handle->node()->GetEventStatus();
+    ScopedEventEmitter emitter(handle->node()->GetEventEmitter());
+    int events = POLLIN | POLLOUT;
+    if (emitter)
+      events = emitter->GetEventStatus();
+
+    if (events & fd_info->events) {
+      fd_info->revents = fd_info->events & events;

[binji, 2013/09/12 01:47:57]

change to "events & fd_info->events" like above?

[noelallen1, 2013/09/12 23:19:03]

Done.

+      event_cnt++;
+      continue;
+    }
+
+    if (NULL == emitter) {
+      fd_info->revents = POLLNVAL;
       event_cnt++;
       continue;
     }
 
     // Otherwise try to track it.
-    err = listener->Track(info->fd, handle->node(), info->events, index);
-    if (err != 0) {
-      info->revents = POLLNVAL;
-      event_cnt++;
-      continue;
+    PollInfo* info = &event_map[emitter.get()];

[binji, 2013/09/12 01:47:57]

this is a lot of extra work to prevent sending the same emitter to WaitOnAny
more than once. I don't see why that should be a problem, though, and allowing
it will simplify this code quite a bit.

[noelallen1, 2013/09/12 23:19:03]

Having a one-to-one mapping between Listener and Emitter registration (getting
rid of EventInfo) greatly simplifies the emitter/listener code which is hard to
reason about.   Having it in would must migrate the complexity to a place where
we need to think about locking and signalling.

+    if (info->index == -1) {
+      EventRequest request;
+      request.emitter = emitter;
+      request.filter = fd_info->events;
+      request.events = 0;
+
+      info->index = requests.size();
+      requests.push_back(request);
     }
-    event_track++;
+    info->fds.push_back(fd_info);
+    requests[info->index].filter |= fd_info->events;
   }
 
-  // If nothing is signaled, then we must wait.
+  // If nothing is signaled, then we must wait on the event map
   if (0 == event_cnt) {
-    std::vector<EventData> events;
-    int ready_cnt;
+    EventListenerPoll wait;
+    Error err = wait.WaitOnAny(&requests[0], requests.size(), timeout);
+    if (err != 0) {
+      errno = err;
+      return -1;
+    }
 
-    bool interrupted = false;
-    events.resize(event_track);
-    listener->Wait(events.data(), event_track, timeout, &ready_cnt);
-    for (index = 0; index < ready_cnt; index++) {
-      struct pollfd* info = &fds[events[index].user_data];
-      if (!info) {
-        interrupted = true;
-        continue;
+    for (size_t rindex = 0; rindex < requests.size(); rindex++) {
+      EventRequest* request = &requests[rindex];
+      if (request->events) {
+        PollInfo* poll_info = &event_map[request->emitter.get()];
+        for (size_t findex = 0; findex < poll_info->fds.size(); findex++) {
+          struct pollfd* fd_info = poll_info->fds[findex];
+          uint32_t events = fd_info->events & request->events;
+          if (events) {
+            fd_info->revents = events;
+            event_cnt++;
+          }
+        }
       }
-
-      info->revents = events[index].events;
-      event_cnt++;
-    }
-    if (0 == event_cnt && interrupted) {
-      errno = EINTR;
-      return -1;
     }
   }
 
   return event_cnt;
 }
 
 
 
 // Socket Functions
 int KernelProxy::accept(int fd, struct sockaddr* addr, socklen_t* len) {
@@ skipping 289 matching lines @@
 
 int KernelProxy::socket(int domain, int type, int protocol) {
   if (AF_INET != domain && AF_INET6 != domain) {
     errno = EAFNOSUPPORT;
     return -1;
   }
 
   MountNodeSocket* sock = NULL;
   switch (type) {
     case SOCK_DGRAM:
-      sock = new MountNodeUDP(socket_mount_.get());
+      sock = new MountNodeUDP(stream_mount_.get());
       break;
 
     case SOCK_STREAM:
-      sock = new MountNodeTCP(socket_mount_.get());
+      sock = new MountNodeTCP(stream_mount_.get());
       break;
 
     default:
       errno = EPROTONOSUPPORT;
       return -1;
   }
 
   ScopedMountNode node(sock);
   if (sock->Init(S_IREAD | S_IWRITE) == 0) {
-    ScopedKernelHandle handle(new KernelHandle(socket_mount_, node));
+    ScopedKernelHandle handle(new KernelHandle(stream_mount_, node));
     return AllocateFD(handle);
   }
 
   // If we failed to init, assume we don't have access.
   return EACCES;
 }
 
 int KernelProxy::socketpair(int domain, int type, int protocol, int* sv) {
   if (NULL == sv) {
     errno = EFAULT;
@@ skipping 29 matching lines @@
     errno = ENOTSOCK;
     return -1;
   }
 
   return 0;
 }
 
 #endif  // PROVIDES_SOCKET_API
 
 }  // namespace_nacl_io