[gambit-list] How do I create a pipeline of processes, as in shell?

Jussi Piitulainen jpiitula at ling.helsinki.fi
Tue Jan 29 08:51:23 EST 2013


Marc Feeley writes:

> On 2013-01-27, at 4:48 PM, Jussi Piitulainen wrote:
...
> > How does one implement the following lines in Gambit-C?
> > 
> > $ sort infile | uniq -c | sort -nr
> > $ sort < infile | uniq -c | sort -nr > outfile
...
> The code below will allow the creation of a pipeline of processes.
> The output of one process in the pipeline is transferred to the
> input of the next process.  Each data transfer from one process to
> the next is achieved by a Gambit thread which reads the output of a
> process and writes it to the input of the next process.  Although it
> "works", it is not as efficient as having the operating system do
> this through file descriptors.  But that could be achieved by
> spawning a shell and asking it to create the pipeline,
> i.e. (open-process '(path: "/bin/sh" arguments: ("-c" "sort infile |
> uniq -c | sort -nr"))).  I assume you are interested in Gambit level
> piping.

Thank you, Marc. You understood correctly that I wanted to do this in
Gambit level. I even expected Gambit to have this built in and ready
to use, since the README on <https://github.com/feeley/gambit> says
Gambit's I/O system supports, among other things, processes and pipes.
Thank you for confirming that I wasn't missing anything obvious.

Your solution is essentially what I was experimenting with, and after
further experimentation, I have decided to be happy with it. I cannot
invest much more effort in this at this time anyway.

My version is below. It represents the whole pipeline as the process
object for the last command in the pipe. There is an optional thunk
for sending input to the pipe. I left some auxiliaries in-line to keep
the global names few. Thanks for port-copy :)

#! /usr/bin/env gsi

;;; (pipe [thunk] command+) => input-port by Jussi Piitulainen
;;; <jpiitula at ling.helsinki.fi>, 29 January 2013; does not contain a
;;; range-check; is short and largely driven by the functionality -
;;; Marc Feeley's version was essentially the same; use, study, adapt,
;;; and copy as you please - if you need my permission, you have it;
;;; no conditions; no warranties. Thank you, gambit-list.

(define (port-copy in-port out-port)
  (let* ((m 4096) (buffer (make-u8vector m)))
    (do ((n (read-subu8vector buffer 0 m in-port 1)
	    (read-subu8vector buffer 0 m in-port 1)))
        ((= n 0))
      (write-subu8vector buffer 0 n out-port))))

;;; (pipe thunk command0 command1 ...) => input-port
;;; (pipe command0 command1 ...) => input-port
;;;
;;; Sends the output of the thunk through a pipeline of processes
;;; specified by the commands. The commands must be valid arguments
;;; for open-process. The pipe is run by reading from input-port, or
;;; it can be drained by asking for its process-status.
;;;
;;; This abstraction leaks: input-port is actually a process-port
;;; which should probably not be written to outside the pipe. (Might
;;; it sometimes make sense to close it as output-port?)

(define (pipe source . commands)
  (if (not (procedure? source)) ;source thunk is optional
      (apply pipe * source commands)
      (let ((first (car commands)) ;one command is required
	    (rest (cdr commands)))
	(let ((first (open-process first)))
	  (thread-start!
	   (make-thread (lambda ()
			  (with-output-to-port first source)
			  (close-output-port first))
			'source-of-pipe))
	  (do ((last first (car rest))
	       (rest (map open-process rest) (cdr rest)))
	      ((null? rest) last)
	    (thread-start!
	     (make-thread (lambda ()
			    (port-copy last (car rest))
			    (close-output-port (car rest)))
			  'piece-of-pipe)))))))

;;; One remaining issue: After the "head" process below terminates as
;;; intended, port-copy fails to write to it and that thread
;;; terminates quietly; when the system collects the thread object, it
;;; also gets to collect its input port, which it closes; the "base64"
;;; process is still around but now it fails to write, which makes it
;;; to terminate with a spurious diagnostic: "broken pipe". The
;;; stderr-redirection: #t masks this message. I wish I knew how to
;;; inform the "base64" process that this particular situation is ok.
;;; The important thing is that the process does terminate, and there
;;; is a way to mask the message, so let it be. Don't want anything
;;; worse to happen.

(define (go)
  (port-copy (pipe (lambda () (display "hi") (newline)) "cat")
	     (current-output-port))
  (port-copy (pipe (list path: "base64" arguments: '("/bin/tcsh")
			 stderr-redirection: #t)
		   (list path: "head" arguments: '("-n" "200"))
		   (list path: "tr" arguments: '("A-Z" "a-z"))
		   (list path: "tr" arguments: '("-c" "b-z" "\n"))
		   (list path: "awk" arguments: '("NF"))
		   "sort"
		   (list path: "uniq" arguments: '("-c"))
		   (list path: "sort" arguments: '("-n" "-r"))
		   (list path: "head" arguments: '("-10")))
	     (current-output-port))
  (port-copy (pipe (lambda () (display "bye") (newline)) "rev" "rev")
	     (current-output-port))
  (write (read (pipe (lambda () (display '(the end))) "cat")))
  (newline))

(go)




More information about the Gambit-list mailing list