mq-impl/stream: Close the port when stopping the fibers.

* gnu/gnunet/mq-impl/stream.scm
  (prepare-port-message-queue)[closed-condition]: Document new uses.
  (prepare-port-message-queue)[start-reader!]: Close port if
  signal-condition! returns #false.
  (prepare-port-message-queue)[start-writer!]: Likewise.
* tests/mq-stream.scm
  (%false-if-broken-pipe): New procedure.
  (false-if-broken-pipe): New macro.
  ("closed for writing --> handle-input! stops (port->message-queue)"):
  Adjust test to resource leak fix.
  (error-handler/regular): New procedure.
  ("port is closed at input eof")
  ("port is closed at output eof")
  ("port is closed at input/output eof"): New tests.

1 files changed, 63 insertions, 2 deletions

diff --git a/tests/mq-stream.scm b/tests/mq-stream.scm
index d9cd667..e16aa1b 100644
--- a/tests/mq-stream.scm
+++ b/tests/mq-stream.scm
@@ -31,6 +31,7 @@
              ((rnrs io ports) #:select (open-bytevector-input-port))
              ((rnrs base) #:select (assert))
              (rnrs hashtables)
+             ((rnrs exceptions) #:select (guard))
              (srfi srfi-26)
              (srfi srfi-43)
              (rnrs io ports)
@@ -532,6 +533,20 @@
    ;; Should make 'yield-many' less fragile.
    #:parallelism 1))
 
+(define (%false-if-broken-pipe thunk)
+  "Call @var{thunk} in an environment where EPIPE system errors are caught.
+If an EPIPE system error is raised, return #f."
+  (guard (c ((and (eq? 'system-error (exception-kind c))
+                  (= EPIPE (car (list-ref (exception-args c) 3))))
+             #f))
+    (thunk)))
+
+(define-syntax-rule (false-if-broken-pipe exp exp* ...)
+  ;; See %false-if-broken-pipe
+  (%false-if-broken-pipe
+   (lambda ()
+     exp exp* ...)))
+
 (test-assert "closed for writing --> handle-input! stops (port->message-queue)"
   (call-with-spawner/wait
    (lambda (spawn)
@@ -569,8 +584,10 @@
      (pk 'waiting)
      (wait end-of-file)
      ;; Attempt to read a message (after buffering a message), even though
-     ;; the connection is half-closed.
-     (put-bytevector beta #vu8(0 4 0 0))
+     ;; the connection is half-closed.  Ignore broken pipe errors here:
+     ;; if a ‘broken pipe’ error happens here, that means ALPHA was closed,
+     ;; which is correct (tested in "port is closed at output").
+     (false-if-broken-pipe (put-bytevector beta #vu8(0 4 0 0)))
      ;; As the 'handle-input!' fiber should have exited already, 'receive!'
      ;; shouldn't be called.
      (yield-many)
@@ -668,4 +685,48 @@
                                         (old-read-waiter p)))))
                     (get-bytevector-some beta))))))
 
+(define (error-handler/regular . e)
+  (match e
+    ('(input:regular-end-of-file) (values))
+    (_ (error "what ~a" e))))
+
+(test-assert "port is closed at input eof"
+  (call-with-spawner/wait
+   (lambda (spawn)
+     (define-values (alpha beta) (two-sockets))
+     (define q (port->message-queue alpha no-handlers error-handler/regular
+                                    #:spawn spawn))
+     (shutdown alpha 0)
+     (yield-many)
+     (sleep 0.05) ;; XXX yield-many above is unsufficient
+     (port-closed? alpha))
+   #:parallelism 1)) ; to make the use of yield-many less fragile
+
+(test-assert "port is closed at output eof"
+  (call-with-spawner/wait
+   (lambda (spawn)
+     (define-values (alpha beta) (two-sockets))
+     (define mq (port->message-queue alpha no-handlers error-handler/regular
+                                     #:spawn spawn))
+     (shutdown alpha 1)
+     ;; XXX It's not possible for the output eof to be waited for currently,
+     ;; so attempt to send a message to wake up the writing fiber.
+     (send-message! mq (bv-slice/read-write #vu8(0 4 0 0)))
+     (yield-many)
+     (sleep 0.05) ;; XXX yield-many above is unsufficient
+     (port-closed? alpha))
+   #:parallelism 1)) ; to make the use of yield-many less fragile
+
+(test-assert "port is closed at input/output eof"
+  (call-with-spawner/wait
+   (lambda (spawn)
+     (define-values (alpha beta) (two-sockets))
+     (define q (port->message-queue alpha no-handlers error-handler/regular
+                                    #:spawn spawn))
+     (shutdown alpha 2)
+     (yield-many)
+     (sleep 0.05) ;; XXX yield-many above is unsufficient
+     (port-closed? alpha))
+   #:parallelism 1)) ; to make the use of yield-many less fragile
+
 (test-end "mq-stream")