Merge pull request #22886 from JuliaLang/jn/io-deadlock

fix an IO deadlock condition (#22832)

disabled test seems to indicate this doesn't fix it fully, but it seems that only reproduces on CI, not on any of my local machines

src/jl_uv.c

@@ -197,16 +197,18 @@ JL_DLLEXPORT void jl_close_uv(uv_handle_t *handle)
     }
 
     if (handle->type == UV_NAMED_PIPE || handle->type == UV_TCP) {
+        uv_stream_t *stream = (uv_stream_t*)handle;
 #ifdef _OS_WINDOWS_
-        if (((uv_stream_t*)handle)->stream.conn.shutdown_req) {
+        if (stream->stream.conn.shutdown_req) {
 #else
-        if (((uv_stream_t*)handle)->shutdown_req) {
+        if (stream->shutdown_req) {
 #endif
             // don't close the stream while attempting a graceful shutdown
             return;
         }
-        if (uv_is_writable((uv_stream_t*)handle)) {
+        if (uv_is_writable(stream) && stream->write_queue_size != 0) {
             // attempt graceful shutdown of writable streams to give them a chance to flush first
+            // TODO: introduce a uv_drain cb API instead of abusing uv_shutdown in this way
             uv_shutdown_t *req = (uv_shutdown_t*)malloc(sizeof(uv_shutdown_t));
             req->data = 0;
             /*
@@ -218,12 +220,12 @@ JL_DLLEXPORT void jl_close_uv(uv_handle_t *handle)
              * b) In case the stream is already closed, in which case uv_close would
              *    cause an assertion failure.
              */
-            uv_shutdown(req, (uv_stream_t*)handle, &jl_uv_shutdownCallback);
+            uv_shutdown(req, stream, &jl_uv_shutdownCallback);
             return;
         }
     }
 
-    if (!uv_is_closing((uv_handle_t*)handle)) {
+    if (!uv_is_closing(handle)) {
         // avoid double-closing the stream
         if (handle->type == UV_TTY)
             uv_tty_set_mode((uv_tty_t*)handle, UV_TTY_MODE_NORMAL);

test/spawn.jl

@@ -298,15 +298,25 @@ let out = Pipe(), echo = `$exename --startup-file=no -e 'print(STDOUT, " 1\t", r
         @test iswritable(out)
         close(out.in)
         @test !isopen(out.in)
-        Sys.iswindows() || @test !isopen(out.out) # it takes longer to propagate EOF through the Windows event system
-        @test_throws ArgumentError write(out, "now closed error")
-        @test isreadable(out)
         @test !iswritable(out)
+        if !Sys.iswindows()
+            # on UNIX, we expect the pipe buffer is big enough that the write queue was immediately emptied
+            # and so we should already be notified of EPIPE on out.out by now
+            # and the other task should have already managed to consume all of the output
+            # it takes longer to propagate EOF through the Windows event system
+            # since it appears to be unwilling to buffer as much data
+            @test !isopen(out.out)
+            @test !isreadable(out)
+        end
+        @test_throws ArgumentError write(out, "now closed error")
         if Sys.iswindows()
-            # WINNT kernel does not provide a fast mechanism for async propagation
+            # WINNT kernel appears to not provide a fast mechanism for async propagation
             # of EOF for a blocking stream, so just wait for it to catch up.
             # This shouldn't take much more than 32ms.
             Base.wait_close(out)
+            # it's closed now, but the other task is expected to be behind this task
+            # in emptying the read buffer
+            @test isreadable(out)
         end
         @test !isopen(out)
     end
@@ -469,3 +479,19 @@ let c = `ls -l "foo bar"`
     @test length(c) == 3
     @test eachindex(c) == 1:3
 end
+
+## Deadlock in spawning a cmd (#22832)
+# FIXME?
+#let stdout = Pipe(), stdin = Pipe()
+#    Base.link_pipe(stdout, julia_only_read=true)
+#    Base.link_pipe(stdin, julia_only_write=true)
+#    p = spawn(pipeline(catcmd, stdin=stdin, stdout=stdout, stderr=DevNull))
+#    @async begin # feed cat with 2 MB of data (zeros)
+#        write(stdin, zeros(UInt8, 1048576 * 2))
+#        close(stdin)
+#    end
+#    sleep(0.5) # give cat a chance to fill the write buffer for stdout
+#    close(stdout.in) # make sure we can still close the write end
+#    @test sizeof(readstring(stdout)) == 1048576 * 2 # make sure we get all the data
+#    @test success(p)
+#end