From 59b2c698ea32397e121b6e1cfa13896aa1071fab Mon Sep 17 00:00:00 2001 From: Hans Dijkema Date: Mon, 8 Jun 2026 09:50:11 +0200 Subject: [PATCH] Port implementation over racket ports. --- .gitignore | 5 + README.md | 12 ++- info.rkt | 11 ++ main.rkt | 179 +++++++++++++++++++++++++++++++++ scribblings/port-channel.scrbl | 145 ++++++++++++++++++++++++++ test/main.rkt | 25 +++++ 6 files changed, 376 insertions(+), 1 deletion(-) create mode 100644 info.rkt create mode 100644 main.rkt create mode 100644 scribblings/port-channel.scrbl create mode 100644 test/main.rkt diff --git a/.gitignore b/.gitignore index 39a4f9c..7cc4715 100644 --- a/.gitignore +++ b/.gitignore @@ -15,3 +15,8 @@ compiled/ # Dependency tracking files *.dep +/scribblings/*.css +/scribblings/*.js +/*.bak +/scribblings/*.bak +/scribblings/*.html diff --git a/README.md b/README.md index da53ea0..61fc86e 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,13 @@ # port-channel -Channels over ports for racket. \ No newline at end of file +`port-channel` wraps a Racket input or output port as an asynchronous channel-like object. + +Input port-channels run a reader thread that reads serialized Racket values from the port, +deserializes complete values and publishes them on an internal async channel. The +`port-channel` itself is a synchronizable event, so `(sync pc)` returns the next complete value. + +Output port-channels expose `port-channel-put`, which queues a value immediately. A writer thread +serializes and flushes queued values to the underlying output port. + +Values are encoded with `racket/serialize`. Custom structs must be declared with +`serializable-struct`. diff --git a/info.rkt b/info.rkt new file mode 100644 index 0000000..a4da3c7 --- /dev/null +++ b/info.rkt @@ -0,0 +1,11 @@ +#lang info + +(define collection "port-channel") +(define pkg-desc "Asynchronous channels over Racket ports using racket/serialize.") +(define version "0.1") +(define pkg-authors '(hnmdijkema)) +(define scribblings '(("scribblings/port-channel.scrbl" ()))) + +(define deps '("base")) +(define build-deps '("racket-doc" "scribble-lib")) + diff --git a/main.rkt b/main.rkt new file mode 100644 index 0000000..d4c5dcf --- /dev/null +++ b/main.rkt @@ -0,0 +1,179 @@ +#lang racket/base + +(require racket/async-channel + racket/serialize) + +(provide port-channel? + port-channel-direction + port-channel-error? + port-channel-error-source + port-channel-error-message + + make-port-channel + port-channel-put + port-channel-get + port-channel-try-get + port-channel-evt + + port-channel-error-get + port-channel-error-try-get + port-channel-error-evt + + close-port-channel + port-channel-wait) + +(struct port-channel-error (source message) #:transparent) + +(struct port-channel + (direction port channel errors thread close-token closing?) + #:property prop:evt + (lambda (pc) + (case (port-channel-direction pc) + [(input) (port-channel-channel pc)] + [else never-evt]))) + +(define failed-token (gensym 'port-channel-failed)) + +(define (put-error! source errors e) + (async-channel-put errors (port-channel-error source (exn-message e)))) + +(define (read-serialized in errors source closing?) + (define serialized + (with-handlers ([exn:fail? + (lambda (e) + (unless (unbox closing?) (put-error! source errors e)) + eof)]) + (read in))) + (cond + [(eof-object? serialized) eof] + [else + (with-handlers ([exn:fail? + (lambda (e) + (put-error! source errors e) + failed-token)]) + (deserialize serialized))])) + +(define (write-serialized out v errors source) + (define serialized + (with-handlers ([exn:fail? + (lambda (e) + (put-error! source errors e) + failed-token)]) + (serialize v))) + (cond + [(eq? serialized failed-token) #t] + [else + (with-handlers ([exn:fail? + (lambda (e) + (put-error! source errors e) + #f)]) + (write serialized out) + (newline out) + (flush-output out) + #t)])) + +(define (make-input-port-channel in source close?) + (define ch (make-async-channel)) + (define errors (make-async-channel)) + (define closing? (box #f)) + (define close-token (gensym 'close-input-port-channel)) + (define reader + (thread + (lambda () + (let loop () + (define v (read-serialized in errors source closing?)) + (cond + [(eof-object? v) + (when close? (with-handlers ([exn:fail? void]) (close-input-port in))) + (async-channel-put ch eof)] + [(eq? v failed-token) (loop)] + [else + (async-channel-put ch v) + (loop)]))))) + (port-channel 'input in ch errors reader close-token closing?)) + +(define (make-output-port-channel out source close?) + (define ch (make-async-channel)) + (define errors (make-async-channel)) + (define closing? (box #f)) + (define close-token (gensym 'close-output-port-channel)) + (define writer + (thread + (lambda () + (let loop () + (define v (async-channel-get ch)) + (cond + [(eq? v close-token) + (set-box! closing? #t) + (when close? (with-handlers ([exn:fail? void]) (close-output-port out)))] + [else + (define ok? (write-serialized out v errors source)) + (when ok? (loop))]))))) + (port-channel 'output out ch errors writer close-token closing?)) + +(define (make-port-channel port + #:direction [direction 'auto] + #:source [source 'port] + #:close? [close? #t]) + (define in? (input-port? port)) + (define out? (output-port? port)) + (case direction + [(input) + (unless in? (raise-argument-error 'make-port-channel "input-port?" port)) + (make-input-port-channel port source close?)] + [(output) + (unless out? (raise-argument-error 'make-port-channel "output-port?" port)) + (make-output-port-channel port source close?)] + [(auto) + (cond + [(and in? (not out?)) (make-input-port-channel port source close?)] + [(and out? (not in?)) (make-output-port-channel port source close?)] + [(and in? out?) + (raise-arguments-error + 'make-port-channel + "port is both input and output; use #:direction 'input or #:direction 'output" + "port" port)] + [else (raise-argument-error 'make-port-channel "port?" port)])] + [else + (raise-argument-error 'make-port-channel "(or/c 'auto 'input 'output)" direction)])) + +(define (port-channel-put pc v) + (unless (eq? (port-channel-direction pc) 'output) + (raise-argument-error 'port-channel-put "output port-channel?" pc)) + (async-channel-put (port-channel-channel pc) v)) + +(define (port-channel-get pc) + (unless (eq? (port-channel-direction pc) 'input) + (raise-argument-error 'port-channel-get "input port-channel?" pc)) + (sync pc)) + +(define (port-channel-try-get pc) + (unless (eq? (port-channel-direction pc) 'input) + (raise-argument-error 'port-channel-try-get "input port-channel?" pc)) + (async-channel-try-get (port-channel-channel pc))) + +(define (port-channel-evt pc) + (unless (eq? (port-channel-direction pc) 'input) + (raise-argument-error 'port-channel-evt "input port-channel?" pc)) + pc) + +(define (port-channel-error-get pc) + (async-channel-get (port-channel-errors pc))) + +(define (port-channel-error-try-get pc) + (async-channel-try-get (port-channel-errors pc))) + +(define (port-channel-error-evt pc) + (port-channel-errors pc)) + +(define (close-port-channel pc) + (set-box! (port-channel-closing? pc) #t) + (case (port-channel-direction pc) + [(input) + (with-handlers ([exn:fail? void]) (close-input-port (port-channel-port pc)))] + [(output) + (async-channel-put (port-channel-channel pc) (port-channel-close-token pc))] + [else (void)])) + +(define (port-channel-wait pc) + (thread-wait (port-channel-thread pc))) diff --git a/scribblings/port-channel.scrbl b/scribblings/port-channel.scrbl new file mode 100644 index 0000000..3d955cb --- /dev/null +++ b/scribblings/port-channel.scrbl @@ -0,0 +1,145 @@ +#lang scribble/manual + +@(require (for-label racket/base + racket/contract + racket/serialize + "../main.rkt")) + +@title{port-channel} +@author[@author+email["Hans Dijkema" "hans@dijkewijk.nl"]] + +@defmodule[port-channel] + +The @racketmodname[port-channel] module wraps a Racket port as an asynchronous, +channel-like object. It is intended as a small building block for stream-based +communication protocols, for example between a parent Racket process and a +subprocess connected through standard input and standard output. + +A port-channel transports serialized Racket values. The implementation uses +@racketmodname[racket/serialize]: values are passed through @racket[serialize], +written to the port, read back as Racket data and reconstructed with +@racket[deserialize]. Ordinary data values work directly. Custom structures must +be defined with @racket[serializable-struct]. Ports, threads, custodians and +ordinary procedures are not serializable values. + +@section{Model} + +A port-channel has one of two directions: + +@itemlist[ + @item{@racket['input]: a reader thread reads complete serialized values from an + input port and puts the deserialized values on an internal asynchronous + channel. The port-channel is itself a synchronizable event, so + @racket[(sync pc)] returns the next complete value.} + @item{@racket['output]: @racket[port-channel-put] queues values on an internal + asynchronous channel. A writer thread serializes and writes queued values + to the output port.}] + +Buffered I/O alone is not enough to provide this behaviour. Synchronizing on a +raw input port only says that some input is available; it does not guarantee that +a complete serialized value is ready. The reader thread converts the byte stream +into complete channel messages. + +@section{Reference} + +@defproc[(make-port-channel [port port?] + [#:direction direction (or/c 'auto 'input 'output) 'auto] + [#:source source any/c 'port] + [#:close? close? any/c #t]) + port-channel?]{ +Creates a port-channel for @racket[port]. With @racket['auto], the direction is +inferred from the port. An input-only port becomes an input port-channel. An +output-only port becomes an output port-channel. If the port is both an input and +an output port, @racket[#:direction] must be supplied explicitly. + +The @racket[source] value is stored in reported @racket[port-channel-error] +values. When @racket[close?] is true, @racket[close-port-channel] closes the +underlying port.} + +@defproc[(port-channel? [v any/c]) boolean?]{ +Returns true when @racket[v] is a port-channel.} + +@defproc[(port-channel-direction [pc port-channel?]) (or/c 'input 'output)]{ +Returns the direction of @racket[pc].} + +@defproc[(port-channel-put [pc port-channel?] [v any/c]) void?]{ +Queues @racket[v] for writing to an output port-channel. The call only enqueues +the value; serialization and port I/O are performed by the writer thread. + +The value must be serializable by @racket[serialize]. If serialization or writing +fails, a @racket[port-channel-error] is published on the error channel.} + +@defproc[(port-channel-get [pc port-channel?]) any/c]{ +Returns the next value from an input port-channel. This is equivalent to +@racket[(sync pc)]. When the underlying input stream reaches end-of-file, +@racket[eof] is published as the final value.} + +@defproc[(port-channel-try-get [pc port-channel?]) any/c]{ +Attempts to get a value from an input port-channel without blocking. Returns +@racket[#f] when no value is currently available.} + +@defproc[(port-channel-evt [pc port-channel?]) evt?]{ +Returns @racket[pc] as a synchronizable event. This is mainly a convenience for +code that wants an explicit event-producing function.} + +@defproc[(close-port-channel [pc port-channel?]) void?]{ +Closes the port-channel. For an input port-channel, this closes the input port. +For an output port-channel, a close marker is queued, so values already queued +before the close marker are written first.} + +@defproc[(port-channel-wait [pc port-channel?]) void?]{ +Waits until the reader or writer thread of @racket[pc] has terminated.} + +@section{Errors} + +@defstruct*[port-channel-error ([source any/c] + [message string?])]{ +Represents an error reported by the reader or writer thread.} + +@defproc[(port-channel-error-get [pc port-channel?]) port-channel-error?]{ +Blocks until the next error is available.} + +@defproc[(port-channel-error-try-get [pc port-channel?]) any/c]{ +Attempts to get the next error without blocking. Returns @racket[#f] when no +error is currently available.} + +@defproc[(port-channel-error-evt [pc port-channel?]) evt?]{ +Returns the error channel as a synchronizable event.} + +@section{Example} + +@racketblock[ +(require port-channel) + +(define-values (in out) (make-pipe)) + +(define reader (make-port-channel in)) +(define writer (make-port-channel out)) + +(port-channel-put writer '(hello 1 2 3)) +(sync reader) +] + +The result is: + +@racketblock[ +'(hello 1 2 3) +] + +@section{Serializable structures} + +@racketblock[ +(require racket/serialize + port-channel) + +(serializable-struct message (id payload) #:transparent) + +(define-values (in out) (make-pipe)) +(define reader (make-port-channel in)) +(define writer (make-port-channel out)) + +(port-channel-put writer (message 1 '(a b c))) +(sync reader) +] + +The received value is a reconstructed @racket[message] structure. diff --git a/test/main.rkt b/test/main.rkt new file mode 100644 index 0000000..33d7eab --- /dev/null +++ b/test/main.rkt @@ -0,0 +1,25 @@ +#lang racket/base + +(require rackunit + racket/serialize + port-channel) + +(serializable-struct msg (id payload) #:transparent) + +(define-values (in out) (make-pipe)) +(define reader (make-port-channel in #:source 'test-reader)) +(define writer (make-port-channel out #:source 'test-writer)) + +(port-channel-put writer '(hello 1 2 3)) +(port-channel-put writer (hasheq 'a 10 'b 20)) +(port-channel-put writer (msg 7 '(a b c))) + +(check-equal? (sync reader) '(hello 1 2 3)) +(check-equal? (sync reader) (hasheq 'a 10 'b 20)) +(check-equal? (sync reader) (msg 7 '(a b c))) + +(close-port-channel writer) +(check-true (eof-object? (sync reader))) + +(module+ main + (displayln "port-channel tests ok"))