fix(hpc): avoid invalid Dask worker interfaces

Author: harryswift01

CodeEntropy/core/dask_clusters.py

@@ -31,55 +31,40 @@ def check_slurm_env(self) -> None:
 
         Some HPC systems require this variable to be unset for correct CPU binding.
         """
-        os.environ.pop("SLURM_CPU_BIND", None)
+        if "SLURM_CPU_BIND" in os.environ:
+            os.environ.pop("SLURM_CPU_BIND")
 
     def system_network_interface(self) -> str:
         """
-        Select the most appropriate network interface for HPC communication.
+        Get best candidate for HPC network interface from commonly known ones.
 
-        If args.hpc_interface is provided, that value is used directly. Otherwise,
-        commonly used HPC interfaces are preferred. Loopback and container interfaces
-        are avoided because Dask workers on other nodes cannot connect to a scheduler
-        advertised on 127.0.0.1.
-
-        Returns:
-            str: Name of selected network interface.
-
-        Raises:
-            RuntimeError: If no suitable non-loopback interface can be found.
+        This deliberately follows the WaterEntropy behaviour and only selects from
+        known HPC-safe interfaces. It avoids selecting arbitrary interfaces such as
+        eno1, which may exist on the master node but not on worker nodes.
         """
-        configured = getattr(self.args, "hpc_interface", None)
-        if configured:
-            return configured
-
-        preferred_nics = ["bond0", "ib0", "hsn0", "eth0"]
+        hpc_nics = ["bond0", "ib0", "hsn0", "eth0"]
         interfaces = list(psutil.net_if_addrs().keys())
 
-        for iface in preferred_nics:
+        for iface in hpc_nics:
             if iface in interfaces:
                 return iface
 
-        for iface in interfaces:
-            if not iface.startswith(("lo", "docker", "veth")):
-                return iface
-
         raise RuntimeError(
-            "Could not find a non-loopback network interface for Dask workers. "
-            f"Available interfaces: {interfaces}. Set 'hpc_interface' in config.yaml."
+            "Could not find a known HPC network interface. "
+            f"Available interfaces: {interfaces}. "
+            "Expected one of: bond0, ib0, hsn0, eth0."
         )
 
     def slurm_directives(self) -> tuple[list[str], list[str]]:
         """
-        Build SLURM job directives and skip list.
+        Process extra SLURM directives and directives to be skipped.
 
         Returns:
-            Tuple[List[str], List[str]]:
-                - Extra SLURM directives
-                - Directives to skip
+            Tuple containing extra directives and skipped directives.
         """
         args = self.args
+
         extra: list[str] = []
-        skip: list[str] = ["--mem"]
 
         if args.hpc_account:
             extra.append(f'--account="{args.hpc_account}"')
@@ -88,14 +73,16 @@ def slurm_directives(self) -> tuple[list[str], list[str]]:
         if args.hpc_constraint:
             extra.append(f'--constraint="{args.hpc_constraint}"')
 
+        skip = ["--mem"]
+
         return extra, skip
 
     def slurm_prologues(self) -> list[str]:
         """
-        Build SLURM job prologue commands for environment setup.
+        Process environment setup commands for the SLURM worker job script.
 
         Returns:
-            List[str]: Shell commands executed before job start.
+            List of shell commands executed before the Dask worker starts.
         """
         args = self.args
         prologue: list[str] = []
@@ -115,10 +102,10 @@ def slurm_prologues(self) -> list[str]:
 
     def configure_cluster(self) -> Client:
         """
-        Configure and launch a SLURM-backed Dask cluster.
+        Configure a SLURM-backed Dask cluster.
 
         Returns:
-            Client: Dask distributed client connected to cluster.
+            Dask distributed client connected to the SLURMCluster.
         """
         args = self.args
 
@@ -140,7 +127,6 @@ def configure_cluster(self) -> Client:
             shebang="#!/bin/bash --login",
             local_directory="$PWD",
             interface=iface,
-            scheduler_options={"interface": iface},
             job_script_prologue=prologue,
         )
 
@@ -155,9 +141,9 @@ def configure_cluster(self) -> Client:
 
     def submit_master(self) -> None:
         """
