Use the bind port as a hint for multiple workers on the same host

Otherwise all the workers would try to bind to the same port, which would cause
an error.

Author: JamesWrigley

docs/src/_changelog.md

@@ -11,6 +11,10 @@ This documents notable changes in DistributedNext.jl. The format is based on
 
 ### Fixed
 - Fixed a cause of potential hangs when exiting the process ([#16]).
+- Fixed cases like `addprocs([("machine 10.1.1.1:9000", 2)])` where the bind
+  port is specified. Previously this would cause errors when the workers all
+  tried to bind to the same port, now all additional workers will treat the bind
+  port as a port hint ([#19]).
 
 ### Added
 - A watcher mechanism has been added to detect when both the Distributed stdlib

src/cluster.jl

@@ -733,8 +733,29 @@ function launch_additional(np::Integer, cmd::Cmd)
     io_objs = Vector{Any}(undef, np)
     addresses = Vector{Any}(undef, np)
 
+    worker_cmd = Cmd(cmd)
+    bind_idx = findfirst(==("--bind-to"), cmd)
+    if !isnothing(bind_idx)
+        # The actual bind spec will be the next argument
+        bind_idx += 1
+
+        bind_addr = worker_cmd[bind_idx]
+        parts = split(bind_addr, ':')
+        if length(parts) == 2
+            port_str = parts[2]
+
+            # If the port is not specified as a port hint then we convert it
+            # to a hint, otherwise the workers will try to bind to the same
+            # port and error.
+            if !startswith(port_str, '[')
+                new_bind_addr = "$(parts[1]):[$(port_str)]"
+                worker_cmd.exec[bind_idx] = new_bind_addr
+            end
+        end
+    end
+
     for i in 1:np
-        io = open(detach(cmd), "r+")
+        io = open(detach(worker_cmd), "r+")
         write_cookie(io)
         io_objs[i] = io.out
     end

test/sshmanager.jl

@@ -64,6 +64,11 @@ end
         @test 8000 >= worker.config.port < 9000
         test_n_remove_pids(new_pids)
 
+        print("\nssh addprocs with multiple workers and port specified\n")
+        new_pids = addprocs_with_testenv([("localhost 127.0.0.1:8000", 2)]; sshflags=sshflags)
+        @test length(new_pids) == 2
+        test_n_remove_pids(new_pids)
+
         print("\nssh addprocs with tunnel\n")
         new_pids = addprocs_with_testenv([("localhost", num_workers)]; tunnel=true, sshflags=sshflags)
         @test length(new_pids) == num_workers