Perform a store-based barrier synchronization across all ranks. This barrier uses a TCP store directly rather than a process group, allowing all ranks to synchronize regardless of which process group they belong to. Args: store: The TCPStore ins
(self, timeout: timedelta = timedelta(minutes=5))
| 523 | logger.info(f"[rank{self._rank}] init process group successfully.") |
| 524 | |
| 525 | def store_based_barrier(self, timeout: timedelta = timedelta(minutes=5)) -> None: |
| 526 | """ |
| 527 | Perform a store-based barrier synchronization across all ranks. |
| 528 | |
| 529 | This barrier uses a TCP store directly rather than a process group, |
| 530 | allowing all ranks to synchronize regardless of which process group |
| 531 | they belong to. |
| 532 | |
| 533 | Args: |
| 534 | store: The TCPStore instance to use for synchronization. |
| 535 | """ |
| 536 | torch.distributed.distributed_c10d._store_based_barrier( |
| 537 | rank=self._rank, |
| 538 | store=self._store, |
| 539 | group_name="parameter_server_barrier", |
| 540 | rendezvous_count=self._world_size, |
| 541 | timeout=timeout, |
| 542 | ) |
| 543 | |
| 544 | def update( |
| 545 | self, |