Unichannel-wait added.

README.md

@@ -40,6 +40,11 @@ For `port-channel`, wrap the input and output endpoints separately:
 (uni-channel-recv reader)
 ```
 
+For output port-channels, `uni-channel-put` queues the value for the writer
+thread. Use `uni-channel-close` followed by `uni-channel-wait` when shutdown must
+wait until queued values have actually been written and the writer thread has
+stopped.
+
 Install and test:
 
 ```sh

main.rkt

@@ -18,6 +18,7 @@
          uni-channel-get-evt
          uni-channel-put-evt
          uni-channel-close
+         uni-channel-wait
          uni-channel-closed?
 
          uni-channel-error-get
@@ -26,7 +27,7 @@
 
 (struct uni-channel
   (kind impl direction get-proc put-proc try-get-proc get-evt-proc put-evt-proc
-        close-proc closed-box error-get-proc error-try-get-proc error-evt-proc)
+        close-proc wait-proc closed-box error-get-proc error-try-get-proc error-evt-proc)
   #:property prop:evt
   (lambda (ch)
     (if (can-get? ch) ((uni-channel-get-evt-proc ch) ch) never-evt)))
@@ -55,6 +56,9 @@
   (mark-closed! ch)
   (void))
 
+(define (default-wait ch)
+  (void))
+
 (define (no-error-get ch)
   (raise-arguments-error 'uni-channel-error-get "uni-channel has no error channel"
                          "kind" (uni-channel-kind ch)))
@@ -71,11 +75,12 @@
                  #:get-evt get-evt-proc
                  #:put-evt put-evt-proc
                  #:close close-proc
+                 #:wait wait-proc
                  #:error-get error-get-proc
                  #:error-try-get error-try-get-proc
                  #:error-evt error-evt-proc)
   (uni-channel kind impl direction get-proc put-proc try-get-proc get-evt-proc put-evt-proc
-               close-proc (box #f) error-get-proc error-try-get-proc error-evt-proc))
+               close-proc wait-proc (box #f) error-get-proc error-try-get-proc error-evt-proc))
 
 (define (make-uni-channel channel)
   (cond
@@ -96,6 +101,7 @@
            #:get-evt (lambda (_ch) ch)
            #:put-evt (lambda (_ch v) (async-channel-put-evt ch v))
            #:close default-close
+           #:wait default-wait
            #:error-get no-error-get
            #:error-try-get no-error-try-get
            #:error-evt no-error-evt))
@@ -110,6 +116,7 @@
            #:get-evt (lambda (_ch) ch)
            #:put-evt (lambda (_ch v) (handle-evt always-evt (lambda (_) (place-channel-put ch v))))
            #:close default-close
+           #:wait default-wait
            #:error-get no-error-get
            #:error-try-get no-error-try-get
            #:error-evt no-error-evt))
@@ -125,6 +132,7 @@
            #:get-evt (lambda (_ch) (port-channel-evt pc))
            #:put-evt (lambda (_ch v) (handle-evt always-evt (lambda (_) (port-channel-put pc v))))
            #:close (lambda (ch) (mark-closed! ch) (close-port-channel pc))
+           #:wait (lambda (_ch) (port-channel-wait pc))
            #:error-get (lambda (_ch) (port-channel-error-get pc))
            #:error-try-get (lambda (_ch) (port-channel-error-try-get pc))
            #:error-evt (lambda (_ch) (port-channel-error-evt pc))))
@@ -160,6 +168,9 @@
   (unless (uni-channel-closed? ch) ((uni-channel-close-proc ch) ch))
   (void))
 
+(define (uni-channel-wait ch)
+  ((uni-channel-wait-proc ch) ch))
+
 (define (uni-channel-closed? ch) (unbox (uni-channel-closed-box ch)))
 
 (define (uni-channel-error-get ch) ((uni-channel-error-get-proc ch) ch))

scribblings/uni-channel.scrbl

@@ -124,6 +124,15 @@ native close operation. For port-channel wrappers this delegates to
 @racket[close-port-channel]. Calling @racket[uni-channel-close] more than once is
 safe.}
 
+@defproc[(uni-channel-wait [ch uni-channel?]) void?]{
+Waits until the underlying channel worker has stopped, when the backend has such
+a worker. For @racket['port] channels this delegates to
+@racket[port-channel-wait]. This is mainly useful for output port-channels: call
+@racket[uni-channel-close] first to enqueue the close marker, then
+@racket[uni-channel-wait] to wait until all queued values have been written and
+the writer thread has stopped. For @racket['async] and @racket['place] channels,
+this returns immediately.}
+
 @defproc[(uni-channel-closed? [ch uni-channel?]) boolean?]{
 Returns true when @racket[uni-channel-close] has been called on @racket[ch].}
 

test/main.rkt

@@ -40,6 +40,7 @@
      (uni-channel-put ch '(sync returns this value))
      (check-equal? (sync (uni-channel-get-evt ch)) '(sync returns this value))
      (uni-channel-close ch)
+     (uni-channel-wait ch)
      (check-true (uni-channel-closed? ch))
      (check-exn exn:fail? (lambda () (uni-channel-put ch 'after-close))))
 
@@ -86,6 +87,7 @@
      (check-exn exn:fail? (lambda () (uni-channel-get writer)))
      (check-exn exn:fail? (lambda () (uni-channel-put reader 'nope)))
      (uni-channel-close writer)
+     (uni-channel-wait writer)
      (check-true (uni-channel-closed? writer))
      (check-true (eof-object? (uni-channel-get reader))))