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

native_client_sdk/src/examples/demo/nacl_io/nacl_io_demo.c

Index: native_client_sdk/src/examples/demo/nacl_io/nacl_io_demo.c
diff --git a/native_client_sdk/src/examples/demo/nacl_io/nacl_io_demo.c b/native_client_sdk/src/examples/demo/nacl_io/nacl_io_demo.c
index 6a08cd22d8e7af3fd6c015d2c823e69214b7a335..525521965addcd4e5ade7eca2a43b3029f1a3c0c 100644
--- a/native_client_sdk/src/examples/demo/nacl_io/nacl_io_demo.c
+++ b/native_client_sdk/src/examples/demo/nacl_io/nacl_io_demo.c
@@ -6,11 +6,18 @@
 #include "nacl_io_demo.h"
 
 #include <assert.h>
+#include <errno.h>
+#include <fcntl.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/ioctl.h>
 #include <sys/mount.h>
+#include <sys/param.h>
+#include <sys/select.h>
+#include <sys/stat.h>
 #include <pthread.h>
+#include <unistd.h>
 
 #include "ppapi/c/pp_errors.h"
 #include "ppapi/c/pp_module.h"
@@ -23,6 +30,7 @@
 #include "ppapi/c/ppp.h"
 #include "ppapi/c/ppp_instance.h"
 #include "ppapi/c/ppp_messaging.h"
+#include "nacl_io/ioctl.h"
 #include "nacl_io/nacl_io.h"
 
 #include "handlers.h"
@@ -37,9 +45,9 @@ typedef struct {
   HandleFunc function;
 } FuncNameMapping;
 
-PP_Instance g_instance = 0;
-PPB_GetInterface g_get_browser_interface = NULL;
-PPB_Messaging* g_ppb_messaging = NULL;
+static PP_Instance g_instance = 0;
+static PPB_GetInterface g_get_browser_interface = NULL;
+static PPB_Messaging* g_ppb_messaging = NULL;
 PPB_Var* g_ppb_var = NULL;
 PPB_VarArray* g_ppb_var_array = NULL;
 PPB_VarDictionary* g_ppb_var_dictionary = NULL;
@@ -70,6 +78,7 @@ static FuncNameMapping g_function_map[] = {
 
 /** A handle to the thread the handles messages. */
 static pthread_t g_handle_message_thread;
+static pthread_t g_echo_thread;
 
 /**
  * Create a new PP_Var from a C string.
@@ -139,13 +148,11 @@ struct PP_Var PrintfToVar(const char* format, ...) {
 }
 
 /**
- * Convert a PP_Var to a C string, given a buffer.
+ * Convert a PP_Var to a C string.
  * @param[in] var The PP_Var to convert.
- * @param[out] buffer The buffer to write to.
- * @param[in] length The length of |buffer|.
- * @return The number of characters written.
+ * @return A newly allocated, NULL-terminated string.
  */
-const char* VarToCStr(struct PP_Var var) {
+static const char* VarToCStr(struct PP_Var var) {
   uint32_t length;
   const char* str = g_ppb_var->VarToUtf8(var, &length);
   if (str == NULL) {
@@ -279,6 +286,64 @@ static void HandleMessage(struct PP_Var message) {
   g_ppb_var->Release(result_var);
 }
 
+
+/**
+ * Helper function used by EchoThread which reads from a file descriptor
+ * and writes all the data that it reads back to the same descriptor.
+ */
+static void EchoInput(int fd) {
+  char buffer[512];
+  while (1) {
+    int rtn = read(fd, buffer, 512);
+    if (rtn > 0) {
+      int wrote = write(fd, buffer, rtn);
+      if (wrote < rtn)
+        fprintf(stderr, "only wrote %d/%d bytes\n", wrote, rtn);

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

might be nice to send these errors to JavaScript, rather than the console.

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

Done.

+    } else {
+      if (rtn < 0 && errno != EAGAIN)
+        fprintf(stderr, "read failed: %d (%s)\n", errno, strerror(errno));
+      break;
+    }
+  }
+}
+
+/**
+ * Worker thread that listens for input on JS pipe nodes and echos all input
+ * back to the same pipe.
+ */
+static void* EchoThread(void* user_data) {
+  int fd1 = open("/dev/jspipe1", O_RDWR | O_NONBLOCK);
+  int fd2 = open("/dev/jspipe2", O_RDWR | O_NONBLOCK);
+  int fd3 = open("/dev/jspipe3", O_RDWR | O_NONBLOCK);
+  int nfds = MAX(fd1, fd2);
+  nfds = MAX(nfds, fd3);
+  while (1) {
+    fd_set readfds;
+    FD_ZERO(&readfds);
+    FD_SET(fd1, &readfds);
+    FD_SET(fd2, &readfds);
+    FD_SET(fd3, &readfds);
+    int rtn = select(nfds + 1, &readfds, NULL, NULL, NULL);
+    if (rtn < 0 && errno != EAGAIN) {
+      fprintf(stderr, "select failed: %s\n", strerror(errno));
+      break;
+    }
+    if (rtn > 0) {
+      if (FD_ISSET(fd1, &readfds))
+        EchoInput(fd1);
+      if (FD_ISSET(fd2, &readfds))
+        EchoInput(fd2);
+      if (FD_ISSET(fd3, &readfds))
+        EchoInput(fd3);
+    }
+
+  }
+  close(fd1);
+  close(fd2);
+  close(fd3);
+  return 0;
+}
+
 /**
  * A worker thread that handles messages from JavaScript.
  * @param[in] user_data Unused.
@@ -318,6 +383,7 @@ static PP_Bool Instance_DidCreate(PP_Instance instance,
         "");      /* data */
 
   pthread_create(&g_handle_message_thread, NULL, &HandleMessageThread, NULL);
+  pthread_create(&g_echo_thread, NULL, &EchoThread, NULL);
   InitializeMessageQueue();
 
   return PP_TRUE;
@@ -341,6 +407,34 @@ static PP_Bool Instance_HandleDocumentLoad(PP_Instance instance,
 
 static void Messaging_HandleMessage(PP_Instance instance,
                                     struct PP_Var message) {
+  /* Special case for jspipe input handling */
+  if (message.type == PP_VARTYPE_DICTIONARY) {

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

this won't work anymore because all messages are dictionaries.

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

Done.

+    struct PP_Var keys = g_ppb_var_dictionary->GetKeys(message);
+    struct PP_Var key = g_ppb_var_array->Get(keys, 0);
+    const char* key_string = VarToCStr(key);
+    char pipe_name[64];
+
+    sprintf(pipe_name, "/dev/%s", key_string);

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

snprintf

+    int fd = open(pipe_name, O_RDONLY);
+    if (fd < 0) {
+      struct PP_Var var = PrintfToVar("Warning: opening %s failed.",
+                                      pipe_name);
+      g_ppb_messaging->PostMessage(g_instance, var);

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

Use PostMessageVar to cleanup the newly created string variable.

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

Done.

+      return;

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

release PP_Vars on failure

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

Done.

+    }
+    struct PP_Var payload = g_ppb_var_dictionary->Get(message, key);
+    if (ioctl(fd, NACL_IOC_HANDLEMESSAGE, (char*)&payload) != 0) {
+      struct PP_Var var = PrintfToVar("Error: ioctl on %s failed: %s",
+                                      pipe_name, strerror(errno));
+      g_ppb_messaging->PostMessage(g_instance, var);
+    }
+    g_ppb_var->Release(keys);
+    g_ppb_var->Release(key);
+    g_ppb_var->Release(payload);
+    close(fd);
+    return;
+  }
+
   g_ppb_var->AddRef(message);
   if (!EnqueueMessage(message)) {
     g_ppb_var->Release(message);