windows: adjust flush handling of file-stream output ports

Some synchronization built into `close-output-port` seems unnecessary
on modern Windows (NT and up). An extra manager thread is needed for a
anonymous pipe for "write ready?" purposes, but not to buffer output.
The Windows documentation is not entirely clear on this point, but
experiments suggest that modern anonymous pipes behave in the obvious
way.

Meanwhile, adjust the io layer used by Racket CS with the
synchronization that would be needed for Windows 95. This adjustment
is questionable, because it doesn't seem likely that we'll ever go
back that far with Racket CS. But the rktio interface to support
flushing synchronization might somehow be needed in a future setting,
or mabe it will turn out that I'm wrong about pipe buffers.

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

@@ -1221,10 +1221,11 @@
   (thread (lambda ()
             (sync (system-idle-evt))
             (if (or force-close?
-                    ;; For Windows, we need a close that doesn't try
+                    ;; For really old Windows, we need a close that doesn't try
                     ;; to flush, because there's no way to avoid
                     ;; buffering at the rktio level:
-                    (eq? 'windows (system-type)))
+                    (and (eq? 'windows (system-type))
+                         (not (regexp-match? "Windows NT" (system-type 'machine)))))
                 (custodian-shutdown-all c)
                 (close-output-port o))))
 

racket/src/io/port/fd-port.rkt

@@ -213,11 +213,25 @@
        #:break newline?
        (void)))]
 
+  ;; in atomic mode, but may leave it temporarily
+  [flush-rktio-buffer-fully
+   (lambda ()
+     (unless (rktio-flushed?)
+       (end-atomic)
+       (sync (rktio-fd-flushed-evt this))
+       (start-atomic)
+       (flush-rktio-buffer-fully)))]
+
   #:static
   [flush-buffer/external
    (lambda ()
      (flush-buffer-fully #f))]
 
+  [rktio-flushed?
+   (lambda ()
+     (or (not bstr)
+         (rktio_poll_write_flushed rktio fd)))]
+
   #:override
   ;; in atomic mode
   [write-out
@@ -258,6 +272,7 @@
   [close
    (lambda ()
      (flush-buffer-fully #f) ; can temporarily leave atomic mode
+     (flush-rktio-buffer-fully) ; can temporarily leave atomic mode
      (when bstr ; <- in case a concurrent close succeeded
        (send fd-output-port this on-close)
        (when flush-handle
@@ -457,6 +472,29 @@
               (rktio_poll_add rktio (fd-evt-fd fde) ps mode)))
            (values #f fde)])]))))
 
+;; ----------------------------------------
+;; Wait on rktio-level flushing. At the time of writing, this is
+;; needed only for Windows so old that Racket CS doesn't run on it,
+;; but here just in case rktio or something else changes.
+
+(struct rktio-fd-flushed-evt (p)
+  #:property
+  prop:evt
+  (poller
+   (lambda (ffe ctx)
+     (define p  (rktio-fd-flushed-evt-p ffe))
+     (cond
+       [(send fd-output-port p rktio-flushed?)
+        (values '(#t) #f)]
+       [else
+        (sandman-poll-ctx-add-poll-set-adder!
+         ctx
+         (lambda (ps)
+           (if (send fd-output-port p rktio-flushed?)
+               (rktio_poll_set_add_nosleep rktio ps)
+               (rktio_poll_add rktio (fd-output-port-fd p) ps RKTIO_POLL_FLUSH))))
+        (values #f (list ffe))]))))
+
 ;; ----------------------------------------
 
 (define (register-fd-close custodian fd fd-refcount flush-handle port)

racket/src/rktio/rktio.h

@@ -278,9 +278,9 @@ intptr_t rktio_write(rktio_t *rktio, rktio_fd_t *fd, const char *buffer, intptr_
    mode. Alternatively, the result can be `RKTIO_WRITE_ERROR` for an
    error. Although `rktio_write` is intended to write only bytes that
    can be fully delivered to the OS, there may be OS limitations that
-   require buffering (e.g., on Windows). Use `rktio_poll_write_flushed`
-   to make sure the data is received by the destination before closing
-   `fd`. */
+   require buffering (e.g., on ancient versions of Windows). Use
+   `rktio_poll_write_flushed` to make sure the data is received by the
+   destination before closing `fd`. */
 
 #define RKTIO_WRITE_ERROR (-2)
 

racket/src/rktio/rktio_fd.c

@@ -732,6 +732,12 @@ int poll_write_ready_or_flushed(rktio_t *rktio, rktio_fd_t *rfd, int check_flush
     int retval;
     Win_FD_Output_Thread *oth = rfd->oth;
 
+    if (check_flushed && rfd->oth->nonblocking) {
+      /* Not Windows 95, so any written data really is in the pipe, as
+         good as flushed, and we don't really need to ask the thread. */
+      return RKTIO_POLL_READY;
+    }
+
     WaitForSingleObject(oth->lock_sema, INFINITE);
     if (oth->nonblocking) {
       if (oth->needflush) {