reuse port if possible

Author: amitmurthy

base/managers.jl

@@ -293,13 +293,35 @@ function connect_w2w(pid::Int, config::WorkerConfig)
     (s,s)
 end
 
+const client_port = Ref{Cushort}(0)
+
+function socket_reuse_port()
+    s = TCPSocket()
+    try
+        client_host = Ref{Cuint}(0)
+        ccall(:jl_tcp_bind, Int32,
+                (Ptr{Void}, UInt16, UInt32, Cuint),
+                s.handle, hton(client_port.x), hton(UInt32(0)), 0) < 0 && throw("bind() error")
+
+        ccall(:jl_tcp_reuseport, Int32, (Ptr{Void}, ), s.handle) < 0 && throw("SO_REUSEPORT error")
+        ccall(:jl_tcp_getsockname_v4, Int32,
+                    (Ptr{Void}, Ref{Cuint}, Ref{Cushort}),
+                    s.handle, client_host, client_port) < 0 && throw("getsockname() error")
+    catch e
+        warn_once("Unable to reuse port : ", e)
+        # provide a clean new socket
+        return TCPSocket()
+    end
+    return s
+end
 
 function connect_to_worker(host::AbstractString, port::Integer)
     # Connect to the loopback port if requested host has the same ipaddress as self.
+    s = socket_reuse_port()
     if host == string(LPROC.bind_addr)
-        s = connect("127.0.0.1", UInt16(port))
+        s = connect(s, "127.0.0.1", UInt16(port))
     else
-        s = connect(host, UInt16(port))
+        s = connect(s, host, UInt16(port))
     end
 
     # Avoid calling getaddrinfo if possible - involves a DNS lookup

base/multi.jl

@@ -778,15 +778,6 @@ notify_empty(rv::RemoteValue) = notify(rv.empty)
 
 ## message event handlers ##
 
-# activity on accept fd
-function accept_handler(server::TCPServer, status::Int32)
-    if status == -1
-        error("an error occured during the creation of the server")
-    end
-    client = accept_nonblock(server)
-    process_messages(client, client)
-end
-
 process_messages(r_stream::TCPSocket, w_stream::TCPSocket) = process_messages(r_stream, w_stream, nothing)
 process_messages(r_stream::TCPSocket, w_stream::TCPSocket, rr_ntfy_join) = @schedule process_tcp_streams(r_stream, w_stream, rr_ntfy_join)
 
@@ -903,29 +894,32 @@ function message_handler_loop(r_stream::AsyncStream, w_stream::AsyncStream, rr_n
         end # end of while
     catch e
         iderr = worker_id_from_socket(r_stream)
-        werr = worker_from_id(iderr)
-        oldstate = werr.state
-        set_worker_state(werr, W_TERMINATED)
-
-
-        # If error occured talking to pid 1, commit harakiri
-        if iderr == 1
-            if isopen(w_stream)
-                print(STDERR, "fatal error on ", myid(), ": ")
-                display_error(e, catch_backtrace())
+        if (iderr < 1)
+            print(STDERR, "Socket from unknown remote worker in worker ", myid())
+        else
+            werr = worker_from_id(iderr)
+            oldstate = werr.state
+            set_worker_state(werr, W_TERMINATED)
+
+            # If error occured talking to pid 1, commit harakiri
+            if iderr == 1
+                if isopen(w_stream)
+                    print(STDERR, "fatal error on ", myid(), ": ")
+                    display_error(e, catch_backtrace())
+                end
+                exit(1)
             end
-            exit(1)
-        end
 
-        # Will treat any exception as death of node and cleanup
-        # since currently we do not have a mechanism for workers to reconnect
-        # to each other on unhandled errors
-        deregister_worker(iderr)
+            # Will treat any exception as death of node and cleanup
+            # since currently we do not have a mechanism for workers to reconnect
+            # to each other on unhandled errors
+            deregister_worker(iderr)
+        end
 
         if isopen(r_stream) close(r_stream) end
         if isopen(w_stream) close(w_stream) end
 
-        if (myid() == 1)
+        if (myid() == 1) && (iderr > 1)
             if oldstate != W_TERMINATING
                 println(STDERR, "Worker $iderr terminated.")
                 rethrow(e)
@@ -977,7 +971,17 @@ function start_worker(out::IO)
     else
         sock = listen(LPROC.bind_port)
     end
-    sock.ccb = accept_handler
+    @schedule begin
+        while true
+            try
+                client = accept(sock)
+                process_messages(client, client)
+            catch e
+                println(STDERR, "Error in accept() : ", e)
+                break
+            end
+        end
+    end
     print(out, "julia_worker:")  # print header
     print(out, "$(dec(LPROC.bind_port))#") # print port
     print(out, LPROC.bind_addr)

src/jl_uv.c

@@ -655,7 +655,36 @@ DLLEXPORT int jl_tcp_quickack(uv_tcp_t *handle, int on)
     }
     return 0;
 }
+
+#endif
+
+
+DLLEXPORT int jl_tcp_reuseport(uv_tcp_t *handle)
+{
+#if defined(SO_REUSEPORT)
+    int fd = (handle)->io_watcher.fd;
+    int yes = 1;
+    if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &yes, sizeof(yes))) {
+        return -1;
+    }
+    return 0;
+#else
+    return -1;
 #endif
+}
+
+DLLEXPORT int jl_tcp_getsockname_v4(uv_tcp_t *handle, uint32_t * ip, uint16_t * port)
+{
+    struct sockaddr_in name;
+    int len = sizeof(name);
+    if (uv_tcp_getsockname(handle, (struct sockaddr *)&name, &len)) {
+        return -1;
+    }
+
+    *ip = ntohl(name.sin_addr.s_addr);
+    *port = ntohs(name.sin_port);
+    return 0;
+}
 
 #ifndef _OS_WINDOWS_