In shell it is easy to run a program, and it is easy to chain programs into a pipeline where each process writes to the following process.
In Gambit-C it appears to be easy to run a program and write to the process and read from the process. I failed to find a way to specify that the output of one process should be connected to the input of another. I have found open-process, process-status. I have found some discussions about running single processes.
How does one implement the following lines in Gambit-C?
$ sort infile | uniq -c | sort -nr $ sort < infile | uniq -c | sort -nr > outfile
Is this non-trivial? I'm looking for something like the following, so that those processes would wait for their input and proceed when it becomes available.
(let ((p (pipe '(path: "sort" arguments: ("infile")) '(path: "uniq" arguments: ("-c")) '(path: "sort" arguments: ("-nr"))))) (let* ((r1 (read-line p)) ...) ...))
(parameterize ((current-input-port (open-input-file "infile")) (current-output-port (open-output-file "outfile"))) (process-status (pipe "sort" '(path: "uniq" arguments: ("-c")) ...)))
I'm _not_ looking for (string-append ...).
Afficher les réponses par date
On 2013-01-27, at 4:48 PM, Jussi Piitulainen jpiitula@ling.helsinki.fi wrote:
In shell it is easy to run a program, and it is easy to chain programs into a pipeline where each process writes to the following process.
In Gambit-C it appears to be easy to run a program and write to the process and read from the process. I failed to find a way to specify that the output of one process should be connected to the input of another. I have found open-process, process-status. I have found some discussions about running single processes.
How does one implement the following lines in Gambit-C?
$ sort infile | uniq -c | sort -nr $ sort < infile | uniq -c | sort -nr > outfile
Is this non-trivial? I'm looking for something like the following, so that those processes would wait for their input and proceed when it becomes available.
(let ((p (pipe '(path: "sort" arguments: ("infile")) '(path: "uniq" arguments: ("-c")) '(path: "sort" arguments: ("-nr"))))) (let* ((r1 (read-line p)) ...) ...))
(parameterize ((current-input-port (open-input-file "infile")) (current-output-port (open-output-file "outfile"))) (process-status (pipe "sort" '(path: "uniq" arguments: ("-c")) ...)))
I'm _not_ looking for (string-append ...).
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.
Marc
(define (port-copy in-port out-port) (let ((buf (make-u8vector 4096))) (let loop () (let ((n (read-subu8vector buf 0 (u8vector-length buf) in-port 1))) ;;(pp n) (if (= n 0) (close-output-port out-port) (begin (write-subu8vector buf 0 n out-port) (loop)))))))
(define (port-copy-in-tread in-port out-port) (thread-start! (make-thread (lambda () (port-copy in-port out-port)))))
(define (process args) (open-process (append args '(stderr-redirection: #t))))
(define (pipe-processes processes) (let ((in-out-port (car processes))) (values in-out-port (if (null? (cdr processes)) in-out-port (receive (out-port in-port) (pipe-processes (cdr processes)) (port-copy-in-tread in-out-port out-port) in-port)))))
(define (pipe . processes) (pipe-processes processes))
(define (go) (receive (out-port in-port) (pipe (process (list path: "du" arguments: '())) (process (list path: "sort" arguments: '("-n" "-r"))) (process (list path: "head" arguments: '("-10")))) (port-copy in-port (current-output-port))))
(go)
Guys,
Why not using the "pipe" system call ?
Regards
On 2013-01-28, at 4:28 PM, Marc Feeley feeley@iro.umontreal.ca wrote:
On 2013-01-27, at 4:48 PM, Jussi Piitulainen jpiitula@ling.helsinki.fi wrote:
In shell it is easy to run a program, and it is easy to chain programs into a pipeline where each process writes to the following process.
In Gambit-C it appears to be easy to run a program and write to the process and read from the process. I failed to find a way to specify that the output of one process should be connected to the input of another. I have found open-process, process-status. I have found some discussions about running single processes.
How does one implement the following lines in Gambit-C?
$ sort infile | uniq -c | sort -nr $ sort < infile | uniq -c | sort -nr > outfile
Is this non-trivial? I'm looking for something like the following, so that those processes would wait for their input and proceed when it becomes available.
(let ((p (pipe '(path: "sort" arguments: ("infile")) '(path: "uniq" arguments: ("-c")) '(path: "sort" arguments: ("-nr"))))) (let* ((r1 (read-line p)) ...) ...))
(parameterize ((current-input-port (open-input-file "infile")) (current-output-port (open-output-file "outfile"))) (process-status (pipe "sort" '(path: "uniq" arguments: ("-c")) ...)))
I'm _not_ looking for (string-append ...).
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.
Marc
(define (port-copy in-port out-port) (let ((buf (make-u8vector 4096))) (let loop () (let ((n (read-subu8vector buf 0 (u8vector-length buf) in-port 1))) ;;(pp n) (if (= n 0) (close-output-port out-port) (begin (write-subu8vector buf 0 n out-port) (loop)))))))
(define (port-copy-in-tread in-port out-port) (thread-start! (make-thread (lambda () (port-copy in-port out-port)))))
(define (process args) (open-process (append args '(stderr-redirection: #t))))
(define (pipe-processes processes) (let ((in-out-port (car processes))) (values in-out-port (if (null? (cdr processes)) in-out-port (receive (out-port in-port) (pipe-processes (cdr processes)) (port-copy-in-tread in-out-port out-port) in-port)))))
(define (pipe . processes) (pipe-processes processes))
(define (go) (receive (out-port in-port) (pipe (process (list path: "du" arguments: '())) (process (list path: "sort" arguments: '("-n" "-r"))) (process (list path: "head" arguments: '("-10")))) (port-copy in-port (current-output-port))))
(go)
Gambit-list mailing list Gambit-list@iro.umontreal.ca https://webmail.iro.umontreal.ca/mailman/listinfo/gambit-list
On 2013-01-28, at 4:57 PM, Richard Prescott rdprescott@gmail.com wrote:
Guys,
Why not using the "pipe" system call ?
You would need to combine this with fork and write quite a bit of C code. Also, the pipe and fork system calls are limited to POSIX systems (you would have to adapt the code substantially for the Windows version).
I think Jussi is interested in a pure Gambit Scheme solution using threads, primarily to explore how this can be done with Gambit. It has the advantage of being portable to all the platforms Gambit runs on.
Marc
Richard Prescott writes:
Guys,
Why not using the "pipe" system call ?
Gambit doesn't seem to expose such system calls. One would need to go down to the FFI level. The Gambit level solution seems adequate to me at the moment.
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@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)
On 2013-01-29, at 8:51 AM, Jussi Piitulainen jpiitula@ling.helsinki.fi wrote:
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.
The word "pipe" in that passage does not refer to the operating system pipes that you get with a system call to "pipe". It refers to a conceptual pipe, i.e. a pair of Scheme ports where date written to one port can be read from the other port. The functionnality is completely implemented within the Scheme process and does not rely on any special feature of the OS. It is particularly interesting for asynchronous interprocess communication. For example:
(receive (port1 port2) (open-u8vector-pipe '(direction: output))
(thread-start! ;; this will be thread #2 (make-thread (lambda () (let loop () (let ((data (read port2))) (if (not (eof-object? data)) (begin (pp (list (current-thread) 'received data)) (loop))))))))
(let ((send (lambda (data) (write data port1) (force-output port1) (pp (list (current-thread) 'sent data)))))
(send '(11 22 33)) (thread-sleep! 1)
(send '(44 55 66)) (thread-sleep! 1)))
;; output: ;; ;; (#<thread #1 primordial> sent (11 22 33)) ;; (#<thread #2> received (11 22 33)) ;; (#<thread #1 primordial> sent (44 55 66)) ;; (#<thread #2> received (44 55 66))
If you want to create an actual OS pipe, you can always do it through the filesystem by executing "mkfifo". For example, you will get the same result as the above code if you replace the first line with:
(shell-command "mkfifo mypipe")
(let* ((port2 (open-input-file "mypipe")) (port1 (open-output-file "mypipe")))
... )
Marc