o hi@sdZddlmZddlmZddlmZmZmZm Z m Z m Z m Z ddl mZe dZeGddde eZd dddd d'ddZ  d(d)ddZddfd*d%d&ZdS)+zl A set of primitive functions for performing collective ops. Each should also handle single rank scenario. ) annotations) dataclass)AnyCallablecastGenericOptionalTypeVarUnionNTc@s2eZdZUded<ded<ded<dZded <dS) SyncPayload Optional[str] stage_nameboolsuccessr payloadNzOptional[Exception] exception)__name__ __module__ __qualname____annotations__rrrv/var/www/html/construction_image-detection-poc/venv/lib/python3.10/site-packages/torch/distributed/collective_utils.pyr s r T)rrrankpg data_or_fnUnion[T, Callable[[], T]]rrrr rintrOptional[dist.ProcessGroup]returnc Cs&|s |dur tdd}d}|dur|dks |durB||krBt|r@z|}Wnty?}z d}|}WYd}~nd}~ww|}t||||d}|dure|g} tj| ||dt| dksaJ| d}|jsd|d } |durz| d |j 7} |j dur| d |j 7} t | |j t t |jS) aK Broadcasts the data payload from rank 0 to all other ranks. Or if a function is passed, execute it in rank 0 and broadcast result to all other ranks. Can be used to broadcast a failure signal to stop all ranks. If the function raises an exception, all ranks will raise. Args: data_or_fn: the data to broadcast or function to execute and broadcast result. success: False to stop all ranks. stage_name: the name of the logical stage for synchronization and debugging rank: rank to broadcast data or execute function and broadcast resutls. pg: the process group for sync Throws: RuntimeError from original exception trace Returns: the value after synchronization Example usage: >> id = broadcast(data_or_fn=allocate_id, rank=0, pg=ext_pg.my_pg) Nz9Data or Function is expected to be None if not successfulrFrrrr)srcgroupzRank z failedz: stage z : exception )AssertionErrorrcallable Exceptionr distbroadcast_object_listlenrrr RuntimeErrorrr r) rrrrrrresync_objbroadcast_list error_msgrrr broadcastsF $     r/list[T]c CsXd}d}d}t|r&z|}Wnty%}z d}|}WYd}~nd}~ww|}t||||d}|durdgt|}t|||ttt|dj}g} g} d} t tt tt|D]-\} } | j|krs| d| d| jd 7} q]| j s| j dur| | | j fq]| | jq]t| dkrt| | | d| S|j std |j |j |jgS) a. A simple all_gather primitive with basic synchronization guard logic, by checking payload from all ranks has the same stage name. Args: data_or_fn: the data to be all gathered across ranks or function to be executed stage_name: the sync stage name for out-of-sync protection pg: the process group for sync Throws: RuntimeError from original exception trace Returns: a list of synced data from all ranks Example usage: >> all_ids = all_gather(data_or_fn=allocate_id, pg=ext_pg.my_pg) NTFr rz)Unexpected stage name received from rank z:  z!all_gather failed with exception )r%r&r r'get_world_sizeall_gather_object_enforce_typerr r enumeratelistrrappendrr)r*)rrrrrrr+r, total_listexception_listret_listr.isprrr all_gatherisb     r=cCst|t|kS)N)type)xyrrrsrAdist.ProcessGroup object_list list[Any]objr type_checkerCallable[[Any, Any], bool]Nonec Csttj|||dt|}|dkrdS|d}td|D]}||||s7td|dt||dt|qdS)aN Similar to plain all_gather_object but with additional type checking AFTER gather is done to ensure basic consistency. If check does not pass, all ranks will fail with exception. This is generally to prevent conditional logic leading to unexpected messages being received. This is considered fatal code error, but due to logic stacks this might happen implicitly in practice. The default check does not check sub type (considered different) or covariance (considered same) but users can pass in custom checker if more complicated check is needed. )r"rNr#zObject type at index z is z, while first object type is )r'all_gather_objectr)range TypeErrorr>)rrCrErFlist_len first_objr;rrrr4sr4) rrrrrr rrrrrr )NN)rrrr rrrr0) rrBrCrDrErrFrGrrH)__doc__ __future__r dataclassesrtypingrrrrrr r torch.distributed distributedr'r r r/r=r4rrrrs$  $  N U