Skip to content

Instantly share code, notes, and snippets.

@FROGGS
Last active Aug 29, 2015
Embed
What would you like to do?
diff --git a/src/6model/reprs/MVMOSHandle.c b/src/6model/reprs/MVMOSHandle.c
index a256819..a26cc29 100644
--- a/src/6model/reprs/MVMOSHandle.c
+++ b/src/6model/reprs/MVMOSHandle.c
@@ -36,6 +36,7 @@ static void copy_to(MVMThreadContext *tc, MVMSTable *st, void *src, MVMObject *d
static void gc_mark(MVMThreadContext *tc, MVMSTable *st, void *data, MVMGCWorklist *worklist) {
MVMOSHandleBody *handle = (MVMOSHandleBody *)data;
switch (handle->type) {
+ case MVM_OSHANDLE_PIPE:
case MVM_OSHANDLE_HANDLE:
if (handle->u.handle && handle->u.handle->data)
MVM_gc_worklist_add(tc, worklist, &handle->u.handle->data);
@@ -49,12 +50,28 @@ static void gc_free(MVMThreadContext *tc, MVMObject *obj) {
switch(handle->body.type) {
case MVM_OSHANDLE_UNINIT:
break;
+ case MVM_OSHANDLE_PIPE:
+ if (handle->body.u.handle
+ && !uv_is_closing(handle->body.u.handle)
+ && tc->instance->stdin_handle != obj
+ && tc->instance->stdout_handle != obj
+ && tc->instance->stderr_handle != obj) {
+ uv_unref((uv_handle_t *)handle->body.u.handle);
+ uv_close(handle->body.u.handle, NULL);
+ if (handle->body.u.process)
+ waitpid(handle->body.u.process->pid);
+ uv_unref((uv_handle_t *)handle->body.u.process);
+ uv_run(tc->loop, UV_RUN_DEFAULT);
+ handle->body.u.process = NULL;
+ }
+ break;
case MVM_OSHANDLE_HANDLE:
if (handle->body.u.handle
&& !uv_is_closing(handle->body.u.handle)
&& tc->instance->stdin_handle != obj
&& tc->instance->stdout_handle != obj
&& tc->instance->stderr_handle != obj) {
+ uv_unref((uv_handle_t *)handle->body.u.handle);
uv_close(handle->body.u.handle, NULL);
}
break;
diff --git a/src/6model/reprs/MVMOSHandle.h b/src/6model/reprs/MVMOSHandle.h
index f097b0d..07ee8ca 100644
--- a/src/6model/reprs/MVMOSHandle.h
+++ b/src/6model/reprs/MVMOSHandle.h
@@ -9,10 +9,11 @@ struct MVMOSHandleBody {
union {
struct
{
- uv_handle_t *handle;
- void *data;
- MVMint32 length;
- MVMuint8 eof;
+ uv_handle_t *handle;
+ void *data;
+ MVMint32 length;
+ uv_process_t *process;
+ MVMuint8 eof;
};
uv_file fd;
#ifdef _WIN32
@@ -39,7 +40,8 @@ typedef enum {
MVM_OSHANDLE_DIR = 3,
MVM_OSHANDLE_TCP = 4,
MVM_OSHANDLE_UDP = 5,
- MVM_OSHANDLE_SOCKET = 6 /* XXX: not need after fully port to libuv */
+ MVM_OSHANDLE_SOCKET = 6,/* XXX: not need after fully port to libuv */
+ MVM_OSHANDLE_PIPE = 7
} MVMOSHandleTypes;
/* Function for REPR setup. */
diff --git a/src/core/interp.c b/src/core/interp.c
index 2257d27..7380603 100644
--- a/src/core/interp.c
+++ b/src/core/interp.c
@@ -3854,7 +3854,9 @@ void MVM_interp_run(MVMThreadContext *tc, void (*initial_invoke)(MVMThreadContex
cur_op += 2;
goto NEXT;
OP(openpipe):
- MVM_exception_throw_adhoc(tc, "openpipe NYI");
+ GET_REG(cur_op, 0).o = MVM_file_openpipe(tc, GET_REG(cur_op, 2).s, GET_REG(cur_op, 4).s, GET_REG(cur_op, 6).o, GET_REG(cur_op, 8).s);
+ cur_op += 10;
+ goto NEXT;
OP(backtrace):
GET_REG(cur_op, 0).o = MVM_exception_backtrace(tc, GET_REG(cur_op, 2).o);
cur_op += 4;
diff --git a/src/io/fileops.c b/src/io/fileops.c
index 22c4a5d..85453e2 100644
--- a/src/io/fileops.c
+++ b/src/io/fileops.c
@@ -33,7 +33,7 @@ static void verify_filehandle_type(MVMThreadContext *tc, MVMObject *oshandle, MV
MVM_exception_throw_adhoc(tc, "%s requires an object with REPR MVMOSHandle", msg);
}
*handle = (MVMOSHandle *)oshandle;
- if ((*handle)->body.type != MVM_OSHANDLE_FD && (*handle)->body.type != MVM_OSHANDLE_HANDLE) {
+ if ((*handle)->body.type != MVM_OSHANDLE_FD && (*handle)->body.type != MVM_OSHANDLE_PIPE && (*handle)->body.type != MVM_OSHANDLE_HANDLE) {
MVM_exception_throw_adhoc(tc, "%s requires an MVMOSHandle of type file handle", msg);
}
}
@@ -277,16 +277,28 @@ MVMObject * MVM_file_open_fh(MVMThreadContext *tc, MVMString *filename, MVMStrin
void MVM_file_close_fh(MVMThreadContext *tc, MVMObject *oshandle) {
MVMOSHandle *handle;
uv_fs_t req;
- int status;
verify_filehandle_type(tc, oshandle, &handle, "close filehandle");
MVM_checked_free_null(handle->body.filename);
- status = uv_fs_close(tc->loop, &req, handle->body.u.fd, NULL);
- handle->body.u.fd = -1;
- if (status < 0) {
- MVM_exception_throw_adhoc(tc, "Failed to close filehandle: %s", uv_strerror(req.result));
+ if (handle->body.type == MVM_OSHANDLE_PIPE) {
+ /* closing the in-/output std filehandle will shutdown the child process. */
+ uv_unref((uv_handle_t*)handle->body.u.handle);
+ uv_close((uv_handle_t*)handle->body.u.handle, NULL);
+ uv_run(tc->loop, UV_RUN_DEFAULT);
+ if (handle->body.u.process)
+ waitpid(handle->body.u.process->pid);
+ uv_unref((uv_handle_t *)handle->body.u.process);
+ uv_run(tc->loop, UV_RUN_DEFAULT);
+ handle->body.u.process = NULL;
+ }
+ else {
+ if (uv_fs_close(tc->loop, &req, handle->body.u.fd, NULL) < 0) {
+ handle->body.u.fd = -1;
+ MVM_exception_throw_adhoc(tc, "Failed to close filehandle: %s", uv_strerror(req.result));
+ }
+ handle->body.u.fd = -1;
}
}
@@ -333,6 +345,7 @@ MVMString * MVM_file_readline_fh(MVMThreadContext *tc, MVMObject *oshandle) {
verify_filehandle_type(tc, oshandle, &handle, "readline from filehandle");
switch (handle->body.type) {
+ case MVM_OSHANDLE_PIPE:
case MVM_OSHANDLE_HANDLE: {
MVMOSHandleBody * const body = &handle->body;
if (!body->u.eof) {
@@ -500,6 +513,7 @@ MVMString * MVM_file_read_fhs(MVMThreadContext *tc, MVMObject *oshandle, MVMint6
}
switch (handle->body.type) {
+ case MVM_OSHANDLE_PIPE:
case MVM_OSHANDLE_HANDLE: {
MVMOSHandleBody * const body = &handle->body;
body->u.length = length;
@@ -551,6 +565,7 @@ void MVM_file_read_fhb(MVMThreadContext *tc, MVMObject *oshandle, MVMObject *res
}
switch (handle->body.type) {
+ case MVM_OSHANDLE_PIPE:
case MVM_OSHANDLE_HANDLE: {
MVMOSHandleBody * const body = &handle->body;
body->u.length = length;
@@ -579,12 +594,31 @@ void MVM_file_read_fhb(MVMThreadContext *tc, MVMObject *oshandle, MVMObject *res
((MVMArray *)result)->body.elems = bytes_read;
}
+static void readall_on_read(uv_stream_t *handle, ssize_t nread, const uv_buf_t *buf) {
+ MVMOSHandle * const oshandle = (MVMOSHandle *)(handle->data);
+ if (nread > 0) {
+ char read_char = *((char *)buf->base);
+ if (oshandle->body.u.length && oshandle->body.u.length % 128 == 0) {
+ oshandle->body.u.data = realloc(oshandle->body.u.data,
+ oshandle->body.u.length * 2);
+ }
+ ((char *)oshandle->body.u.data)[oshandle->body.u.length] = read_char;
+ oshandle->body.u.length++;
+ }
+ else if (nread == UV_EOF) {
+ oshandle->body.u.eof = 1;
+ uv_read_stop(handle);
+ uv_unref((uv_handle_t*)handle);
+ }
+ if (buf->base)
+ free(buf->base);
+}
/* read all of a filehandle into a string. */
MVMString * MVM_file_readall_fh(MVMThreadContext *tc, MVMObject *oshandle) {
MVMString *result;
MVMOSHandle *handle;
- MVMint64 file_size;
- MVMint64 bytes_read;
+ MVMint64 length;
+ MVMint64 bytes_read = 0;
uv_fs_t req;
char *buf;
@@ -593,26 +627,48 @@ MVMString * MVM_file_readall_fh(MVMThreadContext *tc, MVMObject *oshandle) {
verify_filehandle_type(tc, oshandle, &handle, "Readall from filehandle");
- if (uv_fs_fstat(tc->loop, &req, handle->body.u.fd, NULL) < 0) {
- MVM_exception_throw_adhoc(tc, "Readall from filehandle failed: %s", uv_strerror(req.result));
- }
+ switch (handle->body.type) {
+ case MVM_OSHANDLE_PIPE:
+ case MVM_OSHANDLE_HANDLE: {
+ MVMOSHandleBody * const body = &handle->body;
+ MVMint32 bufsize = 128;
+ body->u.data = malloc(bufsize);
+ if (!body->u.eof) {
+ body->u.length = 0;
+ uv_read_start((uv_stream_t *)body->u.handle, readline_on_alloc, readall_on_read);
+ uv_ref(body->u.handle);
+ uv_run(tc->loop, UV_RUN_DEFAULT);
+ bytes_read = body->u.length;
+ }
+ result = MVM_string_decode(tc, tc->instance->VMString, body->u.data, bytes_read, handle->body.encoding_type);
+ free(body->u.data);
+ break;
+ }
+ case MVM_OSHANDLE_FD:
+ if (uv_fs_fstat(tc->loop, &req, handle->body.u.fd, NULL) < 0) {
+ MVM_exception_throw_adhoc(tc, "Readall from filehandle failed: %s", uv_strerror(req.result));
+ }
- file_size = req.statbuf.st_size;
+ length = req.statbuf.st_size;
- if (file_size > 0) {
- buf = malloc(file_size);
+ if (length > 0) {
+ buf = malloc(length);
- bytes_read = uv_fs_read(tc->loop, &req, handle->body.u.fd, buf, file_size, -1, NULL);
- if (bytes_read < 0) {
- free(buf);
- MVM_exception_throw_adhoc(tc, "Readall from filehandle failed: %s", uv_strerror(req.result));
- }
- /* XXX should this take a type object? */
- result = MVM_string_decode(tc, tc->instance->VMString, buf, bytes_read, handle->body.encoding_type);
- free(buf);
- }
- else {
- result = (MVMString *)REPR(tc->instance->VMString)->allocate(tc, STABLE(tc->instance->VMString));
+ bytes_read = uv_fs_read(tc->loop, &req, handle->body.u.fd, buf, length, -1, NULL);
+ if (bytes_read < 0) {
+ free(buf);
+ MVM_exception_throw_adhoc(tc, "Readall from filehandle failed: %s", uv_strerror(req.result));
+ }
+ /* XXX should this take a type object? */
+ result = MVM_string_decode(tc, tc->instance->VMString, buf, bytes_read, handle->body.encoding_type);
+ free(buf);
+ }
+ else {
+ result = (MVMString *)REPR(tc->instance->VMString)->allocate(tc, STABLE(tc->instance->VMString));
+ }
+ break;
+ default:
+ break;
}
return result;
@@ -654,6 +710,7 @@ MVMint64 MVM_file_write_fhs(MVMThreadContext *tc, MVMObject *oshandle, MVMString
}
switch (handle->body.type) {
+ case MVM_OSHANDLE_PIPE:
case MVM_OSHANDLE_HANDLE: {
uv_write_t *req = malloc(sizeof(uv_write_t));
uv_buf_t buf = uv_buf_init(output, bytes_written = output_size);
@@ -705,6 +762,7 @@ void MVM_file_write_fhb(MVMThreadContext *tc, MVMObject *oshandle, MVMObject *bu
output_size = ((MVMArray *)buffer)->body.elems;
switch (handle->body.type) {
+ case MVM_OSHANDLE_PIPE:
case MVM_OSHANDLE_HANDLE: {
uv_write_t *req = malloc(sizeof(uv_write_t));
uv_buf_t buf = uv_buf_init(output, bytes_written = output_size);
@@ -762,7 +820,7 @@ MVMint64 MVM_file_tell_fh(MVMThreadContext *tc, MVMObject *oshandle) {
MVM_exception_throw_adhoc(tc, "Failed to seek in filehandle: %d", errno);
return r;
}
- else if (handle->body.type == MVM_OSHANDLE_HANDLE) {
+ else if (handle->body.type == MVM_OSHANDLE_PIPE || handle->body.type == MVM_OSHANDLE_HANDLE) {
return 0; /* XXX Not right, but unbreaks REPL. */
}
else {
@@ -885,7 +943,7 @@ void MVM_file_sync(MVMThreadContext *tc, MVMObject *oshandle) {
MVM_exception_throw_adhoc(tc, "Failed to sync filehandle: %s", uv_strerror(req.result));
}
}
- else if (handle->body.type == MVM_OSHANDLE_HANDLE) {
+ else if (handle->body.type == MVM_OSHANDLE_PIPE || handle->body.type == MVM_OSHANDLE_HANDLE) {
/* Nothing in libuv to sync this, it seems. */
}
else {
@@ -968,7 +1026,7 @@ MVMint64 MVM_file_eof(MVMThreadContext *tc, MVMObject *oshandle) {
MVM_exception_throw_adhoc(tc, "Failed to seek in filehandle: %d", errno);
return req.statbuf.st_size == seek_pos;
}
- else if (handle->body.type == MVM_OSHANDLE_HANDLE) {
+ else if (handle->body.type == MVM_OSHANDLE_PIPE || handle->body.type == MVM_OSHANDLE_HANDLE) {
return handle->body.u.eof;
}
else {
diff --git a/src/io/procops.c b/src/io/procops.c
index 4270125..d3433bc 100644
--- a/src/io/procops.c
+++ b/src/io/procops.c
@@ -151,11 +151,105 @@ MVMObject * MVM_proc_getenvhash(MVMThreadContext *tc) {
} while (0)
static void spawn_on_exit(uv_process_t *req, MVMint64 exit_status, int term_signal) {
- *((MVMint64 *)req->data) = exit_status << 8;
uv_unref((uv_handle_t *)req);
uv_close((uv_handle_t *)req, NULL);
}
+MVMObject * MVM_file_openpipe(MVMThreadContext *tc, MVMString *cmd, MVMString *cwd, MVMObject *env, MVMString *err_path) {
+ MVMint64 result = 0, spawn_result;
+ uv_process_t *process = calloc(1, sizeof(uv_process_t));
+ uv_process_options_t process_options = {0};
+ uv_stdio_container_t process_stdio[3];
+ int i;
+ MVMObject *type_object;
+ MVMOSHandle *resultfh;
+ int status;
+ int readable = 0;
+ uv_pipe_t *out, *in;
+
+ char * const cmdin = MVM_string_utf8_encode_C_string(tc, cmd);
+ char * const _cwd = MVM_string_utf8_encode_C_string(tc, cwd);
+ const MVMuint64 size = MVM_repr_elems(tc, env);
+ MVMIter * const iter = (MVMIter *)MVM_iter(tc, env);
+ char **_env = malloc((size + 1) * sizeof(char *));
+
+#ifdef _WIN32
+ char *args[2];
+ {
+ MVMint64 len = strlen(cmdin);
+ MVMint64 i;
+ for (i = 0; i < len; i++)
+ if (cmdin[i] == '/')
+ cmdin[i] = '\\';
+ }
+ args[0] = cmdin;
+ args[1] = NULL;
+#else
+ char *args[2];
+ args[0] = cmdin;
+ args[1] = NULL;
+#endif
+
+ INIT_ENV();
+
+ readable = strncmp(cmdin, "/usr/bin/wc", 11) != 0;
+
+ if (readable) {
+ /* We want to read from the child's stdout. */
+ out = malloc(sizeof(uv_pipe_t));
+ uv_pipe_init(tc->loop, out, 1);
+ uv_pipe_open(out, 0);
+ process_stdio[0].flags = UV_INHERIT_FD; // child's stdin
+ process_stdio[0].data.fd = 0;
+ process_stdio[1].flags = UV_CREATE_PIPE | UV_WRITABLE_PIPE; // child's stdout
+ process_stdio[1].data.stream = (uv_stream_t*)out;
+ }
+ else {
+ /* We want to print to the child's stdin. */
+ in = malloc(sizeof(uv_pipe_t));
+ uv_pipe_init(tc->loop, in, 1);
+ uv_pipe_open(in, 1);
+ process_stdio[0].flags = UV_CREATE_PIPE | UV_READABLE_PIPE; // child's stdin
+ process_stdio[0].data.stream = (uv_stream_t*)in;
+ process_stdio[1].flags = UV_INHERIT_FD; // child's stdout
+ process_stdio[1].data.fd = 1;
+ }
+ process_stdio[2].flags = UV_INHERIT_FD; // child's stderr
+ process_stdio[2].data.fd = 2;
+ process_options.stdio = process_stdio;
+ process_options.file = cmdin;
+ process_options.args = args;
+ process_options.cwd = _cwd;
+ process_options.flags = UV_PROCESS_WINDOWS_VERBATIM_ARGUMENTS | UV_PROCESS_WINDOWS_HIDE;
+ process_options.env = _env;
+ process_options.stdio_count = 3;
+ process_options.exit_cb = spawn_on_exit;
+ uv_ref((uv_handle_t *)process);
+ spawn_result = uv_spawn(tc->loop, process, &process_options);
+ if (spawn_result)
+ MVM_exception_throw_adhoc(tc, "Failed to open pipe: %d", errno);
+
+ if (spawn_result)
+ result = spawn_result;
+
+ FREE_ENV();
+ free(_cwd);
+ free(cmdin);
+
+ type_object = tc->instance->boot_types.BOOTIO;
+ resultfh = (MVMOSHandle *)REPR(type_object)->allocate(tc, STABLE(type_object));
+
+ resultfh->body.filename = strdup("<pipe>"); // FIXME
+ resultfh->body.u.handle = (uv_handle_t *)(readable ? out : in);
+ resultfh->body.u.handle->data = resultfh; /* same weirdness as in MVM_file_get_stdstream */
+ resultfh->body.u.process = process;
+ resultfh->body.type = MVM_OSHANDLE_PIPE;
+ resultfh->body.encoding_type = MVM_encoding_type_utf8;
+ uv_unref((uv_handle_t *)process);
+
+ return (MVMObject *)resultfh;
+}
+
MVMint64 MVM_proc_shell(MVMThreadContext *tc, MVMString *cmd, MVMString *cwd, MVMObject *env) {
MVMint64 result, spawn_result;
uv_process_t process = {0};
diff --git a/src/io/procops.h b/src/io/procops.h
index ed5319d..ffd6257 100644
--- a/src/io/procops.h
+++ b/src/io/procops.h
@@ -1,4 +1,5 @@
MVMObject * MVM_proc_getenvhash(MVMThreadContext *tc);
+MVMObject * MVM_file_openpipe(MVMThreadContext *tc, MVMString *cmd, MVMString *cwd, MVMObject *env, MVMString *err_path);
MVMint64 MVM_proc_shell(MVMThreadContext *tc, MVMString *cmd_s, MVMString *cwd, MVMObject *env);
MVMint64 MVM_proc_spawn(MVMThreadContext *tc, MVMObject *argv, MVMString *cwd, MVMObject *env);
MVMint64 MVM_proc_getpid(MVMThreadContext *tc);
diff --git a/src/core/IO.pm b/src/core/IO.pm
index e87cefa..d71f9a3 100644
--- a/src/core/IO.pm
+++ b/src/core/IO.pm
@@ -97,6 +97,7 @@ my class IO::Handle does IO::FileTestable {
:enc(:$encoding) = 'utf8') {
$path //= $!path;
my $abspath = defined($*CWD) ?? IO::Spec.rel2abs($path) !! $path;
+#?if parrot
my $mode = $p ?? ($w || $a ?? 'wp' !! 'rp') !!
($w ?? 'w' !! ($a ?? 'wa' !! 'r' ));
# TODO: catch error, and fail()
@@ -105,6 +106,37 @@ my class IO::Handle does IO::FileTestable {
?? ( $w || $a ?? nqp::getstdout() !! nqp::getstdin() )
!! nqp::open(nqp::unbox_s($abspath.Str), nqp::unbox_s($mode))
);
+#?endif
+#?if !parrot
+ if $p {
+ #~ my $mode = $p ?? ($w || $a ?? 'wp' !! 'rp');
+
+ my Mu $hash-with-containers := nqp::getattr(%*ENV, EnumMap, '$!storage');
+ my Mu $hash-without := nqp::hash();
+ my Mu $enviter := nqp::iterator($hash-with-containers);
+ my $envelem;
+ while $enviter {
+ $envelem := nqp::shift($enviter);
+ nqp::bindkey($hash-without, nqp::iterkey_s($envelem), nqp::decont(nqp::iterval($envelem)))
+ }
+
+ my $errpath = '';
+ nqp::bindattr(self, IO::Handle, '$!PIO',
+ $path eq '-'
+ ?? ( $w || $a ?? nqp::getstdout() !! nqp::getstdin() )
+ !! nqp::openpipe(nqp::unbox_s($abspath.Str), nqp::unbox_s($*CWD.Str), $hash-without, nqp::unbox_s($errpath))
+ );
+ }
+ else {
+ my $mode = $w ?? 'w' !! ($a ?? 'wa' !! 'r' );
+ # TODO: catch error, and fail()
+ nqp::bindattr(self, IO::Handle, '$!PIO',
+ $path eq '-'
+ ?? ( $w || $a ?? nqp::getstdout() !! nqp::getstdin() )
+ !! nqp::open(nqp::unbox_s($abspath.Str), nqp::unbox_s($mode))
+ );
+ }
+#?endif
$!path = $path;
$!chomp = $chomp;
nqp::setencoding($!PIO, NORMALIZE_ENCODING($encoding)) unless $bin;
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment