ȅi#h%SrSSKrSSKrSSKJr SSKJrJr SSKJ r SSK J r J r SSK Jr SSKJs Js Js Jr \R*"5r0r\\\4\S'S\S \4S jr\"S S 55rS \S \ S4Sjr"SS\R>5r \"SS\ 55r!\"SS\ 55r"g)a/ Barrier implementations for synchronizing distributed checkpoint operations. This module provides abstract and concrete barrier implementations that ensure all ranks in a distributed training environment complete their checkpoint operations before proceeding, which is essential for data consistency. N)Counter) dataclassfield) timedelta)AnyOptionalBARRIER_REGISTRY barrier_classreturncN[US5(aU[UR'U$)z0Register a barrier class in the global registry. barrier_type)hasattrr r )r s m/home/james-whalen/.local/lib/python3.13/site-packages/torch/distributed/checkpoint/_experimental/barriers.pyregister_barrierrs&}n--7D334 cN\rSrSr%SrSr\S-\S'\"\ S9r \ \\ 4\S'Sr g) BarrierConfig"a Configuration for barrier construction. This class provides a flexible way to configure different barrier implementations with their specific constructor arguments. The barrier type will be looked up from a registry and instantiated with rank_info and barrier_args. Attributes: barrier_type: A string identifying the barrier type (e.g., "tcp_store"). If None, no barrier will be used. barrier_args: Dictionary of arguments to pass to the barrier constructor. rank_info will be automatically injected as the first argument. Examples: # No barrier BarrierConfig() # TCPStore barrier BarrierConfig( barrier_type="tcp_store", barrier_args={ 'timeout_barrier_init_secs': 30, 'barrier_prefix_list': ['checkpoint'], 'use_checkpoint_barrier_tcpstore_libuv': False, 'tcpstore_port': 12345, 'master_address': 'localhost' } ) Nr )default_factory barrier_args) __name__ __module__ __qualname____firstlineno____doc__r str__annotations__rdictrr__static_attributes__rrrrr"s.< $L#*##(#>L$sCx.>rrbarrier_configBarrierc URcgUR[;a6[SURS[[R 5535e[URnU"S0UR D6$)a Create a barrier instance from BarrierConfig. Args: barrier_config: Configuration for barrier construction. Returns: Barrier instance or None if no barrier type is configured. Raises: ValueError: If the barrier_type is not found in the registry. NzUnknown barrier type: z. Available types: r)r r ValueErrorlistkeysr)r!r s rcreate_barrier_from_configr'Fs""*""*::$^%@%@$AB $%5%:%:%< => @  %^%@%@AM  7>66 77rcv\rSrSrSr\R S\\\ 44Sj5r \R SSj5r Sr g) r"ba Abstract base class for synchronization barriers. A barrier ensures that all ranks in a distributed environment reach a certain point in execution before any rank proceeds further, which is essential for coordinating operations like checkpointing across multiple processes. kwargsc g)z Initialize a barrier. Args: **kwargs: Keyword arguments for specific barrier implementations. Common arguments may include rank information, barrier prefixes, timeout settings, and other barrier-specific configuration. Nr)selfr*s r__init__Barrier.__init__krNcg)z Execute a synchronization barrier. This method uses the barrier_prefix provided during initialization to coordinate synchronization across processes. Nrr,s rexecute_barrierBarrier.execute_barrierwr/rrr N) rrrrrabcabstractmethodrrrr-r2r rrrr"r"bsJ  c3h     rc4\rSrSrSrSrSSjrSSjrSrg) DistBarriera A barrier implementation using PyTorch's distributed barrier for synchronization. This barrier uses the built-in torch.distributed.barrier() function to coordinate synchronization across multiple processes. It's simpler than TCPStoreBarrier but requires an initialized process group. dist_barrierNcN[R"5(d [S5eg)z Initialize a DistBarrier. This barrier requires an initialized PyTorch distributed process group. No additional arguments are needed as it uses the current process group. Raises: AssertionError: If the distributed process group is not initialized. z2DistBarrier requires an initialized process group.N)distis_initializedAssertionErrorr1s rr-DistBarrier.__init__s$""$$ !UV V%rc.[R"5 g)zT Execute a synchronization barrier using the prefix provided during initialization. N)r<barrierr1s rr2DistBarrier.execute_barriers rrr4) rrrrrr r-r2r rrrr8r8s""L W  Wrr8cP\rSrSrSrSrS\S\S\S\S\S \S \S \4S jr SSjr Sr g )TCPStoreBarriera; A barrier implementation using PyTorch's TCPStore for synchronization. This barrier uses a TCP-based distributed key-value store to coordinate synchronization across multiple processes. It uses a single TCP store for all barrier operations, with different prefixes to distinguish between different barrier types. tcp_store global_rankglobal_world_sizebarrier_prefixtimeout_barrier_init_secs%use_checkpoint_barrier_tcpstore_libuv tcpstore_portmaster_address timeout_secsc [RSUUUUUUUU5 [5UlX0lXlX lXl[R"U[U5UR [US9UR S:HS9Ul g)a~ Initialize a TCPStoreBarrier. Args: global_rank: The rank of the current process in the distributed environment. global_world_size: The total number of processes in the distributed environment. barrier_prefix: A string prefix to identify this specific barrier. timeout_barrier_init_secs: Timeout in seconds for initializing the TCPStore. use_checkpoint_barrier_tcpstore_libuv: Whether to use libuv for the TCPStore. tcpstore_port: Port number for the TCPStore. master_address: Address of the master node for the TCPStore. timeout_secs: Maximum time in seconds to wait for all ranks to reach the barrier. zInitializing TCPStore master_address=%s tcpstore_port=%s rank=%s world_size=%s barrier_prefix=%s timeout_barrier_init_secs=%s use_checkpoint_barrier_tcpstore_libuv=%s timeout_secs=%s)secondsr) world_sizetimeout is_masterN) loggerinfor_tcp_store_barrier_seq_barrier_prefix _global_rank_global_world_size _timeout_secsr<TCPStoreintr _tcp_store) r,rGrHrIrJrKrLrMrNs rr-TCPStoreBarrier.__init__s0   G      % 1  07y#-("3)--   ..&?@((A-  rNc URn[RSUUR5 S[S[ 4SjnUR RU"UR5[ URU55 [R"UR URU[ URU5-S9 URU==S- ss'g)a Execute a synchronization barrier using the prefix provided during initialization. The implementation uses a sequence number that is incremented every time a barrier is reached. The sequence number is per barrier prefix to allow different barriers to operate concurrently. z3Executing barrier barrier_prefix=%s timeout_secs=%srankr c SU3$)Nr`r)r`s r _rank_key2TCPStoreBarrier.execute_barrier.._rank_keys$= r)storerQ key_prefixN) rWrTrUrZr\rr]setrXrV store_utilrArY)r,rIrbs rr2TCPStoreBarrier.execute_barriers-- A      !C !C !  d'' ( ++N; < //..T%@%@%P!QQ   ##N3q83r)rWrXrYr]rVrZr4) rrrrrr r\rboolr-r2r rrrrDrDsgL6 6 6  6 $' 6 04 6 6 6 6 p!9rrD)#rr5logging collectionsr dataclassesrrdatetimertypingrrtorch.distributed distributedr<%torch.distributed.elastic.utils.storeelasticutilsrdrh getLoggerrTr rrtyperrrr'ABCr"r8rDrrrrxs ( ::    %'$sDy/&DT  ? ?  ?F8!8 i88 cgg > '  Fe9ge9e9r