suzanne.soy/racket
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

io: convert pipe to object style

Browse Source
This commit is contained in:
Matthew Flatt 2019-02-11 20:31:09 -07:00
parent 78136c0613
commit 35ceb8e3b3
3 changed files with 624 additions and 558 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
pkgs/racket-test-core/tests/racket/file.rktl
View File

@ -583,16 +583,16 @@
(let ([th1 (thread (lambda () (let ([th1 (thread (lambda ()
(display "a" out)))] (display "a" out)))]
[th2 (thread (lambda () [th2 (thread (lambda ()
(display "a" out)))] (display "a" out)))]
[th3 (thread (lambda () [th3 (thread (lambda ()
(display "a" out)))]) (display "a" out)))])
(test #t thread-running? th1) (test #t thread-running? th1)
(test #t thread-running? th2) (test #t thread-running? th2)
(test #t thread-running? th3) (test #t thread-running? th3)
(test 49 read-byte in) (test 49 read-byte in)
(sleep 0.1) (sync (system-idle-evt))
(test 2 + (test 2 +
(if (thread-running? th1) 1 0) (if (thread-running? th1) 1 0)
@ -601,7 +601,7 @@
(test 50 read-byte in) (test 50 read-byte in)
(sleep 0.1) (sync (system-idle-evt))
(test 1 + (test 1 +
(if (thread-running? th1) 1 0) (if (thread-running? th1) 1 0)
@ -610,7 +610,7 @@
(test 51 read-byte in) (test 51 read-byte in)
(sleep 0.1) (sync (system-idle-evt))
(test #f thread-running? th1) (test #f thread-running? th1)
(test #f thread-running? th2) (test #f thread-running? th2)

329
racket/src/io/common/class.rkt
View File

@ -14,6 +14,7 @@
;; | (public [<method-id> <method>] ...) ;; | (public [<method-id> <method>] ...)
;; | (private [<method-id> <method>] ...) ;; | (private [<method-id> <method>] ...)
;; | (override [<method-id> <method>] ...) ;; | (override [<method-id> <method>] ...)
;; | (static [<method-id> <method>] ...) ; not in vtable
;; | (property [<property-expr> <val-expr>] ...) ;; | (property [<property-expr> <val-expr>] ...)
;; <method> = #f ;; <method> = #f
;; | (lambda (<id> ...) <expr> ...+) ;; | (lambda (<id> ...) <expr> ...+)
@ -39,24 +40,31 @@
;; Use ;; Use
;; (send <class-id> <obj-expr> <method-id> <arg-expr> ...) ;; (send <class-id> <obj-expr> <method-id> <arg-expr> ...)
;; to call a method, or ;; to call a method, or
;; (mewthod <class-id> <obj-expr> <method-id>) ;; (method <class-id> <obj-expr> <method-id>)
;; to get a method that expects the object as its first argument. ;; to get a method that expects the object as its first argument.
;; ;;
;; In a method, fields can be accessed directly by name, and `this` is ;; In a method, `field`s, `private`s, and `static`s can be accessed
;; bound to the current object. ;; directly by name, and `this` is bound to the current object. A
;; method overridden in `new` can only access `field`s.
;;
;; Use
;; (with-object <class-id> <object-expr)
;; <body> ...+)
;; to directly reference `field`s and `static`s in the <body>s.
(provide class (provide class
this this
new new
send send
method) method
with-object)
(define-syntax-parameter this (define-syntax-parameter this
(lambda (stx) (lambda (stx)
(raise-syntax-error #f "illegal use outside of a method" stx))) (raise-syntax-error #f "illegal use outside of a method" stx)))
(begin-for-syntax (begin-for-syntax
(struct class-info (struct-info methods-id vtable-id vtable-accessor-id fields methods) (struct class-info (struct-info methods-id vtable-id vtable-accessor-id fields methods statics)
#:property prop:struct-info (lambda (ci) #:property prop:struct-info (lambda (ci)
(class-info-struct-info ci)))) (class-info-struct-info ci))))
@ -85,19 +93,21 @@
[(id expr) [(id expr)
(list #'id #'expr (combine-ids base-id base-id "-" #'id) (combine-ids #'id "set-" base-id "-" #'id "!"))] (list #'id #'expr (combine-ids base-id base-id "-" #'id) (combine-ids #'id "set-" base-id "-" #'id "!"))]
[_ (raise-syntax-error #f (format "bad ~a clause" what) stx e)]))) [_ (raise-syntax-error #f (format "bad ~a clause" what) stx e)])))
(define-values (new-fields new-methods override-methods locals properties) (define-values (new-fields new-methods override-methods locals statics properties)
(let ([l-stx (syntax-case stx () (let ([l-stx (syntax-case stx ()
[(_ _ #:extends _ . rest) #'rest] [(_ _ #:extends _ . rest) #'rest]
[(_ _ . rest) #'rest])]) [(_ _ . rest) #'rest])])
(let loop ([l-stx l-stx] [new-fields null] [new-methods null] [override-methods null] [locals null] [properties null]) (let loop ([l-stx l-stx] [new-fields null] [new-methods null] [override-methods null]
(syntax-case l-stx (field public override private property) [locals null] [statics null] [properties null])
[() (values new-fields new-methods override-methods locals properties)] (syntax-case l-stx (field public override private static property)
[() (values new-fields new-methods override-methods locals statics properties)]
[((field fld ...) . rest) [((field fld ...) . rest)
(loop #'rest (loop #'rest
(add-procs id (syntax->list #'(fld ...)) "field" #:can-immutable? #t) (add-procs id (syntax->list #'(fld ...)) "field" #:can-immutable? #t)
new-methods new-methods
override-methods override-methods
locals locals
statics
properties)] properties)]
[((public method ...) . rest) [((public method ...) . rest)
(loop #'rest (loop #'rest
@ -105,13 +115,16 @@
(add-procs methods-id (syntax->list #'(method ...)) "public") (add-procs methods-id (syntax->list #'(method ...)) "public")
override-methods override-methods
locals locals
statics
properties)] properties)]
[((override method ...) . rest) [((override method ...) . rest)
(loop #'rest new-fields new-methods (syntax->list #'(method ...)) locals properties)] (loop #'rest new-fields new-methods (syntax->list #'(method ...)) locals statics properties)]
[((private method ...) . rest) [((private method ...) . rest)
(loop #'rest new-fields new-methods override-methods (syntax->list #'(method ...)) properties)] (loop #'rest new-fields new-methods override-methods (syntax->list #'(method ...)) statics properties)]
[((static method ...) . rest)
(loop #'rest new-fields new-methods override-methods locals (syntax->list #'(method ...)) properties)]
[((property prop ...) . rest) [((property prop ...) . rest)
(loop #'rest new-fields new-methods override-methods locals (syntax->list #'((#:property . prop) ...)))] (loop #'rest new-fields new-methods override-methods locals statics (syntax->list #'((#:property . prop) ...)))]
[(other . _) [(other . _)
(raise-syntax-error #f "unrecognized" stx #'other)])))) (raise-syntax-error #f "unrecognized" stx #'other)]))))
(define all-fields (if super-ci (define all-fields (if super-ci
@ -121,118 +134,118 @@
(syntax-case override () (syntax-case override ()
[(method-id _) (check-member stx #'method-id (if super-ci (class-info-methods super-ci) null) "method")] [(method-id _) (check-member stx #'method-id (if super-ci (class-info-methods super-ci) null) "method")]
[_ (raise-syntax-error #f "bad override clause" stx override)])) [_ (raise-syntax-error #f "bad override clause" stx override)]))
(with-syntax ([((field-id field-init-expr field-accessor-id field-mutator-maybe-id) ...) all-fields]) (with-syntax ([((field-id field-init-expr field-accessor-id field-mutator-maybe-id) ...) all-fields]
(define wrapped-new-methods [((local-id local-expr) ...) locals]
(for/list ([new-method (in-list new-methods)]) [((static-id static-expr) ...) statics]
(syntax-case new-method () [(local-tmp-id ...) (generate-temporaries locals)]
[(method-id method-init-expr . rest) [(static-tmp-id ...) (generate-temporaries statics)])
#'(method-id (let ([method-id (with-syntax ([local-bindings #'[([field-id field-accessor-id field-mutator-maybe-id] ...)
(bind-fields-in-body ([local-id local-tmp-id] ... [static-id static-tmp-id] ...)]])
([field-id field-accessor-id field-mutator-maybe-id] ...) (define wrapped-new-methods
method-init-expr)]) (for/list ([new-method (in-list new-methods)])
method-id) (syntax-case new-method ()
. rest)]))) [(method-id method-init-expr . rest)
(define all-methods/vtable (if super-ci #'(method-id (let ([method-id (bind-locals-in-body local-bindings method-init-expr)])
(append (for/list ([method (in-list (class-info-methods super-ci))]) method-id)
(syntax-case method () . rest)])))
[(method-id method-init-expr . rest) (define all-methods/vtable (if super-ci
(or (for/or ([override (in-list override-methods)]) (append (for/list ([method (in-list (class-info-methods super-ci))])
(syntax-case override () (syntax-case method ()
[(override-id override-init-expr . _) [(method-id method-init-expr . rest)
(and (eq? (syntax-e #'method-id) (syntax-e #'override-id)) (or (for/or ([override (in-list override-methods)])
(list* #'method-id (syntax-case override ()
#'(let ([method-id [(override-id override-init-expr . _)
(bind-fields-in-body (and (eq? (syntax-e #'method-id) (syntax-e #'override-id))
([field-id field-accessor-id field-mutator-maybe-id] ...) (list* #'method-id
override-init-expr)]) #'(let ([method-id
method-id) (bind-locals-in-body
#'rest))])) local-bindings
method)])) override-init-expr)])
wrapped-new-methods) method-id)
wrapped-new-methods)) #'rest))]))
(define vtable-id (combine-ids #'here id "-vtable")) method)]))
(define all-methods/next (for/list ([method (in-list all-methods/vtable)]) wrapped-new-methods)
(syntax-case method () wrapped-new-methods))
[(method-id method-init-expr method-accessor-id . _) (define vtable-id (combine-ids #'here id "-vtable"))
(with-syntax ([vtable-id vtable-id]) (define all-methods/next (for/list ([method (in-list all-methods/vtable)])
(list #'method-id (syntax-case method ()
#'(method-accessor-id vtable-id) [(method-id method-init-expr method-accessor-id . _)
#'method-accessor-id))]))) (with-syntax ([vtable-id vtable-id])
(with-syntax ([id id] (list #'method-id
[(super-ids ...) (if super-id #'(method-accessor-id vtable-id)
(list super-id) #'method-accessor-id))])))
null)] (with-syntax ([id id]
[quoted-super-id (and super-id #`(quote-syntax #,super-id))] [(super-ids ...) (if super-id
[(vtable-ids ...) (if super-id (list super-id)
null null)]
(list (datum->syntax id 'vtable)))] [quoted-super-id (and super-id #`(quote-syntax #,super-id))]
[vtable-accessor-id (if super-ci [(vtable-ids ...) (if super-id
(class-info-vtable-accessor-id super-ci) null
(combine-ids id id "-vtable"))] (list (datum->syntax id 'vtable)))]
[vtable-id vtable-id] [vtable-accessor-id (if super-ci
[struct:id (combine-ids id "struct:" id)] (class-info-vtable-accessor-id super-ci)
[make-id (combine-ids id "create-" id)] (combine-ids id id "-vtable"))]
[id? (combine-ids id id "?")] [vtable-id vtable-id]
[methods-id methods-id] [struct:id (combine-ids id "struct:" id)]
[(super-methods-ids ...) (if super-ci [make-id (combine-ids id "create-" id)]
(list (class-info-methods-id super-ci)) [id? (combine-ids id id "?")]
null)] [methods-id methods-id]
[(new-field-id/annotated ...) (for/list ([new-field (in-list new-fields)]) [(super-methods-ids ...) (if super-ci
(syntax-case new-field () (list (class-info-methods-id super-ci))
[(id _ _ #f) #'id] null)]
[(id . _) #'[id #:mutable]]))] [(new-field-id/annotated ...) (for/list ([new-field (in-list new-fields)])
[((new-method-id . _) ...) new-methods] (syntax-case new-field ()
[((_ _ rev-field-accessor-id . _) ...) (reverse all-fields)] [(id _ _ #f) #'id]
[((_ _ _ rev-field-mutator-maybe-id) ...) (reverse all-fields)] [(id . _) #'[id #:mutable]]))]
[((method-id method-init-expr/vtable . _) ...) all-methods/vtable] [((new-method-id . _) ...) new-methods]
[((_ method-init-expr/next method-accessor-id) ...) all-methods/next] [((_ _ rev-field-accessor-id . _) ...) (reverse all-fields)]
[((local-id local-expr) ...) locals] [((_ _ _ rev-field-mutator-maybe-id) ...) (reverse all-fields)]
[(local-tmp-id ...) (generate-temporaries locals)] [((method-id method-init-expr/vtable . _) ...) all-methods/vtable]
[((propss ...) ...) properties]) [((_ method-init-expr/next method-accessor-id) ...) all-methods/next]
#`(begin [((propss ...) ...) properties])
(struct id super-ids ... (vtable-ids ... new-field-id/annotated ...) #`(begin
#:omit-define-syntaxes (struct id super-ids ... (vtable-ids ... new-field-id/annotated ...)
#:constructor-name make-id #:omit-define-syntaxes
#:authentic #:constructor-name make-id
propss ... ...) #:authentic
(struct methods-id super-methods-ids ... (new-method-id ...)) propss ... ...)
(define vtable-id (methods-id method-init-expr/vtable ...)) (struct methods-id super-methods-ids ... (new-method-id ...))
(begin (define vtable-id (methods-id method-init-expr/vtable ...))
(define local-tmp-id (let ([local-id (define static-tmp-id (let ([static-id (bind-locals-in-body local-bindings static-expr)])
(bind-fields-in-body ([field-id field-accessor-id field-mutator-maybe-id] ...) static-id))
local-expr)]) ...
(define local-tmp-id (let ([local-id (bind-locals-in-body local-bindings local-expr)])
local-id)) local-id))
(define-syntax (local-id stx) ...
(syntax-case stx () (define-syntax id
[(_ arg (... ...)) (class-info (list (quote-syntax struct:id)
(with-syntax ([this-id (datum->syntax #'here 'this stx)]) (quote-syntax make-id)
(syntax/loc stx (local-tmp-id this-id arg (... ...))))]))) (quote-syntax id?)
... (list (quote-syntax rev-field-accessor-id) ... (quote-syntax vtable-accessor-id))
(define-syntax id (list (maybe-quote-syntax rev-field-mutator-maybe-id) ... #f)
(class-info (list (quote-syntax struct:id) quoted-super-id)
(quote-syntax make-id) (quote-syntax methods-id)
(quote-syntax id?) (quote-syntax vtable-id)
(list (quote-syntax rev-field-accessor-id) ... (quote-syntax vtable-accessor-id)) (quote-syntax vtable-accessor-id)
(list (maybe-quote-syntax rev-field-mutator-maybe-id) ... #f) (list (list (quote-syntax field-id) (quote-syntax field-init-expr)
quoted-super-id) (quote-syntax field-accessor-id) (maybe-quote-syntax field-mutator-maybe-id))
(quote-syntax methods-id) ...)
(quote-syntax vtable-id) (list (list (quote-syntax method-id) (quote-syntax method-init-expr/next)
(quote-syntax vtable-accessor-id) (quote-syntax method-accessor-id))
(list (list (quote-syntax field-id) (quote-syntax field-init-expr) ...)
(quote-syntax field-accessor-id) (maybe-quote-syntax field-mutator-maybe-id)) (list (list (quote-syntax static-id) (quote-syntax static-tmp-id))
...) ...))))))))
(list (list (quote-syntax method-id) (quote-syntax method-init-expr/next)
(quote-syntax method-accessor-id))
...)))))))
(define-syntax (bind-fields-in-body stx) (define-syntax (bind-locals-in-body stx)
(syntax-case stx (lambda case-lambda) (syntax-case stx (lambda case-lambda)
[(_ fields #f) #'#f] [(_ locals #f) #'#f]
[(_ fields (form . rest)) [(_ locals (form . rest))
#'(bind-fields-in-body fields form (form . rest))] (with-syntax ([(_ _ orig) stx])
[(_ fields ctx (lambda (arg ...) body0 body ...)) #'(bind-locals-in-body locals form orig))]
#'(bind-fields-in-body fields ctx (case-lambda [(arg ...) body0 body ...]))] [(_ locals expr) #'expr]
[(_ fields ctx (case-lambda clause ...)) [(_ locals ctx (lambda (arg ...) body0 body ...))
#'(bind-locals-in-body locals ctx (case-lambda [(arg ...) body0 body ...]))]
[(_ locals ctx (case-lambda clause ...))
(with-syntax ([(new-clause ...) (with-syntax ([(new-clause ...)
(for/list ([clause (in-list (syntax->list #'(clause ...)))]) (for/list ([clause (in-list (syntax->list #'(clause ...)))])
(syntax-case clause () (syntax-case clause ()
@ -240,21 +253,25 @@
(with-syntax ([(arg-tmp ...) (generate-temporaries #'(arg ...))]) (with-syntax ([(arg-tmp ...) (generate-temporaries #'(arg ...))])
#'[(this-id arg-tmp ...) #'[(this-id arg-tmp ...)
(syntax-parameterize ([this (make-rename-transformer #'this-id)]) (syntax-parameterize ([this (make-rename-transformer #'this-id)])
(bind-fields (bind-locals
fields locals
this-id ctx this-id ctx
(let-syntax ([arg (make-rename-transformer #'arg-tmp)] ...) (let-syntax ([arg (make-rename-transformer #'arg-tmp)] ...)
body0 body ...)))])]))]) body0 body ...)))])]))])
(syntax/loc (syntax-case stx () [(_ _ _ rhs) #'rhs]) (syntax/loc (syntax-case stx () [(_ _ _ rhs) #'rhs])
(case-lambda new-clause ...)))] (case-lambda new-clause ...)))]
[(_ fields _ expr) [(_ locals _ expr)
#'expr])) #'expr]))
(define-syntax (bind-fields stx) (define-syntax (bind-locals stx)
(syntax-case stx () (syntax-case stx ()
[(_ ([field-id field-accessor-id field-mutator-maybe-id] ...) this-id ctx body) [(_ [([field-id field-accessor-id field-mutator-maybe-id] ...)
([static-id static-tmp-id] ...)]
this-id ctx body)
(with-syntax ([(field-id ...) (for/list ([field-id (in-list (syntax->list #'(field-id ...)))]) (with-syntax ([(field-id ...) (for/list ([field-id (in-list (syntax->list #'(field-id ...)))])
(datum->syntax #'ctx (syntax-e field-id)))]) (datum->syntax #'ctx (syntax-e field-id)))]
[(static-id ...) (for/list ([static-id (in-list (syntax->list #'(static-id ...)))])
(datum->syntax #'ctx (syntax-e static-id)))])
#'(let-syntax ([field-id (make-set!-transformer #'(let-syntax ([field-id (make-set!-transformer
(lambda (stx) (lambda (stx)
(syntax-case stx (set!) (syntax-case stx (set!)
@ -263,6 +280,11 @@
(raise-syntax-error #f "field is immutable" stx))] (raise-syntax-error #f "field is immutable" stx))]
[(_ arg (... ...)) (syntax/loc stx ((field-accessor-id this-id) arg (... ...)))] [(_ arg (... ...)) (syntax/loc stx ((field-accessor-id this-id) arg (... ...)))]
[else (syntax/loc stx (field-accessor-id this-id))])))] [else (syntax/loc stx (field-accessor-id this-id))])))]
...
[static-id (lambda (stx)
(syntax-case stx ()
[(_ arg (... ...))
(syntax/loc stx (static-tmp-id this-id arg (... ...)))]))]
...) ...)
body))])) body))]))
@ -309,8 +331,9 @@
(and (eq? (syntax-e #'override-id) (syntax-e #'id)) (and (eq? (syntax-e #'override-id) (syntax-e #'id))
(with-syntax ([((field-id _ field-accessor-id field-mutator-maybe-id) ...) (with-syntax ([((field-id _ field-accessor-id field-mutator-maybe-id) ...)
(class-info-fields ci)]) (class-info-fields ci)])
#'(bind-fields-in-body #'(bind-locals-in-body
([field-id field-accessor-id field-mutator-maybe-id] ...) [([field-id field-accessor-id field-mutator-maybe-id] ...)
()]
expr)))])) expr)))]))
#'(selector-id vtable-id))]))]) #'(selector-id vtable-id))]))])
(syntax/loc stx (make-id (methods-id method-expr ...) (syntax/loc stx (make-id (methods-id method-expr ...)
@ -325,19 +348,28 @@
(syntax-local-value #'class-id (lambda () #f)))]) (syntax-local-value #'class-id (lambda () #f)))])
(unless (class-info? ci) (unless (class-info? ci)
(raise-syntax-error #f "not a class identifier" stx #'class-id)) (raise-syntax-error #f "not a class identifier" stx #'class-id))
(define method-accessor-id (define make-access
(or (for/or ([method (in-list (class-info-methods ci))]) (or (for/or ([method (in-list (class-info-methods ci))])
(syntax-case method () (syntax-case method ()
[(id _ accessor-id) [(id _ accessor-id)
(and (eq? (syntax-e #'id) (syntax-e #'method-id)) (and (eq? (syntax-e #'id) (syntax-e #'method-id))
#'accessor-id)])) (lambda (o)
(with-syntax ([vtable-accessor-id (class-info-vtable-accessor-id ci)]
[o o])
#'(accessor-id (vtable-accessor-id o)))))]))
(and call?
(for/or ([static (in-list (class-info-statics ci))])
(syntax-case static ()
[(id tmp-id)
(and (eq? (syntax-e #'id) (syntax-e #'method-id))
(lambda (o)
#'tmp-id))])))
(raise-syntax-error #f "cannot find method" stx #'method-id))) (raise-syntax-error #f "cannot find method" stx #'method-id)))
(with-syntax ([vtable-accessor-id (class-info-vtable-accessor-id ci)] (if call?
[method-accessor-id method-accessor-id]) (with-syntax ([proc (make-access #'o)])
(if call?
#'(let ([o obj]) #'(let ([o obj])
((method-accessor-id (vtable-accessor-id o)) o arg ...)) (proc o arg ...)))
#'(method-accessor-id (vtable-accessor-id obj)))))])) (make-access #'obj)))]))
(define-syntax (send stx) (define-syntax (send stx)
(send-or-method stx #t)) (send-or-method stx #t))
@ -350,6 +382,21 @@
[(_ class-id obj method-id) [(_ class-id obj method-id)
(send-or-method stx #f)])) (send-or-method stx #f)]))
(define-syntax (with-object stx)
(syntax-case stx ()
[(_ class-id obj-expr body0 body ...)
(let ([ci (and (identifier? #'class-id)
(syntax-local-value #'class-id (lambda () #f)))])
(unless (class-info? ci)
(raise-syntax-error #f "not a class identifier" stx #'class-id))
(with-syntax ([((field-id _ field-accessor-id field-mutator-maybe-id) ...)
(class-info-fields ci)]
[((static-id static-tmp-id) ...) (class-info-statics ci)])
(with-syntax ([local-bindings #'[([field-id field-accessor-id field-mutator-maybe-id] ...)
([static-id static-tmp-id] ...)]])
#'(let ([o obj-expr])
(bind-locals local-bindings o obj-expr (let () body0 body ...))))))]))
(define-for-syntax (check-member stx id l what) (define-for-syntax (check-member stx id l what)
(or (for/or ([e (in-list l)]) (or (for/or ([e (in-list l)])
(syntax-case e () (syntax-case e ()
@ -372,10 +419,12 @@
[b 2]) [b 2])
(private (private
[other (lambda (q) (list q this))]) [other (lambda (q) (list q this))])
(static
[enbox (lambda (v) (box (vector a v)))])
(public (public
[q #f] [q #f]
[m (lambda (z) (list a (other b)))] [m (lambda (z) (list a (other b)))]
[n (lambda (x y z) (vector a b x y z))])) [n (lambda (x y z) (vector a b (enbox x) y z))]))
(class sub #:extends example (class sub #:extends example
(field (field
@ -395,6 +444,7 @@
(new sub [d 5]) (new sub [d 5])
(send example (new sub) m 'more) (send example (new sub) m 'more)
(set-example-b! ex 6) (set-example-b! ex 6)
(send example ex enbox 88)
(define ex2 (new example (define ex2 (new example
#:override #:override
@ -402,4 +452,7 @@
(box (vector x y z a b)))]) (box (vector x y z a b)))])
[b 'b] [b 'b]
[a 'a])) [a 'a]))
(send example ex2 n 1 2 3)) (send example ex2 n 1 2 3)
(with-object example ex
(list a b (enbox 'c))))

843
racket/src/io/port/pipe.rkt
View File

@ -1,8 +1,7 @@
#lang racket/base #lang racket/base
(require racket/fixnum (require racket/fixnum
"../common/check.rkt" "../common/check.rkt"
"../common/fixnum.rkt" "../common/class.rkt"
"../common/object.rkt"
"../host/thread.rkt" "../host/thread.rkt"
"port.rkt" "port.rkt"
"input-port.rkt" "input-port.rkt"
@ -12,468 +11,482 @@
(provide make-pipe (provide make-pipe
make-pipe-ends make-pipe-ends
pipe-input-port? (rename-out [pipe-input-port?* pipe-input-port?]
pipe-output-port? [pipe-output-port?* pipe-output-port?])
pipe-content-length pipe-content-length)
pipe-write-position
pipe-get-content
pipe-discard-all)
(define (min+1 a b) (if a (min (add1 a) b) b)) (define (min+1 a b) (if a (min (add1 a) b) b))
(struct pipe-data (get-content-length (define pipe-input-port?*
write-position (let ([pipe-input-port?
get-content (lambda (p)
discard-all)) (define cp (->core-input-port p))
(pipe-input-port? p))])
pipe-input-port?))
(define (pipe-input-port? p) (define pipe-output-port?*
(and (input-port? p) (let ([pipe-output-port?
(pipe-data? (core-port-data (->core-input-port p))))) (lambda (p)
(define cp (->core-output-port p))
(define (pipe-output-port? p) (pipe-output-port? p))])
(and (output-port? p) pipe-output-port?))
(pipe-data? (core-port-data (->core-output-port p)))))
(define (pipe-content-length p) (define (pipe-content-length p)
(define cp (define d
(cond (cond
[(pipe-input-port? p) (->core-input-port p)] [(let ([p (->core-input-port p)])
[(pipe-output-port? p) (->core-output-port p)] (and p
(pipe-input-port? p)
p))
=> (lambda (p) (pipe-input-port-d p))]
[(let ([p (->core-output-port p)])
(and p
(pipe-output-port? p)
p))
=> (lambda (p) (pipe-output-port-d p))]
[else [else
(raise-argument-error 'pipe-contact-length "(or/c pipe-input-port? pipe-output-port?)" p)])) (raise-argument-error 'pipe-contact-length "(or/c pipe-input-port? pipe-output-port?)" p)]))
((pipe-data-get-content-length (core-port-data cp)) (atomically
(if (core-input-port? cp) (send pipe-input-port (pipe-data-input d) sync-data)
(compat-input-port-self cp) (send pipe-data d content-length)))
(compat-output-port-self cp))))
;; in atomic mode: ;; ----------------------------------------
(define pipe-write-position
(case-lambda
[(p) ((pipe-data-write-position (core-port-data p)) (compat-output-port-self p))]
[(p pos) ((pipe-data-write-position (core-port-data p)) (compat-output-port-self p) pos)]))
;; in atomic mode: (class pipe-data
(define (pipe-discard-all p) (field
((pipe-data-discard-all (core-port-data p)) (compat-output-port-self p))) [bstr #""]
[len 0]
[limit 0]
[peeked-amt 0] ; peeked but not yet read, effectively extends `limit`
[start 0]
[end 0]
[input #f] ; #f => closed
[output-closed? #f]
[read-ready-sema #f]
[write-ready-sema #f]
[more-read-ready-sema #f] ; for lookahead peeks
[read-ready-evt #f]
[write-ready-evt #f])
;; in atomic mode:x ;; All methods in atomic mode.
(define (pipe-get-content p bstr start-pos) ;; Beware that the input port must be synced to sure that `start`
((pipe-data-get-content (core-port-data p)) (compat-output-port-self p) bstr start-pos)) ;; represents the current position before using these methods.
(static
[content-length
(lambda ()
(define s start)
(define e end)
(if (s . fx<= . e)
(fx- e s)
(fx+ e (fx- len s))))]
(define-constructor (make-pipe-ends [limit #f] [input-name 'pipe] [output-name 'pipe] [input-empty?
#:need-input? [need-input? #t]) (lambda ()
(define bstr (make-bytes (min+1 limit 16))) (fx= start end))]
(define len (bytes-length bstr))
(define-fixnum peeked-amt 0) ; peeked but not yet read effectively extends `limit`
(define-fixnum start 0)
(define-fixnum end 0)
(define write-pos #f) ; to adjust the write position via `file-position` on a string port
(define input-closed? #f)
(define output-closed? #f)
(define (content-length) [output-full?
(if (start . fx<= . end) (lambda ()
(fx- end start) (define l limit)
(fx+ end (fx- len start)))) (and l
(define (input-empty?) (fx= start end)) ((content-length) . >= . (+ l peeked-amt))))]
(define (output-full?)
(and limit
((content-length) . >= . (+ limit peeked-amt))))
(define data ;; Used before/after read:
(pipe-data [check-output-unblocking
;; get-content-length (lambda ()
(method (when (output-full?) (semaphore-post write-ready-sema)))]
(lambda () [check-input-blocking
(atomically (content-length)))) (lambda ()
;; write-position (when (input-empty?) (semaphore-wait read-ready-sema)))]
(method
(case-lambda
;; in atomic mode
[() (or write-pos end)]
[(pos)
;; `pos` must be between `start` and `end`
(if (fx= pos end)
(set! write-pos #f)
(set! write-pos pos))]))
;; get-content
(method
(lambda (to-bstr start-pos)
;; in atomic mode
(define pos (let ([p (fx+ start start-pos)])
(if (p . fx>= . len)
(fx- p len)
p)))
(define end-pos (fx+ pos (bytes-length to-bstr)))
(cond
[(end-pos . fx<= . len)
(bytes-copy! to-bstr 0 bstr pos end-pos)]
[else
(bytes-copy! to-bstr 0 bstr pos len)
(bytes-copy! to-bstr (fx- len pos) bstr 0 (fx- end-pos len))])))
;; discard-all
(method
(lambda ()
;; in atomic mode
(set! peeked-amt 0)
(set! start 0)
(set! end 0)
(set! write-pos #f)))))
(define read-ready-sema (make-semaphore)) ;; Used before/after write:
(define write-ready-sema (and limit (make-semaphore 1))) [check-input-unblocking
(define more-read-ready-sema #f) ; for lookahead peeks (lambda ()
(define read-ready-evt (wrap-evt (semaphore-peek-evt read-ready-sema) (when (and (input-empty?) (not output-closed?)) (semaphore-post read-ready-sema))
(lambda (v) 0))) (when more-read-ready-sema
(define write-ready-evt (if limit (semaphore-post more-read-ready-sema)
(semaphore-peek-evt write-ready-sema) (set! more-read-ready-sema #f)))]
always-evt)) [check-output-blocking
(define progress-sema #f) (lambda ()
(when (output-full?)
(semaphore-wait write-ready-sema)
(when input
(send pipe-input-port input on-output-full))))]
;; Used before/after read: ;; Used after peeking:
(define (check-output-unblocking) [peeked!
(when (output-full?) (semaphore-post write-ready-sema))) (lambda (amt)
(define (check-input-blocking) (when (amt . > . peeked-amt)
(when (input-empty?) (semaphore-wait read-ready-sema))) (check-output-unblocking)
(set! peeked-amt amt)))]))
;; Used before/after write: ;; ----------------------------------------
(define (check-input-unblocking)
(when (and (input-empty?) (not output-closed?)) (semaphore-post read-ready-sema))
(when more-read-ready-sema
(semaphore-post more-read-ready-sema)
(set! more-read-ready-sema #f)))
(define (check-output-blocking)
(when (output-full?) (semaphore-wait write-ready-sema)))
;; Used after peeking: (class pipe-input-port #:extends core-input-port
(define (peeked! amt) (field
(when (amt . > . peeked-amt) [d #f] ; pipe-data
(check-output-unblocking) [progress-sema #f]
(set! peeked-amt amt))) [commit-manager #f])
(define (progress!) (private
(when progress-sema [progress!
(semaphore-post progress-sema) (lambda ()
(set! progress-sema #f))) (when progress-sema
(semaphore-post progress-sema)
(set! progress-sema #f)))]
;; [can leave atomic mode temporarily]
;; After this function returns, complete any commit-changing work
;; before leaving atomic mode again.
[pause-waiting-commit
(lambda ()
(when commit-manager
(commit-manager-pause commit-manager)))]
(define commit-manager #f) ;; [can leave atomic mode temporarily]
[wait-commit
(lambda (progress-evt ext-evt finish)
(cond
[(and (not commit-manager)
;; Try shortcut:
(not (sync/timeout 0 progress-evt))
(sync/timeout 0 ext-evt))
(finish)
#t]
[else
;; General case to support blocking and potentially multiple
;; commiting threads:
(unless commit-manager
(set! commit-manager (make-commit-manager)))
(commit-manager-wait commit-manager progress-evt ext-evt finish)]))]
;; in atomic mode [can leave atomic mode temporarily] [fast-mode!
;; After this function returns, complete any commit-changing work (lambda (amt)
;; before leaving atomic mode again. (unless (or count buffer)
(define (pause-waiting-commit) (with-object pipe-data d
(when commit-manager (define s start)
(commit-manager-pause commit-manager))) (define e end)
(unless (fx= s e)
(set! buffer bstr)
(set! buffer-pos s)
(set! buffer-end (if (s . fx< . e) e len))
(define o offset)
(when o
(set! offset (- (+ o amt) s)))))))]
;; in atomic mode [can leave atomic mode temporarily] [slow-mode!
(define (wait-commit progress-evt ext-evt finish) (lambda ()
(cond (when buffer
[(and (not commit-manager) (with-object pipe-data d
;; Try shortcut: (define pos buffer-pos)
(not (sync/timeout 0 progress-evt)) (define o offset)
(sync/timeout 0 ext-evt)) (when o
(finish) (set! offset (+ o pos)))
#t] (set! start (if (fx= pos len)
[else 0
;; General case to support blocking and potentially multiple pos))
;; commiting threads: (set! buffer #f)
(unless commit-manager (set! buffer-pos buffer-end))))])
(set! commit-manager (make-commit-manager)))
(commit-manager-wait commit-manager progress-evt ext-evt finish)]))
(values (static
;; input ---------------------------------------- [sync-data
(and (lambda ()
need-input? (when buffer
(make-core-input-port (with-object pipe-data d
#:name input-name (define pos buffer-pos)
#:data data (set! start (if (fx= pos len)
#:self self 0
pos)))))]
[on-resize
(lambda ()
(slow-mode!))]
[on-output-full
(lambda ()
(slow-mode!))])
#:prepare-change (override
(method [prepare-change
(lambda () (lambda ()
(pause-waiting-commit))) (with-object pipe-data d
(pause-waiting-commit)))]
#:read-byte [read-in
(method (lambda (dest-bstr dest-start dest-end copy?)
(lambda () (assert-atomic)
(assert-atomic) (slow-mode!)
(cond (with-object pipe-data d
[(input-empty?) (cond
(if output-closed? [(input-empty?)
eof (if output-closed?
;; event's synchronization value is ignored: eof
read-ready-evt)] read-ready-evt)]
[else [else
(define pos start) (check-output-unblocking)
(check-output-unblocking) (define s start)
(unless (fx= 0 peeked-amt) (define e end)
(set! peeked-amt (fxmax 0 (fx- peeked-amt 1)))) (define amt
(define new-pos (fx+ pos 1)) (cond
(if (fx= new-pos len) [(s . fx< . e)
(set! start 0)
(set! start new-pos))
(check-input-blocking)
(progress!)
(bytes-ref bstr pos)])))
#:read-in
(method
(lambda (dest-bstr dest-start dest-end copy?)
(assert-atomic)
(cond
[(input-empty?)
(if output-closed?
eof
read-ready-evt)]
[else
(check-output-unblocking)
(begin0
(cond
[(start . fx< . end)
(define amt (fxmin (fx- dest-end dest-start) (define amt (fxmin (fx- dest-end dest-start)
(fx- end start))) (fx- e s)))
(bytes-copy! dest-bstr dest-start bstr start (fx+ start amt)) (bytes-copy! dest-bstr dest-start bstr s (fx+ s amt))
(set! start (fx+ start amt)) (set! start (fx+ s amt))
(set! peeked-amt (fxmax 0 (fx- peeked-amt amt))) (set! peeked-amt (fxmax 0 (fx- peeked-amt amt)))
amt] amt]
[else [else
(define amt (fxmin (fx- dest-end dest-start) (define amt (fxmin (fx- dest-end dest-start)
(fx- len start))) (fx- len s)))
(bytes-copy! dest-bstr dest-start bstr start (fx+ start amt)) (bytes-copy! dest-bstr dest-start bstr s (fx+ s amt))
(set! start (modulo (fx+ start amt) len)) (set! start (modulo (fx+ s amt) len))
(set! peeked-amt (fxmax 0 (fx- peeked-amt amt))) (set! peeked-amt (fxmax 0 (fx- peeked-amt amt)))
amt]) amt]))
(check-input-blocking) (check-input-blocking)
(progress!))]))) (progress!)
(fast-mode! amt)
amt])))]
#:peek-byte [peek-in
(method (lambda (dest-bstr dest-start dest-end skip progress-evt copy?)
(lambda () (with-object pipe-data d
(assert-atomic) (assert-atomic)
(cond (sync-data)
[(input-empty?) (define content-amt (content-length))
(if output-closed?
eof
read-ready-evt)]
[else
(peeked! 1)
(bytes-ref bstr start)])))
#:peek-in
(method
(lambda (dest-bstr dest-start dest-end skip progress-evt copy?)
(assert-atomic)
(define content-amt (content-length))
(cond
[(and progress-evt
(sync/timeout 0 progress-evt))
#f]
[(content-amt . <= . skip)
(cond
[output-closed? eof]
[else
(unless (or (zero? skip) more-read-ready-sema)
(set! more-read-ready-sema (make-semaphore)))
(define evt (if (zero? skip)
read-ready-evt
(wrap-evt (semaphore-peek-evt more-read-ready-sema)
(lambda (v) 0))))
evt])]
[else
(define peek-start (fxmodulo (fx+ start skip) len))
(cond
[(peek-start . fx< . end)
(define amt (fxmin (fx- dest-end dest-start)
(fx- end peek-start)))
(bytes-copy! dest-bstr dest-start bstr peek-start (fx+ peek-start amt))
(peeked! (+ skip amt))
amt]
[else
(define amt (fxmin (fx- dest-end dest-start)
(fx- len peek-start)))
(bytes-copy! dest-bstr dest-start bstr peek-start (fx+ peek-start amt))
(peeked! (+ skip amt))
amt])])))
#:byte-ready
(method
(lambda (work-done!)
(assert-atomic)
(or output-closed?
(not (fx= 0 (content-length))))))
#:close
(method
(lambda ()
(unless input-closed?
(set! input-closed? #t)
(progress!))))
#:get-progress-evt
(method
(lambda ()
(atomically
(cond (cond
[input-closed? always-evt] [(and progress-evt
(sync/timeout 0 progress-evt))
#f]
[(content-amt . <= . skip)
(cond
[output-closed? eof]
[else
(unless (or (zero? skip) more-read-ready-sema)
(set! more-read-ready-sema (make-semaphore)))
(define evt (if (zero? skip)
read-ready-evt
(wrap-evt (semaphore-peek-evt more-read-ready-sema)
(lambda (v) 0))))
evt])]
[else [else
(unless progress-sema (define peek-start (fxmodulo (fx+ start skip) len))
(set! progress-sema (make-semaphore))) (cond
(semaphore-peek-evt progress-sema)])))) [(peek-start . fx< . end)
(define amt (fxmin (fx- dest-end dest-start)
(fx- end peek-start)))
(bytes-copy! dest-bstr dest-start bstr peek-start (fx+ peek-start amt))
(peeked! (+ skip amt))
amt]
[else
(define amt (fxmin (fx- dest-end dest-start)
(fx- len peek-start)))
(bytes-copy! dest-bstr dest-start bstr peek-start (fx+ peek-start amt))
(peeked! (+ skip amt))
amt])])))]
#:commit [byte-ready
(lambda (work-done!)
(assert-atomic)
(with-object pipe-data d
(or output-closed?
(begin
(sync-data)
(not (fx= 0 (content-length)))))))]
[close
(lambda ()
(with-object pipe-data d
(when input
(slow-mode!)
(set! input #f)
(progress!))))]
[get-progress-evt
(lambda ()
(atomically
(with-object pipe-data d
(cond
[(not input) always-evt]
[else
(slow-mode!)
(unless progress-sema
(set! progress-sema (make-semaphore)))
(semaphore-peek-evt progress-sema)]))))]
[commit
;; Allows `amt` to be zero and #f for other arguments, ;; Allows `amt` to be zero and #f for other arguments,
;; which is helpful for `open-input-peek-via-read`. ;; which is helpful for `open-input-peek-via-read`.
(method (lambda (amt progress-evt ext-evt finish)
(lambda (amt progress-evt ext-evt finish) (assert-atomic)
(assert-atomic) ;; `progress-evt` is a `semepahore-peek-evt`, and `ext-evt`
;; `progress-evt` is a `semepahore-peek-evt`, and `ext-evt` ;; is constrained; we can send them over to different threads
;; is constrained; we can send them over to different threads (cond
(cond [(zero? amt)
[(zero? amt) (progress!)]
(progress!)] [else
[else (wait-commit
(wait-commit progress-evt ext-evt
progress-evt ext-evt ;; in atomic mode, maybe in a different thread:
;; in atomic mode, maybe in a different thread: (lambda ()
(lambda () (with-object pipe-data d
(let ([amt (min amt (content-length))]) (slow-mode!)
(cond (let ([amt (min amt (content-length))])
[(fx= 0 amt) (cond
;; There was nothing to commit; claim success for 0 bytes [(fx= 0 amt)
(finish #"")] ;; There was nothing to commit; claim success for 0 bytes
[else (finish #"")]
(define dest-bstr (make-bytes amt)) [else
(cond (define dest-bstr (make-bytes amt))
[(start . fx< . end) (define s start)
(bytes-copy! dest-bstr 0 bstr start (fx+ start amt))] (define e end)
[else (cond
(define amt1 (fxmin (fx- len start) amt)) [(s . fx< . e)
(bytes-copy! dest-bstr 0 bstr start (fx+ start amt1)) (bytes-copy! dest-bstr 0 bstr s (fx+ s amt))]
(when (amt1 . fx< . amt) [else
(bytes-copy! dest-bstr amt1 bstr 0 (fx- amt amt1)))]) (define amt1 (fxmin (fx- len s) amt))
(set! start (fxmodulo (fx+ start amt) len)) (bytes-copy! dest-bstr 0 bstr s (fx+ s amt1))
(progress!) (when (amt1 . fx< . amt)
(check-input-blocking) (bytes-copy! dest-bstr amt1 bstr 0 (fx- amt amt1)))])
(finish dest-bstr)]))))]))))) (set! start (fxmodulo (fx+ s amt) len))
(progress!)
;; output ---------------------------------------- (fast-mode! amt)
(make-core-output-port (check-input-blocking)
#:name output-name (finish dest-bstr)])))))]))]
#:data data
#:self self
#:evt write-ready-evt [count-lines!
(lambda ()
#:write-out (slow-mode!))]))
;; in atomic mode
(method ;; ----------------------------------------
(lambda (src-bstr src-start src-end nonblock? enable-break? copy?)
(assert-atomic) (class pipe-output-port #:extends core-output-port
(let try-again () (field
(define top-pos (if (fx= start 0) [d d]) ; pipe-data
(fx- len 1)
len)) (override
(define (maybe-grow) [write-out
(cond ;; in atomic mode
[(or (not limit) (lambda (src-bstr src-start src-end nonblock? enable-break? copy?)
((+ limit peeked-amt) . > . (fx- len 1))) (assert-atomic)
;; grow pipe size (with-object pipe-data d
(define new-bstr (make-bytes (min+1 (and limit (+ limit peeked-amt)) (* len 2)))) (send pipe-input-port input sync-data)
(let try-again ()
(define top-pos (if (fx= start 0)
(fx- len 1)
len))
(define (maybe-grow)
(cond (cond
[(fx= 0 start) [(or (not limit)
(bytes-copy! new-bstr 0 bstr 0 (fx- len 1))] ((+ limit peeked-amt) . > . (fx- len 1)))
[else ;; grow pipe size
(bytes-copy! new-bstr 0 bstr start len) (send pipe-input-port input on-resize)
(bytes-copy! new-bstr (fx- len start) bstr 0 end) (define new-bstr (make-bytes (min+1 (and limit (+ limit peeked-amt)) (* len 2))))
(set! start 0) (cond
(set! end (fx- len 1))]) [(fx= 0 start)
(set! bstr new-bstr) (bytes-copy! new-bstr 0 bstr 0 (fx- len 1))]
(set! len (bytes-length new-bstr)) [else
(try-again)] (bytes-copy! new-bstr 0 bstr start len)
[else (pipe-is-full)])) (bytes-copy! new-bstr (fx- len start) bstr 0 end)
(define (pipe-is-full) (set! start 0)
(wrap-evt write-ready-evt (lambda (v) #f))) (set! end (fx- len 1))])
(define (apply-limit amt) (set! bstr new-bstr)
(if limit (set! len (bytes-length new-bstr))
(min amt (- (+ limit peeked-amt) (content-length))) (try-again)]
amt)) [else (pipe-is-full)]))
(cond (define (pipe-is-full)
[(fx= src-start src-end) ;; => flush (wrap-evt write-ready-evt (lambda (v) #f)))
0] (define (apply-limit amt)
[write-pos ; set by `file-position` on a bytes port (if limit
(define amt (apply-limit (fxmin (fx- end write-pos) (min amt (- (+ limit peeked-amt) (content-length)))
(fx- src-end src-start)))) amt))
(cond (cond
[(fx= amt 0) (pipe-is-full)] [(fx= src-start src-end) ;; => flush
[else 0]
(check-input-unblocking) [(and (end . fx>= . start)
(bytes-copy! bstr write-pos src-bstr src-start (fx+ src-start amt)) (end . fx< . top-pos))
(let ([new-write-pos (fx+ write-pos amt)]) (define amt (apply-limit (fxmin (fx- top-pos end)
(if (fx= new-write-pos end) (fx- src-end src-start))))
(set! write-pos #f) ; back to normal mode
(set! write-pos new-write-pos)))
(check-output-blocking)
amt])]
[(and (end . fx>= . start)
(end . fx< . top-pos))
(define amt (apply-limit (fxmin (fx- top-pos end)
(fx- src-end src-start))))
(cond
[(fx= amt 0) (pipe-is-full)]
[else
(check-input-unblocking)
(bytes-copy! bstr end src-bstr src-start (fx+ src-start amt))
(let ([new-end (fx+ end amt)])
(set! end (if (fx= new-end len) 0 new-end)))
(check-output-blocking)
amt])]
[(fx= end top-pos)
(cond
[(fx= start 0)
(maybe-grow)]
[else
(define amt (fxmin (fx- start 1)
(fx- src-end src-start)))
(cond (cond
[(fx= amt 0) (pipe-is-full)] [(fx= amt 0) (pipe-is-full)]
[else [else
(check-input-unblocking) (check-input-unblocking)
(bytes-copy! bstr 0 src-bstr src-start (fx+ src-start amt)) (bytes-copy! bstr end src-bstr src-start (fx+ src-start amt))
(set! end amt) (let ([new-end (fx+ end amt)])
(set! end (if (fx= new-end len) 0 new-end)))
(check-output-blocking) (check-output-blocking)
amt])])] amt])]
[(end . fx< . (fx- start 1)) [(fx= end top-pos)
(define amt (apply-limit (fxmin (fx- (fx- start 1) end) (cond
(fx- src-end src-start)))) [(fx= start 0)
(cond (maybe-grow)]
[(fx= amt 0) (pipe-is-full)] [else
(define amt (fxmin (fx- start 1)
(fx- src-end src-start)))
(cond
[(fx= amt 0) (pipe-is-full)]
[else
(check-input-unblocking)
(bytes-copy! bstr 0 src-bstr src-start (fx+ src-start amt))
(set! end amt)
(check-output-blocking)
amt])])]
[(end . fx< . (fx- start 1))
(define amt (apply-limit (fxmin (fx- (fx- start 1) end)
(fx- src-end src-start))))
(cond
[(fx= amt 0) (pipe-is-full)]
[else
(check-input-unblocking)
(bytes-copy! bstr end src-bstr src-start (fx+ src-start amt))
(set! end (fx+ end amt))
(check-output-blocking)
amt])]
[else [else
(check-input-unblocking) (maybe-grow)]))))]
(bytes-copy! bstr end src-bstr src-start (fx+ src-start amt))
(set! end (fx+ end amt))
(check-output-blocking)
amt])]
[else
(maybe-grow)]))))
#:count-write-evt-via-write-out [get-write-evt
(method (get-write-evt-via-write-out (lambda (out v bstr start)
(lambda (op v bstr start) (port-count! out v bstr start)))]
(port-count! op v bstr start)))
#:close [close
;; in atomic mode ;; in atomic mode
(method (lambda ()
(lambda () (with-object pipe-data d
(unless output-closed? (unless output-closed?
(set! output-closed? #t) (set! output-closed? #t)
(when write-ready-sema (when write-ready-sema
(semaphore-post write-ready-sema)) (semaphore-post write-ready-sema))
(when more-read-ready-sema (when more-read-ready-sema
(semaphore-post more-read-ready-sema)) (semaphore-post more-read-ready-sema))
(semaphore-post read-ready-sema))))))) (semaphore-post read-ready-sema))))]))
;; ----------------------------------------
(define (make-pipe-ends [limit #f] [input-name 'pipe] [output-name 'pipe])
(define len (min+1 limit 16))
(define read-ready-sema (make-semaphore))
(define write-ready-sema (and limit (make-semaphore 1)))
(define write-ready-evt (if limit
(semaphore-peek-evt write-ready-sema)
always-evt))
(define d (new pipe-data
[bstr (make-bytes len)]
[len len]
[limit limit]
[read-ready-sema read-ready-sema]
[write-ready-sema write-ready-sema]
[read-ready-evt (wrap-evt (semaphore-peek-evt read-ready-sema)
(lambda (v) 0))]
[write-ready-evt write-ready-evt]))
(define input (new pipe-input-port
[name input-name]
[d d]))
(define output (new pipe-output-port
[name output-name]
[evt write-ready-evt]
[d d]))
(set-pipe-data-input! d input)
(values input output))
(define/who (make-pipe [limit #f] [input-name 'pipe] [output-name 'pipe]) (define/who (make-pipe [limit #f] [input-name 'pipe] [output-name 'pipe])
(check who #:or-false exact-positive-integer? limit) (check who #:or-false exact-positive-integer? limit)