zj*5UddlmZddlZddlZddlmZmZddlmZddl Z ddl m Z ddl m Z ddlmZmZmZdd lmZmZmZmZmZmZmZmZmZmZmZmZmZered Z d e!d <dd l"m#Z#gZ$da%dZ&ia'dZ(ia)ia*da+gdZ,da-da.ej/dZ0da1dZ2dZ3dZ4dZ5dZ6dZ7dZ8dZ9dZ:dZ; d8dZdde0ddfd9d+Z?d:d-Z@d;d/ZAiaBd0ZCd1ZDd2ZEd3ZFd) annotationsN) TYPE_CHECKINGLiteral) TypeAlias)core)in_dynamic_mode)Group_add_new_groupis_initialized) _c_concat _c_identity_c_lookup_table_c_softmax_with_cross_entropy)_c_softmax_with_multi_label_cross_entropy_c_split_Linear_linear _mp_allreduce_parallel_embedding_parallel_linear_set_var_distributedsplit)glooncclxcclbkclflagcxr _BackendList) NCCLConfigcZtstjatSN) _global_envpaddle distributed ParallelEnvm/lsinfo/ai/hellotax_ai/data_center/backend/venv/lib/python3.11/site-packages/paddle/distributed/collective.py_get_global_envr*:s$ 7(4466 r( _default_pg)rrheterrrri)secondsc |adSr")_default_backend)backends r)_set_default_backendr1Xsr(c |adSr")_default_store)stores r)_set_default_storer5]s NNNr(c ttvrPt}t|jdt t |jtt<tS)Nr)_global_env_gid _group_mapr*r ranklistrange world_size)genvs r)_get_group_mapr>bsOj((  &+ Iq$uT_5566' ' ?# r(c4ttSr")r>r7r'r(r)_get_global_groupr@ls   O ,,r(ctSr"_group_map_by_namer'r(r)_get_group_map_by_namerDps r(cdts JdttS)NzZCall paddle.distributed.init_parallel_env first to initialize the distributed environment.)r rD_default_group_namer'r(r)_get_default_grouprGus;    5  " # #$7 88r(c0|tvsJ|t|<dSr")r8)gidgroups r)_set_group_maprK~s j JsOOOr(c0|tvsJ|t|<dSr"rB)namerJs r)_set_group_map_by_namerNs% ) ) ) ) )$tr(c0|tvsJ|t|<dSr")_group_map_backend)rJr0s r)_set_group_map_backendrQs% * * * * * 'ur(ctr3tdz atttjdzSt t ttjdzS)Nr )r_start_ring_idmaxr*nringslenr>r'r(r) _new_ring_idrXseH!O$5$5$##$$s?+<+<+CQ'G'GGGr(c d} t} |tvsJd|d|dkr#tj||||} n|dkr+tj||||| j||} n|dkr)tj|| j|||} nX|dkr#tj ||||} n/|dkr)tj ||||| j|} | S)NzUnsupported backend: .rrrrr) r*_valid_backend_listrProcessGroupGloocreateProcessGroupNCCL pg_timeoutProcessGroupCustom device_typeProcessGroupBKCLProcessGroupFlagcx) r0r4r9r< group_name pg_optionsgroup_idnccl_comm_init_option nccl_configpgr=s r)_new_process_group_implrjs= B   D ) ) ) )+M7+M+M+M ) ) )&  " ) )%z8 L L F    " ) )     O !    F    $ + + 4#T:x   F    " ) )%z8 L L H    $ + +     O !    Ir(c |adSr") _custom_gid)rIs r)_set_custom_gidrms KKKr(rankslist[int] | Noner0Literal['nccl'] | Nonetimeoutdatetime.timedeltargintrhNCCLConfig | Nonereturnr c trMtrtn t}tt |z}|dkrj|t |dkrUt }|j}|j} |tn|}|| }t |t | ks Jdt |} t|}| dkr?||vr;|dkrdn| |} t|t| | |d||| } nd} d} t| ||| |} | t|<| t |<|t"| <t%| | S|sd }|d ks Jd t'j}t}||vrtd||}|t |<nWt|}| |}t |}t|||}|t |<|d krt)j}||_||_fd |D|_j|_d|_t)jrBt)jj}t)j|||nft)j rBt)j!j}t)j"|||ntGd |StrtIj%dgdntIj&dgdd}tHj'(|dtHj')||S)a Creates a new distributed communication group. Args: ranks (list): The global ranks of group members. backend (str): The backend used to create group, only nccl is supported now. timeout (datetime.timedelta, optional): The waiting timeout for store relevant options, default is 30 minutes. Returns: Group: The group instance. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env: DISTRIBUTED) >>> import paddle >>> paddle.distributed.init_parallel_env() >>> tindata = paddle.randn(shape=[2, 3]) >>> gp = paddle.distributed.new_group([2, 4, 6]) >>> paddle.distributed.all_reduce(tindata, group=gp, sync_op=False) r,Nr zQSize of new group must be less than or equal to that of the default global group.r)rerfrgrh)rirMrz,backend other than nccl is not supported yetc*g|]}j|Sr')trainer_endpoints).0ir=s r) znew_group..3s.***./&q)***r(zno cuda device foundint32)dtypeT)sync_op)*rrlrXrFstrrWrGr9rnr/sortedindexrjr3r rCr8rPr r*rParallelStrategynranks local_rankrzcurrent_endpointrVis_compiled_with_cuda CUDAPlace device_idNCCLParallelContextinit_with_ring_idis_compiled_with_xpuXPUPlaceBKCLParallelContextAssertionErrorr$ to_tensorfullr% all_reducewait)rnr0rqrgrhrIrd global_group global_rank global_rankssizer9rirJring_idgp group_rank group_sizestrategyplacetmpr=s @r) new_grouprsB((>> import paddle >>> print(paddle.distributed.is_available()) )ris_compiled_with_distr'r(r) is_availablerTs  % ' ''r(Nonec&tj}t}|j}|j}|j}|dkr$tj|d||dS|dkrd}|jD]}||z }|d dz }tj | d }tj|tj|d|||dS|dkrP|j} t"j| d |tj|d||| dS|d krd}|jD]}||z }|d dz }tj | d }tj|tj|d|||dSdS) Nr0rz ring_id:{}zUTF-8)encodingr:r)rcreate_or_get_global_tcp_storer*r9r<rCommContextManagercreate_gloo_comm_contextrzformathashlibmd5encode hexdigest set_device_idcreate_nccl_comm_contextrar$device set_devicecreate_xccl_comm_contextcreate_bkcl_comm_context) r0r4 global_envr9r<dev_id endpoints_strendpointendpoints_str_hashdev_types r)_init_parallel_envresb  / 1 1E ""J ?D&J  !F& 88 3j      F   "4 & &H X %MM,,S111 $[  ' 2 2  )++  --f555 88 3j*<      F  )   H!7!7v!7!7888 88 3j(      F   "4 & &H X %MM,,S111 $[  ' 2 2  )++  --f555 88 3j*<       r(ctSr"_shutdown_group_map_by_namer'r(r)_get_shutdown_group_map_by_namers &&r(c|t|<dSr"r)pg_namerJs r)"_update_shutdown_group_map_by_namers+0(((r(ct|=dSr"r)rs r)"_delete_shutdown_group_map_by_namers#G,,,r(c8tdSr")rclearr'r(r)!_clear_shutdown_group_map_by_namers%%'''''r(rJ Group | Nonecxt}|gtD]D\}}|j8||vr4|tkr)|jt ||EdS|j9|j|vr2|jt |j|dSdSdSr")rrDitems process_grouprFshutdownrrM)rJshutdown_groupsrris r)shutdown_process_grouprs577O }13399;; @ @KGR ,?22222 ))+++27B??? @ @   + /11   ( ( * * * .uz5 A A A A A , +11r(c(t}|@|D]}|jt dS|j8|j|vr1|jt |jdSdSdSr")rvaluesrrestartrrMr)rJrris r)restart_process_grouprs577O }!((** ' 'B   $ $ & & & &)+++++   *uz_/L/L   ' ' ) ) ) .uz : : : : : + */L/Lr()rrN) rnror0rprqrrrgrsrhrtrur )rur)r0rrurr")rJrrur)I __future__rdatetimertypingrrtyping_extensionsrr$ paddle.baserpaddle.frameworkrcommunication.groupr r r fleet.layers.mpu.mp_opsr rrrrrrrrrrrrr__annotations__paddle.base.libpaddler __all__r#r*r8r7rCrPrFr[r3r/ timedelta_default_timeoutrTr1r5r>r@rDrGrKrNrQrXrjrlrmrrrrrrrrrrr'r(r)rs#"""""" (''''' ,,,,,,FFFFFFFFFF 1%&NOLOOOO000000   $III%8%d333  --- 999 %%% ((( HHH"))))^  #&*"2!"%) @@@@@F(((("( ( ( ( V!''' 111 --- ((( BBBBB, ; ; ; ; ; ; ;r(