Skip to content
This repository has been archived by the owner. It is now read-only.
Permalink
Browse files

Add argument to uv_pipe_init for IPC

  • Loading branch information...
ry committed Sep 29, 2011
1 parent 72d8d76 commit 32151d54ef1aeee65c8689a4d44537d86d7267fe
@@ -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.
@@ -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;
}

@@ -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);
@@ -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++;

@@ -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);
@@ -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);
@@ -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);
@@ -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)));
@@ -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);
@@ -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);
@@ -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);

@@ -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);
@@ -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;

0 comments on commit 32151d5

Please sign in to comment.
You can’t perform that action at this time.