Add argument to uv_pipe_init for IPC

joyent · Sep 29, 2011 · 32151d5 · 32151d5
1 parent 72d8d76
commit 32151d5
Show file tree

Hide file tree

Showing 13 changed files with 79 additions and 59 deletions.
diff --git a/include/uv.h b/include/uv.h
@@ -642,9 +642,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.

diff --git a/src/unix/pipe.c b/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;
 }
 

diff --git a/src/unix/process.c b/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);

diff --git a/src/win/pipe.c b/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++;
 

diff --git a/test/benchmark-pound.c b/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);

diff --git a/test/benchmark-pump.c b/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);

diff --git a/test/benchmark-spawn.c b/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);

diff --git a/test/echo-server.c b/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)));

diff --git a/test/run-tests.c b/test/run-tests.c
@@ -61,7 +61,7 @@ static int ipc_helper() {
   int r;
   uv_buf_t buf;
 
-  r = uv_pipe_init(uv_default_loop(), &channel);
+  r = uv_pipe_init(uv_default_loop(), &channel, 1);
   ASSERT(r == 0);
 
   uv_pipe_open(&channel, 0);

diff --git a/test/test-ipc.c b/test/test-ipc.c
@@ -22,6 +22,7 @@
 #include "uv.h"
 #include "task.h"
 
+#include <stdio.h>
 #include <string.h>
 
 static char exepath[1024];
@@ -31,6 +32,8 @@ 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");
@@ -47,9 +50,17 @@ static uv_buf_t on_alloc(uv_handle_t* handle, size_t 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.... */
 
-  printf("got %d bytes\n", (int)nread);
+  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);
@@ -62,7 +73,7 @@ TEST_IMPL(ipc) {
   uv_process_options_t options;
   uv_process_t process;
 
-  r = uv_pipe_init(uv_default_loop(), &channel);
+  r = uv_pipe_init(uv_default_loop(), &channel, 1);
   ASSERT(r == 0);
 
   memset(&options, 0, sizeof(uv_process_options_t));
@@ -77,8 +88,6 @@ TEST_IMPL(ipc) {
   options.args = args;
   options.exit_cb = exit_cb;
   options.stdin_stream = &channel;
-  options.stdout_stream = NULL;
-  options.stderr_stream = NULL;
 
   r = uv_spawn(uv_default_loop(), &process, options);
   ASSERT(r == 0);

diff --git a/test/test-ping-pong.c b/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);
 

diff --git a/test/test-pipe-bind-error.c b/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);

diff --git a/test/test-spawn.c b/test/test-spawn.c
@@ -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 = &in;