spawn: Allow libuv IOServers to be passed as extra FDs

Extracted from #59271, but I now have an independent use case.
This allows libuv `IOServer`s (unix domain servers, tcp servers,
etc) to be passed as extra fd arguments to child processes.

Author: Keno

base/cmd.jl

@@ -188,7 +188,7 @@ if OS_HANDLE !== RawFD
 end
 setup_stdio(stdio::Union{DevNull,OS_HANDLE,RawFD}, ::Bool) = (stdio, false)
 
-const Redirectable = Union{IO, FileRedirect, RawFD, OS_HANDLE}
+const Redirectable = Union{IO, IOServer, FileRedirect, RawFD, OS_HANDLE}
 const StdIOSet = NTuple{3, Redirectable}
 
 struct CmdRedirect <: AbstractCmd

base/process.jl

@@ -84,7 +84,7 @@ function _uv_hook_close(proc::Process)
     nothing
 end
 
-const SpawnIO  = Union{IO, RawFD, OS_HANDLE, SyncCloseFD} # internal copy of Redirectable, removing FileRedirect and adding SyncCloseFD
+const SpawnIO  = Union{IO, IOServer, RawFD, OS_HANDLE, SyncCloseFD} # internal copy of Redirectable, removing FileRedirect and adding SyncCloseFD
 const SpawnIOs = Memory{SpawnIO} # convenience name for readability (used for dispatch also to clearly distinguish from Vector{Redirectable})
 
 function as_cpumask(cpus::Vector{UInt16})
@@ -341,13 +341,13 @@ close_stdio(stdio::SyncCloseFD) = close_stdio(stdio.fd)
 
 spawn_opts_swallow(stdios::StdIOSet) = Redirectable[stdios...]
 spawn_opts_inherit(stdios::StdIOSet) = Redirectable[stdios...]
-spawn_opts_swallow(in::Redirectable=devnull, out::Redirectable=devnull, err::Redirectable=devnull) =
-    Redirectable[in, out, err]
+spawn_opts_swallow(in::Redirectable=devnull, out::Redirectable=devnull, err::Redirectable=devnull, extra::Redirectable...) =
+    Redirectable[in, out, err, extra...]
 # pass original descriptors to child processes by default, because we might
 # have already exhausted and closed the libuv object for our standard streams.
 # ref issue #8529
-spawn_opts_inherit(in::Redirectable=RawFD(0), out::Redirectable=RawFD(1), err::Redirectable=RawFD(2)) =
-    Redirectable[in, out, err]
+spawn_opts_inherit(in::Redirectable=RawFD(0), out::Redirectable=RawFD(1), err::Redirectable=RawFD(2), extra::Redirectable...) =
+    Redirectable[in, out, err, extra...]
 
 function eachline(cmd::AbstractCmd; keep::Bool=false)
     out = PipeEndpoint()

base/stream.jl

@@ -202,12 +202,7 @@ end
 
 function PipeEndpoint(fd::OS_HANDLE)
     pipe = PipeEndpoint()
-    iolock_begin()
-    err = ccall(:uv_pipe_open, Int32, (Ptr{Cvoid}, OS_HANDLE), pipe.handle, fd)
-    uv_error("pipe_open", err)
-    pipe.status = StatusOpen
-    iolock_end()
-    return pipe
+    return open_pipe!(pipe, fd)
 end
 if OS_HANDLE != RawFD
     PipeEndpoint(fd::RawFD) = PipeEndpoint(Libc._get_osfhandle(fd))
@@ -283,8 +278,8 @@ end
 lock(s::LibuvStream) = lock(s.lock)
 unlock(s::LibuvStream) = unlock(s.lock)
 
-setup_stdio(stream::LibuvStream, ::Bool) = (stream, false)
-rawhandle(stream::LibuvStream) = stream.handle
+setup_stdio(stream::Union{LibuvStream, LibuvServer}, ::Bool) = (stream, false)
+rawhandle(stream::Union{LibuvStream, LibuvServer}) = stream.handle
 unsafe_convert(::Type{Ptr{Cvoid}}, s::Union{LibuvStream, LibuvServer}) = s.handle
 
 function init_stdio(handle::Ptr{Cvoid})

stdlib/Sockets/src/PipeServer.jl

@@ -24,6 +24,23 @@ function PipeServer()
     return pipe
 end
 
+function PipeServer(handle::OS_HANDLE)
+    pipe = PipeServer()
+    return Base.open_pipe!(pipe, handle)
+end
+
+function Base.open_pipe!(p::PipeServer, handle::OS_HANDLE)
+    iolock_begin()
+    if p.status != StatusInit
+        error("pipe is already in use or has been closed")
+    end
+    err = ccall(:uv_pipe_open, Int32, (Ptr{Cvoid}, OS_HANDLE), p.handle, handle)
+    uv_error("pipe_open", err)
+    p.status = StatusOpen
+    iolock_end()
+    return p
+end
+
 ## server functions ##
 
 accept(server::PipeServer) = accept(server, PipeEndpoint())

test/spawn.jl

@@ -1053,3 +1053,31 @@ end
     @assert eltype(args) != String
     @test Cmd(["ls", args...]) == `ls -l /tmp`
 end
+
+# Test passing a pipe server as an addition fd
+@testset "Pipe server as additional fd" begin
+    mktempdir() do dir
+        path = joinpath(dir, "test.sock")
+        server = Sockets.PipeServer()
+        bind(server, path)
+        Base.errormonitor(@async begin
+            local client
+            while true
+                try
+                    client = Sockets.connect(path)
+                    break
+                catch e
+                    isa(e, Base.IOError) || rethrow(e)
+                end
+                sleep(1)
+            end
+            println(client, "Hello Socket!")
+            closewrite(client)
+        end)
+        buf = IOBuffer()
+        proc = run(`$(Base.julia_cmd()) -e 'using Sockets; s = listen(Sockets.PipeServer(RawFD(3))); c = accept(s); print(read(c, String))'`, devnull, buf, stderr, server)
+        close(server)
+        @test success(proc)
+        @test String(take!(buf)) == "Hello Socket!\n"
+    end
+end