fix file-stream read/write on concurrent close

Includes repairs for both Racket CS and traditional Racket.

pkgs/racket-test-core/tests/racket/file.rktl

@@ -1166,6 +1166,68 @@
                   (peek-bytes 1 end i)))))
   (delete-file tempfilename))
 
+;;------------------------------------------------------------
+;; File-stream ports and blocking behavior
+
+(let ()
+  (define-values (s i o e) (subprocess #f #f #f (find-exe) "-e" "(read)"))
+
+  (thread (lambda ()
+            (sync (system-idle-evt))
+            (close-input-port i)))
+
+  (err/rt-test
+   (peek-bytes-avail! (make-bytes 10) 0 #f i)
+   exn:fail?)
+
+  (close-output-port o)
+  (close-input-port e)
+  (subprocess-wait s))
+
+(let ()
+  (define-values (s i o e) (subprocess #f #f #f (find-exe) "-e" "(read)"))
+
+  (thread (lambda ()
+            (sync (system-idle-evt))
+            (close-input-port i)))
+
+  (test 0 peek-bytes-avail! (make-bytes 10) 0 (port-progress-evt i) i)
+
+  (close-output-port o)
+  (close-input-port e)
+  (subprocess-wait s))
+
+(let ()
+  (define-values (s i o e) (subprocess #f #f #f (find-exe) "-e" "(read)"))
+
+  (thread (lambda ()
+            (sync (system-idle-evt))
+            (close-input-port i)))
+
+  ;; Short not get stuck:
+  (sync (port-progress-evt i))
+
+  (close-output-port o)
+  (close-input-port e)
+  (subprocess-wait s))
+
+(let ()
+  (define-values (s i o e) (subprocess #f #f #f (find-exe) "-e" "(let loop () (write-bytes (make-bytes 1024)) (loop))"))
+
+  (thread (lambda ()
+            (sync (system-idle-evt))
+            (close-output-port o)))
+
+  (err/rt-test
+   (let loop ()
+     (write-bytes-avail #"hello" o)
+     (loop))
+   exn:fail?)
+
+  (close-input-port i)
+  (close-input-port e)
+  (subprocess-wait s))
+
 ;;------------------------------------------------------------
 
 ;; Test custom output port

racket/src/io/port/peek-via-read-port.rkt

@@ -36,7 +36,8 @@
   [close-peek-buffer
    (lambda ()
      (purge-buffer)
-     (set! bstr #""))]
+     (set! bstr #"")
+     (progress!))]
 
   [buffer-adjust-pos
    (lambda (i)

racket/src/racket/src/port.c

@@ -5174,6 +5174,8 @@ fd_flush_done(Scheme_Object *port)
 
   op = scheme_output_port_record(port);
 
+  if (op->closed) return 1;
+
   fop = (Scheme_FD *)op->port_data;
 
   return !fop->flushing;
@@ -5296,6 +5298,9 @@ static intptr_t flush_fd(Scheme_Output_Port *op,
                                           (Scheme_Object *)op, 0.0,
                                           enable_break);
         END_ESCAPEABLE();
+
+        if (op->closed)
+          return 0;
       } else if (len == RKTIO_WRITE_ERROR) {
         if (consume_buffer) {
           /* Drop unsuccessfully flushed bytes. This isn't the
@@ -5448,12 +5453,15 @@ fd_close_output(Scheme_Output_Port *port)
   if (fop->bufcount)
     flush_fd(port, NULL, 0, 0, 0, 0);
 
-  if (fop->flushing && !scheme_force_port_closed)
+  if (fop->flushing && fop->bufcount && !scheme_force_port_closed) {
     wait_until_fd_flushed(port, 0);
+    if (port->closed)
+      return;
+  }
 
   if (!scheme_force_port_closed && fop->fd) {
     /* Check for flushing at the rktio level (not to be confused
-       with pulmber flushes): */
+       with plumber flushes): */
     while (!rktio_poll_write_flushed(scheme_rktio, fop->fd)) {
       scheme_block_until(end_fd_flush_done, end_fd_flush_needs_wakeup, (Scheme_Object *)fop, 0.0);
     }