guile pipeline patch

[Rutger van Beusekom <rutger.van.beusekom@verum.com>]

[-- Attachment #1: Type: text/plain, Size: 1126 bytes --]

Hi,

This patch replaces open-process with piped-process in posix.c and
reimplement open-process with piped-process in popen.scm. This allows
setting up a pipeline in guile scheme using the new pipeline procedure
in popen.scm and enables its use on operating systems which happen to
lack the capability to fork, but do offer the capability captured by
start_child (see posix.c).

The snippet below demonstrates the backwards compatibility of this patch
as well as the feature it offers:

(use-modules (ice-9 popen))
(use-modules (ice-9 rdelim))
(use-modules (ice-9 receive))

(receive (from to pid)
    ((@@ (ice-9 popen) open-process) OPEN_BOTH "rev")
  (display "dlrow olleh" to) (close to)
  (display (read-string from)) (newline)
  (display (status:exit-val (cdr (waitpid pid)))) (newline))

(receive (from to pids)
    (pipeline '(("echo" "dlrow olleh") ("rev")))
  (display (read-string from))
  (display (map waitpid pids)) (newline))

(let ((p (pipe)))
  (piped-process "echo" '("foo" "bar")
                 (cons (port->fdes (car p))
                       (port->fdes (cdr p))))
  (display (read-string (car p))))


[-- Warning: decoded text below may be mangled, UTF-8 assumed --]
[-- Attachment #2: PATCH --]
[-- Type: text/x-diff, Size: 7478 bytes --]

From 6cf03c09da190f0e797768b598f9e347168e3f19 Mon Sep 17 00:00:00 2001
From: Rutger van Beusekom <rutger.van.beusekom@verum.com>
Date: Mon, 2 Mar 2020 10:38:57 +0100
Subject: [PATCH] Allow client code to create pipe pairs when opening a
 process.

* libguile/posix.c (scm_piped_process): Replace open_process by piped_process.
* module/ice-9/popen.scm (pipe->fdes): Convert pipe pair to fdes pair.
(open-process): Implement open-process with piped-process.
(pipeline): Implement a pipeline with piped-process.
---
 libguile/posix.c       | 84 +++++++++++++++++++-----------------------
 module/ice-9/popen.scm | 32 +++++++++++++++-
 2 files changed, 68 insertions(+), 48 deletions(-)

diff --git a/libguile/posix.c b/libguile/posix.c
index a1520abc4..f9ffd5d8e 100644
--- a/libguile/posix.c
+++ b/libguile/posix.c
@@ -1368,55 +1368,56 @@ start_child (const char *exec_file, char **exec_argv,
 
 #ifdef HAVE_START_CHILD
 static SCM
-scm_open_process (SCM mode, SCM prog, SCM args)
-#define FUNC_NAME "open-process"
+scm_piped_process (SCM prog, SCM args, SCM from, SCM to)
+#define FUNC_NAME "piped-process"
+/* SCM_DEFINE (scm_piped_process, "piped-process", 2, 2, 0, */
+/*            (SCM prog, SCM args, SCM from, SCM to), */
+/* "Execute the command indicated by @var{prog} with arguments @var(args),\n" */
+/* "optionally connected by an input and an output pipe.\n" */
+/* "@var(from) and @var(to) are either #f or a valid file descriptor\n" */
+/* "of an input and an output pipe, respectively.\n" */
+/* "\n" */
+/* "This function returns the PID of the process executing @var(prog)." */
+/* "\n" */
+/* "Example:\n" */
+/* "(let ((p (pipe)))\n" */
+/* "  (piped-process \"echo\" '(\"foo\" \"bar\")\n" */
+/* "                 (cons (port->fdes (car p))\n" */
+/* "                       (port->fdes (cdr p))))\n" */
+/* "  (display (read-string (car p))))\n" */
+/* "(let ((p (pipe)))\n" */
+/* "  (read-string (piped-process \"echo\" '(\"foo\" \"bar\")\n" */
+/* "                              (port->fdes (car p)))))\n") */
+/* #define FUNC_NAME scm_piped_process */
 {
-  long mode_bits;
   int reading, writing;
-  int c2p[2]; /* Child to parent.  */
-  int p2c[2]; /* Parent to child.  */
+  int c2p[2] = {}; /* Child to parent.  */
+  int p2c[2] = {}; /* Parent to child.  */
   int in = -1, out = -1, err = -1;
   int pid;
   char *exec_file;
   char **exec_argv;
-  SCM read_port = SCM_BOOL_F, write_port = SCM_BOOL_F;
 
   exec_file = scm_to_locale_string (prog);
   exec_argv = scm_i_allocate_string_pointers (scm_cons (prog, args));
 
-  mode_bits = scm_i_mode_bits (mode);
-  reading = mode_bits & SCM_RDNG;
-  writing = mode_bits & SCM_WRTNG;
+  reading = !scm_is_eq (from, SCM_UNDEFINED);
+  writing = !scm_is_eq (to, SCM_UNDEFINED);
 
   if (reading)
     {
-      if (pipe (c2p))
-        {
-          int errno_save = errno;
-          free (exec_file);
-          errno = errno_save;
-          SCM_SYSERROR;
-        }
+      c2p[0] = scm_to_int (scm_car (from));
+      c2p[1] = scm_to_int (scm_cdr (from));
       out = c2p[1];
     }
-  
+
   if (writing)
     {
-      if (pipe (p2c))
-        {
-          int errno_save = errno;
-          free (exec_file);
-          if (reading)
-            {
-              close (c2p[0]);
-              close (c2p[1]);
-            }
-          errno = errno_save;
-          SCM_SYSERROR;
-        }
+      p2c[0] = scm_to_int (scm_car (to));
+      p2c[1] = scm_to_int (scm_cdr (to));
       in = p2c[0];
     }
-  
+
   {
     SCM port;
 
@@ -1449,23 +1450,12 @@ scm_open_process (SCM mode, SCM prog, SCM args)
       SCM_SYSERROR;
     }
 
-  /* There is no sense in catching errors on close().  */
   if (reading)
-    {
-      close (c2p[1]);
-      read_port = scm_i_fdes_to_port (c2p[0], scm_mode_bits ("r0"),
-                                      sym_read_pipe,
-                                      SCM_FPORT_OPTION_NOT_SEEKABLE);
-    }
+    close (c2p[1]);
   if (writing)
-    {
-      close (p2c[0]);
-      write_port = scm_i_fdes_to_port (p2c[1], scm_mode_bits ("w0"),
-                                       sym_write_pipe,
-                                       SCM_FPORT_OPTION_NOT_SEEKABLE);
-    }
+    close (p2c[0]);
 
-  return scm_values_3 (read_port, write_port, scm_from_int (pid));
+  return scm_from_int (pid);
 }
 #undef FUNC_NAME
 
@@ -1529,8 +1519,8 @@ SCM_DEFINE (scm_system_star, "system*", 0, 0, 1,
                          SCM_UNDEFINED);
 #endif
 
-  res = scm_open_process (scm_nullstr, prog, args);
-  pid = scm_to_int (scm_c_value_ref (res, 2));
+  res = scm_piped_process (prog, args, SCM_BOOL_F, SCM_BOOL_F);
+  pid = scm_to_int (scm_c_value_ref (res, 0));
   SCM_SYSCALL (wait_result = waitpid (pid, &status, 0));
   if (wait_result == -1)
     SCM_SYSERROR;
@@ -2371,7 +2361,7 @@ SCM_DEFINE (scm_gethostname, "gethostname", 0, 0, 0,
 static void
 scm_init_popen (void)
 {
-  scm_c_define_gsubr ("open-process", 2, 0, 1, scm_open_process);
+  scm_c_define_gsubr ("piped-process", 2, 2, 0, scm_piped_process);
 }
 #endif /* HAVE_START_CHILD */
 
diff --git a/module/ice-9/popen.scm b/module/ice-9/popen.scm
index 2afe45701..ad1f64c7c 100644
--- a/module/ice-9/popen.scm
+++ b/module/ice-9/popen.scm
@@ -22,9 +22,10 @@
   #:use-module (rnrs bytevectors)
   #:use-module (ice-9 binary-ports)
   #:use-module (ice-9 threads)
+  #:use-module (srfi srfi-1)
   #:use-module (srfi srfi-9)
   #:export (port/pid-table open-pipe* open-pipe close-pipe open-input-pipe
-            open-output-pipe open-input-output-pipe))
+            open-output-pipe open-input-output-pipe pipe->fdes piped-process pipeline))
 
 (eval-when (expand load eval)
   (load-extension (string-append "libguile-" (effective-version))
@@ -84,6 +85,17 @@
 (define port/pid-table (make-weak-key-hash-table))
 (define port/pid-table-mutex (make-mutex))
 
+(define (pipe->fdes)
+  (let ((p (pipe)))
+   (cons (port->fdes (car p))
+         (port->fdes (cdr p)))))
+
+(define (open-process mode command . args)
+  (let* ((from (and (or (equal? mode OPEN_READ) (equal? mode OPEN_BOTH)) (pipe->fdes)))
+         (to (and (or (equal? mode OPEN_WRITE) (equal? mode OPEN_BOTH)) (pipe->fdes)))
+         (pid (piped-process command args from to)))
+    (values (and from (fdes->inport (car from))) (and to (fdes->outport (cdr to))) pid)))
+
 (define (open-pipe* mode command . args)
   "Executes the program @var{command} with optional arguments
 @var{args} (all strings) in a subprocess.
@@ -176,3 +188,21 @@ information on how to interpret this value."
   "Equivalent to @code{open-pipe} with mode @code{OPEN_BOTH}"
   (open-pipe command OPEN_BOTH))
 
+
+(define (pipeline procs)
+  "Execute a pipeline of @code(procs) -- where a proc is a list of a
+command and its arguments as strings -- returning an input port to the
+end of the pipeline, an output port to the beginning of the pipeline and
+a list of PIDs of the @code(procs)"
+  (let* ((to (pipe->fdes))
+         (pipes (map (lambda _ (pipe->fdes)) procs))
+	 (pipeline (fold (lambda (from proc prev)
+                           (let* ((to (car prev))
+                                  (pids (cdr prev)))
+                             (cons from (cons (piped-process (car proc) (cdr proc) from to) pids))))
+                         `(,to)
+                         pipes
+                         procs))
+	 (from (car pipeline))
+	 (pids (cdr pipeline)))
+    (values (fdes->inport (car from)) (fdes->outport (cdr to)) pids)))
-- 
2.25.1