[NaCl SDK] nacl_io: Add flow control the JavaScript pipes.

native_client_sdk/src/libraries/nacl_io/devfs/jspipe_node.cc

Index: native_client_sdk/src/libraries/nacl_io/devfs/jspipe_node.cc
diff --git a/native_client_sdk/src/libraries/nacl_io/devfs/jspipe_node.cc b/native_client_sdk/src/libraries/nacl_io/devfs/jspipe_node.cc
index 7979e33a29d8cc09a19b3e459e3664ba84c35386..5e457f3343ad4f704f4c09435fbe9ec02824f29b 100644
--- a/native_client_sdk/src/libraries/nacl_io/devfs/jspipe_node.cc
+++ b/native_client_sdk/src/libraries/nacl_io/devfs/jspipe_node.cc
@@ -4,94 +4,94 @@
 
 #include "nacl_io/devfs/jspipe_node.h"
 
-//#include <cstdio>
 #include <cstring>
 
 #include "nacl_io/devfs/dev_fs.h"
 #include "nacl_io/error.h"
 #include "nacl_io/ioctl.h"
 #include "nacl_io/kernel_handle.h"
+#include "nacl_io/log.h"
 #include "nacl_io/pepper_interface.h"
 
+#define TRACE(format, ...) LOG_TRACE("jspipe: " format, ##__VA_ARGS__)
+#define ERROR(format, ...) LOG_TRACE("jspipe: " format, ##__VA_ARGS__)
+
+namespace {
+const size_t kPostMessageBufferSize = 512*1024;
+}
+
 namespace nacl_io {
 
-#define MAX_MESSAGE_SIZE (64*1024)
+JSPipeNode::JSPipeNode(Filesystem* filesystem)
+  : Node(filesystem),
+    pipe_(new JSPipeEventEmitter(filesystem_->ppapi(), kPostMessageBufferSize))
+{
+}
+
+JSPipeEventEmitter* JSPipeNode::GetEventEmitter() {
+  return pipe_.get();
+}
+
+Error JSPipeNode::Read(const HandleAttr& attr,
+                       void* buf,
+                       size_t count,
+                       int* out_bytes) {
+  int ms = attr.IsBlocking() ? -1 : 0;
+
+  EventListenerLock wait(GetEventEmitter());
+  Error err = wait.WaitOnEvent(POLLIN, ms);
+  if (err == ETIMEDOUT)
+    err = EWOULDBLOCK;
+  if (err)
+    return err;
+
+  return GetEventEmitter()->Read_Locked(static_cast<char*>(buf), count,
+                                        out_bytes);
+}
+
 Error JSPipeNode::Write(const HandleAttr& attr,
                         const void* buf,
                         size_t count,
                         int* out_bytes) {
-  PepperInterface* ppapi = filesystem_->ppapi();
-  MessagingInterface* iface = ppapi->GetMessagingInterface();
-  VarInterface* var_iface = ppapi->GetVarInterface();
-  VarArrayInterface* array_iface = ppapi->GetVarArrayInterface();
-  VarArrayBufferInterface* buffer_iface = ppapi->GetVarArrayBufferInterface();
-  if (!iface || !var_iface || !array_iface || !buffer_iface)
-    return ENOSYS;
-
-  if (name_.empty())
-    return EIO;
-
-  // Limit the size of the data we send with PostMessage to MAX_MESSAGE_SIZE
-  if (count > MAX_MESSAGE_SIZE)
-      count = MAX_MESSAGE_SIZE;
-
-  // Copy data in a new ArrayBuffer
-  PP_Var buffer = buffer_iface->Create(count);
-  memcpy(buffer_iface->Map(buffer), buf, count);
-  buffer_iface->Unmap(buffer);
-
-  // Construct string var containing the name of the pipe
-  PP_Var string = var_iface->VarFromUtf8(name_.c_str(), name_.size());
-
-  // Construct two element array containing string and array buffer
-  PP_Var array = array_iface->Create();
-  array_iface->Set(array, 0, string);
-  array_iface->Set(array, 1, buffer);
-
-  // Release our handles to the items that are now in the array
-  var_iface->Release(string);
-  var_iface->Release(buffer);
-
-  // Send array via PostMessage
-  iface->PostMessage(ppapi->GetInstance(), array);
-
-  // Release the array
-  var_iface->Release(array);
-
-  *out_bytes = count;
-  return 0;
+  int ms = attr.IsBlocking() ? -1 : 0;
+  TRACE("write timeout=%d", ms);
+
+  EventListenerLock wait(GetEventEmitter());
+  Error err = wait.WaitOnEvent(POLLOUT, ms);
+  if (err == ETIMEDOUT)
+    err = EWOULDBLOCK;
+  if (err)
+    return err;
+
+  return GetEventEmitter()->Write_Locked(static_cast<const char*>(buf),
+                                         count, out_bytes);
 }
 
 Error JSPipeNode::VIoctl(int request, va_list args) {
-  switch (request) {
-    case TIOCNACLPIPENAME: {
-      // name can only be set once
-      if (!name_.empty())
-        return EIO;
+  AUTO_LOCK(node_lock_);
 
+  switch (request) {
+    case NACL_IOC_PIPE_SETNAME: {
       const char* new_name = va_arg(args, char*);
-      // new name must not be empty
-      if (!new_name || strlen(new_name) == 0)
-        return EIO;
-
-      name_ = new_name;
+      return GetEventEmitter()->SetName(new_name);
+    }
+    case NACL_IOC_PIPE_GETISPACE: {
+      int* space = va_arg(args, int*);
+      *space = GetEventEmitter()->GetISpace();
       return 0;
     }
-    case TIOCNACLINPUT: {
-      // This ioctl is used to deliver data from javascipt into the pipe.
-      struct tioc_nacl_input_string* message =
-          va_arg(args, struct tioc_nacl_input_string*);
-      const char* buffer = message->buffer;
-      int num_bytes = message->length;
-      int wrote = 0;
-      HandleAttr data;
-      // Write to the underlying pipe.  Calling JSPipeNode::Write
-      // would write using PostMessage.
-      int error = PipeNode::Write(data, buffer, num_bytes, &wrote);
-      if (error || wrote != num_bytes)
-        return EIO;
+    case NACL_IOC_PIPE_GETOSPACE: {
+      int* space = va_arg(args, int*);
+      *space = GetEventEmitter()->GetOSpace();
       return 0;
     }
+    case NACL_IOC_HANDLEMESSAGE: {
+      struct PP_Var* message = va_arg(args, struct PP_Var*);
+      return GetEventEmitter()->HandleJSMessage(*message);
+    }
+    default:
+      TRACE("unknown ioctl: %#x", request);
+      break;
   }
 
   return EINVAL;