Small refactor for Distributed:managers.jl (JuliaLang/julia#34556)

- update `Sockets.bind` to handle client sockets too
- use `Sockets.bind` in `Distributed.bind_client_port` 
- make it compatible to IPv6

Author: ssikdar1

src/managers.jl

@@ -489,57 +489,52 @@ end
 
 const client_port = Ref{Cushort}(0)
 
-function socket_reuse_port()
+function socket_reuse_port(iptype)
     if ccall(:jl_has_so_reuseport, Int32, ()) == 1
-        s = TCPSocket(delay = false)
+        sock = TCPSocket(delay = false)
 
         # Some systems (e.g. Linux) require the port to be bound before setting REUSEPORT
         bind_early = Sys.islinux()
 
-        bind_early && bind_client_port(s)
-        rc = ccall(:jl_tcp_reuseport, Int32, (Ptr{Cvoid},), s.handle)
+        bind_early && bind_client_port(sock, iptype)
+        rc = ccall(:jl_tcp_reuseport, Int32, (Ptr{Cvoid},), sock.handle)
         if rc < 0
             # This is an issue only on systems with lots of client connections, hence delay the warning
             nworkers() > 128 && @warn "Error trying to reuse client port number, falling back to regular socket" maxlog=1
 
             # provide a clean new socket
             return TCPSocket()
         end
-        bind_early || bind_client_port(s)
-        return s
+        bind_early || bind_client_port(sock, iptype)
+        return sock
     else
         return TCPSocket()
     end
 end
 
-# TODO: this doesn't belong here, it belongs in Sockets
-function bind_client_port(s::TCPSocket)
-    Sockets.iolock_begin()
-    @assert s.status == Sockets.StatusInit
-    host_in = Ref(hton(UInt32(0))) # IPv4 0.0.0.0
-    err = ccall(:jl_tcp_bind, Int32, (Ptr{Cvoid}, UInt16, Ptr{Cvoid}, Cuint, Cint),
-                s, hton(client_port[]), host_in, 0, false)
-    Sockets.iolock_end()
-    uv_error("tcp_bind", err)
-
-    _addr, port = getsockname(s)
+function bind_client_port(sock::TCPSocket, iptype)
+    bind_host = iptype(0)
+    Sockets.bind(sock, bind_host, client_port[])
+    _addr, port = getsockname(sock)
     client_port[] = port
-    return s
+    return sock
 end
 
 function connect_to_worker(host::AbstractString, port::Integer)
-    s = socket_reuse_port()
-    connect(s, host, UInt16(port))
-
     # Avoid calling getaddrinfo if possible - involves a DNS lookup
     # host may be a stringified ipv4 / ipv6 address or a dns name
     bind_addr = nothing
     try
-        bind_addr = string(parse(IPAddr,host))
+        bind_addr = parse(IPAddr,host)
     catch
-        bind_addr = string(getaddrinfo(host))
+        bind_addr = getaddrinfo(host)
     end
-    (s, bind_addr)
+
+    iptype = typeof(bind_addr)
+    sock = socket_reuse_port(iptype)
+    connect(sock, bind_addr, UInt16(port))
+
+    (sock, string(bind_addr))
 end
 
 

test/managers.jl

@@ -0,0 +1,13 @@
+using Test
+
+using Distributed: bind_client_port
+using Sockets
+
+sock = bind_client_port(TCPSocket(), typeof(IPv4(0)))
+addr, port = getsockname(sock)
+@test addr == ip"0.0.0.0"
+
+
+sock = bind_client_port(TCPSocket(), typeof(IPv6(0)))
+addr, port = getsockname(sock)
+@test addr == ip"::"

test/runtests.jl

@@ -10,3 +10,5 @@ cmd = `$test_exename $test_exeflags $disttestfile`
 if !success(pipeline(cmd; stdout=stdout, stderr=stderr)) && ccall(:jl_running_on_valgrind,Cint,()) == 0
     error("Distributed test failed, cmd : $cmd")
 end
+
+include("managers.jl")