o * i,@sddlmZddlZddlmZmZddlZddlmZ ddlm Z er0ddlm Z ddl m Z GdddZGd d d Zd d Zd dZddZddZddZd1ddZd2d3ddZd4d5d d!Zd"d#Zd4d$d%Z &d6d7d*d+Zd2d3d,d-Zd2d8d/d0ZdS)9) annotationsN) TYPE_CHECKINGLiteral) framework)Tensor) ProcessGroupc@seZdZdZ  d-d.ddZed/ddZed0ddZed/ddZed1ddZ ed2ddZ ed/ddZ ed3ddZ ed/d d!Z d4d#d$Zd5d'd(Zd5d)d*Zd3d+d,ZdS)6Groupz/ The abstract representation of group. N rank_in_groupintidranks list[int]pgProcessGroup | Nonename str | NonereturnNonecCs8||_|dkr t|nd|_||_||_||_||_dS)Nr)_rank_in_grouplen _world_size_id_ranks_pg_name)selfr r r rrrr/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/paddle/distributed/communication/group.py__init__"s  zGroup.__init__cC|jSN)rrrrrrank1z Group.rankcCr r!)rr"rrrr 5r$z Group.rankscCs t|jSr!)rrr"rrrnranks9 z Group.nrankscCr r!)rr"rrrr=r$z Group.namercCr r!)rr"rrr process_groupAr$zGroup.process_groupcCr r!)rr"rrr world_sizeEr$zGroup.world_sizestrcCs |jSr!)rrr"rrrbackendIr&z Group.backendcCr r!)rr"rrrr Mr$zGroup.idboolcCs |jdkrdS|jdkrdSdS)NrFT)r#r%r"rrr is_memberQs  zGroup.is_memberr#int | Literal[-1]cCs|r |j|SdS)Nr)r-r indexrr#rrrget_group_rankXs zGroup.get_group_rankcCs|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)r-r r0rrrget_global_rank^s zGroup.get_global_rankcCsZd|jd|jd|jd}|dtt|j7}|d7}||jr(|j7}|Sd7}|S)Nzrank: z , nranks: z, id: z , ranks: z, z; name: r)r#r%r joinmapr)r r)rZ debug_strrrr__repr__oszGroup.__repr__)NN) r r r r r r rrrrrr)rr )rr )rr)rr)rr)rr+)r#r rr.)__name__ __module__ __qualname____doc__rpropertyr#r r%rr'r(r*r r-r1r2r5rrrrrs2             rc@seZdZdZiZdS) _GroupManagerrN)r7r8r9global_group_idgroup_map_by_idrrrrr<ysr<cCs tjtjvr tdtjtjS)Nz$The global group is not initialized.)r<r=r> RuntimeErrorrrrr_get_global_group~s  r@cCs.|jtjvrtd|jd|tj|j<dS)NzThe group with id z already exist.)r r<r>r?grouprrr_add_new_groups rCcCs |jtjkSr!)r r<r=rArrr_is_global_groups rDcCs4t}|r|std|d|jdSdS)NzCurrent global rank z is not in group TF)distZget_rankr-warningswarnr)rB global_rankrrr_warn_cur_rank_not_in_groups rIcCs,||}|dksJd|d|j|S)NrzThe input rank z# can not be found inside the group )r1r)rHrBZ group_rankrrr_get_or_throw_group_ranks  rJrr+cCs tjtjvS)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 )r<r=r>rrrris_initializeds rKrB Group | NonercCsT|durtn|}|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.)r@r r<r>rDclearrArrrdestroy_process_groups  rNr r cCs*|tjvr tj|Std|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)r<r>rFrG)r rrr get_groups  rOcCsLtr tj|Sd}tj|fit}|j|d|gid|giddS)NZc_sync_calc_streamXOut)typeinputsoutputs)rin_dynamic_modepaddle_C_opsZsync_calc_stream LayerHelperlocals append_op)tensorop_typehelperrrr_sync_calc_streams  r^cCsVtr tj|g|Sd}tj|fit}|j|d|gid|gid|iddS)NZc_sync_comm_streamrPrQring_idrRrSrTattrs)rrUrVrWZsync_comm_streamrXrYrZ)r[r_r\r]rrr_sync_comm_streams rbTr[ruse_calc_streamcCsD|dur |s dS|rt|dS|durdn|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)r-r^r rb)r[rBrcr_rrrwaits  rdcCs |dur |s dStr7|durtn|}t}t|tjr'|j}n | }|j|}| dS|dur=dn|j }t j dgddd}tr^t jjs\t j||d|SdSd}t|tsitdtj|fit}|j|d |gid |gid|id 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() NrZint32)Zdtyper_barrierz,The type of 'group' for barrier must be int.rPrQr`)r-rrUr@Z_current_expected_place isinstanceZCPUPlacer'rfZ get_device_idrdr rVfullcoreZis_compiled_with_xpuZ _legacy_C_opsr ValueErrorrXrYrZ)rBZplacetaskZ device_idr_Zbarrier_tensorr\r]rrrrfEs:      rfr)cCs.t|rtd|durt}|jS|}|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 specifiedN)rIr?r@r*rArrr get_backends rlr6r!)rBrLrr)r)r r rr)NT)r[rrBrLrcr+rr)rBrLrr)) __future__rrFtypingrrrVZpaddle.distributed distributedrErrZpaddle.base.corerrr<r@rCrDrIrJrKrNrOr^rbrdrfrlrrrrs2     \   *  &: