[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..8b670d74f36aae6c0e5ad2cb5d89c7432ba5a8c8 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,186 @@
 
 #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)
-Error JSPipeNode::Write(const HandleAttr& attr,
-                        const void* buf,
-                        size_t count,
-                        int* out_bytes) {
+JSPipeNode::JSPipeNode(Filesystem* filesystem)
+  : Node(filesystem),
+    pipe_(new JSPipeEventEmitter(filesystem_->ppapi(), kPostMessageBufferSize)),
+    messaging_iface_(NULL),
+    var_iface_(NULL),
+    array_iface_(NULL),
+    buffer_iface_(NULL),
+    dict_iface_(NULL) {
   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 (ppapi) {
+    messaging_iface_ = ppapi->GetMessagingInterface();
+    var_iface_ = ppapi->GetVarInterface();
+    array_iface_ = ppapi->GetVarArrayInterface();
+    buffer_iface_ = ppapi->GetVarArrayBufferInterface();
+    dict_iface_ = ppapi->GetVarDictionaryInterface();
+  } else {
+    TRACE("missing PPAPI provider");
+  }
+}
 
-  if (name_.empty())
+JSPipeEventEmitter* JSPipeNode::GetEventEmitter() {
+  return pipe_.get();
+}
+
+Error JSPipeNode::HandleJSWrite(struct PP_Var message) {
+  TRACE("HandleJSWrite");
+  if (message.type != PP_VARTYPE_ARRAY_BUFFER) {
+    TRACE("HandleJSWrite expected ArrayBuffer but got %d.", message.type);
+    return EINVAL;
+  }
+  uint32_t length;
+  if (buffer_iface_->ByteLength(message, &length) != PP_TRUE)
+    return EINVAL;
+
+  char* buffer = (char*)buffer_iface_->Map(message);
+
+  // Write data to the input fifo
+  int wrote = GetEventEmitter()->HandleJSWrite(buffer, length);

[binji, 2014/05/01 20:22:31]

returns size_t

+  buffer_iface_->Unmap(message);
+  if (wrote != (int) length) {
+    LOG_ERROR("Only wrote %d of %d bytes to pipe", wrote, length);

[binji, 2014/05/01 20:22:31]

ERROR instead of LOG_ERROR?

[Sam Clegg, 2014/05/01 22:16:55]

File local macro that also print the name of the pipe in the error message.

     return EIO;
+  }
+  TRACE("done HandleWrite: %d", length);
+  return 0;
+}
 
-  // Limit the size of the data we send with PostMessage to MAX_MESSAGE_SIZE
-  if (count > MAX_MESSAGE_SIZE)
-      count = MAX_MESSAGE_SIZE;
+Error JSPipeNode::HandleJSAck(PP_Var message) {
+  if (message.type != PP_VARTYPE_INT32) {
+    TRACE("HandleAck integer object expected but got %d.", message.type);
+    return EINVAL;
+  }
+  GetEventEmitter()->HandleJSAck(message.value.as_int);

[binji, 2014/05/01 20:22:31]

This will never be able to ack above 2**31

[Sam Clegg, 2014/05/01 22:16:55]

True, this is a current know issue that also effect nassh.

We can probably work around it by using two integers on the JS side if it causes
a problem.

+  return 0;
+}
 
-  // Copy data in a new ArrayBuffer
-  PP_Var buffer = buffer_iface->Create(count);
-  memcpy(buffer_iface->Map(buffer), buf, count);
-  buffer_iface->Unmap(buffer);
+Error JSPipeNode::HandleJSMessage(struct PP_Var message) {
+  Error err = 0;
+  if (!messaging_iface_ || !var_iface_ || !array_iface_ || !buffer_iface_) {
+    TRACE("HandleJSMessage: missing PPAPI interfaces");
+    return ENOSYS;
+  }
+
+  // Verify that we have an array with size two.
+  if (message.type != PP_VARTYPE_ARRAY) {
+    TRACE("HandleJSMessage passed non-array var");
+    return EINVAL;
+  }
+
+  if (array_iface_->GetLength(message) != 2) {
+    TRACE("HandleJSMessage passed array of size %d",
+          array_iface_->GetLength(message));
+    return EINVAL;
+  }
 
-  // Construct string var containing the name of the pipe
-  PP_Var string = var_iface->VarFromUtf8(name_.c_str(), name_.size());
+  PP_Var message_type_var = array_iface_->Get(message, 0);
+  PP_Var payload = array_iface_->Get(message, 1);
+  if (message_type_var.type != PP_VARTYPE_STRING) {
+    err = EINVAL;
+  } else {
+    uint32_t length;
+    const char* message_type_string;
+    message_type_string = var_iface_->VarToUtf8(message_type_var, &length);
+    std::string message_type(message_type_string, length);
+
+    TRACE("HandleJSMessage %s", message_type.c_str());
+    if (message_type == "write") {
+      err = HandleJSWrite(payload);
+    } else if (message_type == "ack") {
+      err = HandleJSAck(payload);
+    } else {
+      TRACE("Unknown message type: %s", message_type.c_str());
+      err = EINVAL;
+    }
+  }
 
-  // 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);
+  var_iface_->Release(message_type_var);
+  var_iface_->Release(payload);
+  return err;
+}
 
-  // Release our handles to the items that are now in the array
-  var_iface->Release(string);
-  var_iface->Release(buffer);
+Error JSPipeNode::Read(const HandleAttr& attr,
+                       void* buf,
+                       size_t count,
+                       int* out_bytes) {
+  int ms = attr.IsBlocking() ? -1 : 0;
 
-  // Send array via PostMessage
-  iface->PostMessage(ppapi->GetInstance(), array);
+  EventListenerLock wait(GetEventEmitter());
+  Error err = wait.WaitOnEvent(POLLIN, ms);
+  if (err == ETIMEDOUT)
+    err = EWOULDBLOCK;
+  if (err)
+    return err;
 
-  // Release the array
-  var_iface->Release(array);
+  *out_bytes = GetEventEmitter()->Read_Locked(static_cast<char*>(buf), count);
+  return 0;
+}
 
-  *out_bytes = count;
+Error JSPipeNode::Write(const HandleAttr& attr,
+                        const void* buf,
+                        size_t count,
+                        int* out_bytes) {
+  int ms = attr.IsBlocking() ? -1 : 0;
+  TRACE("write %d", ms);
+
+  EventListenerLock wait(GetEventEmitter());
+  Error err = wait.WaitOnEvent(POLLOUT, ms);
+  if (err == ETIMEDOUT)
+    err = EWOULDBLOCK;
+  if (err)
+    return err;
+
+  *out_bytes = GetEventEmitter()->Write_Locked(static_cast<const char*>(buf),
+                                               count);
   return 0;
 }
 
 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 HandleJSMessage(*message);
+    }
+    default:
+      TRACE("unknown ioctl: %#x", request);
+      break;
   }
 
   return EINVAL;