[NaCl SDK] Support non blocking TCP/UDP

native_client_sdk/src/libraries/nacl_io/mount_node_tty.cc

Index: native_client_sdk/src/libraries/nacl_io/mount_node_tty.cc
diff --git a/native_client_sdk/src/libraries/nacl_io/mount_node_tty.cc b/native_client_sdk/src/libraries/nacl_io/mount_node_tty.cc
index 534e53a28dac1605d6bea6ee82a1a8ff026d6163..567c1fb92e1dbf11f138563edb40e146dc25c6a5 100644
--- a/native_client_sdk/src/libraries/nacl_io/mount_node_tty.cc
+++ b/native_client_sdk/src/libraries/nacl_io/mount_node_tty.cc
@@ -32,14 +32,17 @@
 
 namespace nacl_io {
 
-MountNodeTty::MountNodeTty(Mount* mount) : MountNodeCharDevice(mount),
-                                           is_readable_(false),
-                                           did_resize_(false),
-                                           rows_(DEFAULT_TTY_ROWS),
-                                           cols_(DEFAULT_TTY_COLS) {
+
+MountNodeTty::MountNodeTty(Mount* mount)
+    : MountNodeCharDevice(mount),
+      emitter_(new EventEmitter),
+      rows_(DEFAULT_TTY_ROWS),
+      cols_(DEFAULT_TTY_COLS) {
   output_handler_.handler = NULL;
-  pthread_cond_init(&is_readable_cond_, NULL);
   InitTermios();
+
+  // Output will never block
+  emitter_->RaiseEvents_Locked(POLLOUT);
 }
 
 void MountNodeTty::InitTermios() {
@@ -70,14 +73,15 @@ void MountNodeTty::InitTermios() {
   termios_.c_cc[VEOL2] = 0;
 }
 
-MountNodeTty::~MountNodeTty() {
-  pthread_cond_destroy(&is_readable_cond_);
+EventEmitter* MountNodeTty::GetEventEmitter() {
+  return emitter_.get();
 }
 
 Error MountNodeTty::Write(size_t offs,
                      const void* buf,
                      size_t count,
                      int* out_bytes) {
+
   AUTO_LOCK(output_lock_);
   *out_bytes = 0;
 
@@ -98,21 +102,17 @@ Error MountNodeTty::Write(size_t offs,
   return 0;
 }
 
+
 Error MountNodeTty::Read(size_t offs, void* buf, size_t count, int* out_bytes) {
-  AUTO_LOCK(node_lock_);
-  did_resize_ = false;
-  while (!is_readable_) {
-    pthread_cond_wait(&is_readable_cond_, node_lock_.mutex());
-    if (!is_readable_ && did_resize_) {
-      // If an async resize event occured then return the failure and
-      // set EINTR.
-      *out_bytes = 0;
-      return EINTR;
-    }
-  }
+  EventListenerLock wait(GetEventEmitter());
+  *out_bytes = 0;
 
-  size_t bytes_to_copy = std::min(count, input_buffer_.size());
+  // If interrupted, return
+  Error err = wait.WaitOnEvent(POLLIN, -1);
+  if (err != 0)
+    return err;
 
+  size_t bytes_to_copy = std::min(count, input_buffer_.size());
   if (IS_ICANON) {
     // Only read up to (and including) the first newline
     std::deque<char>::iterator nl = std::find(input_buffer_.begin(),
@@ -137,12 +137,15 @@ Error MountNodeTty::Read(size_t offs, void* buf, size_t count, int* out_bytes) {
   // mark input as no longer readable if we consumed
   // the entire buffer or, in the case of buffered input,
   // we consumed the final \n char.
+  bool avail;
   if (IS_ICANON)
-    is_readable_ =
-        std::find(input_buffer_.begin(),
-                  input_buffer_.end(), '\n') != input_buffer_.end();
+    avail = std::find(input_buffer_.begin(),
+                      input_buffer_.end(), '\n') != input_buffer_.end();
   else
-    is_readable_ = input_buffer_.size() > 0;
+    avail = input_buffer_.size() > 0;
+
+  if (!avail)
+    emitter_->ClearEvents_Locked(POLLIN);
 
   return 0;
 }
@@ -160,7 +163,7 @@ Error MountNodeTty::Echo(const char* string, int count) {
 }
 
 Error MountNodeTty::ProcessInput(struct tioc_nacl_input_string* message) {
-  AUTO_LOCK(node_lock_);
+  AUTO_LOCK(emitter_->GetLock())
 
   const char* buffer = message->buffer;
   size_t num_bytes = message->length;
@@ -220,12 +223,7 @@ Error MountNodeTty::ProcessInput(struct tioc_nacl_input_string* message) {
       input_buffer_.push_back(c);
 
     if (c == '\n' || c == termios_.c_cc[VEOF] || !IS_ICANON)
-      is_readable_ = true;
-  }
-
-  if (is_readable_) {
-    RaiseEvent(POLLIN);
-    pthread_cond_broadcast(&is_readable_cond_);
+      emitter_->RaiseEvents_Locked(POLLIN);
   }
 
   return 0;
@@ -260,11 +258,10 @@ Error MountNodeTty::Ioctl(int request, char* arg) {
       }
       kill(getpid(), SIGWINCH);
 
-      // Wake up any thread waiting on Read
       {
-        AUTO_LOCK(node_lock_);
-        did_resize_ = true;
-        pthread_cond_broadcast(&is_readable_cond_);
+        // Wake up any thread waiting on Read with EINTR
+        AUTO_LOCK(emitter_->GetLock())
+        emitter_->RaiseEvents_Locked(POLLERR);
       }
       return 0;
     }