^j#dUdZddlZddlZddlmZddlmZmZddlm Z ddl m Z m Z ddl mZddlmcmcmcmZej*ZiZeeefed<ded efd ZeGd d Zd ed e dfdZGddej>Z eGdde Z!eGdde Z"y)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_classreturncDt|dr|t|j<|S)z0Register a barrier class in the global registry. barrier_type)hasattrr r )r s ~/var/www/ramen.bs-engineer-server.com/venv/lib/python3.12/site-packages/torch/distributed/checkpoint/_experimental/barriers.pyregister_barrierrs#}n-7D334 cLeZdZUdZdZedzed<eeZ eee fed<y) BarrierConfiga& 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____doc__r str__annotations__rdictrrrrrr"s.< $L#*##(#>L$sCx.>rrbarrier_configBarrierc |jy|jtvr6td|jdttj t|j}|di|j S)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)rr s rcreate_barrier_from_configr$Fs""*""*::$^%@%@$AB $%5%:%:%< => @  %^%@%@AM  7>66 77rcjeZdZdZej deeeffdZ ej ddZ y)ra( 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 y)a+ 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__zBarrier.__init__krNcy)z Execute a synchronization barrier. This method uses the barrier_prefix provided during initialization to coordinate synchronization across processes. Nrr(s rexecute_barrierzBarrier.execute_barrierwr*rr N) rrrrabcabstractmethodrrrr)r-rrrrrbsJ  c3h     rc(eZdZdZdZ ddZddZy) DistBarriera2 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_barrierNcBtjs tdy)a/ 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_initializedAssertionErrorr,s rr)zDistBarrier.__init__s!""$ !UV V%rc,tjy)zd Execute a synchronization barrier using the prefix provided during initialization. N)r5barrierr,s rr-zDistBarrier.execute_barriers rr.)rrrrr r)r-rrrr2r2s""L W  Wrr2cDeZdZdZdZdedededededed ed efd Zdd Z y )TCPStoreBarrieraS 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 &tjd|||||||| t|_||_||_||_||_tj|t||j t||j dk(|_ y)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_secsr5TCPStoreintr _tcp_store) r(r=r>r?r@rArBrCrDs rr)zTCPStoreBarrier.__init__s0   G      % 1  07y#-("3)--   ..&?@((A-  rNc |j}tjd||jdtdt fd}|j j||jt |j|tj|j |j|t |j|z|j|xxdz cc<y)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 d|S)NrUr)rUs r _rank_keyz2TCPStoreBarrier.execute_barrier.._rank_keys$= r)storerG key_prefixN) rMrJrKrPrRrrSsetrNrL store_utilr9rO)r(r?rWs rr-zTCPStoreBarrier.execute_barriers-- A      !C !C !  d'' ( ++N; < //..T%@%@%P!QQ   ##N3q83rr.) rrrrr rRrboolr)r-rrrr;r;sgL6 6 6  6 $' 6 04 6 6 6 6 p!9rr;)#rr/logging collectionsr dataclassesrrdatetimertypingrrtorch.distributed distributedr5%torch.distributed.elastic.utils.storeelasticutilsrXr\ getLoggerrJr rrtyperrrr$ABCrr2r;rrrrks ( ::    %'$sDy/&DT  ? ?  ?F8!8 i88 cgg > '  Fe9ge9e9r