已合并
Remove judgement of done in watchdog to slove the stucking problem of error exit #13113
AtomGit-Bot创建于 2024年7月23日
Remove judgement of done in watchdog to slove the stucking problem of error exit #13113
已合并
从refs/pull/13113/head合入到master
共 4 个文件变更+57-15
| @@ -1,5 +1,6 @@ | |||
| 1 | import os | 1 | import os |
| 2 | import unittest | 2 | import unittest |
| 3 | +import time | ||
| 3 | import torch.distributed.run as launch | 4 | import torch.distributed.run as launch |
| 4 | from torch_npu.testing.testcase import run_tests, TestCase | 5 | from torch_npu.testing.testcase import run_tests, TestCase |
| 5 | from torch_npu.testing.common_distributed import skipIfUnsupportMultiNPU | 6 | from torch_npu.testing.common_distributed import skipIfUnsupportMultiNPU |
| @@ -11,15 +12,12 @@ def path(script): | |||
| 11 | 12 | ||
| 12 | class ElasticLaunchTest(TestCase): | 13 | class ElasticLaunchTest(TestCase): |
| 13 | 14 | ||
| 14 | - def test_communicate_npu_watchdog_timeout(self): | 15 | + def test_npu_watchdog_timeout(self): |
| 15 | try: | 16 | try: |
| 16 | error = None | 17 | error = None |
| 17 | launch.main( | 18 | launch.main( |
| 18 | [ | 19 | [ |
| 19 | - "--run-path", | ||
| 20 | - "--nnodes=1", | ||
| 21 | "--nproc-per-node=2", | 20 | "--nproc-per-node=2", |
| 22 | - "--monitor-interval=1", | ||
| 23 | path("watchdog/watchdog_base.py"), | 21 | path("watchdog/watchdog_base.py"), |
| 24 | ] | 22 | ] |
| 25 | ) | 23 | ) |
| @@ -30,5 +28,25 @@ class ElasticLaunchTest(TestCase): | |||
| 30 | raise RuntimeError("Test case fail") | 28 | raise RuntimeError("Test case fail") |
| 31 | 29 | ||
| 32 | 30 | ||
| 31 | + | ||
| 32 | + def test_npu_watchdog_quick_exit(self): | ||
| 33 | + start_time = time.time() | ||
| 34 | + try: | ||
| 35 | + launch.main( | ||
| 36 | + [ | ||
| 37 | + "--nproc-per-node=2", | ||
| 38 | + path("watchdog/watchdog_quick_exit.py"), | ||
| 39 | + ] | ||
| 40 | + ) | ||
| 41 | + except Exception: | ||
| 42 | + print("Program fail and exit") | ||
| 43 | + | ||
| 44 | + end_time = time.time() | ||
| 45 | + excution_time = end_time - start_time | ||
| 46 | + if excution_time > 120: | ||
| 47 | + print(f"Excution time using time.time(): {excution_time} seconds") | ||
| 48 | + raise RuntimeError("Test case fail") | ||
| 49 | + | ||
| 50 | + | ||
| 33 | if __name__ == "__main__": | 51 | if __name__ == "__main__": |
| 34 | run_tests() | 52 | run_tests() |
| @@ -12,15 +12,12 @@ def main(): | |||
| 12 | device = torch.device('npu:{}'.format(local_rank)) | 12 | device = torch.device('npu:{}'.format(local_rank)) |
| 13 | torch.npu.set_device(device) | 13 | torch.npu.set_device(device) |
| 14 | 14 | ||
| 15 | - dist.init_process_group(backend='hccl', timeout=datetime.timedelta(seconds=1)) | 15 | + dist.init_process_group(backend='hccl', rank=rank, world_size=2, timeout=datetime.timedelta(seconds=10)) |
| 16 | - tensor = torch.tensor([1024]).npu(non_blocking=True) | 16 | + tensor = torch.tensor(1).npu() |
| 17 | 17 | ||
| 18 | - for i in range(10): | 18 | + dist.all_reduce(tensor) |
| 19 | + if rank == 0: | ||
| 19 | dist.all_reduce(tensor) | 20 | dist.all_reduce(tensor) |
| 20 | - if rank == 1: | ||
| 21 | - time.sleep(10) | ||
| 22 | - | ||
| 23 | - dist.destroy_process_group() | ||
| 24 | 21 | ||
| 25 | 22 | ||
| 26 | if __name__ == "__main__": | 23 | if __name__ == "__main__": |
| @@ -0,0 +1,26 @@ | |||
| 1 | +import os | ||
| 2 | +import time | ||
| 3 | +import datetime | ||
| 4 | +import torch.distributed as dist | ||
| 5 | +import torch | ||
| 6 | +import torch_npu | ||
| 7 | + | ||
| 8 | + | ||
| 9 | +def main(): | ||
| 10 | + rank = int(os.environ['RANK']) | ||
| 11 | + local_rank = int(os.environ['LOCAL_RANK']) | ||
| 12 | + device = torch.device('npu:{}'.format(local_rank)) | ||
| 13 | + torch.npu.set_device(device) | ||
| 14 | + dist.init_process_group(backend='hccl', rank=rank, world_size=2, timeout=datetime.timedelta(seconds=120)) | ||
| 15 | + tensor = torch.tensor(1).npu() | ||
| 16 | + dist.all_reduce(tensor) | ||
| 17 | + if rank == 0: | ||
| 18 | + x1 = torch.randn(3).float().npu() | ||
| 19 | + x2 = torch.randn(1).long().npu() | ||
| 20 | + x3 = torch.randn(1).float().npu() | ||
| 21 | + y = torch.addcmul(x1, x2, x3) | ||
| 22 | + dist.all_reduce(tensor) | ||
| 23 | + | ||
| 24 | + | ||
| 25 | +if __name__ == "__main__": | ||
| 26 | + main() | ||
| @@ -796,9 +796,8 @@ const std::vector<uint64_t>& ProcessGroupHCCL::groupRanks() const | |||
| 796 | void ProcessGroupHCCL::workCleanupLoop() | 796 | void ProcessGroupHCCL::workCleanupLoop() |
| 797 | { | 797 | { |
| 798 | bool needSetDevice = true; | 798 | bool needSetDevice = true; |
| 799 | - bool done = false; | ||
| 800 | std::list<ProcessGroupHCCL::WorkHCCL> completedWorkList; | 799 | std::list<ProcessGroupHCCL::WorkHCCL> completedWorkList; |
| 801 | - while (!done || !terminateProcessGroup_.load()) { | 800 | + while (!terminateProcessGroup_.load()) { |
| 802 | std::unique_lock<std::mutex> lock(workMetaListMutex_); | 801 | std::unique_lock<std::mutex> lock(workMetaListMutex_); |
| 803 | // We busy-poll the work vector every kWatchdogThreadSleepMillis | 802 | // We busy-poll the work vector every kWatchdogThreadSleepMillis |
| 804 | // milliseconds as long as the atomic is True. | 803 | // milliseconds as long as the atomic is True. |
| @@ -875,8 +874,10 @@ void ProcessGroupHCCL::workCleanupLoop() | |||
| 875 | ++it; | 874 | ++it; |
| 876 | } | 875 | } |
| 877 | } | 876 | } |
| 878 | - | 877 | + } |
| 879 | - done = workMetaList_.empty(); | 878 | + if (terminateProcessGroup_.load()) { |
| 879 | + std::unique_lock<std::mutex> lock(workMetaListMutex_); | ||
| 880 | + workMetaList_.clear(); | ||
| 880 | } | 881 | } |
| 881 | } | 882 | } |
| 882 | 883 | ||