Fix subprocess spawning on windows.

Also fix (:read stream :all)

src/core/ev.c

@@ -955,7 +955,7 @@ JanetListenerState *janet_listen(JanetStream *stream, JanetListener behavior, in
     JanetListenerState *state = janet_listen_impl(stream, behavior, mask, size, user);
     if (!(stream->flags & JANET_STREAM_IOCP)) {
         if (NULL == CreateIoCompletionPort(stream->handle, janet_vm_iocp, (ULONG_PTR) stream, 0)) {
-            janet_panic("failed to listen for events");
+            janet_panicf("failed to listen for events: %V", janet_ev_lasterr());
         }
         stream->flags |= JANET_STREAM_IOCP;
     }
@@ -1587,7 +1587,7 @@ JanetAsyncStatus ev_machine_read(JanetListenerState *s, JanetAsyncEvent event) {
         case JANET_ASYNC_EVENT_READ: {
             JanetBuffer *buffer = state->buf;
             int32_t bytes_left = state->bytes_left;
-            int32_t read_limit = bytes_left < 0 ? 4096 : bytes_left;
+            int32_t read_limit = bytes_left > 4096 ? 4096 : bytes_left;
             janet_buffer_extra(buffer, read_limit);
             ssize_t nread;
 #ifdef JANET_NET
@@ -1923,6 +1923,10 @@ int janet_make_pipe(JanetHandle handles[2]) {
      */
     JanetHandle rhandle, whandle;
     UCHAR PipeNameBuffer[MAX_PATH];
+    SECURITY_ATTRIBUTES saAttr;
+    memset(&saAttr, 0, sizeof(saAttr));
+    saAttr.nLength = sizeof(saAttr);
+    saAttr.bInheritHandle = TRUE;
     sprintf(PipeNameBuffer,
             "\\\\.\\Pipe\\JanetPipeFile.%08x.%08x",
             GetCurrentProcessId(),
@@ -1931,17 +1935,17 @@ int janet_make_pipe(JanetHandle handles[2]) {
                   PipeNameBuffer,
                   PIPE_ACCESS_INBOUND | FILE_FLAG_OVERLAPPED,
                   PIPE_TYPE_BYTE | PIPE_NOWAIT,
-                  1,             /* Number of pipes */
+                  255,           /* Max number of pipes for duplication. */
                   4096,          /* Out buffer size */
                   4096,          /* In buffer size */
                   120 * 1000,    /* Timeout in ms */
-                  NULL);
+                  &saAttr);
     if (!rhandle) return -1;
     whandle = CreateFileA(
                   PipeNameBuffer,
                   GENERIC_WRITE,
                   0,
-                  NULL,
+                  &saAttr,
                   OPEN_EXISTING,
                   FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OVERLAPPED,
                   NULL);
@@ -2058,7 +2062,7 @@ Janet janet_cfun_stream_read(int32_t argc, Janet *argv) {
     double to = janet_optnumber(argv, argc, 3, INFINITY);
     if (janet_keyeq(argv[1], "all")) {
         if (to != INFINITY) janet_addtimeout(to);
-        janet_ev_readchunk(stream, buffer, -1);
+        janet_ev_readchunk(stream, buffer, INT32_MAX);
     } else {
         int32_t n = janet_getnat(argv, 1);
         if (to != INFINITY) janet_addtimeout(to);

src/core/os.c

@@ -527,10 +527,24 @@ static void close_handle(JanetHandle handle) {
 /* Create piped file for os/execute and os/spawn. Need to be careful that we mark
    the error flag if we can't create pipe and don't leak handles. *handle will be cleaned
    up by the calling function. If everything goes well, *handle is owned by the calling function,
-   (if it is set) and the returned JanetFile owns the other end of the pipe, which will be closed
+   (if it is set) and the returned handle owns the other end of the pipe, which will be closed
    on GC or fclose. */
 static JanetHandle make_pipes(JanetHandle *handle, int reverse, int *errflag) {
     JanetHandle handles[2];
+#ifdef JANET_EV
+
+    /* non-blocking pipes */
+    if (janet_make_pipe(handles)) goto error;
+    if (reverse) swap_handles(handles);
+#ifdef JANET_WINDOWS
+    if (!SetHandleInformation(handles[0], HANDLE_FLAG_INHERIT, 0)) goto error;
+#endif
+    *handle = handles[1];
+    return handles[0];
+
+#else
+
+    /* Normal blocking pipes */
 #ifdef JANET_WINDOWS
     SECURITY_ATTRIBUTES saAttr;
     memset(&saAttr, 0, sizeof(saAttr));
@@ -548,6 +562,8 @@ static JanetHandle make_pipes(JanetHandle *handle, int reverse, int *errflag) {
     *handle = handles[1];
     return handles[0];
 #endif
+
+#endif
 error:
     *errflag = 1;
     return JANET_HANDLE_NONE;
@@ -736,7 +752,6 @@ static Janet os_execute_impl(int32_t argc, Janet *argv, int is_spawn) {
     startupInfo.cb = sizeof(startupInfo);
     startupInfo.dwFlags |= STARTF_USESTDHANDLES;
     saAttr.nLength = sizeof(saAttr);
-    saAttr.bInheritHandle = TRUE;
 
     JanetBuffer *buf = os_exec_escape(exargs);
     if (buf->count > 8191) {
@@ -862,9 +877,6 @@ static Janet os_execute_impl(int32_t argc, Janet *argv, int is_spawn) {
     if (status) {
         os_execute_cleanup(envp, child_argv);
         janet_panicf("%p: %s", argv[0], strerror(errno));
-    } else if (is_spawn) {
-        /* Get process handle */
-        os_execute_cleanup(envp, child_argv);
     } else {
         /* Wait to complete */
         os_execute_cleanup(envp, child_argv);

test/suite0009.janet

@@ -83,9 +83,16 @@
 
 (assert-error "bad arity to ev/call" (ev/call inc 1 2 3))
 
+(assert (os/execute [(dyn :executable) "-e" `(+ 1 2 3)`] :xp) "os/execute self")
+
 # Subprocess
+(let [p (os/spawn [(dyn :executable) "-e" `(file/read stdin :line)`] :px {:in :pipe})]
+  (:write (p :in) "hello!")
+  (assert-no-error "pipe stdin to process" (os/proc-wait p)))
+
 (let [p (os/spawn [(dyn :executable) "-e" `(print "hello")`] :p {:out :pipe})]
-  (assert (deep= @"hello\n" (:read (p :out) :all)) "capture stdout from os/spawn")
-  (os/proc-wait p))
+  (os/proc-wait p)
+  (def x (:read (p :out) 1024))
+  (assert (deep= "hello" (string/trim x)) "capture stdout from os/spawn"))
 
 (end-suite)