-        Submit a SLURM job that runs a master Dask orchestration process.
+        Submit a SLURM job that runs the master CodeEntropy process.
 
-        This generates a temporary SLURM script and submits it via `sbatch`.
+        This generates a temporary SLURM script and submits it via sbatch.
         """
         cli = list(sys.argv[1:])
         if "--submit" in cli:

tests/unit/CodeEntropy/core/dask_clusters/test_dask_clusters.py

@@ -24,7 +24,6 @@ def args_helper(args_list):
     parser.add_argument("--hpc-nodes", type=int, default=4)
     parser.add_argument("--hpc-processes", type=int, default=20)
     parser.add_argument("--hpc-walltime", type=str, default="24:00:00")
-    parser.add_argument("--hpc-interface", type=str, default=None)
     parser.add_argument("--hpc-modules", nargs="+", default=None)
 
     parser.add_argument("--conda-env", type=str, default="codeentropy")
@@ -170,7 +169,17 @@ def test_slurm_prologues_includes_hpc_modules():
 
 
 @mock.patch("psutil.net_if_addrs")
-def test_system_network_interface_prefers_ib0(net_if_addrs):
+def test_system_network_interface_prefers_bond0(net_if_addrs):
+    net_if_addrs.return_value = {"bond0": [], "ib0": [], "eth0": []}
+
+    args = args_helper([])
+    manager = HPCDaskManager(args)
+
+    assert manager.system_network_interface() == "bond0"
+
+
+@mock.patch("psutil.net_if_addrs")
+def test_system_network_interface_prefers_ib0_when_bond0_missing(net_if_addrs):
     net_if_addrs.return_value = {"ib0": [], "eth0": []}
 
     args = args_helper([])
@@ -180,33 +189,37 @@ def test_system_network_interface_prefers_ib0(net_if_addrs):
 
 
 @mock.patch("psutil.net_if_addrs")
-def test_system_network_interface_uses_configured_interface(net_if_addrs):
-    net_if_addrs.return_value = {"lo": [], "ib0": []}
+def test_system_network_interface_prefers_hsn0_when_bond0_and_ib0_missing(
+    net_if_addrs,
+):
+    net_if_addrs.return_value = {"hsn0": [], "eth0": []}
 
-    args = args_helper(["--hpc-interface", "custom0"])
+    args = args_helper([])
     manager = HPCDaskManager(args)
 
-    assert manager.system_network_interface() == "custom0"
+    assert manager.system_network_interface() == "hsn0"
 
 
 @mock.patch("psutil.net_if_addrs")
-def test_system_network_interface_falls_back_to_non_loopback_interface(net_if_addrs):
-    net_if_addrs.return_value = {"lo": [], "ens5": [], "docker0": []}
+def test_system_network_interface_prefers_eth0_when_only_eth0_known_interface(
+    net_if_addrs,
+):
+    net_if_addrs.return_value = {"eth0": [], "eno1": []}
 
     args = args_helper([])
     manager = HPCDaskManager(args)
 
-    assert manager.system_network_interface() == "ens5"
+    assert manager.system_network_interface() == "eth0"
 
 
 @mock.patch("psutil.net_if_addrs")
-def test_system_network_interface_raises_without_non_loopback_interface(net_if_addrs):
-    net_if_addrs.return_value = {"lo": [], "docker0": [], "veth123": []}
+def test_system_network_interface_raises_without_known_hpc_interface(net_if_addrs):
+    net_if_addrs.return_value = {"lo": [], "docker0": [], "eno1": []}
 
     args = args_helper([])
     manager = HPCDaskManager(args)
 
-    with pytest.raises(RuntimeError, match="Could not find a non-loopback"):
+    with pytest.raises(RuntimeError, match="Could not find a known HPC network"):
         manager.system_network_interface()
 
 
@@ -424,7 +437,7 @@ def test_configure_cluster_writes_job_script(
     slurm_cluster.assert_called_once()
     _, kwargs = slurm_cluster.call_args
     assert kwargs["interface"] == "ib0"
-    assert kwargs["scheduler_options"] == {"interface": "ib0"}
+    assert "scheduler_options" not in kwargs
 
     cluster_instance.scale.assert_called_once_with(jobs=4)
     client.assert_called_once_with(cluster_instance)