finish broker selector

Author: JBlaschke

src/NetworkInterfaceControllers.jl

@@ -1,5 +1,7 @@
 module NetworkInterfaceControllers
 
+using Sockets
+
 include("interfaces.jl")
 using .Interfaces
 export Interface, get_interface_data
@@ -42,20 +44,19 @@ function broker_ip_port(ipv)
             x->NameSelector.best_interfaces(
                 x,
                 NICPreferences.BROKER_INTERFACE.name,
-                NICPreferences.BROKER_INTERFACE.match_strategy) |>
-            only
+                NICPreferences.BROKER_INTERFACE.match_strategy
+            ) |> only
     return iface.ip, NICPreferences.BROKER_INTERFACE.port
 end
 
 function start_broker(ipv)
-    ip, port = broker_port(ipv)
+    ip, port = broker_ip_port(ipv)
+
+    t::Task = @task Broker.start_server(ip, UInt32(port))
 
-    t::Task = @task Broker.start_server(
-        iface.ip, UInt32(NICPreferences.BROKER_INTERFACE.port)
-    )
     # Run the server right away
     schedule(t)
-    return iface.ip, NICPreferences.BROKER_INTERFACE.port, t
+    return ip, port, t
 end
 
 function broker_ip_string(ipv::Int)::String
@@ -90,25 +91,26 @@ end
 
 broker_startup_string() = broker_startup_string(4)
 
-function broker_query_string(ip::String, port::Int)::String
-    runtime_str = julia_runtime_str()
-    import_str  = "using NetworkInterfaceControllers.Broker, Sockets"
-    query_str   = "Broker.query(ip\"$(ip)\", UInt32($(port)), ifaces)"
-
-    return "$(runtime_str) -e '$(import_str); $(query_str)'"
-end
+# function broker_query_string(ip::String, port::Int)::String
+#     runtime_str = julia_runtime_str()
+#     import_str  = "using NetworkInterfaceControllers.Broker, Sockets"
+#     query_str   = "Broker.query(ip\"$(ip)\", UInt32($(port)), ifaces)"
+# 
+#     return "$(runtime_str) -e '$(import_str); $(query_str)'"
+# end
 
 export start_broker, broker_ip_port, broker_ip_string, broker_port_string
-export broker_startup_string, broker_query_string
+export broker_startup_string#, broker_query_string
 
 function best_interface_hwloc_closest(
         data::Interface; pid::Union{T, Nothing}=nothing
     ) where T <: Integer
 end
 
 function best_interface_broker(
-        data::Interface; broker_port::Union{T, Nothing}=nothing
-    ) where T <: Integer
+        data::Vector{Interface}, ipv::Type{V};
+        broker_port::Union{T, Nothing}=nothing
+    ) where {T <: Integer, V <: IPAddr}
 
     if isnothing(broker_port)
         @debug "Getting broker port from NICPreferences.BROKER_INTERFACE"
@@ -118,9 +120,22 @@ function best_interface_broker(
 
     # Default to `localhost` if a suitable environment variable containing the
     # broker address is not set
-    broker_addr = "localhost"
-    for env_add in NICPreferences.BROKER_HOST_ENV
+    broker_addr        = "localhost"
+    broker_addr_source = "default"
+    for broker_addr_source in NICPreferences.BROKER_HOST_ENV
+        if broker_addr_source in keys(ENV)
+            @debug "`$(broker_addr_source)` find in environment => using as broker address"
+            broker_addr = ENV[broker_addr_source]
+            break # break on first occurrence
+        end
     end
+
+    # Interpret broker_addr as a hostlist
+    broker_addr = Hostlists.Hostlist(broker_addr) |> first
+    @debug "Using broker server address = `$(broker_addr)` (from `ENV[$(broker_addr_source)]`)"
+
+    ip, port = broker_ip_port(ipv)
+    return Broker.query_broker(ip, UInt32(port), data)
 end
 
 function best_interfaces(data::Vector{Interface})

src/broker.jl

@@ -154,7 +154,7 @@ function query_broker(
         try
             conn_txn = connect(ip, txn_port.port)
         catch e
-            if (e isa Base.IOError) && (e.code == -61)
+            if (e isa Base.IOError) && (e.code in (-61, -111))
                 @debug "Server not ready, retyring"
                 sleep(timeout)
             else

src/nic_preferences.jl

@@ -169,7 +169,7 @@ const _broker_mode_default      = setting(:mode, USE_DISABLED)
 const _broker_interface_default = setting(:interface, ".*", MATCH_REGEX, 3000)
 const _broker_mode      = @load_preference("broker_mode")
 const _broker_interface = @load_preference("broker_interface")
-const BORKER_HOST_ENV   = @load_preference("broker_host_env", [])
+const BROKER_HOST_ENV   = @load_preference("broker_host_env", [])
 
 const NAME_SELECTOR  = ModeSettings(@val_or_default(:_name_selector_mode))
 const HWLOC_SELECTOR = ModeSettings(@val_or_default(:_hwloc_selector_mode))
@@ -209,7 +209,6 @@ function configure!(;
     @assert check_interface(preferred_interface)
     @assert check_interface(broker_interface)
 
-    println(broker_host_env)
     @set_preferences!(
         "name_selector_mode"       => name_selector_mode,
         "preferred_interface"      => preferred_interface,