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

Marc Feeley feeley at iro.umontreal.ca
Tue Jan 29 09:59:26 EST 2013


On 2013-01-29, at 8:51 AM, Jussi Piitulainen <jpiitula at 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






More information about the Gambit-list mailing list