diff --git a/monai/apps/nnunet/nnunetv2_runner.py b/monai/apps/nnunet/nnunetv2_runner.py index 5d5c82801a..dd6baeaa61 100644 --- a/monai/apps/nnunet/nnunetv2_runner.py +++ b/monai/apps/nnunet/nnunetv2_runner.py @@ -17,6 +17,7 @@ import re import shlex import subprocess +import threading from typing import Any import monai @@ -740,17 +741,39 @@ def train_parallel( f"log '.txt' inside '{os.path.join(self.nnunet_results, self.dataset_name)}'" ) for stage in all_cmds: - processes = [] - for device_id in stage: - if not stage[device_id]: + threads = [] + for device_id, device_cmds in stage.items(): + if not device_cmds: continue - cmd_str = "; ".join(shlex.join(cmd) for cmd, _ in stage[device_id]) - env = stage[device_id][0][1] - logger.info(f"Current running command on GPU device {device_id}:\n{cmd_str}\n") - processes.append(subprocess.Popen(cmd_str, shell=True, env=env, stdout=subprocess.DEVNULL)) - # finish this stage first - for p in processes: - p.wait() + logger.info( + f"Current running commands on GPU device {device_id}:\n" + + "\n".join(shlex.join(cmd) for cmd, _ in device_cmds) + + "\n" + ) + # one worker per device runs that device's commands sequentially, in order; + # different devices run concurrently. No shell is invoked (shell=False). + thread = threading.Thread(target=self._run_device_commands, args=(device_cmds,)) + thread.start() + threads.append(thread) + # finish this stage before starting the next + for thread in threads: + thread.join() + + @staticmethod + def _run_device_commands(device_cmds: list) -> None: + """ + Run one device's argument-list commands sequentially without invoking a shell. + + Args: + device_cmds: a list of ``(cmd, env)`` pairs, where ``cmd`` is an argument list and + ``env`` is the environment for that command. + + Each command is run in its original order with ``shell=False``, keeping its associated + environment. A non-zero exit status does not stop the remaining commands for the device, + matching the previous ``"; ".join(...)`` shell behavior. + """ + for cmd, env in device_cmds: + subprocess.run(cmd, env=env, stdout=subprocess.DEVNULL, check=False) def validate_single_model(self, config: str, fold: int, **kwargs: Any) -> None: """ diff --git a/tests/apps/nnunet/test_nnunetv2_runner_command.py b/tests/apps/nnunet/test_nnunetv2_runner_command.py index 506c30fad0..09ac997c93 100644 --- a/tests/apps/nnunet/test_nnunetv2_runner_command.py +++ b/tests/apps/nnunet/test_nnunetv2_runner_command.py @@ -74,5 +74,40 @@ def test_validate_emits_bare_val_flag(self): self.assertNotIn("True", cmd) +class TestRunDeviceCommands(unittest.TestCase): + """``train_parallel`` runs each device's commands without a shell (see issue #8992).""" + + def test_runs_in_order_without_shell(self): + cmd_a = ["python", "-m", "nnunetv2", "train", "0"] + cmd_b = ["python", "-m", "nnunetv2", "train", "1"] + device_cmds = [(cmd_a, {"CUDA_VISIBLE_DEVICES": "0"}), (cmd_b, {"CUDA_VISIBLE_DEVICES": "0"})] + + with mock.patch("monai.apps.nnunet.nnunetv2_runner.subprocess.run") as run: + nnUNetV2Runner._run_device_commands(device_cmds) + + # both commands ran, in order + self.assertEqual(run.call_count, 2) + self.assertEqual(run.call_args_list[0].args[0], cmd_a) + self.assertEqual(run.call_args_list[1].args[0], cmd_b) + for call, (expected_cmd, expected_env) in zip(run.call_args_list, device_cmds): + # first positional arg is the argument list itself (not a joined string) + self.assertIsInstance(call.args[0], list) + self.assertEqual(call.args[0], expected_cmd) + # no shell is invoked + self.assertFalse(call.kwargs.get("shell", False)) + # each command keeps its own environment + self.assertEqual(call.kwargs.get("env"), expected_env) + + def test_continues_after_nonzero_exit(self): + # a non-zero exit for the first command must not stop the remaining commands. + device_cmds = [(["cmd", "0"], {}), (["cmd", "1"], {})] + with mock.patch("monai.apps.nnunet.nnunetv2_runner.subprocess.run") as run: + nnUNetV2Runner._run_device_commands(device_cmds) + # check=False is used so no exception is raised and both commands are attempted. + self.assertEqual(run.call_count, 2) + for call in run.call_args_list: + self.assertFalse(call.kwargs.get("check", False)) + + if __name__ == "__main__": unittest.main()