Stream command processing (long message)

[John D. Ramsdell <linus!ramsdell@mitre-bedford.ARPA>]

While I am an avid fan of Scheme, I find myself frustrated by how
difficult it is to program some stream processing tasks in Scheme
compared with the difficulty of programming the same tasks in UNIX.
Your typical UNIX hacker can whip together some concoction of awk, grep,
sed, yacc, etc. to solve these kinds of problems very quickly.  But it
is quite a pain in Scheme.  Even more painful is the fact that it is
very hard to use someone else's stream processing programs.

I would like to propose having two standard top level environments for
Scheme.  One is the one we all know and love, and the other would be the
stream command processor (SCP).  This top level environment would map a
stream like syntax into a simple composition of Scheme functions.  The
map would follow a simple convention, so that inexperienced users could
easily modify or creat new functions to be used in the composition.

I have no suggestion as to the particular syntax to be used; I will use
a UNIX like syntax with no implied endorsement.  I will give an example
of a simple map from this syntax to compositions of Scheme functions
using continuation streams (CSTREAMS).  The continuation is not the same
one you get when you use call-with-current-continuation.  Cstreams are
defined by:

; Continuation streams are procedures of one argument
; which is the procedure's continuation.
; A continuation is a procedure of two arguments,
; a stream element, and a the next stream.

; cstream = (lambda (continuation)
;               Compute element and next-cstream.
;               (continuation element next-cstream))

A port is turned into a cstream with:

(define (read-cstream port)
 (lambda (c)
  (c (read port) (read-cstream port))))

and a cstream is turned into something that works as an argument to
call-with-output-port using:

(define (print-cstream cstream)
 (lambda (port)
  (letrec
   ((continuation
     (lambda (element cstream)
      (if (ecse? element)
          'print-done
          (begin
           (print element port)
           (cstream continuation))))))
   (cstream continuation))))

Cstream manipulation functions have the signature:

(lambda (input-cstream . arguments) ...)  ==> output-cstream.

For example, the cstream mapping function looks like:

; => a cstream that results from mapping elements of cstream
; with a procedure of one argument.
(define (map-cstream cstream procedure)
 (letrec
  ((next-cstream
    (lambda (cstream)
     (lambda (c)
      (cstream
       (lambda (element cstream)
        (if (ecse? element)
            (ecs c)
            (c (procedure element) (next-cstream cstream)))))))))
  (next-cstream cstream)))


The SCP would translate a command that looks like

map-cstream list <lists.scm >more-parens.scm

into the expression

(call-with-output-file
    "more-parens.scm"
    (map-cstream
        (call-with-input-file
            "lists.scm"
            read-cstream)
        list))

For a more complex example, SCP would translate the following:

filter-cstream symbol? <lists.scm | map-cstream list >single-level-list.scm

into

(call-with-output-file
    "single-level-list.scm"
    (filter-cstream
        (map-cstream
            (call-with-input-file
                "lists.scm"
                read-cstream)
            list)
        symbol?))

The set of stream functions are easily extended by writting functions that
expect a cstream for their first argument and return a cstream.

I am not particularly attached to my representation of streams, and the
streams in Sussman and Abelson's book would also do the trick.  I am
most interested in an commonly aggreed upon stream processing system
that encourages the interchange of stream functions, just as the Revised
Revised Report on Scheme encourages interchange of Scheme functions.
Then Scheme hackers will definitely be kings of stream processing.

John

Here is some code to play with:
-----------
;;; -*- Mode: LISP -*-

; Continuation Streams.

; Continuation streams are procedures of one argument
; which is the procedure's continuation.
; A continuation is a procedure of two arguments,
; a stream element, and a the next stream.

; cstream = (lambda (continuation)
;               Compute element and next-cstream.
;               (continuation element next-cstream))

(define (error-cstream message)
 (lambda (c)
  c                                             ; ignore continuation.
  (error message)))

(define *ecse*					; Used to identify the empty continuation stream.
 '*ecse*)

; Is the cstream element the empty cstream element?
(define (ecse? x)
 (eq? x *ecse*))

(define past-end-of-cstream
 (error-cstream "Read past end of cstream"))

; the empty continuation stream.
(define ecs
 (lambda (c) (c *ecse* past-end-of-cstream)))

(define (ecs? cstream)
 (cstream 
  (lambda (element cstream)
   cstream                                      ; ignore cstream.
   (ecse? element))))

; => cstream which is cstream0 appended to cstream1.
(define (append-cstreams cstream0 cstream1)
 (letrec
  ((next-cstream
    (lambda (cstream)
     (lambda (c)
      (cstream
       (lambda (element cstream)
        (if (ecse? element)
            (cstream1 c)
            (c element (next-cstream cstream)))))))))
  (next-cstream cstream0)))

; => cstream which is a fair merge of cstream0 and cstream1.
(define (interleave-cstreams cstream0 cstream1)
 (lambda (c)
  (cstream0
   (lambda (element cstream)
    (if (ecse? element)
        (cstream1 c)
        (c element (interleave-cstreams cstream1 cstream)))))))

; => a cstream that results from mapping elements of cstream
; with a procedure of one argument.
(define (map-cstream cstream procedure)
 (letrec
  ((next-cstream
    (lambda (cstream)
     (lambda (c)
      (cstream
       (lambda (element cstream)
        (if (ecse? element)
            (ecs c)
            (c (procedure element) (next-cstream cstream)))))))))
  (next-cstream cstream)))

; Returns a cstream containing elements that satisfy the predicate.
(define (filter-cstream cstream predicate)
 (letrec
  ((next-cstream
    (lambda (cstream)
     (lambda (c)
      (cstream
       (lambda (element cstream)
        (if (ecse? element)
            (ecs c)
            (let ((next-cstream (next-cstream cstream)))
             (if (predicate element)
                 (c element next-cstream)
                 (next-cstream c))))))))))
  (next-cstream cstream)))

(define (list->cstream l)			; => cstream with elements in l.
 (if (null? l)
     ecs
     (let ((element (car l))
           (l (cdr l)))
      (lambda (c) (c element (list->cstream l))))))

(define (cstream->list cstream)			; => list of elements in the cstream.
 (let ((l '()))                                 ; The reverse of the result is held here.
  (letrec
   ((continuation
     (lambda (element cstream)
      (if (not (ecse? element))
          (begin
           (set! l (cons element l))
           (cstream continuation))))))
   (cstream continuation))
  (reverse! l)))

(define (read-cstream port)
 (lambda (c)
  (c (read port) (read-cstream port))))

(define (print-cstream cstream)
 (lambda (port)
  (letrec
   ((continuation
     (lambda (element cstream)
      (if (ecse? element)
          'print-done
          (begin
           (print element port)
           (cstream continuation))))))
   (cstream continuation))))

; For more, see J. D. Ramsdell "An Implementation of a Domain Calculus
; Query Language Using Continuation Streams",  M84-56, MITRE Corp.
; Bedford, MA, December 1984.