mvm-openpipe.diff

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

rakudo-openpipe.diff

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;