Add argument to uv_pipe_init for IPC, unix impl

include/uv.h

@@ -648,9 +648,14 @@ struct uv_pipe_s {
   UV_HANDLE_FIELDS
   UV_STREAM_FIELDS
   UV_PIPE_PRIVATE_FIELDS
+  int ipc; /* non-zero if this pipe is used for passing handles */
 };
 
-int uv_pipe_init(uv_loop_t*, uv_pipe_t* handle);
+/*
+ * Initialize a pipe. The last argument is a boolean to indicate if
+ * this pipe will be used for handle passing between processes.
+ */
+int uv_pipe_init(uv_loop_t*, uv_pipe_t* handle, int ipc);
 
 /*
  * Opens an existing file descriptor or HANDLE as a pipe.

src/unix/pipe.c

@@ -29,10 +29,12 @@
 #include <unistd.h>
 #include <stdlib.h>
 
-int uv_pipe_init(uv_loop_t* loop, uv_pipe_t* handle) {
+
+int uv_pipe_init(uv_loop_t* loop, uv_pipe_t* handle, int ipc) {
   uv__stream_init(loop, (uv_stream_t*)handle, UV_NAMED_PIPE);
   loop->counters.pipe_init++;
   handle->pipe_fname = NULL;
+  handle->ipc = ipc;
   return 0;
 }
 

src/unix/process.c

@@ -1,4 +1,3 @@
-
 /* Copyright Joyent, Inc. and other Node contributors. All rights reserved.
  *
  * Permission is hereby granted, free of charge, to any person obtaining a copy
@@ -63,6 +62,34 @@ static void uv__chld(EV_P_ ev_child* watcher, int revents) {
   }
 }
 
+
+/*
+ * Used for initializing stdio streams like options.stdin_stream. Returns
+ * zero on success.
+ */
+static int uv__process_init_pipe(uv_pipe_t* handle, int fds[2]) {
+  if (handle->type != UV_NAMED_PIPE) {
+    errno = EINVAL;
+    return -1;
+  }
+
+  if (handle->ipc) {
+    if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) < 0) {
+      return -1;
+    }
+  } else {
+    if (pipe(fds) < 0) {
+      return -1;
+    }
+  }
+
+  uv__cloexec(fds[0], 1);
+  uv__cloexec(fds[1], 1);
+
+  return 0;
+}
+
+
 #ifndef SPAWN_WAIT_EXEC
 # define SPAWN_WAIT_EXEC 1
 #endif
