Αia.SSKJr SSKrSSKrSSKrSSKrSSKrSSKJrJ r SSK J r SSK J r Jr SSKJrJr SSKrSSKrSSKJrJr SSKJr SS KJrJrJrJrJrJrJ r J!r!J"r"J#r# SS K$J%r%J&r&J'r' SS K(J)r) SS K*J+r+ SS K,J-r-J.r/J0r0J1r1 SSK2J3r3 SSK4J5r5 SSK6J7r7 SSK8J9r9 \(aSSK:J;r; SSKJr> SSK?J@r@ /rA\0RrBSrCSrD\RS5rF\RS5rG\/R\RS*Sj55rI\/R\RS+S,Sjj55rJ"SS\35rKSqL"SS 5rMS!rNS"rOS#rPS$rQS%rRS&rSS-S.S'jjrTS-S/S(jjrUS-S/S)jjrVg)0) annotationsN) OrderedDict namedtuple)contextmanager)ManagerProcess) TYPE_CHECKINGAny) _legacy_C_ops framework)get_all_custom_device_type) Group_default_group_name_get_group_map_by_name_new_process_group_impl_set_default_backend_set_default_store_set_group_map_set_group_map_backend_set_group_map_by_name_valid_backend_list)_add_new_group_get_global_groupis_initialized)wait_server_ready) check_backend)_set_expected_placebasecorein_dynamic_mode)Layer) deprecated)parallel_helper)getenv_or_backup) Generator)Tensor) NCCLConfig) _StateDictc[5n[RR5RUl[RR5R Ul[RR5R Ul[RR5RUlU$N)ParallelStrategypaddle distributed ParallelEnvnranks local_ranktrainer_endpointscurrent_endpoint)strategys [/var/www/html/banglarbhumi/venv/lib/python3.13/site-packages/paddle/distributed/parallel.py _build_default_parallel_strategyr6Ps!H((446==HO ,,88:EEH&&(::  &&(99  Oc b/nUR5Hup#/n/nUH`nURUR5 UR[R"U[ R "URSS9/S95 Mb [R"U5nURXsU/5 M U$)Nint64dtype)xshape)itemsappendr=r-reshapenpprodconcat) var_groupscoalesced_grads_and_grad_varsgroup_id grad_varsflattened_vars g_var_shapesg_varcoalesced_grads r5_coalesce_tensorsrL]s$&!)//1 E    ,  ! !BGGEKKw$G#H  ~6%,,  5  2 )(r7c[R"URS9n[R"5R SSU0XS.SU0S9 g)Nr:reshape2X)OutXShaper=typeinputsoutputsattrs)r _create_tensorr;_dygraph_tracertrace_op)r<r=x_shapes r5_reshape_inplacer[psJ&&QWW5G (( Qx- )r7c F[5(aUHunnnUVs/sHn[R"USS9PM nnSnUSU4- nUS- n[R"X/UQ76 [ X#5H&uptUR US9 URU:XaM&e M ggs snf)Nr9r:sections)axisr)r=)r rArBr splitzipreshape_r=)rErKorigin_grad_vars grad_shapesg_shape grad_var_lenrVrJs r5_split_tensorsrg{s +    @K?JGw/{ E j,/ /E [ E    I5 I"%&6"DW-{{g---#E+  sBcSnSn[5nUSRnUHnURn[U[R5(a'[ R RRUn[R"URSS9[R"U5-nX1:aXVR:XaX8- nOUnURnUS- nURU/5RU5 M [U5$)Nrr9r:r#)rr; isinstancerDataTyper-pirdatatype_to_vartyperArBr= size_of_dtype setdefaultr?rL) vars group_size group_idxmemory_counterrDr;var var_dtypebytess r5 build_groupsrvs INJ GMMEII i / / ;;IFI 1D4F4F 5    &5II+=  #N"NIIE NIi,33C8 Z ((r7c X/nUR5R5GHupx[U[RR 5(d[ SURS35eU(a$[US5(aUR(aMsU(d[USS5(aMO&[USS5(a[USS5(dMUR[RRR:XaMURUR!55 GM [#U5S:XagU(a[%US5n U H&un n[&R(R+XUS S 9 M( U Hjupn U V s/sHn [,R."U S S 9PM nn [&R0R2R55R7S SU 0SU 0USS.S9 Ml gUH2nUR95n[&R(R+XUS S 9 M4 gs sn f)NThe data type of 'z' must be core.eager.Tensoris_distributedno_syncFexpertriT)srcgroupsync_opr9r:r`rOrP)r^r_rR)_obtain_parameters_buffersr>rireagerr' TypeErrornamehasattrrygetattrrSVarDescVarTypeVOCABr?detachlenrvr-r. broadcastrArBrr rXrY contiguous)model comm_groupsrc_rankis_model_parallel fuse_paramsis_moe_sharding_parallel model_vars_paramcoalesced_vars coalesced_var origin_vars var_shapesv_shapevar_lenrss r5sync_params_buffersrsJ446<<>%!2!233$UZZL0KL  u.//E4H4H'ui//0 ui//x99 ::--33 3 %,,.)5?6 :!%j2CD#1 M1a    ( (:t ) $27E 2M ?I?IGw/z  KK ! ! 1 1 3 < <]+ ,#*A6 =  7EC.."C    ( (T ) s H'cZ^\rSrSr%SrS\S'S\S'S\S'S\S 'S \S 'S \S 'SSU4S jjjrSSjrSr\ SSj5r SSjr \ "SSS9S5r \ "SSS9S5rSS Sjjr\R"S!S"Sjj5r\r\rSrU=r$)# DataParallelaX Run the dygraph module with data parallelism. Currently, DataParallel class only supports to run the dynamic graph with multi-process. Now supports two ways to start training: 1. start by ``paddle.distributed.spawn`` method, for example: ``python demo.py`` (spawn need to be called in ``__main__`` method) 2. start by ``paddle.distributed.launch`` module, for example: ``python -m paddle.distributed.launch --gpus=0,1 demo.py`` . And the content of `demo.py` is the code of examples. Args: layers(Layer): The module that should be executed by data parallel. strategy(ParallelStrategy, optional): (deprecated) The strategy of data parallelism, contains environment configuration related to parallel execution. Default: None. comm_buffer_size(int, optional): It limits the memory size(MB) of one buffer parameters' gradient which is the input of communication calling(e.g NCCLAllReduce). Default: 25. last_comm_buffer_size(float, optional): It limits memory size(MB) of last buffer in communication calling. Making the last communication buffer size small is useful to improve performance. Default: 1. find_unused_parameters(bool, optional): Whether to traverse the entire backward graph from the all tensors in the return value of the wrapped model's forward function. For parameters not involved in loss calculation, their gradients will be marked as ready in advance to prepare reduce. Please note that all forward outputs derived from the wrapped model parameters must participate in the calculation of loss and subsequent gradient calculations. If not, serious error will occur. Note that setting the find_unused_parameters to True will affect computing performance. Therefore, if all parameters are sure to participate in the loss calculation and the autograd graph construction, please set it False. Default: False. Returns: Layer: The data paralleled module. Examples: .. code-block:: python :name: dp-example >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle >>> import paddle.nn as nn >>> import paddle.optimizer as opt >>> import paddle.distributed as dist >>> class LinearNet(nn.Layer): ... def __init__(self): ... super().__init__() ... self._linear1 = nn.Linear(10, 10) ... self._linear2 = nn.Linear(10, 1) ... def forward(self, x): ... return self._linear2(self._linear1(x)) >>> def train(): ... # 1. initialize parallel environment ... dist.init_parallel_env() ... # 2. create data parallel layer & optimizer ... layer = LinearNet() ... dp_layer = paddle.DataParallel(layer) ... loss_fn = nn.MSELoss() ... adam = opt.Adam( ... learning_rate=0.001, parameters=dp_layer.parameters()) ... # 3. run layer ... inputs = paddle.randn([10, 10], 'float32') ... outputs = dp_layer(inputs) ... labels = paddle.randn([10, 1], 'float32') ... loss = loss_fn(outputs, labels) ... loss.backward() ... adam.step() ... adam.clear_grad() >>> if __name__ == '__main__': ... # 1. start by ``paddle.distributed.spawn`` (default) ... dist.spawn(train, nprocs=2) ... # 2. start by ``paddle.distributed.launch`` ... # train() .. note:: ``PyLayer`` is not supported in DataParallel. To solve problems of this kind, it's recommended to skip gradient synchronization among multiple cards by 'no_sync', and manually implement 'all_reduce' before model optimization. There is an example showing specific implementation processing. Examples: .. code-block:: python :name: dp-pylayer-example >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import numpy >>> import paddle >>> import paddle.distributed as dist >>> from paddle.autograd import PyLayer >>> from paddle.distributed.fleet.utils.hybrid_parallel_util import fused_allreduce_gradients >>> class cus_tanh(PyLayer): ... @staticmethod ... def forward(ctx, x): ... y = paddle.tanh(x) ... ctx.save_for_backward(y) ... return y ... @staticmethod ... def backward(ctx, dy): ... y, = ctx.saved_tensor() ... grad = dy * (1 - paddle.square(y)) ... return grad >>> class SimpleNet(paddle.nn.Layer): ... def __init__(self): ... super().__init__() ... self.linear = paddle.nn.Linear(2, 2) ... def forward(self, inputs): ... inputs = cus_tanh.apply(inputs) ... return self.linear(inputs) >>> if __name__ == '__main__': ... dist.init_parallel_env() ... model = SimpleNet() ... model = paddle.DataParallel(model) ... opt = paddle.optimizer.SGD(learning_rate=0.01, parameters=model.parameters()) ... for step in range(10): ... x_data = numpy.random.randn(2, 2).astype(numpy.float32) ... x = paddle.to_tensor(x_data) ... x.stop_gradient = False ... # step 1 : skip gradient synchronization by 'no_sync' ... with model.no_sync(): ... y_pred = model(x) ... loss = y_pred.mean() ... loss.backward() ... # step 2 : fuse + allreduce manually before optimization ... fused_allreduce_gradients(list(model.parameters()), None) ... opt.step() ... opt.clear_grad() boolfind_unused_parametersgrad_need_sync Group | Noner}r'rtintcomm_buffer_sizelast_comm_buffer_sizec @>[T U]UR5S-5 [5(dS5eXlXPlSUlX`l[RRUl UbX l O[5Ul URRS:Ga_[R cS5e[5(aURc(["R$R&R)5O URUl[+UR["R$R&R,5(dS5eURR/5VVs/sH6upxUR15(aM[2R4"SUS35PM8 nn[7URS S 9 [9US -S -5Ul[9US -S -5UlUR?5 g[2R4"S 5 gs snnf) N_data_parallelzBIt's not supported to construct DataParallel in static graph mode.Tr#zsParallelContext must be initialized before. You should use init_parallel_env() beforeconstructing the DataParallel.z:ProcessGroup must be an instance of Group in DataParallel.zparam [z<] is not contiguous, please check it and make it contiguous.F)rizThe program will return to single-card operation. Please check 1, whether you use spawn or fleetrun to start the program. 2, Whether it is a multi-card program. 3, Is the current environment multi-card.) super__init__ full_namer _layersrrr}rrr'rt _strategyr6r0r$__parallel_ctx__clz__r-r. collective_get_default_grouprirnamed_parameters is_contiguouswarningswarnrrrr init_reducer) selflayersr4rrrr}rr __class__s r5rDataParallel.__init__s ))+.>>?   P   &<#" **  %N=?DN >> 1 $"88D 1 D   zz)&&11DDF "JJ 2 2 = = C CPOP(,||'D'D'F (G  ..0 HMM!$'cd(G   % @$'(84(?$(F$GD ! *-%,t3*D &     MME )s 2HHc /n[5nUR5HnURSS9HupEUbXR;aMURU5 [ XPR 5(d&[ SURSUR S35eUR(dMsURX545 M M [[SUVVs/sHupEUPM snn55n[U5S:dS5eS nUVVs/sH!up5[US S5(aMU"U5PM# nnn[5(a[R "UUUR"UR$/5Ul[R("U[[+UR&55UUR,R.UR"UR$/UR05Ulggs snnfs snnf) NF)include_sublayersrxz ' must be ''c&[USS5(+$)NrzF)r)r<s r5+DataParallel.init_reducer..sgaE::r7rzWThis model does not have any parameters to train, and does not need to use DataParallelc[U[RRRR 5(a UR $g)NF)rir-nnlayercommon Embedding_sparse)sublayers r5check_layer_sparse5DataParallel.init_reducer..check_layer_sparses2(FIIOO$:$:$D$DEE'''r7rz)set sublayersraddrirtrr trainabler?listfilterrrr reager_assign_group_by_sizerr group_indices EagerReducerreversedr} process_groupr_reducer) r layers_param params_setrrrtrainable_parametersris_sparse_gradients r5rDataParallel.init_reducers U (H$555N=E$7u%!%88#,UZZL DNNCSSTU??? ''(9:O) $ :'34|81|4  '(1,  0 , $0 #/5)U3 ) x (#/    !%!@!@$"++T-B-BC"D  !--$Xd0012" ((++T-B-BC++ DM 15$ s G$G*! G*cx[RRn[X5(aU/$[U[[ 45(a'[ R"[URU56$[U[5(a5[ R"[URUR556$/$r+) rrr'rirtuple itertoolschainmap _find_tensordictvalues)robjvar_types r5rDataParallel._find_tensor s::$$ c $ $5L cD%= ) )??C(9(93$?@ @ c4 ??C(9(93::<$HI I r7c#^# URnSUlSv Xlg!Xlf=f7f)aQ A context manager to stop gradient synchronization. Within no_sync(), gradients of parameters will only be accumulated on model and not synchronized util the first forward-backward out of this context. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle >>> import paddle.nn as nn >>> import paddle.distributed as dist >>> class SimpleNet(nn.Layer): ... def __init__(self): ... super().__init__() ... self._linear = nn.Linear(10, 1) ... def forward(self, x): ... return self._linear(x) >>> dist.init_parallel_env() >>> model = SimpleNet() >>> dp_model = paddle.DataParallel(model) >>> inputs_1 = paddle.randn([10, 10], 'float32') >>> inputs_2 = paddle.ones([10, 10], 'float32') >>> with dp_model.no_sync(): ... # gradients will not be synchronized ... dp_model(inputs_1).backward() >>> # synchronization happens here >>> dp_model(inputs_2).backward() FN)r)rtmp_grad_need_syncs r5rzDataParallel.no_syncs1J"00# 5 "4 "4 s-"-*-c.UR"U0UD6nURRS:ah[R"5R (aDUR (a3URR[URU555 U$)Nr#) rrr0r rX _has_gradrrprepare_for_backwardrr)rrTkwargsrUs r5forwardDataParallel.forwardCsi,,1&1 NN ! !A %))+55## MM . .tD4E4Eg4N/O Pr7z2.0.0z/This method does not need to be called anymore.)sincereasoncU$)zd Deprecated method, now ``scale_loss`` is an empty method, keep this method just for compatibility. r])rlosss r5 scale_lossDataParallel.scale_lossMs  r7cg)zp Deprecated method, now ``apply_collective_grads`` is an empty method, keep this method just for compatibility. Nr]rs r5apply_collective_grads#DataParallel.apply_collective_gradsWs r7c8URRUUUS9$)aS Get all parameters and persistable buffers of current layer and its sub-layers. And set them into a dict Parameters: destination(dict, optional) : If provide, all the parameters and persistable buffers will be set to this dict . Default: None include_sublayers(bool, optional) : If true, also include the parameters and persistable buffers from sublayers. Default: True Returns: dict: a dict contains all the parameters and persistable buffers. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle >>> import paddle.distributed as dist >>> dist.init_parallel_env() >>> emb = paddle.nn.Embedding(10, 10) >>> emb = paddle.DataParallel(emb) >>> state_dict = emb.state_dict() >>> paddle.save(state_dict, "paddle_dy.pdparams") ) destinationrstructured_name_prefix)r state_dict)rrrrs r5rDataParallel.state_dictas*B||&&#/#9'  r7c6URRXS9 g)a Set parameters and persistable buffers from state_dict. All the parameters and buffers will be reset by the tensor in the state_dict Parameters: state_dict(dict) : Dict contains all the parameters and persistable buffers. use_structured_name(bool, optional) : If true, use structured name as key, otherwise, use parameter or buffer name as key. Default: True Returns: None Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle >>> import paddle.distributed as dist >>> dist.init_parallel_env() >>> emb = paddle.nn.Embedding(10, 10) >>> emb = paddle.DataParallel(emb) >>> state_dict = emb.state_dict() >>> paddle.save(state_dict, "paddle_dy.pdparams") >>> para_state_dict = paddle.load("paddle_dy.pdparams") >>> emb.set_state_dict(para_state_dict) )use_structured_nameN)rset_state_dict)rrrs r5rDataParallel.set_state_dictsD ##  $ r7) rrrrrrr}rrrt)Nr#FN)rr!r4zParallelStrategy | Nonerrrfloatrrr}rreturnNone)rr)rzGenerator[None, None, None])rTr rr rr')NT)rz_StateDict | Nonerrrstrrr))T)rr)rrrr)__name__ __module__ __qualname____firstlineno____doc____annotations__rrrrrzrr"rrrr deprecate_stat_dictrset_dict load_dict__static_attributes__ __classcell__)rs@r5rrsuPd!   -1 "'(',"II*I I % I !% II IIV8t)5)5VOO*."&&( % &%  % !$ %  % N""BF# $# ;?# # ## LHIr7rc\rSrSrSrSr\SSj5r\SSj5r\SSj5r \SSj5r \SSj5r \SS j5r \SS j5r \SS j5r\r\r\ rS rg )r/ia; .. note:: This API is not recommended, if you need to get rank and world_size, it is recommended to use ``paddle.distributed.get_rank()`` and ``paddle.distributed.get_world_size()`` . This class is used to obtain the environment variables required for the parallel execution of ``paddle.nn.Layer`` in dynamic mode. The parallel execution in dynamic mode needs to be started using ``paddle.distributed.launch`` or ``paddle.distributed.spawn`` . Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle >>> import paddle.distributed as dist >>> def train(): ... # 1. initialize parallel environment ... dist.init_parallel_env() ... # 2. get current ParallelEnv ... parallel_env = dist.ParallelEnv() ... print("rank: ", parallel_env.rank) ... print("world_size: ", parallel_env.world_size) >>> if __name__ == '__main__': ... # 1. start by ``paddle.distributed.spawn`` (default) ... dist.spawn(train, nprocs=2) ... # 2. start by ``paddle.distributed.launch`` ... train() # Print result in process 1: rank: 1 world_size: 2 # Print result in process 2: rank: 2 world_size: 2 cN[[R"SS55Ul[[R"SS55Ul[ 5nU(a[ US5OSUl[[R"SS55UlURS:waJS URS 3n[R"US5RS 5n[US5Ul O[R"5(a:[R"S S5RS 5n[US5Ul OS[R"5(a9[R"S S5RS 5n[US5Ul [SS5RS 5Ul[R"SS5Ul[[R"SS55UlUR"S:dS5eUR"S:dS5eg)NPADDLE_TRAINER_ID0PADDLE_TRAINERS_NUM1rrPADDLE_PG_TIMEOUT1800000FLAGS_selected_s,FLAGS_selected_gpusFLAGS_selected_xpusPADDLE_TRAINER_ENDPOINTSPADDLE_CURRENT_ENDPOINTFLAGS_nccl_nringsz.nccl_nrings must be an integer greater than 0. zEnccl_nrings should be less than 9, which is enough in most scenarios.)rosgetenv_rank _world_sizer r _device_type _pg_timeoutr` _device_idris_compiled_with_cudais_compiled_with_xpur%_trainer_endpoints_current_endpoint_nrings)rcustom_device_typesFLAGS_selected_custom_devicesselected_custom_devices selected_gpus selected_xpuss r5rParallelEnv.__init__s#6<= ryy)>DE8:+>C#A& 'B ryy)? ||a < ||a S r7cUR$)a Rank of current trainer. Its value is equal to the value of the environment variable ``PADDLE_TRAINER_ID`` . The default value is 0. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> # execute this command in terminal: export PADDLE_TRAINER_ID=0 >>> import paddle.distributed as dist >>> env = dist.ParallelEnv() >>> print("The rank is %d" % env.rank) The rank is 0 )r$rs r5rankParallelEnv.rank s&zzr7cUR$)a  The number of trainers (number of processes participating in current job). Its value is equal to the value of the environment variable ``PADDLE_TRAINERS_NUM`` . The default value is 1. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> # execute this command in terminal: export PADDLE_TRAINERS_NUM=4 >>> import paddle.distributed as dist >>> env = dist.ParallelEnv() >>> print("The world_size is %d" % env.world_size) The world_size is 4 )r%rs r5 world_sizeParallelEnv.world_sizes&r7cUR$)a The ID of selected GPU card for parallel training. Its value is equal to the value of the environment variable ``FLAGS_selected_gpus`` . The default value is 0. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> # execute this command in terminal: export FLAGS_selected_gpus=1 >>> import paddle.distributed as dist >>> env = dist.ParallelEnv() >>> print("The device id are %d" % env.device_id) The device id are 1 )r(rs r5 device_idParallelEnv.device_id4s$r7cUR$)z The type of custom device for parallel training. Its value is equal to the value of paddle.device.get_all_custom_device_type() . The default value is None. )r&rs r5 device_typeParallelEnv.device_typeHs   r7cUR$)a@ The endpoint of current trainer, it is in the form of (node IP + port). Its value is equal to the value of the environment variable ``PADDLE_CURRENT_ENDPOINT`` . The default value is "". Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> # execute this command in terminal: export PADDLE_CURRENT_ENDPOINT=127.0.0.1:6170 >>> import paddle.distributed as dist >>> env = dist.ParallelEnv() >>> print("The current endpoint are %s" % env.current_endpoint) The current endpoint are 127.0.0.1:6170 )r,rs r5r3ParallelEnv.current_endpointRs$%%%r7cUR$)a The endpoints of all trainer nodes in the task, which are used to broadcast the NCCL ID when NCCL2 is initialized. Its value is equal to the value of the environment variable ``PADDLE_TRAINER_ENDPOINTS`` . The default value is "". Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> # execute this command in terminal: export PADDLE_TRAINER_ENDPOINTS=127.0.0.1:6170,127.0.0.1:6171 >>> import paddle.distributed as dist >>> env = dist.ParallelEnv() >>> print("The trainer endpoints are %s" % env.trainer_endpoints) The trainer endpoints are ['127.0.0.1:6170', '127.0.0.1:6171'] )r+rs r5r2ParallelEnv.trainer_endpointsfs(&&&r7cUR$)a Nrings of current trainer. Its value is equal to the value of the environment variable ``FLAGS_nccl_nrings`` . The default value is 1. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> # execute this command in terminal: export FLAGS_nccl_nrings=1 >>> import paddle.distributed as dist >>> env = dist.ParallelEnv() >>> print("The nrings is %d" % env.nrings) The nrings is 1 )r-rs r5nringsParallelEnv.nrings|s$||r7cUR$)a timeout of process group. Its value is equal to the value of the environment variable ``PADDLE_PG_TIMEOUT`` . The default value is 30 minutes. Examples: .. code-block:: python >>> # execute this command in terminal: export PADDLE_PG_TIMEOUT=1800000 >>> import paddle.distributed as dist >>> env = dist.ParallelEnv() >>> # the pg_timeout of process group 1800000 )r'rs r5 pg_timeoutParallelEnv.pg_timeouts r7)r,r(r&r-r'r$r+r%N)rr)rr)rz list[str])rrrr r rpropertyr5r8r;r>r3r2rErHr1r0dev_idrr]r7r5r/r/s)V$ L(  (&!!&&&''*&  $J F Fr7r/c0[c [5q[$r+)_global_parallel_envr/r]r7r5_get_global_parallel_envrNs#*} r7cXSSKJn U"[U5US9nUR5 SnUR SS5(dUR 5(dF[ R"U5 UR SS5(aM/UR 5(dMFUR5 g)Nr)KVServer)sizerunningF) *paddle.distributed.fleet.utils.http_serverrPrstartget should_stoptimesleepstop)port http_server_drQrP http_server wait_secondss r5_start_kv_serverr_sC3t940KL   Iu - -[5L5L5N5N <    Iu - -[5L5L5N5Nr7c[U5 US;a4[R"5(d [R"5(dUS:Xagg)N)autoncclbkclheterflagcxxcclFT)rrr)r*)backends r5 _is_cpuonlyrhs>'>>  ' ' ) )T-F-F-H-H F r7c@[US5nUc[SUS35eg)Nz:paddle.distributed initialize error, environment variable z is needed, but not set.)r% ValueError)var_namerss r5_check_var_existsrls8 8T *C { $$,:-E G  r7c/n[S/SQ5n[R"5nUR5HDnUR U5nUR U5nXE:XaM,UR U"X4U55 MF U$)NFLAGS)r current_value default_value)rrglobalskeysrV get_defaultr?)retrn global_flagskeyvaluerps r5_get_modified_flagsrxsr C w J KE<<>L  "  %$005 % JJuS7 8 # Jr7c@[U5S:a[RRS5 UHJn[RR[ U55 [RRS5 ML [RRS5 gg)NrzH======================= Modified FLAGS detected =======================  zH======================================================================= )rsysstderrwriter)modified_flagsflags r5_print_modified_flagsrst >Q  W #D JJ  SY ' JJ  T "#  W r7c [5n[U5 [5q[nURS:a[ R "S5 g[RRSS5nUS:Xa/[R"5(aS[RS'Sn[U5nU(dE[R"5(d+[R"5(dUS:Xd [S 5eUS:XaS URS 3n[!U5 OmU(d0[R"5(a[!S 5 US:XaS OUnO6U(d/[R"5(a[!S5 US:XaSOUn[!S5 [!S5 [!S5 US:Xa,[R""URUR$5nOU(a[R&"5nOu[R"5(a![R("UR$5nO:[R"5(a [R*"UR$5n[-W5 SnU[.;GaM[15(Ga=[2[55;a[55[2$[7U5 [9[R:"S55n[9[R:"S55n US:a X:aU S:dS5e[R:"SS5n [R:"SS5n U (aU (aSR=X/5OSn U c[R:"SS5n U c[?S5RAS5Sn U (dS5eU RAS5up[9U 5n US:Hn [9[R:"SS55n[RB"5n[EU5 US;a)[RFRIUR$5 SSK%J&n [OUUUU [2SU"US 5S!9n[Q[SU 55n[UUSUU[2S"9n[W[2U5 [YSU5 [[Xs5 []U5 [^R`"S#5 U$URbVs1sHnURAS5SiM nn[9[R:"S$S%55nU(d U(dUS&:XaURbSRAS5n[e5nURg5nS'US('URhS:XaZS)UR0nUS&:Xa S)[kU50n[m[n[9US5UU4S*9nS#Ul8S#US('URs5 [u5n[^Rv"5(a[ R "S+5 URUl<URhUl=URbUl1UR|Ul>UR~Ul?U(a,[^R`"[R"UU55 OUS&:Xa6[^R`"[R"UUR$55 O[R"5(a,[^R`"[R"UU55 OE[R"5(a+[^R`"[R"UU55 US&:waLURbSSnURUR|5 U(dURzS:Xa [U5 [^R"5 U(dUS&:Xa'URhS:XaS'WS('WR=5 U$U(a[URbS/5 [R"5nURhUl4URUlHWSUlI[9US5UlJS,nS-nUUlKUUlL[R"U5n U R5 URhS:XaS'WS('WR=5 U$s snf).a Initialize parallel training environment in dynamic graph mode. Note: Now initialize both `NCCL` and `GLOO` contexts for communication. Args: backend (string): A string represents the backend used by DataParallel, should be one of 'gloo'(for cpu), 'nccl'(for cuda), 'bkcl'(for xpu), 'auto'(auto detect). The auto detection prefer 'nccl', 'bkcl' than 'gloo'. Returns: None Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:GPU, env:DISTRIBUTED) >>> import paddle >>> import paddle.nn as nn >>> import paddle.optimizer as opt >>> import paddle.distributed as dist >>> class LinearNet(nn.Layer): ... def __init__(self): ... super().__init__() ... self._linear1 = nn.Linear(10, 10) ... self._linear2 = nn.Linear(10, 1) ... def forward(self, x): ... return self._linear2(self._linear1(x)) >>> def train(): ... # 1. initialize parallel environment ... dist.init_parallel_env() ... # 2. create data parallel layer & optimizer ... layer = LinearNet() ... dp_layer = paddle.DataParallel(layer) ... loss_fn = nn.MSELoss() ... adam = opt.Adam( ... learning_rate=0.001, parameters=dp_layer.parameters()) ... # 3. run layer ... inputs = paddle.randn([10, 10], 'float32') ... outputs = dp_layer(inputs) ... labels = paddle.randn([10, 1], 'float32') ... loss = loss_fn(outputs, labels) ... loss.backward() ... adam.step() ... adam.clear_grad() >>> if __name__ == '__main__': ... dist.spawn(train) zlCurrently not a parallel execution environment, `paddle.distributed.init_parallel_env` will not do anything.NPADDLE_DISTRI_BACKENDrarercrfzAIf you want to use CPU-only version, please use 'gloo' as backendrrrrbrrrrrr#zrank must be non-negative and world_size must be the maximum rank plus one. Moreover, at least two processes are required to create a process group. MASTER_ADDR MASTER_PORT: PADDLE_MASTERrrzThe environment variable 'MASTER_ADDR' and 'MASTER_PORT' must be specified, for example 'export MASTER_ADDR=127.0.0.1' and 'export MASTER_ADDR=54612'. Or you can start your trainingwith paddle.distributed.run module.FLAGS_stop_check_timeout900)rbrfrcre)message2nccl_configdefault) pg_options nccl_config)pgrTPADDLE_WITH_GLOOrrdFrS_worker)targetargsz.The parallel environment has been initialized.ii)Orxrr/rMr8rrr"environrVrr*rhr)NotImplementedErrorr>rl CustomPlacer;CPUPlace CUDAPlaceXPUPlacerrr rrrrr#joinr%r`create_or_get_global_tcp_storerCommContextManager set_device_id&paddle.distributed.fleet.base.topologyrrrrangerrrrrr$_set_parallel_ctxr2rrr5rrr_daemonrUr,_is_parallel_ctx_initializedr0r1r3rEGLOOParallelContextHeterParallelContextNCCLParallelContextBKCLParallelContextremover_init_parallel_ctxGlooParallelStrategyrank_num ip_addressip_port init_seconds run_secondsGlooParallelContextinit)!rr~ parallel_envrg is_cpu_onlyr/placer}r5r8 master_addr master_port endpoints is_masterstop_check_timeout default_storerrranksinode_num init_gloo ep_rank_0managerr\rQr]r4other_endpoints gloo_strategydefault_init_timeout_secondsdefault_run_timeout_secondsgloos! r5init_parallel_envrsjp)*N.) '='L" z  jjnn4f=G(t88::.4 *+g&K   % % ' '  $ $ & & f ! O  &l667q 9 & 78t99;; 3 4 '6 1fwG!:!:!  2 A ii t4 ii t4 { HHk/ 0    /48I  ()CDJJ3OI  2 y $-??3#7 +& AI  +Eu!MN;;= =) 8 8  # # 1 1,2H2H I %     +  U:&'dAu2EF2E:q% u.u))$/ )5)G)GH)GA Q)GHHBII0#67Ii7g#5 2215;;C@ )  #( i    !|667D'!!3x=1!')A,'=K"&K '+M) $     !H3355 FG"--HO&++H!-!?!?H , = =H"))HO))  $ $Xu 5  G ))  % %h 0F0F G   # # % %))  $ $Xu 5   " " $ $))  $ $Xu 5 '"44Q7x889x22a7 o .&&( w')|/@/@A/E$) i $ L! <99!<=>113 ).. !-!8!8 #,Q<  #IaL 1 '+$&-#%A "$? !'' 6    !',M) $     LiIs cc[5(aU(a UR$UbS5e[5R$)ao Returns the rank of current trainer in the given group, ranks are consecutive integers in [0, ``world_size``). If none of the group is given, the global group will be used as default. Args: group (Group, optional): The communication group you want to get rank of current trainer, use global group as default if group is None. Returns: (int) The rank of current trainer in the given group. Return -1 if the process is not part of the given group. Warning: Argument ``group`` only supports in dygraph mode. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> # Execute this script using distributed launch with one card configs. >>> import paddle >>> import paddle.distributed as dist >>> dist.init_parallel_env() >>> print("The rank is %d" % dist.get_rank()) The rank is 0 *Only support group argument in eager mode.)r r5rNr}s r5get_rankrs86Uzz =FFF= # % * **r7c[5(aUc[5(a [5n[5(aU(a UR$UbS5e[ 5R$)ah Returns the number of trainers (number of processes participating in current job) in the given group. If none of the group is given, the global group will be used as default. Args: group (Group, optional): The communication group you want to check world size, use global group as default if group is None. Returns: (int) The number of trainers in the given group. Return -1 if the process if not part of the given group. Warning: Argument ``group`` only supports in dygraph mode. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> # Execute this script using distributed launch with one card configs. >>> import paddle >>> import paddle.distributed as dist >>> dist.init_parallel_env() >>> print("The world_size is %d" % dist.get_world_size()) The world_size is 1 r)r rrr8rNrs r5get_world_sizer"sX6em   %'EU =FFF= # % 0 00r7)roz list[Tensor]rprrz-list[list[Tensor | list[Tensor] | list[int]]])NrFTF)rr!rrrrrrrrrrrrr+)rzNCCLConfig | Nonerr)r}rrr)W __future__rrr"r{rXr collectionsrr contextlibrmultiprocessingrrtypingr r numpyrAr-r r paddle.base.corer paddle.distributed.collectiverrrrrrrrrr&paddle.distributed.communication.grouprrr5paddle.distributed.fleet.base.private_helper_functionr%paddle.distributed.fleet.launch_utilsrpaddle.frameworkrrimperative_baserr paddle.nn.layerr! paddle.utilsr"rr$ backup_envr%collections.abcr&r'paddle.base.libpaddler(paddle.nn.layer.layersr)__all__r,r6rL dygraph_onlyr[rgno_gradrvrrrMr/rNr_rhrlrxrrrrr]r7r5rs# /%,  +7    @ "#()01 (( )&  ..& ) )$')2))2  $#%* ? ??? ?  ? # ? ??DB5BNmm`      Up+D#1r7