zj,ddlmZddlZddlmZmZddlZddlmZ ddlm Z er ddlm Z ddl m Z GddZGd d Zd Zd Zd ZdZdZd$dZd%d&dZd'd(dZdZd'dZ d)d*d Zd%d&d!Zd%d+d#ZdS),) annotationsN) TYPE_CHECKINGLiteral) framework)Tensor) ProcessGroupceZdZdZ d d!dZed"dZed#dZed"dZed$dZ ed%dZ ed"dZ ed&dZ ed"dZ d'dZd(dZd(dZd&dZdS))Groupz/ The abstract representation of group. N rank_in_groupintidranks list[int]pgProcessGroup | Nonename str | NonereturnNonec||_|dkrt|nd|_||_||_||_||_dS)Nr)_rank_in_grouplen _world_size_id_ranks_pg_name)selfr r rrrs v/lsinfo/ai/hellotax_ai/data_center/backend/venv/lib/python3.11/site-packages/paddle/distributed/communication/group.py__init__zGroup.__init__"sH,)6!););3u:::  c|jSN)rrs r rankz Group.rank1s ""r"c|jSr$)rr%s r rz Group.ranks5s {r"c*t|jSr$)rrr%s r nranksz Group.nranks9s4;r"c|jSr$)rr%s r rz Group.name=s zr"rc|jSr$)rr%s r process_groupzGroup.process_groupA xr"c|jSr$)rr%s r world_sizezGroup.world_sizeEs r"strc4|jSr$)rrr%s r backendz Group.backendIsx}}r"c|jSr$)rr%s r r zGroup.idMr-r"boolc:|jdkrdS|jdkrdSdS)NrFT)r&r)r%s r is_memberzGroup.is_memberQs' 9q==5 ;??5tr"r&int | Literal[-1]cb|r|j|SdS)Nr)r7rindexrr&s r get_group_rankzGroup.get_group_rankXs/ >>   :##D)) )2r"cH|r |j|SdS)a# Get the global rank of a process within a group. Args: rank (int): The local rank within the group. Returns: If the current process is a member of the group, returns the corresponding global rank; otherwise returns -1. r)r7rr;s r get_global_rankzGroup.get_global_rank^s( >>   :d# #2r"cd|jd|jd|jd}|dt t |jz }|dz }||jr|jndz }|S)Nzrank: z , nranks: z, id: z , ranks: z, z; name: r)r&r)r joinmapr0rr)r debug_strs r __repr__zGroup.__repr__osv OTY O O$+ O OTW O O O  TYYs3 33444 Z $)7TYY7 r")NN) r r r r rrrrrrrr)rr )rr)rr)rr)rr0rr4)r&r rr8)__name__ __module__ __qualname____doc__r!propertyr&rr)rr,r/r2r r7r<r>rCr"r r r s#'     ###X#X   X XX   X XX "r"r ceZdZdZiZdS) _GroupManagerrN)rErFrGglobal_group_idgroup_map_by_idrJr"r rLrLysOOOOr"rLctjtjvrtdtjtjS)Nz$The global group is not initialized.)rLrMrN RuntimeErrorrJr"r _get_global_grouprQ~s3$M,IIIABBB  ()F GGr"c|jtjvrtd|jd|tj|j<dS)NzThe group with id z already exist.)r rLrNrPgroups r _add_new_grouprUsC x=000IIIIJJJ.3M!%(+++r"c,|jtjkSr$)r rLrMrSs r _is_global_grouprWs 8}4 44r"ctj}|r5|s!tjd|d|jdSdS)NzCurrent global rank z is not in group TF)distget_rankr7warningswarnr)rT global_ranks r _warn_cur_rank_not_in_groupr^s[-//K U__&& M; M M M M   t 5r"cf||}|dksJd|d|j|S)NrzThe input rank z# can not be found inside the group )r<r)r]rT group_ranks r _get_or_throw_group_rankrasF%%k22J ???V+VV%*VV ?? r"rr4c2tjtjvS)a@ Check whether the distributed environment has been initialized Returns: `True` if distributed environment has been initialized, otherwise `False`. Warning: This API only supports the dygraph mode. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env: DISTRIBUTED) >>> import paddle >>> print(paddle.distributed.is_initialized()) False >>> paddle.distributed.init_parallel_env() >>> print(paddle.distributed.is_initialized()) True )rLrMrNrJr"r is_initializedrcs2  (M,I IIr"rT Group | Nonerc|tn|}|jtjvsJd|jdt |r tjdStj|j=dS)a Destroy a given group for communication Args: group (Group, optional): The group to be destroyed. All of process groups, including the default group, will be destroyed and the distributed environment will be deinitialized. Returns : None Warning: This API only supports the dygraph mode. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env: DISTRIBUTED) >>> import paddle >>> import paddle.distributed as dist >>> dist.init_parallel_env() >>> group = dist.new_group([0, 1]) >>> dist.destroy_process_group(group) >>> print(dist.is_initialized()) True >>> dist.destroy_process_group() >>> print(dist.is_initialized()) False NzDestroy group with id z is invalid.)rQr rLrNrWclearrSs r destroy_process_grouprgs@$)=    eE 8}4 4 4 47777 5 4 44%++-----  )%( 3 3 3r"r r cv|tjvrtj|Stjd|ddS)a Get group instance by group id. Args: id (int): the group id. Default value is 0. Returns: Group: the group instance. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env: DISTRIBUTED) >>> import paddle >>> import paddle.distributed as dist >>> dist.init_parallel_env() >>> gid = paddle.distributed.new_group([2,4,6]) >>> paddle.distributed.get_group(gid.id) zGroup z is not initialized.N)rLrNr[r\)r s r get_grouprisA0 ] ***,R00 M32333444 4r"ctjrtj|Sd}tj|fit }||d|gid|gidS)Nc_sync_calc_streamXOut)typeinputsoutputs)rin_dynamic_modepaddle_C_opssync_calc_stream LayerHelperlocals append_op)tensorop_typehelpers r _sync_calc_streamr{s ""  }--f555&&w;;&((;;&?VH%      r"ctjr!tj|g|Sd}tj|fit }||d|gid|gid|idS)Nc_sync_comm_streamrlrmring_idrnrorpattrs)rrqrrrssync_comm_streamrurvrw)rxr~ryrzs r _sync_comm_streamrs ""  }--vh@@@&&w;;&((;;&?VH%g&      r"Trxruse_calc_streamc||sdS|rt|dS|dn|j}t||dS)a wait to sync stream for group. Args: tensor (Tensor): The Tensor used before sync. group (Group): The Group instance to perform sync. use_calc_stream (bool): Whether to use calculation stream (True) or communication stream (False). Default to True. Returns: None. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env: DISTRIBUTED) >>> import paddle >>> paddle.distributed.init_parallel_env() >>> tindata = paddle.randn(shape=[2, 3]) >>> paddle.distributed.all_reduce(tindata, sync_op=True) >>> paddle.distributed.wait(tindata) Nr)r7r{r r)rxrTrr~s r waitrs_8 !2!2+&!!!!!}!!%(&'*****r"cZ||sdStjr|tn|}tj}t |tjr|j}n.| }|j|}| dS|dn|j }tj dgdd}tjrGtjjs"tj||d|SdSd}t |t"st%dtj|fit)}||d |gid |gid|i dS) a* Barrier among all participators in the group. Args: group (Group): The group instance return by new_group or None for global default group. Returns: None. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env: DISTRIBUTED) >>> import paddle >>> from paddle.distributed import init_parallel_env >>> paddle.set_device(f'gpu:{paddle.distributed.ParallelEnv().dev_id}') >>> init_parallel_env() >>> paddle.distributed.barrier() Nrint32)dtyper~barrierz,The type of 'group' for barrier must be int.rlrmr)r7rrqrQ_current_expected_place isinstanceCPUPlacer,r get_device_idrr rrfullcoreis_compiled_with_xpu _legacy_C_opsr ValueErrorrurvrw)rTplacetask device_idr~barrier_tensorryrzs r rrEs, !2!2 "" ',}!###%133 eY/ 0 0 :&..00DD++--I&..y99D =aaehG[!aw777N "" $99;; '// 7    '3'' MKLL L&w;;&((;;.)*^,-g&      r"r0cpt|rtd|tn|}|jS)a Get the backend of given group. Args: group (Group): The group to work on. Use the global group as default. Returns: Returns the name of the given group backend. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env: DISTRIBUTED) >>> import paddle >>> paddle.distributed.init_parallel_env() >>> paddle.distributed.get_backend() NCCL zInvalid group specified)r^rPrQr2rSs r get_backendrs@(#5))64555#(=    eE =r"rDr$)rTrdrr)r)r r rr )NT)rxrrTrdrr4rr)rTrdrr0) __future__rr[typingrrrrpaddle.distributed distributedrYrrpaddle.base.corerr rLrQrUrWr^rarcrgrir{rrrrrJr"r rs#""""")))))))) !!!!!!.------YYYYYYYYx HHH 444 555JJJJ8'4'4'4'4'4T<         IM#+#+#+#+#+L7 7 7 7 7 tr"