@@ -89,43 +116,19 @@ int uv_spawn(uv_loop_t* loop, uv_process_t* process,
 
   process->exit_cb = options.exit_cb;
 
-  if (options.stdin_stream) {
-    if (options.stdin_stream->type != UV_NAMED_PIPE) {
-      errno = EINVAL;
-      goto error;
-    }
-
-    if (pipe(stdin_pipe) < 0) {
-      goto error;
-    }
-    uv__cloexec(stdin_pipe[0], 1);
-    uv__cloexec(stdin_pipe[1], 1);
+  if (options.stdin_stream &&
+      uv__process_init_pipe(options.stdin_stream, stdin_pipe)) {
+    goto error;
   }
 
-  if (options.stdout_stream) {
-    if (options.stdout_stream->type != UV_NAMED_PIPE) {
-      errno = EINVAL;
-      goto error;
-    }
-
-    if (pipe(stdout_pipe) < 0) {
-      goto error;
-    }
-    uv__cloexec(stdout_pipe[0], 1);
-    uv__cloexec(stdout_pipe[1], 1);
+  if (options.stdout_stream &&
+      uv__process_init_pipe(options.stdout_stream, stdout_pipe)) {
+    goto error;
   }
 
-  if (options.stderr_stream) {
-    if (options.stderr_stream->type != UV_NAMED_PIPE) {
-      errno = EINVAL;
-      goto error;
-    }
-
-    if (pipe(stderr_pipe) < 0) {
-      goto error;
-    }
-    uv__cloexec(stderr_pipe[0], 1);
-    uv__cloexec(stderr_pipe[1], 1);
+  if (options.stderr_stream &&
+      uv__process_init_pipe(options.stderr_stream, stderr_pipe)) {
+    goto error;
   }
 
   /* This pipe is used by the parent to wait until
@@ -154,7 +157,7 @@ int uv_spawn(uv_loop_t* loop, uv_process_t* process,
     goto error;
   }
 # else
-  if (pipe(signal_pipe) < 0) {
+  if (socketpair(AF_UNIX, SOCK_STREAM, 0, signal_pipe) < 0) {
     goto error;
   }
   uv__cloexec(signal_pipe[0], 1);

src/win/pipe.c

@@ -51,13 +51,14 @@ static void uv_unique_pipe_name(char* ptr, char* name, size_t size) {
 }
 
 
-int uv_pipe_init(uv_loop_t* loop, uv_pipe_t* handle) {
+int uv_pipe_init(uv_loop_t* loop, uv_pipe_t* handle, int ipc) {
   uv_stream_init(loop, (uv_stream_t*)handle);
 
   handle->type = UV_NAMED_PIPE;
   handle->reqs_pending = 0;
   handle->handle = INVALID_HANDLE_VALUE;
   handle->name = NULL;
+  handle->ipc = ipc;
 
   loop->counters.pipe_init++;
 

test/benchmark-pound.c

@@ -222,7 +222,7 @@ static void tcp_make_connect(conn_rec* p) {
 static void pipe_make_connect(conn_rec* p) {
   int r;
 
-  r = uv_pipe_init(loop, (uv_pipe_t*)&p->stream);
+  r = uv_pipe_init(loop, (uv_pipe_t*)&p->stream, 0);
   ASSERT(r == 0);
 
   r = uv_pipe_connect(&((pipe_conn_rec*)p)->conn_req, (uv_pipe_t*)&p->stream, TEST_PIPENAME, connect_cb);

test/benchmark-pump.c

@@ -253,7 +253,7 @@ static void maybe_connect_some() {
     } else {
       pipe = &pipe_write_handles[max_connect_socket++];
 
-      r = uv_pipe_init(loop, pipe);
+      r = uv_pipe_init(loop, pipe, 0);
       ASSERT(r == 0);
 
       req = (uv_connect_t*) req_alloc();
@@ -277,7 +277,7 @@ static void connection_cb(uv_stream_t* s, int status) {
     ASSERT(r == 0);
   } else {
     stream = (uv_stream_t*)malloc(sizeof(uv_pipe_t));
-    r = uv_pipe_init(loop, (uv_pipe_t*)stream);
+    r = uv_pipe_init(loop, (uv_pipe_t*)stream, 0);
     ASSERT(r == 0);
   }
 
@@ -396,7 +396,7 @@ HELPER_IMPL(pipe_pump_server) {
 
   /* Server */
   server = (uv_stream_t*)&pipeServer;
-  r = uv_pipe_init(loop, &pipeServer);
+  r = uv_pipe_init(loop, &pipeServer, 0);
   ASSERT(r == 0);
   r = uv_pipe_bind(&pipeServer, TEST_PIPENAME);
   ASSERT(r == 0);

test/benchmark-spawn.c

@@ -113,7 +113,7 @@ static void spawn() {
   options.args = args;
   options.exit_cb = exit_cb;
 
-  uv_pipe_init(loop, &out);
+  uv_pipe_init(loop, &out, 0);
   options.stdout_stream = &out;
 
   r = uv_spawn(loop, &process, options);

test/echo-server.c

@@ -151,7 +151,7 @@ static void on_connection(uv_stream_t* server, int status) {
   case PIPE:
     stream = malloc(sizeof(uv_pipe_t));
     ASSERT(stream != NULL);
-    r = uv_pipe_init(loop, (uv_pipe_t*)stream);
+    r = uv_pipe_init(loop, (uv_pipe_t*)stream, 0);
     ASSERT(r == 0);
     break;
 
@@ -248,7 +248,7 @@ static int pipe_echo_start(char* pipeName) {
   server = (uv_handle_t*)&pipeServer;
   serverType = PIPE;
 
-  r = uv_pipe_init(loop, &pipeServer);
+  r = uv_pipe_init(loop, &pipeServer, 0);
   if (r) {
     fprintf(stderr, "uv_pipe_init: %s\n",
         uv_strerror(uv_last_error(loop)));

test/run-tests.c

@@ -24,6 +24,7 @@
 
 #include "runner.h"
 #include "task.h"
+#include "uv.h"
 
 /* Actual tests and helpers are defined in test-list.h */
 #include "test-list.h"
@@ -48,12 +49,44 @@ int main(int argc, char **argv) {
 }
 
 
+static int ipc_helper() {
+  /*
+   * This is launched from test-ipc.c. stdin is a duplex channel that we
+   * over which a handle will be transmitted. In this initial version only
+   * data is transfered over the channel. XXX edit this comment after handle
+   * transfer is added.
+   */
+  uv_pipe_t channel;
+  uv_write_t write_req;
+  int r;
+  uv_buf_t buf;
+
+  r = uv_pipe_init(uv_default_loop(), &channel, 1);
+  ASSERT(r == 0);
+
+  uv_pipe_open(&channel, 0);
+
+  buf = uv_buf_init("hello\n", 6);
+  r = uv_write(&write_req, (uv_stream_t*)&channel, &buf, 1, NULL);
+  ASSERT(r == 0);
+
+  r = uv_run(uv_default_loop());
+  ASSERT(r == 0);
+
+  return 0;
+}
+
+
 static int maybe_run_test(int argc, char **argv) {
   if (strcmp(argv[1], "--list") == 0) {
     print_tests(stdout);
     return 0;
   }
 
+  if (strcmp(argv[1], "ipc_helper") == 0) {
+    return ipc_helper();
+  }
+
   if (strcmp(argv[1], "spawn_helper1") == 0) {
     return 1;
   }

test/test-ipc.c

@@ -0,0 +1,102 @@
+/* Copyright Joyent, Inc. and other Node contributors. All rights reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to
+ * deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+ * sell copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+ * IN THE SOFTWARE.
+ */
+
+#include "uv.h"
+#include "task.h"
+
+#include <stdio.h>
+#include <string.h>
+
+static char exepath[1024];
+static size_t exepath_size = 1024;
+static char* args[3];
+static uv_pipe_t channel;
+
+static int exit_cb_called;
+
+static uv_write_t write_req;
+
+
+static void exit_cb(uv_process_t* process, int exit_status, int term_signal) {
+  printf("exit_cb\n");
+  exit_cb_called++;
+  ASSERT(exit_status == 1);
+  ASSERT(term_signal == 0);
+  uv_close((uv_handle_t*)process, NULL);
+}
+
+
+static uv_buf_t on_alloc(uv_handle_t* handle, size_t suggested_size) {
+  return uv_buf_init(malloc(suggested_size), suggested_size);
+}
+
+
+static void on_read(uv_stream_t* pipe, ssize_t nread, uv_buf_t buf) {
+  int r;
+  uv_buf_t outbuf;
+  /* listen on the handle provided.... */
+
+  if (nread) {
+    outbuf = uv_buf_init("world\n", 6);
+    r = uv_write(&write_req, pipe, &outbuf, 1, NULL);
+    ASSERT(r == 0);
+
+    fprintf(stderr, "got %d bytes\n", (int)nread);
+  }
+
+  if (buf.base) {
+    free(buf.base);
+  }
+}
+
+
+TEST_IMPL(ipc) {
+  int r;
+  uv_process_options_t options;
+  uv_process_t process;
+
+  r = uv_pipe_init(uv_default_loop(), &channel, 1);
+  ASSERT(r == 0);
+
+  memset(&options, 0, sizeof(uv_process_options_t));
+
+  r = uv_exepath(exepath, &exepath_size);
+  ASSERT(r == 0);
+  exepath[exepath_size] = '\0';
+  args[0] = exepath;
+  args[1] = "ipc_helper";
+  args[2] = NULL;
+  options.file = exepath;
+  options.args = args;
+  options.exit_cb = exit_cb;
+  options.stdin_stream = &channel;
+
+  r = uv_spawn(uv_default_loop(), &process, options);
+  ASSERT(r == 0);
+
+  uv_read_start((uv_stream_t*)&channel, on_alloc, on_read);
+
+  r = uv_run(uv_default_loop());
+  ASSERT(r == 0);
+
+  ASSERT(exit_cb_called == 1);
+  return 0;
+}

test/test-list.h

@@ -20,6 +20,7 @@
  */
 
 TEST_DECLARE   (tty)
+TEST_DECLARE   (ipc)
 TEST_DECLARE   (tcp_ping_pong)
 TEST_DECLARE   (tcp_ping_pong_v6)
 TEST_DECLARE   (tcp_ref)
@@ -110,6 +111,7 @@ HELPER_DECLARE (pipe_echo_server)
 
 TASK_LIST_START
   TEST_ENTRY  (tty)
+  TEST_ENTRY  (ipc)
 
 
   TEST_ENTRY  (tcp_ref)

test/test-ping-pong.c

@@ -204,7 +204,7 @@ static void pipe_pinger_new() {
   pinger->pongs = 0;
 
   /* Try to connec to the server and do NUM_PINGS ping-pongs. */
-  r = uv_pipe_init(uv_default_loop(), &pinger->stream.pipe);
+  r = uv_pipe_init(uv_default_loop(), &pinger->stream.pipe, 0);
   pinger->stream.pipe.data = pinger;
   ASSERT(!r);
 

test/test-pipe-bind-error.c

@@ -45,12 +45,12 @@ TEST_IMPL(pipe_bind_error_addrinuse) {
   uv_pipe_t server1, server2;
   int r;
 
-  r = uv_pipe_init(uv_default_loop(), &server1);
+  r = uv_pipe_init(uv_default_loop(), &server1, 0);
   ASSERT(r == 0);
   r = uv_pipe_bind(&server1, TEST_PIPENAME);
   ASSERT(r == 0);
 
-  r = uv_pipe_init(uv_default_loop(), &server2);
+  r = uv_pipe_init(uv_default_loop(), &server2, 0);
   ASSERT(r == 0);
   r = uv_pipe_bind(&server2, TEST_PIPENAME);
   ASSERT(r == -1);
@@ -79,7 +79,7 @@ TEST_IMPL(pipe_bind_error_addrnotavail) {
   uv_pipe_t server;
   int r;
 
-  r = uv_pipe_init(uv_default_loop(), &server);
+  r = uv_pipe_init(uv_default_loop(), &server, 0);
   ASSERT(r == 0);
   r = uv_pipe_bind(&server, BAD_PIPENAME);
 
@@ -100,7 +100,7 @@ TEST_IMPL(pipe_bind_error_inval) {
   uv_pipe_t server;
   int r;
 
-  r = uv_pipe_init(uv_default_loop(), &server);
+  r = uv_pipe_init(uv_default_loop(), &server, 0);
   ASSERT(r == 0);
   r = uv_pipe_bind(&server, TEST_PIPENAME);
   ASSERT(r == 0);
@@ -123,7 +123,7 @@ TEST_IMPL(pipe_listen_without_bind) {
   uv_pipe_t server;
   int r;
 
-  r = uv_pipe_init(uv_default_loop(), &server);
+  r = uv_pipe_init(uv_default_loop(), &server, 0);
   ASSERT(r == 0);
   r = uv_listen((uv_stream_t*)&server, SOMAXCONN, NULL);
   ASSERT(r == -1);

test/test-spawn.c

@@ -67,7 +67,7 @@ static void kill_cb(uv_process_t* process, int exit_status, int term_signal) {
 }
 
 
-uv_buf_t on_alloc(uv_handle_t* handle, size_t suggested_size) {
+static uv_buf_t on_alloc(uv_handle_t* handle, size_t suggested_size) {
   uv_buf_t buf;
   buf.base = output + output_used;
   buf.len = OUTPUT_SIZE - output_used;
@@ -138,7 +138,7 @@ TEST_IMPL(spawn_stdout) {
 
   init_process_options("spawn_helper2", exit_cb);
 
-  uv_pipe_init(uv_default_loop(), &out);
+  uv_pipe_init(uv_default_loop(), &out, 0);
   options.stdout_stream = &out;
 
   r = uv_spawn(uv_default_loop(), &process, options);
@@ -169,8 +169,8 @@ int r;
 
   init_process_options("spawn_helper3", exit_cb);
 
-  uv_pipe_init(uv_default_loop(), &out);
-  uv_pipe_init(uv_default_loop(), &in);
+  uv_pipe_init(uv_default_loop(), &out, 0);
+  uv_pipe_init(uv_default_loop(), &in, 0);
   options.stdout_stream = &out;
   options.stdin_stream = ∈