|
18 | 18 | import shlex |
19 | 19 | import subprocess |
20 | 20 | import warnings |
| 21 | +from concurrent.futures import ThreadPoolExecutor |
21 | 22 | from typing import Any |
22 | 23 |
|
23 | 24 | import monai |
@@ -711,7 +712,11 @@ def train_parallel( |
711 | 712 | **kwargs: Any, |
712 | 713 | ) -> None: |
713 | 714 | """ |
714 | | - Create the line command for subprocess call for parallel training. |
| 715 | + Launch subprocesses for parallel training. |
| 716 | +
|
| 717 | + The commands for each GPU run sequentially on that device, while different devices run in |
| 718 | + parallel. Each stage waits for all of its devices to finish before the next stage starts. |
| 719 | +
|
715 | 720 | Note: to set the number of GPUs to use, use ``gpu_id_for_all`` instead of the `CUDA_VISIBLE_DEVICES` |
716 | 721 | environment variable. |
717 | 722 |
|
@@ -742,19 +747,19 @@ def train_parallel( |
742 | 747 | f"log '.txt' inside '{os.path.join(self.nnunet_results, self.dataset_name)}'" |
743 | 748 | ) |
744 | 749 | for stage in all_cmds: |
745 | | - max_cmds_per_gpu = max((len(cmds) for cmds in stage.values()), default=0) |
746 | | - for cmd_index in range(max_cmds_per_gpu): |
747 | | - processes = [] |
748 | | - for device_id, gpu_cmds in stage.items(): |
749 | | - if cmd_index >= len(gpu_cmds): |
750 | | - continue |
751 | | - cmd, env = gpu_cmds[cmd_index] |
| 750 | + device_cmds = [(device_id, gpu_cmds) for device_id, gpu_cmds in stage.items() if gpu_cmds] |
| 751 | + if not device_cmds: |
| 752 | + continue |
| 753 | + |
| 754 | + def _run_device_commands(item): |
| 755 | + device_id, gpu_cmds = item |
| 756 | + for cmd, env in gpu_cmds: |
752 | 757 | cmd_str = shlex.join(cmd) |
753 | 758 | logger.info(f"Current running command on GPU device {device_id}:\n{cmd_str}\n") |
754 | | - processes.append(subprocess.Popen(cmd, shell=False, env=env, stdout=subprocess.DEVNULL)) |
755 | | - # finish this command round on all GPUs before starting the next command |
756 | | - for p in processes: |
757 | | - p.wait() |
| 759 | + subprocess.Popen(cmd, shell=False, env=env, stdout=subprocess.DEVNULL).wait() |
| 760 | + |
| 761 | + with ThreadPoolExecutor(max_workers=len(device_cmds)) as executor: |
| 762 | + list(executor.map(_run_device_commands, device_cmds)) |
758 | 763 |
|
759 | 764 | def validate_single_model(self, config: str, fold: int, **kwargs: Any) -> None: |
760 | 765 | """ |
|
0 commit comments