+ j#VaRt 0tRt^RIt^RIt^RIHt^RIHtHt^RI H t ^RI H t H t ^RIHt^RIHuHuHuHt]P,!4t/t]^kRRlt]!RR 44tR R lt!R R ]P84t]!RR]44t]!RR]44tR#)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)AnyOptionalc0V^8dQhR\R\/#) barrier_classreturn)type)formats"t/home/luca/screen-checking/.venv/lib/python3.14/site-packages/torch/distributed/checkpoint/_experimental/barriers.py __annotate__rsDTcP\VR4'dV\VP&V#)z0Register a barrier class in the global registry. barrier_type)hasattrBARRIER_REGISTRYr)r s&rregister_barrierrs&}n--7D334 rcFa]tRt^"toRtRt]!]R7tV3Rlt Rt Vt R#) 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' } ) N)default_factorycV<V^8dQh/S[R,;R&S[S[S[3,;R&#)r Nr barrier_args)strdictr)r __classdict__s"rrBarrierConfig.__annotate__"s/@*#ABsCx.>Cr) __name__ __module__ __qualname____firstlineno____doc__rrrr__annotate_func____static_attributes____classdictcell__rs@rrr"s#< $L#(#>LCrrc>V^8dQhR\R\R,/#)r barrier_configr Barrier)rr)r s"rrrFs!88!8 i8rc  VPfR#VP\9d7\RVP R\\P 44 24h\VP,pV!R/VP B#)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)rr ValueErrorlistkeysr)r*r s& rcreate_barrier_from_configr0Fs""*""*::$^%@%@$AB $%5%:%:%< => @  %^%@%@AM  7>66 77rca]tRt^btoRt]P V3RlRl4t]P V3RlRl4tRt Vt R#)r+a 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. c6<V^8dQhRS[S[S[3,/#)r kwargs)rrr)r rs"rrBarrier.__annotate__ls  c3h rc R#)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)selfr3s&,r__init__Barrier.__init__krc<V^8dQhRR/#r r Nr)r rs"rrr4xs   rc R#)z Execute a synchronization barrier. This method uses the barrier_prefix provided during initialization to coordinate synchronization across processes. Nrr6s&rexecute_barrierBarrier.execute_barrierwr9rrN) r r!r"r#r$abcabstractmethodr7r>r&r'r(s@rr+r+bs>      rr+cLa]tRt^toRtRtV3RlRltV3RlRltRtVt R#) 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_barrierc<V^8dQhRR/#r;r)r rs"rrDistBarrier.__annotate__s W W  Wrc R\P!4'g \R4hR#)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_initializedAssertionErrorr=s&rr7DistBarrier.__init__s$""$$ !UV V%rc<V^8dQhRR/#r;r)r rs"rrrFsrc 0\P!4R#)zT Execute a synchronization barrier using the prefix provided during initialization. N)rHbarrierr=s&rr>DistBarrier.execute_barriers rrN r r!r"r#r$rr7r>r&r'r(s@rrCrCs'"L W WrrCcLa]tRt^toRtRtV3RlRltV3RlRltRtVt R#) 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_storecJ<V^8dQhRS[RS[RS[RS[RS[RS[RS[RS[/#) r global_rankglobal_world_sizebarrier_prefixtimeout_barrier_init_secs%use_checkpoint_barrier_tcpstore_libuv tcpstore_portmaster_address timeout_secs)intrbool)r rs"rrTCPStoreBarrier.__annotate__s[6 6 6 6  6 $' 6 04 6 6 6 6 rc  "\PRVVVVVVVV4 \4VnW0nWnW nWn\P!V\V4VP \VR7VP ^8HR7Vn R#)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)seconds) world_sizetimeout is_masterN) loggerinfor_tcp_store_barrier_seq_barrier_prefix _global_rank_global_world_size _timeout_secsrHTCPStorer]r _tcp_store) r6rUrVrWrXrYrZr[r\s &&&&&&&&&rr7TCPStoreBarrier.__init__s0   G      % 1  07y#-("3)--   ..&?@((A-  rc<V^8dQhRR/#r;r)r rs"rrr_s!9!9!9rc  VPp\PRVVP4RRlpVPP V!VP 4\VPV,44\P!VPVPV\VPV,4,R7VPV;;,^, uu&R#)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=%sc0V^8dQhR\R\/#)r rankr )r]r)r s"rr5TCPStoreBarrier.execute_barrier..__annotate__s ! !C !C !rc RV 2#rrrrus&r _rank_key2TCPStoreBarrier.execute_barrier.._rank_keys$= r)storerb key_prefixN) rhrerfrkrmsetrirrg store_utilrNrj)r6rWrvs& rr>TCPStoreBarrier.execute_barriers-- A      !  d'' ( ++N; < //..T%@%@%P!QQ   ##N3q83r)rhrirjrmrgrkNrPr(s@rrRrRs&L6 6 p!9!9rrRcbV^8dQh/^\9d\\\3,;R&#)r r)__conditional_annotations__rrr )r s"rrrs&.'&$sDy/&/r)!r~r$r@logging collectionsr dataclassesrrdatetimertypingrrtorch.distributed distributedrH%torch.distributed.elastic.utils.storeelasticutilsrxr{ getLoggerrerrrr0ABCr+rCrRr)r~s@rrs ( ::    %'&  ? ?  ?F88 cgg > '  Fe9ge9e9r