From df4f7971aa1200a525d68aaa44db45fcb20a3b04 Mon Sep 17 00:00:00 2001
From: Jan Nieuwenhuizen <janneke@gnu.org>
Date: Tue, 26 Jun 2018 20:34:07 +0200
Subject: [PATCH] updates from verum/gaiag.

---
 gash/gash.scm |  32 +++---
 gash/io.scm   |   2 +-
 gash/job.scm  | 113 ++++++++++----------
 gash/pipe.scm | 277 +++++++++++++++++++++++++-------------------------
 gash/util.scm |   6 +-
 5 files changed, 222 insertions(+), 208 deletions(-)

diff --git a/gash/gash.scm b/gash/gash.scm
index c384fed..b5841a6 100644
--- a/gash/gash.scm
+++ b/gash/gash.scm
@@ -1,24 +1,24 @@
 (define-module (gash gash)
 
-  :use-module (srfi srfi-1)
-  :use-module (srfi srfi-26)
+  #:use-module (srfi srfi-1)
+  #:use-module (srfi srfi-26)
 
-  :use-module (ice-9 ftw)
-  :use-module (ice-9 getopt-long)
-  :use-module (ice-9 local-eval)
-  :use-module (ice-9 match)
-  :use-module (ice-9 rdelim)
-  :use-module (ice-9 readline)
-  :use-module (ice-9 buffered-input)
-  :use-module (ice-9 regex)
+  #:use-module (ice-9 ftw)
+  #:use-module (ice-9 getopt-long)
+  #:use-module (ice-9 local-eval)
+  #:use-module (ice-9 match)
+  #:use-module (ice-9 rdelim)
+  #:use-module (ice-9 readline)
+  #:use-module (ice-9 buffered-input)
+  #:use-module (ice-9 regex)
 
-  :use-module (gash job)
-  :use-module (gash pipe)
-  :use-module (gash peg)
-  :use-module (gash io)
-  :use-module (gash util)
+  #:use-module (gash job)
+  #:use-module (gash pipe)
+  #:use-module (gash peg)
+  #:use-module (gash io)
+  #:use-module (gash util)
 
-  :export (main))
+  #:export (main))
 
 (define (remove-shell-comments s)
   (string-join (map
diff --git a/gash/io.scm b/gash/io.scm
index 6f64b34..e5dc37a 100644
--- a/gash/io.scm
+++ b/gash/io.scm
@@ -1,6 +1,6 @@
 (define-module (gash io)
 
-  :export (stdout stderr))
+  #:export (stdout stderr))
 
 (define (output port o)
   (map (lambda (o) (display o port)) o)
diff --git a/gash/job.scm b/gash/job.scm
index 0af578d..c5de569 100644
--- a/gash/job.scm
+++ b/gash/job.scm
@@ -1,21 +1,22 @@
 (define-module (gash job)
-  :use-module (srfi srfi-1)
-  :use-module (srfi srfi-8)
-  :use-module (srfi srfi-9)
-  :use-module (srfi srfi-26)
+  #:use-module (srfi srfi-1)
+  #:use-module (srfi srfi-8)
+  #:use-module (srfi srfi-9)
+  #:use-module (srfi srfi-26)
 
-  :use-module (gash io)
-  :use-module (gash util)
+  #:use-module (gash io)
+  #:use-module (gash util)
 
-  :export (job-control-init
-	   jobs report-jobs
-	   new-job
-	   job-add-process
-	   add-to-process-group
-	   wait
-	   fg
-	   bg
-	   setup-process))
+  :export (bg
+           fg
+           job-add-process
+           job-control-init
+           job-debug-id
+           job-setup-process
+           jobs
+           new-job
+           report-jobs
+           wait))
 
 (define-record-type <process>
   (make-process pid command status)
@@ -25,14 +26,21 @@
   (status process-status set-process-status!))
 
 (define-record-type <job>
-  (make-job id pgid processes)
+  (make-job id pgid processes debug-id)
   job?
   (id job-id)
   (pgid job-pgid set-job-pgid!)
-  (processes job-processes set-job-processes!))
+  (processes job-processes set-job-processes!)
+  (debug-id job-debug-id))
+
+(define debug-id
+  (let ((id -1))
+    (lambda ()
+      (set! id (1+ id))
+      (number->string id))))
 
 (define (new-job)
-  (let ((job (make-job (+ 1 (length job-table)) #f '())))
+  (let ((job (make-job (+ 1 (length job-table)) #f '() (debug-id))))
     (set! job-table (cons job job-table))
     job))
 
@@ -82,27 +90,37 @@
     (every (cut member <> '(Done Terminated)) state)))
 
 (define (add-to-process-group job pid)
-  (let* ((pgid (job-pgid job))
-         (pgid (or pgid pid)))
-    (setpgid pid pgid)
+  (let* ((interactive? (isatty? (current-error-port)))
+         (pgid (if interactive?
+                   (or (job-pgid job) pid)
+                   (getpgrp))))
+    (set-job-pgid! job pgid)
+    (when interactive? (setpgid pid pgid))
     pgid))
 
 (define (job-add-process fg? job pid command)
   (let ((pgid (add-to-process-group job pid)))
     (set-job-pgid! job pgid)
-    (when fg? (tcsetpgrp (current-error-port) pgid))
+    (stderr "job-add-process fg?=~a\n" fg?)
+    (when (and #f fg?) ;; FIXME
+      (tcsetpgrp (current-error-port) pgid))
     (set-job-processes! job (cons (make-process pid command #f) (job-processes job)))))
 
+(define (job-setup-process fg? job)
+  (when (isatty? (current-error-port))
+    (when (and #f fg?)
+      (tcsetpgrp (current-error-port) (add-to-process-group job (getpid))))
+    (map (cut sigaction <> SIG_DFL)
+         (list SIGINT SIGQUIT SIGTSTP SIGTTIN SIGTTOU SIGCHLD))))
+
 (define (job-control-init)
-  (let* ((interactive? (isatty? (current-error-port)))
-         (pgid (getpgrp))
-         (pid (getpid)))
-    (when interactive?
+  (when (isatty? (current-error-port))
+    (let ((pgid (getpgrp)))
       (while (not (eqv? (tcgetpgrp (current-error-port)) pgid))
-        (kill (- pgid) SIGTTIN)) ;; oops we are not in the foreground
-      (map (cut sigaction <> SIG_IGN)
-           (list SIGINT SIGQUIT SIGTSTP SIGTTIN SIGTTOU))
-      (sigaction SIGCHLD SIG_DFL)
+        (kill (- pgid) SIGTTIN))) ;; oops we are not in the foreground
+    (map (cut sigaction <> SIG_IGN) (list SIGINT SIGQUIT SIGTSTP SIGTTIN SIGTTOU))
+    (sigaction SIGCHLD SIG_DFL)
+    (let ((pid (getpid)))
       (setpgid pid pid) ;; create new process group for ourself
       (tcsetpgrp (current-error-port) pid))))
 
@@ -120,24 +138,24 @@
         (reap-jobs)))))
 
 (define (wait job)
-  (let loop ()
-    (let* ((pid-status (waitpid (- (job-pgid job)) WUNTRACED))
-           (pid (car pid-status))
-           (status (cdr pid-status)))
-      (job-update job pid status)
-      (if (job-running? job) (loop))))
-  (tcsetpgrp (current-error-port) (getpid))
+  (when (job-running? job)
+    (let loop ()
+      (let* ((pid-status (waitpid (- (job-pgid job)) WUNTRACED))
+             (pid (car pid-status))
+             (status (cdr pid-status)))
+        (job-update job pid status)
+        (if (job-running? job) (loop)))))
   (unless (job-completed? job)
     (newline) (display-job job))
   (reap-jobs)
-  (last (job-status job)))
+  (or (and (every zero? (job-status job)) 0) 1))
 
 (define (fg index)
   (let ((job (job-index index)))
     (cond (job
            (let ((pgid (job-pgid job)))
-	     (tcsetpgrp (current-error-port) pgid)
-	     (kill (- (job-pgid job)) SIGCONT))
+             (tcsetpgrp (current-error-port) pgid)
+             (kill (- (job-pgid job)) SIGCONT))
            (stdout (job-command job))
            (wait job))
           (#t
@@ -146,14 +164,7 @@
 (define (bg index)
   (let ((job (job-index index)))
     (cond (job
-	   (map (cut set-process-status! <>  #f) (job-processes job))
-	   (kill (- (job-pgid job)) SIGCONT))
-	  (#t
-	   (stderr "fg: no such job " index)))))
-
-(define (setup-process fg? job)
-  (when (isatty? (current-error-port))
-    (when fg? (tcsetpgrp (current-error-port) (add-to-process-group job (getpid))))
-    (map (cut sigaction <> SIG_DFL)
-         (list SIGINT SIGQUIT SIGTSTP SIGTTIN SIGTTOU SIGCHLD)))
-  (fdes->inport 0) (map fdes->outport '(1 2))) ;; reset stdin/stdout/stderr
+           (map (cut set-process-status! <>  #f) (job-processes job))
+           (kill (- (job-pgid job)) SIGCONT))
+          (#t
+           (stderr "fg: no such job " index)))))
diff --git a/gash/pipe.scm b/gash/pipe.scm
index a3564ec..1ea845a 100644
--- a/gash/pipe.scm
+++ b/gash/pipe.scm
@@ -1,173 +1,176 @@
 (define-module (gash pipe)
 
-  :use-module (ice-9 popen)
-  :use-module (ice-9 rdelim)
+  #:use-module (ice-9 curried-definitions)
+  #:use-module (ice-9 popen)
+  #:use-module (ice-9 rdelim)
 
-  :use-module (srfi srfi-1)
-  :use-module (srfi srfi-8)
-  :use-module (srfi srfi-9)
-  :use-module (srfi srfi-26)
+  #:use-module (srfi srfi-1)
+  #:use-module (srfi srfi-8)
+  #:use-module (srfi srfi-9)
+  #:use-module (srfi srfi-26)
 
-  :use-module (gash job)
+  #:use-module (gash job)
+  #:use-module (gash io)
 
-  :export (pipeline substitute))
+  #:export (handle-error pipeline pipeline->string substitute))
+
+;; TODO
+(define %debug-level 0)
+
+(define (handle-error job error)
+  (let ((status (wait job)))
+    (when (not (zero? status))
+      (format (current-error-port) "ERROR: exit: ~a: ~s" status error)
+      (exit status))
+    status))
 
 (define (pipe*)
   (let ((p (pipe)))
     (values (car p) (cdr p))))
 
 ;;              lhs        rhs
-;; [source] w -> r [filter] w -> r [sink]
+;; [source] w[1] -> r[0] [filter] w[1] -> r[0] [sink]
+;;          w[2]            ->            r[3] [sink]
 
 (define (exec* command) ;; list of strings
   (catch #t (lambda () (apply execlp (cons (car command) command)))
     (lambda (key . args) (format (current-error-port) "~a\n" (caaddr args))
             (exit #f))))
 
-(define (spawn-source fg? job command)
-  (receive (r w) (pipe*)
-    (let ((pid (primitive-fork)))
-      (cond ((= 0 pid)
-	     (close r)
-	     (setup-process fg? job)
-	     (move->fdes w 1)
-             (if (procedure? command)
-		 (begin
-		   (close-port (current-output-port))
-		   (set-current-output-port w)
-		   (command)
-		   (exit 0))
-		 (exec* command)))
-            (#t
-             (job-add-process fg? job pid command)
-             (close w)
-             r)))))
+(define ((tee-n file-names) inputs outputs)
+  (let* ((files  (map open-output-file file-names))
+         (tees (zip files inputs outputs)))
+    (let loop ((tees tees))
+      (loop (filter-map (lambda (tee)
+                          (let ((file (first tee))
+                                (input (second tee))
+                                (output (third tee)))
+                            (when (char-ready? input)
+                              (let ((char (read-char input)))
+                                (if (not (eof-object? char))
+                                    (begin (display char file)
+                                           (display char output)
+                                           (list file input output))
+                                    #f)))))
+                        tees)))
+    (map close outputs)))
 
-(define (spawn-filter fg? job src command)
-  (receive (r w) (pipe*)
-    (let ((pid (primitive-fork)))
-      (cond ((= 0 pid)
-             (setup-process fg? job)
-             (if src (move->fdes src 0))
-             (close r)
-             (move->fdes w 1)
-	     (if (procedure? command)
-		 (begin
-		   (close-port (current-input-port))
-		   (close-port (current-output-port))
-		   (set-current-input-port src)
-		   (set-current-output-port w)
-		   (command)
-		   (exit 0))
-		 (exec* command)))
-            (#t
-             (job-add-process fg? job pid command)
-             (close w)
-             r)))))
-
-(define (spawn-sink fg? job src command)
-  (let ((pid (primitive-fork)))
+(define* (spawn fg? job command #:optional (input '()))
+  ;;(format #t "spawn: ~a\n" (length input))
+  (let* ((ofd '(1 2)) ;; output file descriptors 1, ...
+         (ifd (cond
+               ((= (length input) 0) '())
+               ((= (length input) 1) '(0))))
+         (pipes (map (lambda (. _) (pipe)) ofd))
+         (r (map car pipes))
+         (w (map cdr pipes))
+         (pid (primitive-fork)))
+    ;;(format (current-error-port) "INPUT: ~a\n" (length input))
+    ;;(format (current-error-port) "OUTPUT: ~a\n" (length w))
     (cond ((= 0 pid)
-           (setup-process fg? job)
-           (if src (move->fdes src 0))
-	   (if (procedure? command)
-	       (begin
-		   (close-port (current-input-port))
-		   (set-current-input-port src)
-		   (command)
-		   (exit 0))
-	       (exec* command)))
+           (job-setup-process fg? job)
+           (map close r)
+           (if (procedure? command)
+               (begin
+                 (when (pair? input)
+                   (close-port (current-input-port))
+                   (set-current-input-port (car input)))
+                 (when (pair? w)
+                   (close-port (current-output-port))
+                   (set-current-output-port (car w)))
+                 ;;(format (current-error-port) "INPUT: ~a\n" (length input))
+                 ;;(format (current-error-port) "OUTPUT: ~a\n" (length w))
+                 (if (thunk? command) (command)
+                     (command input w))
+                 (exit 0))
+               (begin
+                 (map dup->fdes w ofd)
+                 (map dup->fdes input ifd)
+                 (exec* command))))
           (#t
            (job-add-process fg? job pid command)
-           (and src (close src))))))
-
-
-(define* (spawn fg? job command #:optional (input '()) (output 0))
-  ;;(format #t "spawn: ~a ~a\n" (length input) output)
-  (let* ((ofd (iota output 1)) ;; output file descriptors 1, ...
-	 (count (length input))
-	 (start (1+ output))
-	 (ifd (cond
-	       ((= count 0) '())
-	       ((= count 1) '(0))
-	       ((#t (cons 0 (iota (1- count) start))))))
-	 (ifd (if (pair? input) (cons 0 ifd) ifd))
-	 ;;(foo (format #t "ifd: ~a\n" ifd))
-	 ;;(foo (format #t "ofd: ~a\n" ofd))
-	 (pipes (map (lambda (. _) (pipe)) ofd))
-	 (r (map car pipes))
-	 (w (map cdr pipes))
-	 (pid (primitive-fork)))
-      (cond ((= 0 pid)
-	     (setup-process fg? job)
-	     (map close r)
-	     (map move->fdes w ofd)
-	     (map move->fdes input ifd)
-             (if (procedure? command)
-		 (begin
-		   (when (pair? input)
-		     (close-port (current-input-port))
-		     (set-current-input-port (car input)))
-		   (when (pair? w)
-		     (close-port (current-output-port))
-		     (set-current-output-port (car w)))
-		   (command)
-		   (exit 0))
-		 (exec* command)))
-            (#t
-             (job-add-process fg? job pid command)
-             (map close w)
-             r))))
-
-(define (pipeline+ fg? open? . commands)
-  (let* ((job (new-job))
-	 (ports (if (> (length commands) 1)
-		   (let loop ((input (spawn fg? job (car commands) '() 1)) ;; spawn-source
-			      (commands (cdr commands)))
-		     (if (null? (cdr commands))
-			 (spawn fg? job (car commands) input (if open? 1 0)) ;; spawn-sink
-			 (loop (spawn fg? job (car commands) input 1) ;; spawn-filter
-			       (cdr commands))))
-		   (spawn fg? job (car commands) `((current-input-port))))))
-    (if fg? (wait job) (values job ports))))
+           (map close w)
+           r))))
 
 (define (pipeline fg? . commands)
-  (apply pipeline+ (cons* fg? #f commands)))
+  (when (> %debug-level 0)
+    (format (current-error-port) "pipeline[~a]: COMMANDS: ~s\n" fg? commands))
+  (receive (r w)
+      (pipe*)
+    (move->fdes w 2)
+    (let* ((error-port (set-current-error-port w))
+           (job (new-job))
+           (debug-id (job-debug-id job))
+           (commands
+            (if (zero? %debug-level) commands
+                (fold-right (lambda (command id lst)
+                              (let ((file (string-append debug-id "." id)))
+                                (cons* command `("tee" ,file) lst)))
+                            '() commands (map number->string (iota (length commands))))))
+           (foo (when (> %debug-level 0) (with-output-to-file debug-id (cut format #t "COMMANDS: ~s\n" commands))))
+           (ports (if (> (length commands) 1)
+                      (let loop ((input (spawn fg? job (car commands) '())) ;; spawn-source
+                                 (commands (cdr commands)))
+                        (if (null? (cdr commands))
+                            (spawn fg? job (car commands) input) ;; spawn-sink
+                            (loop (spawn fg? job (car commands) input) ;; spawn-filter
+                                  (cdr commands))))
+                      (spawn fg? job (car commands) '())))) ;; spawn-sink
+      (when fg?
+        (let loop ((input ports)
+                   (output (list (current-output-port) error-port)))
+          (let ((line (map read-line input)))
+            (let* ((input-available? (lambda (o ln) (and (not (eof-object? ln)) o)))
+                   (line (filter-map input-available? line line))
+                   (output (filter-map input-available? output line))
+                   (input (filter-map input-available? input line)))
+              (when (pair? input)
+                (map display line output)
+                (map newline output)
+                (loop input output)))))
+        (wait job))
+      (move->fdes error-port 2)
+      (set-current-error-port error-port)
+      (close w)
+      (values job (append ports (list r))))))
+
+(define (pipeline->string . commands)
+  (receive (job ports)
+      (apply pipeline #f commands)
+    (let ((output (read-string (car ports))))
+      (wait job)
+      output)))
 
 ;;(pipeline #f '("head" "-c128" "/dev/urandom") '("tr" "-dc" "A-Z0-9") (lambda () (display (read-string))))
 ;;(pipeline #f '("head" "-c128" "/dev/urandom") '("tr" "-dc" "A-Z0-9") '("cat"))
 ;;(pipeline #f (lambda () (display 'foo)) '("grep" "o") '("tr" "o" "e"))
 
 ;; (pipeline #f
-;; 	  (lambda () (display "\nbin\nboot\nroot\nusr\nvar"))
-;; 	  '("tr" "u" "a")
-;; 	  (lambda () (display (string-map (lambda (c) (if (eq? c #\o) #\e c)) (read-string))))
-;; 	  '("cat")
-;; 	  (lambda () (display (read-string))))
+;;    (lambda () (display "\nbin\nboot\nroot\nusr\nvar"))
+;;    '("tr" "u" "a")
+;;    (lambda () (display (string-map (lambda (c) (if (eq? c #\o) #\e c)) (read-string))))
+;;    '("cat")
+;;    (lambda () (display (read-string))))
 
 ;; (receive (job ports)
-;;     (pipeline+ #f #t
-;; 	      (lambda () (display "\nbin\nboot\nroot\nusr\nvar"))
-;; 	      '("tr" "u" "a")
-;; 	      (lambda () (display (string-map (lambda (c) (if (eq? c #\o) #\e c)) (read-string))))
-;; 	      '("cat"))
-;;   (display (read-string (car ports))))
+;;     (pipeline #f
+;;               (lambda ()
+;;                 (display "foo")
+;;                 (display "bar" (current-error-port)))
+;;               '("tr" "o" "e"))
+;;   (map (compose display read-string) ports))
 
+;; _
+;;  \
+;;   -
+;; _/
 
-(define (pipeline->string . commands)
-  (let* ((fg? #f)
-         (job (new-job))
-         (output (read-string
-                  (if (> (length commands) 1)
-                      (let loop ((src (spawn-source fg? job (car commands)))
-                                 (commands (cdr commands)))
-                        (if (null? (cdr commands))
-                            (spawn-filter fg? job src (car commands))
-                            (loop (spawn-filter fg? job src (car commands))
-                                  (cdr commands))))
-                      (spawn-filter fg? job #f (car commands))))))
-    (wait job)
-    output))
+;; (display (pipeline->string
+;;   (lambda () (display "\nbin\nboot\nroot\nusr\nvar"))
+;;   '("tr" "u" "a")
+;;   (lambda () (display (string-map (lambda (c) (if (eq? c #\o) #\e c)) (read-string))))
+;;   '("cat")
+;;   (lambda () (display (read-string)) (newline))))
 
 ;; _
 ;;  \
diff --git a/gash/util.scm b/gash/util.scm
index a0ecc52..ce1c8b5 100644
--- a/gash/util.scm
+++ b/gash/util.scm
@@ -1,8 +1,8 @@
 (define-module (gash util)
-  :use-module (srfi srfi-1)
-  :use-module (srfi srfi-26)
+  #:use-module (srfi srfi-1)
+  #:use-module (srfi srfi-26)
 
-  :export (disjoin conjoin))
+  #:export (disjoin conjoin))
 
 (define (disjoin . predicates)
   (lambda (. arguments)