Add options to os/pipe for custom blocking behavior.

Useful for create pipelines on unix.

Also add bundle/whois to help diagnose where bundle installed files
came from.

src/boot/boot.janet

@@ -4339,6 +4339,19 @@
     (print "add " absdest)
     absdest)
 
+  (defn bundle/whois
+    "Given a file path, figure out which bundle installed it."
+    [path]
+    (var ret nil)
+    (def rpath (bundle-rpath path))
+    (each bundle-name (bundle/list)
+      (def files (get (bundle/manifest bundle-name) :files []))
+      (def has-file (index-of rpath files))
+      (when has-file
+        (set ret bundle-name)
+        (break)))
+    ret)
+
   (defn bundle/add-file
     "Add files during an install relative to `(dyn *syspath*)`"
     [manifest src &opt dest chmod-mode]

src/core/ev.c

@@ -469,7 +469,7 @@ static Janet janet_stream_next(void *p, Janet key) {
 static void janet_stream_tostring(void *p, JanetBuffer *buffer) {
     JanetStream *stream = p;
     /* Let user print the file descriptor for debugging */
-    janet_formatb(buffer, "<core/stream handle=%d>", stream->handle);
+    janet_formatb(buffer, "[fd=%d]", stream->handle);
 }
 
 const JanetAbstractType janet_stream_type = {
@@ -2699,6 +2699,7 @@ static volatile long PipeSerialNumber;
  * mode = 0: both sides non-blocking.
  * mode = 1: only read side non-blocking: write side sent to subprocess
  * mode = 2: only write side non-blocking: read side sent to subprocess
+ * mode = 3: both sides blocking - for use in two subprocesses (making pipeline from external processes)
  */
 int janet_make_pipe(JanetHandle handles[2], int mode) {
 #ifdef JANET_WINDOWS
@@ -2712,6 +2713,11 @@ int janet_make_pipe(JanetHandle handles[2], int mode) {
     memset(&saAttr, 0, sizeof(saAttr));
     saAttr.nLength = sizeof(saAttr);
     saAttr.bInheritHandle = TRUE;
+    if (mode == 3) {
+        /* No overlapped IO involved, just call CreatePipe */
+        if (!CreatePipe(handles, handles + 1, &saAttr, 0)) return -1;
+        return 0;
+    }
     sprintf(PipeNameBuffer,
             "\\\\.\\Pipe\\JanetPipeFile.%08x.%08x",
             (unsigned int) GetCurrentProcessId(),
@@ -2757,8 +2763,8 @@ int janet_make_pipe(JanetHandle handles[2], int mode) {
     if (pipe(handles)) return -1;
     if (mode != 2 && fcntl(handles[0], F_SETFD, FD_CLOEXEC)) goto error;
     if (mode != 1 && fcntl(handles[1], F_SETFD, FD_CLOEXEC)) goto error;
-    if (mode != 2 && fcntl(handles[0], F_SETFL, O_NONBLOCK)) goto error;
-    if (mode != 1 && fcntl(handles[1], F_SETFL, O_NONBLOCK)) goto error;
+    if (mode != 2 && mode != 3 && fcntl(handles[0], F_SETFL, O_NONBLOCK)) goto error;
+    if (mode != 1 && mode != 3 && fcntl(handles[1], F_SETFL, O_NONBLOCK)) goto error;
     return 0;
 error:
     close(handles[0]);

src/core/os.c

@@ -2680,16 +2680,24 @@ JANET_CORE_FN(os_open,
 }
 
 JANET_CORE_FN(os_pipe,
-              "(os/pipe)",
+              "(os/pipe &opt flags)",
               "Create a readable stream and a writable stream that are connected. Returns a two-element "
               "tuple where the first element is a readable stream and the second element is the writable "
-              "stream.") {
+              "stream. `flags` is an optional set of keyword flags used to set whether the reader or writer ends of the pipe "
+              "can be used for blocking IO. \n\n"
+              "* :W - sets the writable end of the pipe to a blocking stream.\n"
+              "* :R - sets the readable end of the pipe to a blocking stream.\n\n"
+              "By default, both ends of the pipe are non-blocking for use the the `ev` module.") {
     (void) argv;
-    janet_fixarity(argc, 0);
+    janet_arity(argc, 0, 1);
     JanetHandle fds[2];
-    if (janet_make_pipe(fds, 0)) janet_panicv(janet_ev_lasterr());
-    JanetStream *reader = janet_stream(fds[0], JANET_STREAM_READABLE, NULL);
-    JanetStream *writer = janet_stream(fds[1], JANET_STREAM_WRITABLE, NULL);
+    int flags = 0;
+    if (argc > 0 && !janet_checktype(argv[0], JANET_NIL)) {
+        flags = (int) janet_getflags(argv, 0, "WR");
+    }
+    if (janet_make_pipe(fds, flags)) janet_panicv(janet_ev_lasterr());
+    JanetStream *reader = janet_stream(fds[0], (flags & 2) ? 0 : JANET_STREAM_READABLE, NULL);
+    JanetStream *writer = janet_stream(fds[1], (flags & 1) ? 0 : JANET_STREAM_WRITABLE, NULL);
     Janet tup[2] = {janet_wrap_abstract(reader), janet_wrap_abstract(writer)};
     return janet_wrap_tuple(janet_tuple_n(tup, 2));
 }