Αi/%SSKJr SSKrSSKrSSKrSSKrSSKJr SSKJ r SSK J r J r J r Jr SSKrSSKrSSKJr SSKJrJrJr SSKJr SSKJr SS KJr SS KJr SS K J!r! SS K"J#r#J$r$J%r%J&r&J'r'J(r( SS KJ)r) SSK*J+r+J,r- SSK.J/r0 SSK1J2r2 SSK3J4r4 SSK5J6r6 SSK7J8r8 SSK9J:r:J;r;Jr>J?r? SSK@JArAJBrBJCrC SSKDJErE SSKFJGrGJHrH SSKIJJrJ SSKKJLrLJMrMJNrN SSKOJPrPJQrQJRrRJSrSJTrTJUrUJVrV SSKWJXrXJYrYJZrZJ[r[J\r\ SSK]J^r^J_r_ SSK`JaraJbrbJcrc \ (atSSKdJereJfrf SS KgJhrh SS!KJiri SS"KjJkrkJlrlJmrmJnrn SS#KoJprp SS$K"Jqrq SS%KJrrr SS&KsJtrt SS'KuJvrv SS(KwJxrx SS)KyJzrz SS*K{J|r|J}r}J~r~JrJrJrJrJrJr \ S+rS,\S-'"S.S/\S0S19rS`S2jrS3r"S4S5\EGR5rSaSbS6jjr/"S7S8\5rS9rS:rScS;jr"S<S=\5rSaS>jrS?rS@rSdSAjrSeSBjrSaSfSCjjrSgSDjr"SESF\J5r"SGSH5r"SISJ\5r"SKSL\5r"SMSN\5r"SOSP\5rShSiSQjjrSjSRjr"SSST5r"SUSV\-GRB5r"SWSX5rSkSlSYjjrSmSZjr"S[S\5rSnSoS]jjrS^rS_rg)p) annotationsN) OrderedDict) MethodType) TYPE_CHECKINGAnyLiteral TypedDict)_C_opsnnpir)amp_global_state)OptimizerState)PyLayer) unique_name)switch_to_static_graph)EagerParamBaseVariabledefault_main_programin_dygraph_mode in_pir_mode use_pir_api)fleet)Enginestrategy shard_tensor) ProcessMesh)$mark_as_sharding_propagation_skip_op)get_default_distributed_context)DistributedOperator)convert_to_dims_mappingfuse_param_func get_dist_attr split_meshsplit_param_functo_list)align alignmentget_current_device_type)core)DistributedBatchSampler_InfiniteIterableSampler) Optimizer)_enable_auto_dp_fake_replicate_grad_to_partialin_auto_dp_mode)_cal_local_shape _dist_reshape_dtensor_from_local_NdMeshAlltoAll_only_reshard_mesh_shape_reshard_mesh_shape_specific_alltoall_dim)check_placements_equalget_shard_specplacemetns_to_dist_status to_dim_map to_placements)determinate_rng rng_state)ShardingOptimizerStage1get_mesh_comm_listget_placement_with_sharding)CallableSequence) TypeAlias)Tensor) DTypeLikeNestedNumericSequence PlaceLike TensorLike) GradScaler)Program) Placement)DistributedInputSpec) DataLoader)Metric)Layer) _AMPConfig_DPOptimizationConfig_FusedPassesConfig_GradientMergeConfig_MPOptimizationConfig_PipelineConfig_RecomputeConfig_ShardingConfig_SPOptimizationConfigtrainevalpredictrE_Modecp\rSrSr%S\S'S\S'S\S'S\S 'S \S 'S \S 'S\S'S\S'S\S'Srg)_ConfigrYshardingrT fused_passesrUgradient_mergerWpipelinerRamprX recomputerVmp_optimizationrSdp_optimizationrZsp_optimizationN)__name__ __module__ __qualname____firstlineno____annotations____static_attributes__rld/var/www/html/banglarbhumi/venv/lib/python3.13/site-packages/paddle/distributed/auto_parallel/api.pyraras7!!((,,!! ##......rsraF)totalc*[R"5nUR5(aIUR5(a/UR UR 5R 55 U$SnU$UR UR 55 U$N)r* DenseTensoris_dist_is_initialized_share_data_with _local_value get_tensor)tensor lodtensors rt _to_lodtensorrs  "I ~~  ! ! # #  & &v':':'<'G'G'I J  I  ""6#4#4#67 rscLURU5(aU[U5S$grw) startswithlen)sprefixs rt _get_suffixrs%||FVrsc.\rSrSrSrSr\S5rSrg)DistAttraS DistAttr specifies how tensors are distributed or sliced on ProcessMesh. Args: mesh(paddle.distributed.ProcessMesh): The `ProcessMesh` object describes the Cartesian topology of the used processes. sharding_specs(list[str|None]): The specification describing how to shard the Tensor. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> mesh = dist.ProcessMesh([[2, 4, 5], [0, 1, 3]], dim_names=['x', 'y']) >>> dist_attr = dist.DistAttr(mesh=mesh, sharding_specs=['x', 'y']) >>> print(dist_attr) cf[U[R5(d [S5e[U[5(d [S5e[ SU55(dS5eX l/nUHnnUcURS5 MXAR;a[SUSURS35eURURRU55 Mp [RRU5 Xl X0l URS 5 URS 5 g) Nz?The mesh must be an instance of paddle.distributed.ProcessMesh.z/The sharding_specs must be an instance of list.c3X# UH n[U[5=(d USLv M" g7frw) isinstancestr).0dim_names rt $DistAttr.__init__..s+ * x % 9T)9 9*s(*z@The dimension name in sharding_specs must be an instance of str.zInvalid sharding dimension 'z%'. Available dimensions in mesh are: . process_mesh dims_mapping)rr*r ValueErrorlistall_sharding_specsappend dim_namesindexTensorDistAttr__init__rrmark_annotated)selfmeshsharding_specsrrs rtrDistAttr.__init__s/$ 0 011Q .$//NO O *    N N N . &H##B'>>1$6xjA==A^^>> import paddle >>> import paddle.distributed as dist >>> mesh = dist.ProcessMesh([[2, 4, 5], [0, 1, 3]], dim_names=['x', 'y']) >>> # dense tensor >>> a = paddle.to_tensor([[1,2,3], ... [5,6,7]]) >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> # distributed tensor >>> d_tensor = dist.shard_tensor(a, mesh, [dist.Shard(0), dist.Shard(1)]) >>> print(d_tensor) N stop_gradientTzinput tensor is not pir value.zAshard_tensor() input data only supported dense tensor type right.z:Get an uninitialized param with an unregistered init_func.)dtypeplacercv^^TRHnUR5(dMS5e UU4SjnU$)NzLazy init not support partial reshard. Notice that: shard a param to partial won't save any memory, but will increase the communication cost!c >[R"5TRR;ag[ [R"5TRTR S9n[ U5 T"X5 SSS5 g!,(df  g=f)Nr placements)distget_rankr process_idsr>rr?)varblockrng_name origin_hookparams rt _init_func8shard_tensor..lazy_init_hook.._init_funcWsc}}e.@.@.L.LL. %*%7%7#(#3#3 H #8,#C/-,,s - A?? B )r is_partial)rr placementrs`` rtlazy_init_hook$shard_tensor..lazy_init_hookOsB!&!1!1I(3355[5"2 0"!rsr)rrr)paddle framework_current_expected_place_get_paddle_placegetattrrrtyper Valueis_dense_tensor_typerrzrrF to_tensorin_dynamic_mode from_tensor__dict__r set_init_funcr r persistabler:ndimshard_tensor_static) datarrrrrr~r dist_paramorigin_init_func dist_tensorrs rtrrsJl }  88:    . .u 5Eot<  ##%%$dSYY 788 , 8((** O * dN + +D4H4H4J4J??. L .F fmm , ,F%%F dN + + "0(33!%// J (.';';J $  ,#)#4#4 ((":@  --jK )/(<(E$$66u=E"MM !% M+7*D*DM '  ! ! d# j) o. ./   !-->>!  K):!(<(J(JK %&7&:&F&FK # rsc J[R"5(aUR5up#pEUcUR5$[RR 5n[RR U5n/n[U5H\upUR[R"UR5UU UUS95 USR5RS5 M^ U$UR5unn nn[RRUUUUU 5$)NrrT)rr saved_tensorr|rrr enumeraterrFr}_unsafe_set_skip_check_meshr moe_sub_mesh_tensors) r grad_tensor global_mesh local_dimsrrroutirglobal_placementss rtbackward _moe_global_mesh_tensor.backwards  ! ! # #  " GK_&"//11((@@B((::5A%.%?MAJJ '446!+)3'7"' G&&(DDTJ&@   "  ! ==55 !  rsrlrwrmrnrorp staticmethodrrrrrlrsrtrrs. 88t$$rsrcUbUbUc [S5e[X5n[U5nU[U5:a[R "5XB'X44$)NzMthe args global_mesh, global_placements and local_mesh_dim should all be set.)rr$rrr Replicate)rr sub_mesh_dim sub_mesh_listrs rt$_get_sub_meshes_and_local_placementsrsdl26G6O [  {9M-.c*++)-)9&  **rsc[U5n[U5HMupEUR5(dMUR5nX6S:XaM6X6nXqRU-X6'MO U$)Nr)rris_shardget_dimr) local_shaperr global_shaperr shard_dimlocal_dim_sizes rt_cal_global_shapersm $L#J/     !))+I&",)4N&4zz#&FL # 0 rsc [R"U5n[XU5upE[R"UR 5R UR5n[R"U[R"5:H5nUSRS:XaSnOXsSnXn [R"5(Ga)USRS:Xa[USRUSU5n OXR5Rn [!XU5n /n [#U5HupUR%5R'S5 [)UR*U5(aUR,XM:waBU R/[1XU U55 U SR%5R'S5 MU R/U5 M [2R5U UUUUU U5$[R6R95(ad[!U R:X5n [R<R?UUUUUU 5nUSR@Ul USRBUl!U$[ES5e)NrTrzEdtensor_from_local_list() are only supported in dynamic and pir mode.)#rrrnparrayrreshaperwhererrsizerrr2r|rrr}rr9rrrreshardrapplyrr _local_shaper rrrNotImplementedError)rrrlocal_mesh_dimrrr local_coordlocal_tensor_idxrlocal_tensor_shaperresharded_local_tensor_listrr~rs rtrr s\z*J(L .)%O((4++,44TZZ@K((;$--/9:K1~a&6q9$6L  q>  ! #!1!!$**OA,>@P"  "3@@BHH ((:*M &(#"#45IA     ; ;D A*6+<+<>NOO&&/*<<+22FA$68HI,*,::4@+226:6',, '            % % ' ''  % %t mm::        %6a$8$F$F !"3A"6"B"B ! S  rscB\rSrSr\SSj5r\S5rSrg)_moe_sub_mesh_tensorsiUNc UR[R"U5UU[R"U5UUR5 [R "5(GaUcUcUR 5$UbUc [S5eURnXQR:wa [S5e[XaR5(dSUSURS35e[URXV5n[U5HFupU R5(dMU R5n Xn XSRU -X'MH [RR!5n [RR#U 5n /n[U5Hmunn[R$"UR 5UUUU S9nUR'5R)S5 UR*UlUR-U5 Mo U$[RR/5(aP[R0R3UUUUU5nUH%nUR*UlUR4UlM' U$g) NzAthe args global_mesh and global_placements should be set togetherzAthe global_mesh should be the same as dist_tensor's process_mesh.zthe global_placements (z,) is not equal to dist_tensor's placements (z).rrT)rrrrrrr|rrr9rr2rrrrrrrFr}rrrrr rr)rrrrrrrori_meshrrrrrrrrrr local_tensorss rtr_moe_sub_mesh_tensors.forwardVs  MM/ *   MM+ &        ! ! # #"'8'@"//11&*;*C$['33":"::$[.%'='=..?-@@lmxnDnDmEEGH /%%{ '00@&ANC ))++$-$5$5$7 )4)?*Q-?-E-Ec-JJ$. 'B((@@B((::5A$&!%.%?MAz#)==#002(%/#3# $L!++-II$O1<1J1JL.%,,\:&@)(    ) ) + +"MM>> ! M!. -8-F-F *+6+B+B (!.! ,rsctUR5unnnnnn[RR5n[RR U5nUn [ R "U R5RU R5n [ R"U [R"5:H5n U SRS:XaSn OXSn Xn [R"5(ae[RR5n[RR U5n[R"U R!5UU UUS9nU$[RR#5(a:[%U R&X5n[R(R+UUUUUU5$g)Nrr)rrrrrrrrrrr rrr rrFr|rrr r r)rrrrrrrrrrrrr local_gradrrs rtr_moe_sub_mesh_tensors.backwards            88:  2259hht//088D hh{dmmo=> q>  ! # *:1=  2  ! ! # #$$<<>E$$66u=E"MM'')!!, M!    ) ) + ++''K==77 !   ,rsrlNNNNNrrlrsrtrrUs<H!H!T++rsrc[R"U5n[XU5upE[RR 5(a[ RUUUUUU5$[RR5(aP[RRUUUUU5nUH%nURUl URUl M' U$[S5e)zO Get the local part of the ``dist_tensor`` on the specific ``local_mesh_dim``. z7moe_sub_mesh_tensors is only supported in dynamic mode.)rrrrrrrr rr rrrr)rrrrrrrrs rtrrs  &78(L)%O''))$**             % % ' ' ::       *L)4)B)BL &'2'>'>L $*! E  rsc[R"5(a]UR5SLa UR5(a [ S5e[R R RXU5$[RR5(a [RRXU5$[S5e)NTz#The input should be a local tensor.z?dtensor_from_local() are only supported in dynamic or pir mode.) rrryrzrbaser*dtensor_from_localrrr RuntimeError)rrrs rtr!r!s     !T )l.J.J.L.LBC C{{22       % % ' '}}// JOO M  rscb[R"5(aHUR5SLa [S5e[RR R XU5$[RR5(a [RR XU5$[S5e)NF)The input should be a distributed tensor.z=dtensor_to_local() are only supported in dynamic or pir mode.) rrryrr r*dtensor_to_localrrr r")rrrs rtr%r%s     E )HI I{{00JOO    % % ' '}}--kLL K  rsc*U"U0UD6n[XQU5$)ab Construct a Distributed Tensor from a function of arguments. Args: fn (callable): A callable function that creates and returns a tensor, such as paddle.ones, paddle.zeros, etc. mesh(paddle.distributed.ProcessMesh): The `ProcessMesh` object describes the Cartesian topology of the used processes. placements(list[paddle.distributed.Placement]): the placements describe how to place the tensor on ProcessMesh, it can be Shard, Replicate and Partial. *args (tuple): A tuple of arguments to be passed to the ``fn`` function. **kwargs (dict): A dict of arguments to be passed to the ``fn`` function. Returns: Tensor: A Tensor constructed from ``fn`` with distributed attributes. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> # Create a distributed attribute >>> mesh = dist.ProcessMesh([0, 1], dim_names=["x"]) >>> # Call the function dtensor_from_fn with dist_attr parameter >>> d_tensor = dist.dtensor_from_fn(paddle.ones, mesh, [dist.Replicate()], shape=[1]) >>> print(d_tensor) r)fnrrargskwargsr~s rtdtensor_from_fnr*s!B  F j 11rsc F[XU5(a[XRX5$[RR 5(GaU[ X RSS9up4n[R"5nX6l Xl URS5 URS5 [U5S:a*UR5HupxURXx5 M [U5S:a=/n UR5HupzU R!U5 M UR#U 5 [%XU5n U b[&R("XX+5$[+XU5(a[XRX5$[R,RR/X5$[15(a [R2R/XU5$[5U[65(d SUS35e[9XUR5n [;5n [=5nU R?5RA[BRD"SRGS S /55URHURURJURLURNS 9n[QX5nU R?5RSS S U/0SU/0S9n[UU5nUURVl URVRS5 SURVl,URVR[UR\5nUUl/URS5 URVRaUR\5nUUl/URS5 URcU5 [eU5 U$)a Reshard a distributed ``paddle.Tensor`` with given distributed attributes. Args: dist_tensor(Tensor): the distributed tensor to be resharded. mesh(paddle.distributed.ProcessMesh): The `ProcessMesh` object describes the Cartesian topology of the used processes. placements(list[paddle.distributed.Placement]): the placements describe how to place the tensor on ProcessMesh, it can be Shard, Replicate and Partial. Returns: Tensor: A Distributed Tensor resharded with distributed attributes. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> mesh = dist.ProcessMesh([0, 1], dim_names=["x"]) >>> # dense tensor >>> a = paddle.ones([10, 20]) >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> # distributed tensor >>> d_tensor = dist.shard_tensor(a, mesh, [dist.Partial()]) >>> out_d_tensor = dist.reshard(d_tensor, mesh, [dist.Replicate()]) >>> print(out_d_tensor) T)return_split_factorrrrz@in dy2static mode, reshard's input should be Variable, but got []r reshard_apitmp)namerrrrrassignXOut)rinputsoutputs)3r6r3rrrrr;rr*rmulti_dims_mappingrrritems_set_split_factorr_set_partial_dimsr8r5r r7r r rr rrr:rr current_block create_varrgenerate_with_ignorable_keyjoinrrrrr! append_opr dist_attrchunk_idget_input_dist_attrr0rget_output_dist_attradd_dist_op_for_programr)rrrrpartial_status split_factorr?dimsfr_ alltoall_dimr main_programdefault_dist_ctxout_vartarget_dims_mappingtrans_opdist_opinput_dist_attroutput_dist_attrs rtr r >sJF :>>[*;*;TNN ''))6O ((d6 2 l'') '3$!%  0  0 | q '--/++C40 ~  "D(..0 C 1  ' ' --kL  #"((:  {* = = .. {{'' ?? }}$$[ CC+x00 N{m[\ ] 0(+:J:JK+- :<,,.9988-/0####!!#//%33:  6nK--/99+'WI&:  &h/)-&((8%&"!++??    (;$&&~6",,AA',,O(;%''7009,X6rsc^^^Tc [S5e[T[5(d [S5eS Sjn[R"5(aUc"UR SS9HupgU"UT5 M O*UR SS9HupgU"XgT5 U"UT5 M TbUR UU4Sj5 TbURUU4Sj5 U$[S5e) aP Converts all layer's parameters to DistTensor parameters according to the `shard_fn` specified. It could also control the conversion of input or output of the layer by specifying the `input_fn` and `output_fn`. (i.e. convert the input to `paddle.Tensor` with distributed attributes, convert output back to `paddle.Tensor` without distributed attributes.) The `shard_fn` should have the following signature: def shard_fn(layer_name, layer, process_mesh) -> None The `input_fn` should have the following signature: def input_fn(inputs, process_mesh) -> list(paddle.Tensor) In general, the type of `input_fn` return value is paddle.Tensor with distributed attributes. The `output_fn` should have the following signature: def output_fn(outputs, process_mesh) -> list(paddle.Tensor) In general, the type of `output_fn` return value is paddle.Tensor with distributed attributes. Args: layer (paddle.nn.Layer): The Layer object to be shard. process_mesh (paddle.distributed.ProcessMesh): The `ProcessMesh` information to be place the input `layer`. shard_fn (Callable): The function to shard layer parameters across the `process_mesh`. If not specified, by default we replicate all parameters of the layer across the `process_mesh`. input_fn (Callable): Specify how the input of the layer is sharded. The `input_fn` will be registered for the Layer as a `forward pre-hook`. By default we do not shard the input. output_fn (Callable): Specify how the output of the layer is sharded or convert it back to `paddle.Tensor` without distributed attributes. The `output_fn` will be registered for the Layer as `forward post-hook`. By default we do not shard or convert the output. Returns: Layer: A layer that contains parameters/buffers that are all `paddle.Tensor` with distributed attributes. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> mesh = dist.ProcessMesh([0, 1], dim_names=["x"]) >>> class MLP(paddle.nn.Layer): ... def __init__(self): ... super().__init__() ... self.fc1 = paddle.nn.Linear(8, 8) ... self.fc2 = paddle.nn.Linear(8, 8) ... ... def forward(self, input): ... return self.fc2(self.fc1(input)) >>> def shard_fn(layer_name, layer, process_mesh): ... if layer_name == 'fc1': ... layer.weight = dist.shard_tensor(layer.weight, process_mesh, [dist.Shard(0)]) >>> layer = MLP() >>> layer = dist.shard_layer(layer, mesh, shard_fn) >>> print(layer) >>> # This case need to be executed in multi-card environment >>> # export CUDA_VISIBLE_DEVICES=0,1 >>> # python -m paddle.distributed.launch {test_case}.py z,The argument `process_mesh` cannot be empty.z;The argument `process_mesh` is not `dist.ProcessMesh` type.c URR5Hup#Ub}UR5(dh[[ UR 55Vs/sH!n[ RR5PM# nnURU[X1U55 MM URR5Hup&Ub}UR5(dh[[ UR 55Vs/sH!n[ RR5PM# nnURU[XaU55 MM gs snfs snfrw) _parametersr7ryrangerrr distributedr add_parameterr_buffersregister_buffer)layerrkeyrrHrbuffers rt"replicate_layer_params_and_buffers7shard_layer..replicate_layer_params_and_bufferss( ++113JC #3u{{#344&&0024## j9 4!>>//1KC!&..*:*:#3v||#455&&0025%% z: 2s (E;(E T) include_selfc>T"UT5$rwrl)rHr4input_fnrs rtshard_layer..?s (6<"@rsc>T"UT5$rwrl)rHr4r5 output_fnrs rtrbrcDs 9Wl+KrszB`paddle.distributed.shard_layer` only supports dynamic graph mode.)rZznn.LayerrrreturnNone) rrrrrnamed_sublayersregister_forward_pre_hookregister_forward_post_hookr)rZrshard_fnrarer]r0 sublayerss ` `` rt shard_layerrms\GHH lK 0 0 I  * <  $)#8#8d#8#K29lK$L$)#8#8d#8#K,739lK $L    + +@    , ,K  " P  rsc`[R"5(aD[U[R5=(a# [ US5=(a UR 5$[U[R RRR5=(a UR5SL$)z Check if an input is a dist_tensor in both dynamic and static modes. Args: tensor: The input to check Returns: bool: True if the input is a dist_tensor, False otherwise ryN) rrrrFhasattrryr  libpaddler rr?)r~s rtis_dist_tensorrqOs vv}} - ! * !   vv{{4488>> ? /  "$. rsc^\rSrSrSSjrSrSrSrSrSr Sr S r S r S r S rS rSrSrSrSrSU4SjjrSrSrSrU=r$)_ShardOptimizeridc@UcS5e[U[RR[RR45(dS5e[R R RURR5Ul XRS'SUl [US5(aGURb:[UR[RR 5(aSUl X lSUlSUlX0lUR"c[+S5Ul[UR"[*[,[.[045(dS5e[UR"[,[.[045(a5UR35 UR"R5UR$5 [UR"[.5(aG[75(d8UR8R:HnUR"R=U5 M [UR"[05(aUR8R:HnUR"R?U5 M UR8R:HnUR"R=U5 M S [@RBS '/Ul"/Ul#/Ul$SUl%SUl&SUl'[Q5Ul)SUl*SUl+g) Nz)The argument `optimizer` cannot be empty.zR`paddle.distributed.ShardOptimizer` only supports AdamW and SGD optimizer for now. _inner_optF _grad_clipTrzgshard_fn must be an instance of one of: _ShardingStage0, ShardingStage1, ShardingStage2, ShardingStage31skip_sharding3_output_reshard),rr optimizerAdamWSGDr  layer_helper LayerHelper __class__rmhelperr _shard_cliprorvr ClipGradByGlobalNorm _shard_fn_sharding_axis_sharding_degreegradient_accumulation_steps_ShardingStage0ShardingStage1ShardingStage2ShardingStage3'_set_and_check_sharding_prop_from_param_set_sharding_axisr1ru_parameter_list_register_hook_for_param_grad_shard_parameterosenvironfuse_param_view param_storage grad_storage_sharding_group _mp_groupdo_tensor_fusion_onceStrategy _strategyenable_tensor_fusionenable_sharding_overlap)rryrkrrs rtr_ShardOptimizer.__init__es$ 7 $ ((..0@0@0D0DE    a  ";;33??    ( ( '0 l#  I| , ,$$09//1O1OPP#D !" $+F( >> !,Q/DN NN nnn M    v     NN^^^L    8 8 : NN - -d.A.A B dnnn 5 5o>O>O88<The sharding degree of all parameters must be equal currently.)rautoget_mesh get_dim_sizer_sharding_mesh_dimr_meshrrrurryrrrrrrsetrdim_size)rr param_listrrrrFrs rtr7_ShardOptimizer._set_and_check_sharding_prop_from_paramsjj))+ $/$<$<11%D !^^ ! !$(NN$8$8$E$E11%D ![   __44 E==??%%D))Jj)<)<=t~~NN&/ &;NC!)T^^<<.1+'< ../ DC D t''(3{/F/F+GGH%%t''(3NN((44, MM$"5"56$:O:OOUO; rscURSS:XagURnURURRR 5;aURRURnURn[ UR [[45(acUR RX5URRUR'X RRURlURRR 5GHnURRUUnUR5(a![ U[R5(dMV[R"5(a URn[ UR [[[ 45(a-UR XAU5URRUU'OUR5(aSU;a UR"nOK[%['UR(R55Vs/sHn[*R,"5PM nn[/UUR(US9URRUU'[R"5(dGMWURRUUlGM gs snf)Nrr.betarr)rr0ru_master_weightskeysrrrrshard_master_weight _accumulatorsryr rrrrrrUrrrrr) rr target_name master_weightr[ accumulatororigin_accumulator_namerrHs rt_shard_accumulator"_ShardOptimizer._shard_accumulatorsH ;;q>Q  jj ::88==? ? OO;;EJJGM',,K$..>>*JKKNN66uL// ;DO// ;@??00557C//77<[IK""$$Z SYY-O-O%%''*5*:*:' PNN3{;--c2;?==??S(&+%5%5 &+3u/A/A/G/G+H%I&%I!NN,%I#& %'!&!3!3'1OO11#6{C%%''1--c2I8,&sK'cUR5(a[UR[[45(a[U[ R 5(d|URn[R"5X R'[R"XRU5nUR5RUR55 ggggrw)ryrrrrr rrrrrr rr}r{)rr new_placement out_params rt_reset_placements!_ShardOptimizer._reset_placementss ==??z NN^^<  eSYY// % 0 0 59^^5E 112 LL--}   "33I4H4H4JK 0  ?rsc[U[5(aURS5nUH0nURR X/5 UR U5 M2 g)Nparams)rdictgetru_create_accumulatorsr)rr parametersps rtr$_ShardOptimizer._create_accumulators%sK j$ ' '#1JA OO 0 0 <  # #A &rscURRX5 UR(Ga*URHnUR 5 SUlM UR (d[[UR55HnURUR5URR-nU[URRS5-nXe-n[ R"R%URUXg5nURR&R)XRU5R+5 M gg[-U[.5(dUSnUH[up[15R2(a,[-UR4[6[845(aSU lUR=U 5 M] g)Nrr)ru_finish_updaterrzero_check_inrrUrrr_numelrnranksmaxrankrr view_slice process_group all_gatherwaitrrr use_master_gradrrr main_gradr) rrparameters_and_gradsrr shard_sizebeginend slice_bufferrrHs rtr_ShardOptimizer._finish_update.s &&uC  $ $ $ $ 1 1 ""$() %!2//s4#7#789A**1-446//667'T-A-A-F-F)JJE,C#)==#;#;**1-u$L((66AA$&8&8&;df:02D99';H'E$-#%55*NN^^$D;;#'AK&&q) -rsc /nUH'up4URX0RSX4545 M) [R"X5$Ngrad)rrr-apply_gradients)r params_gradsnew_params_gradsrrs rtr_ShardOptimizer.apply_gradientsOsF'KE  # #vu;< (((@@rscPURR5n/n[URRS[5(a%URRH nX#S- nM OURRnUHKnUR (aM[ US5(aURbUs $M:URcMIUs $ [SUR555(aU$[URRS[5(dURRHnUR (aM[ US5(aOURb[SUR35e[R"U[RS9UlMvURb[SUR35e[R"XDRS9UlM OURR HnUSHnUR (aM[ US5(aOURb[SUR35e[R"U[RS9UlMvURb[SUR35e[R"XDRS9UlM M UR#5 URR%SS9 URR5$) z Create and shard the optimizer states e.g., accumulators and master_weights before load_state_dict. If training has already started or the optimizer states are already created and sharded, do nothing. rrrc3V# UHupUS;dM UR5v M! g7f))master_weights LR_SchedulerN)ry)rkvs rtr-_ShardOptimizer.state_dict..qs, *:: AIIKK*s ))z gradient should be None, but is )rF) set_to_zero)ru state_dictrrrrrorranyr7rr zeros_likefloat32r _param_groupsstep clear_grad)rrr param_grouprs rtr_ShardOptimizer.state_dictXs __//1  doo55a8$ ? ?#>> (33  ?88JE""uk**??.%%/::)%%   "((*    $//99!u>OP'-&7&7V^^'EOzz-(>uzzlK"(!2!25 !LEJ!9$ $<< (2E** uk22 ??6","B5??BS T#+1*;*;!+!::1","B5::, O#&,%6%6uKK%P !3 =$  ""u"5))++rsc P[5(GaUSR5(GaUSRnUSRnUSnURnS SjnSn[R R 5n SU R;aGU RS5n [U 5H'n URU"U 5R:XdM%U n O Sn [SU55(aBURU"U5R:Xa"URU"U5R:waSn U (a[RRRUUR U"U5[R""[R$R&5[R""[R$R&5/5nURn[[)U5S- SS5H^n [+X;[R"5(dM&[R,"5X;'[R."XURU5nM` UR0S:a[35(aXPR0-nU (aI[RRRXUR U[R,"5/5nUSU4nUR4R7X5 UR8(Ga"[;USS 5(Ga USR<n USR<S:XaUR>U RA5URBRD-nU[GURBRHS5-nX-n[JRLROUR>U UU5n[JRPRSUR>U UURBSS 9nSUR>U l*gSUR>U l*ggg) Nr.rc[RR5nSUR;aUR SU5nU$)u 获得pp_idx的mesh pp)rrrrget_mesh_with_dim)pp_idxrs rtr5_ShardOptimizer._append_optimize_op..get_meshs9zz**,4>>)11$?D rsrFc3V# UHn[U[R5v M! g7frw)rrPartial)rrs rtr6_ShardOptimizer._append_optimize_op..s!EO 9dll33Z')Trlast_idxgroupsync_opr)+in_auto_parallel_align_moderyrrrrrrrrUrrr auto_parallel moe_utilsr3rr ReduceTypekRedSumrrrr rrru_append_optimize_oprrorrrrrrrrr rrVris_sync)rrparam_and_gradrmeshsr grad_meshrippr pp_degreer change_meshrrrrrtasks rtr#_ShardOptimizer._append_optimize_ops{ ' ) )q!))++'*55J"1%22E!!$D))I C**--/K{,,,'44T: y)A((HQK,C,CC*  KEO""hsm&?&??__ (?(??" ))33AAJJSM T__%<%<= T__%<%<= "__ 3z?Q.B7jmT\\::$(NN$4JM<<.?.?LD8//!38I8I888))33AA**i$..2B1C-Q/6N ++EB  ' ' '~a(*55$Q'00!!$--2**3/668//667'T-A-A-F-F)JJE,C#)==#;#;**3/$L"--88**3/$"22 % 9D 7;D&&s+36;D&&s+3)6 (rsclS[RS'SUlURR 5 g)NrwFLAGS_enable_tensor_fusionT)rrrr_enable_tensor_fusionrs rtr %_ShardOptimizer._enable_tensor_fusions)36 /0$(! ,,.rsc[US5(a"URRSS5(agSUl[ U[ R R5(d[S[U535eXl g)Nconfig to_staticFTz+`layers` must be `paddle.nn.Layer` but got ) rorrrrrr rQr"r_layers)rlayerss rt_enable_sharding_overlap(_ShardOptimizer._enable_sharding_overlapsg 68 $ $):):;)N)N '+$&&))//22=d6l^L  rscUR5URR-nU[URRS5-nX2-n[ R RXU5n[ RRUU[ RRRURSS9R5 g)NrFoprr) rrrrrrr rrVreduce_scatterReduceOpSUMr)rrrrrreduce_scattereds rt_reduce_scatter_gradients)_ShardOptimizer._reduce_scatter_gradients s!((*d.B.B.I.II S!5!5!:!:A>> !==33LM))  !!**..&& * $&rsc UR(d [S5e0nURR5H&nURSS9HnX![ U5'M M( [ UR 5[ UR5:wa8[S[ UR 5S[ UR5S35e[[ UR 55GHnURURU5 SnSn[ UR U5UR-nS [RR5R;a0U[RR5RS 5-nUR UR!5H8upU S R#U"UURUUR$55 M: U[ UR 5S - :dGM!['[)UR UR+555S n UR-[ U 55nUR/U"UR0US -UR$55 GM g) Nz_Sharding overlap requires an initialized model. Call `_enable_sharding_overlap()` to set model.F)include_sublayersz"Length mismatch: fuse_param_view (z) vs grad_storage ()c^^^^[RR5UUU4Sj5nU$)Nc>T=RS- slTRT:XaTR5TR-nU[TRS5-nX!-n[ R RTX#5n[ RRUT[ RRRTSS9nUTl gg)Nr.rFr) rrrrrrr rrVrrr comm_task) rHrrrrr comm_grouprparam_group_lens rt fuse_commT_ShardOptimizer._async_sharding_comm..fuse_comm_hook_func..fuse_comm)s!))Q.)#,,?%1%8%8%:j>O>O%O *S!-D D#0+1==+C+C(%,( &11@@,(%11::>>",$) A 26 .@rsrautogradno_grad)r%rr$r&s``` rtfuse_comm_hook_funcA_ShardOptimizer._async_sharding_comm..fuse_comm_hook_func(s*((*6+6&! rscZ^^[RR5UU4Sj5nU$)Nc,>TR(dTR5TR-nU[TRS5-nX!-n[ R RTX#5n[ RRTUTSS9nSTlgg)NrFrT) rrrrrrr rrVr)rHrrrrrr$rs rtr&Z_ShardOptimizer._async_sharding_comm..fuse_all_gather_hook_func..fuse_comm@s)00%2%9%9%;z?P?P%P *S!-D D#0'-}}'?'?)5(  &11<<)(",$) =  15 -1rsr()rr$r&s`` rtfuse_all_gather_hook_funcG_ShardOptimizer._async_sharding_comm..fuse_all_gather_hook_func?s*((*5+5"! rsrrr.)rr"rlridrrrrUrrrrrrrr7_register_backward_hookrnextitervaluesrrir) r param2layerrZrrr+r0r%r0view first_params rt_async_sharding_comm$_ShardOptimizer._async_sharding_comms@||B  \\++-E%%%>%*BqE"?. t## $D,=,=(> >4S9M9M5N4OObcfgkgxgxcybzz{| s4//01A  * *4+<+3t++,q00"4(<([R"UR5n[[ 5[ UR -T-nX-S- U-U-$Nr.)rprodr r(r)r'r)rr  align_sizesharding_degrees rtget_padded_size?_ShardOptimizer._build_fuse_param_view..get_padded_sizexs[775--.D134%&!"  &*z9ZG Grsr)rrF)rrT)r0rzerosrrrrr#rrr|flatten_r1_slicerr rr} _set_dimsr r4rrr{r_cleardevicecuda empty_cache)rparams_and_gradsrArBtotal_buffer_size param2indexrrH param_buffer grad_dtype grad_bufferviewsr padded_sizer param_shaper tmp_paramtmp_grads ` rt_build_fuse_param_view&_ShardOptimizer._build_fuse_param_viewss  H (HE&7 # !7 7 ) ||$%-=a-@-C-I-I  % ^^ ll*;)*!!H 'EO OO  $ $ & MM   * * ,o,r[11rsc UR(Ga\[RR5n[ US5nUHCn[R "[ U55n[R"5U;dM=XPlME SUR;aURRS5nURUUl [ US5nUHCn[R "[ U55n[R"5U;dM=XPl ME SUlUVs/sHoSPM n nURRR n U S:aSn U S-S-n S/[#U 5-n U V s0sHoR$U R&_M nn U V s/sHoR)5PM nn [*R,"XX/5n[/5n[1U5H3unnUH'nUR3U/5R5UU5 M) M5 UR75HunnUR9UURR:5unnnUR<R5U5 UR>R5U5 UR@R5U5 M URB(aURE5 URFRHbSURFRHl%URURFRHl&SUR;a5URS:a%URURFRHl'/n/n[Q[#UR<55GHnURB(dURSUR@U5 UR<UR75GHunnUS n US nUR>URU5URR:-nU[WURRXS5-nUU-n [WUU5n![[UU RU5-U 5n"U!U":aM[\R^RaUR>UU!U"5n#[cU#U Rd[Rf"5/5n#UU#lU RhU#l4U RjU#l5SU#l6U RnU#l7U RhU#l4U RpU#l8U RrU#l9U RtU#l:U RvU#l;UR5U#5 [\R^RaUR@UU!U"5n$[cU$U Rd[Rf"5/5n$UR5U$5 GM URB(dGMqUUS l<UR@URzcGMUR@URzR}5 GM [[UU55n%U%$s snfs sn fs sn f) a 1. Tensor Fusion - Groups params/grads into contiguous param_storage/grad_storage buffers - Supports non-uniform partitioning across GPUs - Uses view_slice to access individual params/grads each step 2. Reduce_scatter Overlap - Overlaps grad reduce_scatter with backward 3. All_gather Overlap - Overlaps param all_gather with forward - Strategically scatters all_gather during forward (Launching all all_gather at once blocks overlap with other sync/comm ops) dpmpFriTr.rrr)ArrrrrA new_groupsortedrr _dim_namesr_shape _mp_degreerrrccomm_buffer_size_MBrr0rr|r*eager_assign_group_by_sizerr setdefaultrr7rWrrrrrr:rurvshould_comm_on_shard_dimsharding_groupmp_grouprUrrrrminrr rr4rrr need_clipr trainable optimize_attr regularizerdo_model_averageis_distributedrr#rrzip)&rrr shard_groupsrr$ mp_mesh_axis mp_groupsp_grrb group_sizeis_sparse_gradientr shape_dict dense_params group_indices var_groups group_idxindicesrrLrrr new_params new_gradsr0r8rr rank_beginrank_end param_begin param_end new_paramnew_gradrs& rt_tensor_fusion_ShardOptimizer._tensor_fusions  % % %%%..0D-dD9L%!^^F5M: ==?e++5(&t&#44T: "&++l";.tT: &E!%u !>J}}%/)3'*/D &,89LSa&LJ9"&.."9"9"M"M "Q&&)#,t3d:J"'3z?!: ?IJze**ekk1zJJ>HIjU..0jLI!;;:2JM%J&/ &>" 7 A)))R8?? QP!'? 0:/?/?/A+ + //$((// #! $$++O<""))-8!!((60B++))+))5FJ**C<@..DOO..7  s4//01A//..t/@/@/CD"2215;;= dW W &&q)002++223(#d.B.B.G.G*KK % 2!%4   6A )+"MM44&&q);  0&&^^%& "& */*=*= '&+oo #(, %&+oo #*/*=*= '*/*=*= '(-(9(9 %-2-C-C *+0+?+? (!!),"==33%%a(+y/e004>>3C2D  *S>V+++*+ 2'$$Q'11=%%a(22779g2j J :;K: KIs%Y,8 Y1Y6c/nSnUGHgupE[R"UR5nUnURn[ 5n UR R n URH;n U R5(dMU R5n U RU 5 M= U"X5n URURR5(dl[R"5X`R'U S:waEXRS:wa3U RU 5 [R"U 5X`R'[UR5HupXR:XaMXS:Xa[R"5Xm'M7U R5(aMN[R"5Xm'U"X5n U R5(dMU S:Xa[R"5Xm'MU RU 5 [R"U 5Xm'M URU:wa![R "XUR U5nUR#XG45 GMj U$)a Optimizes gradient placements for parameters in dynamic sharding mode to minimize redundant allreduce operations during gradient clipping. This function adjusts tensor placements across mesh axes based on priority rules, prioritizing sharding for dimensions marked in `_sharding_axis`. For each axis in the mesh: 1. Preserves existing `Shard(dim)` placements for any axis. 2. Converts `Partial()` placements to Shard(dim) where possible, falling back to `Replicate()` if sharding isn't feasible. 3. Maintains `Replicate()` placements unchanged. Processes axes in order of `_sharding_axis` first before other mesh axes in their natural order. e.g. a) sharding_axis = 0, tensor rank = 2, placements: [Partial(), Partial(), Repliacate()] -> [Shard(0), Shard(1), Repliacate()] b) sharding_axis = 0, tensor rank = 2, placements: [Partial(), Shard(0), Partial() ] -> [Shard(1), Shard(0), Repliacate()] c`[[U55HnXS:XaM X!;dMUs $ g)Nr.r)rUr) tensor_shapeshard_dims_set tensor_dims rtget_first_can_shard_dimR_ShardOptimizer._fused_comm_before_apply_optimize..get_first_can_shard_dimjs5#C $56 +q03%% 7 rsrr.)rrrr rrrrraddrrrShardrrr r)rrrrrrnew_placementsrrr mesh_shaperr mesh_axiss rt!_fused_comm_before_apply_optimize1_ShardOptimizer._fused_comm_before_apply_optimizeUs" (KE!]]4??;NH,,L UN**00J"__ %%''!*!2!2!4J"&&z2- 1NJ??4#6#67@@BB6:nn6F223# 3F3F(G1(L"&&z2:>**Z:PN#6#67)2$//(B$  3 33(A-040@N- ))++040@N-!8$"J!++--%+8<8HN5+..z:8< :8NN5))C,.0<<.?.?P  # #U$5 6e(h rsc>[R"5(aS[UR[5(a4UR (aUR U5nOURU5n[TU]%XX45$rw) rrrrrrrrsuper_apply_optimize)rlossstartup_programrparam_group_idxr~s rtr_ShardOptimizer._apply_optimizesp  ! ! # # NNN) ) ((#22<@ #EE   w& <  rscSUR;a.US:XaURU$[URSU5$[e)Nru)rrAttributeError)ritems rt __getattr___ShardOptimizer.__getattr__sA 4== (|#}}T**4==6= = rscUS:Xa#[U5RS3n[U5e[URX5$)Nruz._inner_opt is READ ONLY)rrmrsetattrru)rrvaluemsgs rt __setattr___ShardOptimizer.__setattr__s> < $Z(())ABC % %t44rs)rrarrrrrrrrrrrrrrr>r)rmrnrorprrrrrrrrrr rrr:rWrrrrrrr __classcell__r~s@rtrsrsdsC-J4l61p L'*BAK,ZT> KKM9   "{{,,00II   YK .: *rscUR5(aYUR5nUR5S:XaURXU5 U$[R "XR U5$[UUR US9$)Nrr)ryrr0rrr rr)rr~rrrs rt_apply_placement#_ShardingStageBase._apply_placementsu >>  '')BwwyL($$VJ? <<(:(:JG G ##!  rsc,[XR5$rw)r0r)rrs rt'_reshard_fake_replicate_grad_to_partial:_ShardingStageBase._reshard_fake_replicate_grad_to_partials.t5H5HIIrsc^^SmUU4Sjn[5R(a(STlTRU5 S[5lgTRT5 g)NcUR5(agSn[UR5H(up#[U[R 5(dM&UnM* Ub![ X5n[XRU5$U$rw) ryrrrrrrBr r)rpartial_mesh_axisrrrs rt _reshard_gradG_ShardingStageBase._register_hook_for_param_grad.._reshard_grad"sp||~~$(!,5doo,F(I!)T\\::,5)-G%0%@&N#4):):NKKKrsc>[R"U[R5nUR5 TRcT"U5TlgTRR T"U55 grw)rcastr _clear_dataradd_)rrVrrs rt_main_grad_hookI_ShardingStageBase._register_hook_for_param_grad.._main_grad_hook0sP{{48H    &"/"9$$]8%<=rsT)r rr register_hook$already_register_final_backward_hook)rrrrs ` @rtr0_ShardingStageBase._register_hook_for_param_grad!sJ  >   - -"EO    0FJ   C    .rsrN)rrFrrFrfrF)r~rFrrFrr)r~rFrrFrrrfrF)rrFrfrF) rmrnrorprrr rrrrrrrrlrsrtrrsd* ,)'',2' 'R :  %+ 9=  J/rsrcF^\rSrSrSSU4SjjjrSSjrSrU=r$)ri@c2>[TU]X!5 SUlg)Nr)rrrrrrr~s rtr_ShardingStage0.__init__As 1rscRUS:Xa [5(aURU5$U$r)r1r)rr[rr~s rt__call___ShardingStage0.__call__Gs' &=_..??G G rs)rrwrz int | strrzProcessMesh | Nonerfrgr[rrrFr~rFrfrF)rmrnrorprrrrrrs@rtrr@s2GK!*2D  rsrcJ^\rSrSrSrSSU4SjjjrSSjrSrU=r$) riNa A builtin shard_fn for shard_optimizer interface, users can pass it to shard_optimizer to implement sharding optimization with stage 1. Args: sharding_mesh_dim(int|str): The sharding dimension in the mesh. mesh(None|paddle.distributed.ProcessMesh): If mesh is not None, the `ProcessMesh` object describes the Cartesian topology of the used processes for dense type parameters. Note: Currently, only one mesh configuration is supported for all dense parameters. If there is a need for multiple mesh configurations, please configure them yourself in the upper layer networking code. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> mesh = dist.ProcessMesh([0, 1], dim_names=["x"]) >>> class MLP(paddle.nn.Layer): ... def __init__(self): ... super().__init__() ... self.fc1 = paddle.nn.Linear(8, 8) ... self.fc2 = paddle.nn.Linear(8, 8) ... ... def forward(self, input): ... return self.fc2(self.fc1(input)) >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> layer = MLP() >>> batch = paddle.rand(shape=[8, 8]) >>> opt = paddle.optimizer.AdamW(parameters=layer.parameters()) >>> opt = dist.shard_optimizer(opt, dist.ShardingStage1("x", mesh)) >>> for _ in range(5): >>> loss = layer(batch) >>> loss.backward() >>> opt.step() >>> opt.clear_grad() >>> # This case need to be executed in multi-card environment >>> # python -m paddle.distributed.launch --gpus=0,1 {test_case}.py c$>[TU]X!5 grwrrrs rtrShardingStage1.__init__u 1rscUR5(dU$UR(dSU;a[X R5nOK[ [ UR R55Vs/sHn[R"5PM nnUS:Xa [5(aURU5nURX2U5$s snf)Nrr) ryrrBrrUrrrrrr1rrrr[rr~rrHs rtrShardingStage1.__call__|s}}M((V3->4U>> import paddle >>> import paddle.distributed as dist >>> mesh = dist.ProcessMesh([0, 1], dim_names=["x"]) >>> class MLP(paddle.nn.Layer): ... def __init__(self): ... super().__init__() ... self.fc1 = paddle.nn.Linear(8, 8) ... self.fc2 = paddle.nn.Linear(8, 8) ... ... def forward(self, input): ... return self.fc2(self.fc1(input)) >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> layer = MLP() >>> batch = paddle.rand(shape=[8, 8]) >>> opt = paddle.optimizer.AdamW(parameters=layer.parameters()) >>> opt = dist.shard_optimizer(opt, dist.ShardingStage2("x", mesh)) >>> for _ in range(5): >>> loss = layer(batch) >>> loss.backward() >>> opt.step() >>> opt.clear_grad() >>> # This case need to be executed in multi-card environment >>> # python -m paddle.distributed.launch --gpus=0,1 {test_case}.py c$>[TU]X!5 grwrrs rtrShardingStage2.__init__rrsc2UR5(a|SU;a[X R5nOK[[ UR R 55Vs/sHn[R"5PM nn[UUR US9$U$s snf)Nrr) ryrBrrUrrrrrrrs rtrShardingStage2.__call__s ==??S 8.. #3u'9'9'?'?#@AANN$A ''%   sBrlrwrrrrs@rtrrs>$R$(2$2!2  22rsrcV^\rSrSrSrSS U4SjjjrSrSrS SjrSr U=r $) ria A builtin shard_fn for shard_optimizer interface, users can pass it to shard_optimizer to implement sharding optimization with stage 3. Args: sharding_mesh_dim(int|str): The sharding dimension name in the mesh. mesh(None|paddle.distributed.ProcessMesh): If mesh is not None, the `ProcessMesh` object describes the Cartesian topology of the used processes for dense type parameters. Note: Currently, only one mesh configuration is supported for all dense parameters. If there is a need for multiple mesh configurations, please configure them yourself in the upper layer networking code. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> mesh = dist.ProcessMesh([0, 1], dim_names=["x"]) >>> class MLP(paddle.nn.Layer): ... def __init__(self): ... super().__init__() ... self.fc1 = paddle.nn.Linear(8, 8) ... self.fc2 = paddle.nn.Linear(8, 8) ... ... def forward(self, input): ... return self.fc2(self.fc1(input)) >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> layer = MLP() >>> batch = paddle.rand(shape=[8, 8]) >>> opt = paddle.optimizer.AdamW(parameters=layer.parameters()) >>> opt = dist.shard_optimizer(opt, dist.ShardingStage3("x", mesh)) >>> for _ in range(5): >>> loss = layer(batch) >>> loss.backward() >>> opt.step() >>> opt.clear_grad() >>> # This case need to be executed in multi-card environment >>> # python -m paddle.distributed.launch --gpus=0,1 {test_case}.py c$>[TU]X!5 grwrrs rtrShardingStage3.__init__rrscUR5(a}URbp/n[[URR55H'nUR [ R"55 M) URX R5 UR5(ad[XR5n[ R"XRU5nUR5RUR55 ggrw)is_denserrUrrrrr _to_dist_ryrBrr rr}r{)rrrrHr shard_params rtrShardingStage3._shard_parameters >>   6J3tzz//01!!$.."232 OOJ 3 ==??8**N,,))>K     / / 0F0F0H I rsc|UR5(aURn[X R[R 5(a![R "5X R'[R"XRU5nUR5RUR55 ggrw) ryrrrrrrr rr}r{)rrrrs rt_unshard_parameter!ShardingStage3._unshard_parameter s ==??"--N.)<)<=tzzJJ6:nn6F223 U,>,>OI     / / 0D0D0F G rscUR5(dU$US:Xa[5(a [S5eSU;a9URn[ SU55(a[ X R 5nO9URRVs/sHn[R"5PM nnURX2U5$s snf)Nrz3Sharding Stage 3 does not support auto dp mode yet.rc3V# UHn[U[R5v M! g7frw)rrr)rrs rtr*ShardingStage3.__call__..# sE*Q:a00*r) ryr1r"rrrBrrrrrrrs rtrShardingStage3.__call__ s}}M &=_..E   ))JE*EEE8.. 5:4F4F4L4LM4Lq$..*4LJM$$VJ??NsCrlrwrr) rmrnrorprrrrrrrrrs@rtrrsL$R$(2$2!2  22J"H@@rsrc[XU5$)a Warp the global view optimizer to distributed view. Note: The `shard_fn` should have the following signature: def shard_fn(accumulator_name, param, accumulator) -> sharded_accumulator Args: optimizer (paddle.optimizer.Optimizer): The optimizer to be sharded. shard_fn (Callable, optional): The function to shard accumulators. If not specified, we simply pass down the dist attr of the params. Returns: An optimizer with distributed view. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> mesh = dist.ProcessMesh([0, 1], dim_names=["x"]) >>> class MLP(paddle.nn.Layer): ... def __init__(self): ... super().__init__() ... self.fc1 = paddle.nn.Linear(8, 8) ... self.fc2 = paddle.nn.Linear(8, 8) ... ... def forward(self, input): ... return self.fc2(self.fc1(input)) >>> layer = MLP() >>> batch = paddle.rand(shape=[8, 8]) >>> opt = paddle.optimizer.AdamW(parameters=layer.parameters()) >>> opt = dist.shard_optimizer(opt) >>> for _ in range(5): >>> loss = layer(batch) >>> loss.backward() >>> opt.step() >>> opt.clear_grad() >>> # This case need to be executed in multi-card environment >>> # python -m paddle.distributed.launch --gpus=0,1 {test_case}.py )rs)ryrkrs rtshard_optimizerr, s` 90K LLrsc,Sn[X5UlU$)a Warp the global view grad_scaler to distributed view. Args: scaler (paddle.amp.GradScaler): The GradScaler to be sharded. Returns: A GradScaler with distributed view. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> mesh = dist.ProcessMesh([0, 1], dim_names=["x"]) >>> class MLP(paddle.nn.Layer): ... def __init__(self): ... super().__init__() ... self.fc1 = paddle.nn.Linear(8, 8) ... self.fc2 = paddle.nn.Linear(8, 8) ... ... def forward(self, input): ... return self.fc2(self.fc1(input)) >>> layer = MLP() >>> batch = paddle.rand(shape=[8, 8]) >>> opt = paddle.optimizer.AdamW(parameters=layer.parameters()) >>> layer, opt = paddle.amp.decorate(layer, opt, level='O2') >>> scaler = paddle.amp.GradScaler(init_loss_scaling=1024) >>> scaler = dist.shard_scaler(scaler) >>> opt = dist.shard_optimizer(opt) >>> for _ in range(5): >>> with paddle.amp.auto_cast(True): >>> loss = layer(batch) >>> scaled = scaler.scale(loss) >>> scaled.backward() >>> scaler.step(opt) >>> scaler.update() >>> opt.clear_grad() >>> # This case need to be executed in multi-card environment >>> # python -m paddle.distributed.launch --gpus=0,1 {test_case}.py c UR(dgUR[U5nUS[RLa [ S5eUS[R La [ S5eSnSn[R"[R"S/5R[R55Ul 0n[USS5(a[UR S["5(aUR HnUSHnUR%5nUcM[USS5"5(dM2UcUR&b UR&nOUc.UR)5(aUR&b UR&nUR&U;aU/XXR&'MXXR&R+U5 M M OUR,HnUR%5nUcM[USS 5"5(dM2Uc UR&nUc!UR)5(a UR&nUR&U;aU/XXR&'MXXR&R+U5 M UR/5GHZup/n /n [R"[R"S/5R[R55n [R"[R"S/5R[R55n[R"[R"S/5R[R55nUR0R35(aUR0R55nO UR0nU HnUR6[8R:R<R>[R@[8R:R<RB[RD4;aU R+U5 MU R+U5 M [GU 5(a[HRJ"U U5upS [RLRO5;aD[RP"U S 5n [RR"U 5n [RP"U S 5n [HRT"X5n [GU 5(a[HRJ"U U5upS [RLRO5;aD[RP"US 5n[RR"U5n[RP"US 5n[HRT"X5n [VRX"XU RZ5n [HRT"URU 5Ul GM] URR&U:XaNUR]5H9n[VRX"URUURRZ5n M; OZUb[_US 5(d [aS5e[VRX"URUURRZ5Ul [RUS'g)NstatezMunscale_() has already been called on this optimizer since the last update().z(unscale_() is being called after step().rrrrzcgNFrlrlrsrtrb6shard_scaler..unscale_method.. srscgrrlrlrsrtrbr sUrsxpuint32boolranksz:Invalid current_process_mesh: must be a valid ProcessMesh.)1_enable_optimizer_statesr2rUNSCALEDr"STEPPEDrrrrastypebool_ _found_infrrrr _grad_ivarrrzrrr7_scaleryr|rr*VarDescVarTypeFP16float16BF16bfloat16rr check_finite_and_unscale_rI get_devicersum bitwise_orrr rrror)rryoptimizer_statesrc_meshcurrent_process_meshmesh2param_gradsrrtgt_gradrH param_gradstemp_param_grads_halftemp_param_grads_fp32temp_found_inftemp_found_inf_halftemp_found_inf_fp32 temp_scalerrs rtunscale_method$shard_scaler..unscale_method s|| 00I? 7 #~'>'> >_ W %)?)? ?IJ J# **288QC=+?+?+IJ 9ot 4 4  # #A &: : #00"8_E$//1H ,#$&7 %, ( 5 5 A'/'<')>?()>)>?FFxP!3$/446NA$& !$& !#--bhhsm.B.B288.LMN"("2"2! $$RXX.# #)"2"2! $$RXX.# {{""$$![[557 ![[ #::LL((--NNLL((--OO " *006)006$()))/)I)I)*& FMM4466,2KK-w-)-3JJ7L,M),2KK-v-)"(!2!2""()))/)I)I)*& FMM4466*0+++W+'+1**5H*I'*0+++V+'"(!2!2"""\\.*C*CN%//PDOE7J ?? ' '+? ? 0 5 5 7 LLOO\4??3M3M!8 $+7$g44!P#ll$**DO $2#:#: rs)r_unscale)scalerr(s rt shard_scalerr,_ sZ[;z!8FO MrscB\rSrSr%SrS\S'S\S'S\S'S SjrS rg) FusePassesi/ z8 A helper class for users to configure the fuse passes. renable gemm_epilogue dropout_addNcSUlSUlSUlUbDUR5H/up#[ X5(a[ XU5 M#[ SU35e gg)NFzUnknown fuse pass )r/r0r1r7rorr)r config_dictr[rs rtrFusePasses.__init__8 sa "   ")//1 4%%Du-$'9#%?@@ 2 #rs)r1r/r0rw)rmrnrorprrqrrrrlrsrtr.r./ s  L Arsr.c^\rSrSrSrS S U4SjjjrSr\SSj5r\SSj5r \SSj5r \SSj5r \SS j5r \SS j5r S rU=r$)riD a The `Strategy` object is used to configure the parallelization and optimization strategies for static graph. Currently supports configuring ``sharding``, ``fused_passes``, ``gradient_merge`` and ``pipeline``. More strategies will be supported in the future. ``sharding`` is used to configure the sharding states of the optimizer, for saving the GPU memory. ``fused_passes`` is used to configure the fusion of the computation in the model. ``gradient_merge`` is used to configure the gradient merge strategy in training. ``pipeline`` is used to configure the pipeline parallelism strategy. Args: config(dict|None, optional): The user-defined configurations. If ``config`` is None, use default configurations. If it is a dict, the items inside the dict will be used to set the configurations, and the others remain the default values. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> strategy = dist.Strategy() >>> strategy.sharding.enable = True >>> strategy.sharding.stage = 2 >>> strategy.sharding.degree = 2 >>> strategy.gradient_merge.enable = True >>> strategy.gradient_merge.k_steps = 2 >>> strategy.gradient_merge.avg = False >>> strategy.pipeline.enable = True >>> strategy.pipeline.schedule_mode = "1F1B" # default is "1F1B" >>> strategy.pipeline.micro_batch_size = 2 c>Ub?[U[5(a[R"U5UlO[ SU35e0Ul[ RRn[TU])X R5 URR[ RRS5n[ R"U5UlURR[ RRS5n[ R "U5UlURR[ RR$S5n[ R&"U5UlURR[ RR*S5n[ R,"U5UlURR[ RR0S5n[3U5UlURR[ RR6S5n[ R8"U5UlURR[ RR<S5n[ R>"U5Ul URR[ RRBS5n[ RD"U5Ul#URR[ RRHS5n[ RJ"U5Ul&URRSS5Ul'g)Nz%Expected a dictionary. But received: full_graphT)(rrrr _config_dictr auto_strategy constantsBASErrrSHARDINGShardingConfig _shardingGRADIENT_MERGEGradientMergeConfig_gradient_mergePIPELINEPipelineConfig _pipelineAMP AMPConfig_amp FUSED_PASSESr. _fused_passes RECOMPUTERecomputeConfig _recomputeMP_OPTIMIZATIONMPOptimizationConfig_mp_optimizationDP_OPTIMIZATIONDPOptimizationConfig_dp_optimizationSP_OPTIMIZATIONSPOptimizationConfig_sp_optimization _full_graph)rrcategoryr3r~s rtrStrategy.__init__q s  &$''$(MM&$9! ;F8D!#D  **// #4#45''++  # # , ,d '55kB''++  # # 2 2D  -@@M''++  # # , ,d '55kB''++M,C,C,G,GN !++K8 ''++  # # 0 0$ ( 4''++  # # - -t (77 D''++  # # 3 3T !. B B; O''++  # # 3 3T !. B B; O''++  # # 3 3T !. B B; O,,00tDrsc SSKn[RRn[RR U5nUR 5Hn[ X[X55 M URRURl SURR;aSURl SURR;aSURl UR"UR5UlUR"UR"5UlUR"UR&5UlUR"UR*5UlUR"UR.5UlUR"UR25UlUR"UR65UlUR"UR:5Ulg)z_ NOTE(lizhiyu): This is a template function to get `dist.Strategy` from `fleet.auto.Strategy`. rNfused_gemm_epilogue_passTfused_dropout_add_pass)rr9r:r;get_category_default_configrrrrdr/rIfused_passes_listr0r1rrgrGrcr>rerArfrDrhrLrirOrjrRrkrU)rlegacy_strategyrrW base_configr[s rt_from_legacy_strategyStrategy._from_legacy_strategy sq  **//#--II  ##%C Dw< =&$3$@$@$G$G! &++== >04D   , $++== >.2D   *MM/"5"56 '?'?@#}}_-K-KL'?'?@--(A(AB $ o.M.M N $ o.M.M N $ o.M.M NrscUR$)z Whether to use AST mode. )rVrs rtr7Strategy.full_graph s rscUR$)a ``sharding`` is used to configure the sharding states of the optimizer, containing following configs: ``enable`` (bool): whether to enable sharding. Default: False. ``stage`` (int): can be set to 1, 2 or 3. 1 indicates the optimizer states segmentation, 2 indicates optimizer states and gradient segmentation, 3 indicates the segmentation of optimizer states, gradient and parameters. Default: 1. ``degree`` (int): the number of segmentation pieces. Default: 8. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> strategy = dist.Strategy() >>> strategy.sharding.enable = True >>> strategy.sharding.stage = 2 >>> strategy.sharding.degree = 2 )r>rs rtrcStrategy.sharding s4~~rscUR$)a ``gradient_merge`` is used to configure the gradient merge strategy in training, containing following configs: ``enable`` (bool): whether to enable gradient merge. Default: False. ``k_steps`` (int): the number of steps for merging gradients. Default: 1. ``avg`` (bool): whether to average the gradients of each step. Default: True. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> strategy = dist.Strategy() >>> strategy.gradient_merge.enable = True >>> strategy.gradient_merge.k_steps = 2 >>> strategy.gradient_merge.avg = True )rArs rtreStrategy.gradient_merge s0###rscUR$)a ``fused_passes`` is used to configure the fusion of the computation in the model, containing following configs: ``enable`` (bool): whether to enable fused passes. Default: False. ``gemm_epilogue`` (bool): whether to fuse ``matmul`` and ``add`` computation in the ``Linear`` layer. Default: False "dropout_add" (bool): whether to fuse ``dropout`` and ``add`` computation. Default: False. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> strategy = dist.Strategy() >>> strategy.fused_passes.enable = True >>> strategy.fused_passes.gemm_spilogue = True >>> strategy.fused_passes.dropout_add = True )rIrs rtrdStrategy.fused_passes s2!!!rscUR$)a ``pipeline`` is used to configure the pipeline parallelism, containing following configs: ``enable`` (bool): whether to enable pipeline parallelism. Default: False. ``schedule_mode`` (str): the scheduling mode of pipeline parallelism. Default: "1F1B". ``micro_batch_size`` (int): the size of each micro-batch inside a mini-batch. Default: 1. ``accumulate_steps`` (int): number of steps for accumulating. Default: 1. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> strategy = dist.Strategy() >>> strategy.pipeline.enable = True >>> strategy.pipeline.micro_batch_size = 2 )rDrs rtrfStrategy.pipeline& s2~~rscUR$)a6 ``amp`` is used to configure the amp, containing following configs: ``enable`` (bool): whether to enable AMP. Default: False. ``dtype``, (str): the data type of AMP. Default: "float16". ``level``, (str): the level of AMP. Default: "O1". ``init_loss_scaling``, (float): the initial value of loss scaling. Default: 32768.0 ``incr_every_n_steps``, (int): the number of steps for increasing loss scaling. Default: 1000 ``decr_every_n_nan_or_inf``, (int): the number of steps for decreasing loss scaling. Default: 2 ``incr_ratio``, (float): the ratio for increasing loss scaling. Default: 2.0 ``decr_ratio``, (float): the ratio for decreasing loss scaling. Default: 2.0 ``use_dynamic_loss_scaling``, (bool): whether to use dynamic loss scaling. Default: False ``custom_white_list``, (list): the list of names for which AMP will be applied. Default: [] ``custom_black_list``, (list): the list of names for which AMP will not be applied. Default: [] ``custom_black_varnames``, (list): the list of names for which AMP will not be applied. Default: [] ``use_fp16_guard``, (bool): whether to use fp16 guard. Default: False ``use_bf16_guard``, (bool): whether to use bf16 guard. Default: False ``use_master_grad``, (bool): whether to use master grad. Default: False Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> strategy = dist.Strategy() >>> strategy.amp.enable = True >>> strategy.amp.dtype = "float16" >>> strategy.amp.level = "O2" )rGrs rtrg Strategy.ampA sDyyrs) rGr8rRrVrIrArOrDrLr>rUrw)rz_Config | Nonerfrgrfr)rfzauto_strategy.ShardingConfig)rfz!auto_strategy.GradientMergeConfig)rfr.)rfzauto_strategy.PipelineConfig)rfzauto_strategy.AMPConfig)rmrnrorprrr`rr7rcrerdrfrgrrrrs@rtrrD s*X9E9Ev OD  6$$2""44!!rsrc$\rSrSrSrSSSjjrSSjrSSjrSSjrSr SSS jjr SSS jjr SSS jjr SSS jjr S rSr\S Sj5rSSjrS!S"SjjrSrS#SjrSrSrSrSrSrSrSrg)$ DistModelif aL `DistModel` is the model converted from a ``paddle.nn.layer`` with distributed tensors as its parameters. It contains the static graph converted from a ``paddle.nn.layer`` whose parameters are distributed tensors (constructed from ``paddle.distributed.shard_tensor``), and provides the APIs for training, evaluation and prediction with the static graph. It is suggested to generate DistModel by ``paddle.distributed.to_static``, not directly by ``paddle.distributed.DistModel``. Please first set the DistModel to "train", "eval" or "predict" mode with ``train()/eval()/predict()`` method and then use the ``__call__`` method for training, evaluation and prediction respectively. For more details of the usage, please refer to the sample code in ``paddle.distributed.to_static``. Args: layer(paddle.nn.Layer): The layer in dygraph mode, whose parameters are distributed tensors generated by ``shard_tensor``. loader(ShardDataLoader|paddle.io.DataLoader): The data loader used in dygraph mode, used to infer inputs_spec and labels_spec. loss(Loss|Callable|None, optional): The loss function for training or evaluating the model. Can be a `paddle.nn.Layer` instance or any callable function. If loss is not None, DistModel will be set to "train" (when the optimizer is also not None) or "eval" mode (when optimizer is None) in default. If it is None, DistModel will be set to "predict" mode in default. Default: None. optimizer(paddle.optimizer.Optimizer|None, optional): The optimizer for training. If both optimizer and loss are set, DistModel will be set to "train" mode in default. Default: None. strategy(paddle.distributed.Strategy|None, optional): Configs for parallel strategies and optimization settings (e.g. sharding, pipeline parallelism). Default: None. input_spec(list[list[paddle.distributed.DistributedInputSpec]]|None, optional): The custom input specs specify the shape, dtype, and name information of model inputs and labels. If it is not None, the input specs and label specs will be inferred from the custom input specs. The custom input specs should be a list containing two sublists: the first sublist represents theinput specs, and the second sublist represents the label specs. Default: None. NcURU5UlUR5R5VV s0sHupXR_M sn nUlUR R5VV s0sHupX_M sn nUl[R"S5(ac[RRR[R5RS5 [R R#SS9 U(aUR$R&(a[)U[*5(a[-US5(a[-US5(a[/5(a[)UR0[25(dS5e[)UR0[25(a/UR0n UR4n [7XUR5nO[R8"S5 [;XXFURS 9UlSUl0Ul Ub)US UR<l!US UR<l"O[)U[F5(a=UR<RIU5uUR<l!UR<l"O^URJRLn UR<ROURPSU 5uUR<l!UR<l"[RRTRVRYS 5S Ul-URZ(d<UbUbUR]5 gUbUR_5 gURa5 ggs sn nfs sn nf) NPOD_NAMEz0Distribute training by paddle.distributed.launchT) is_collectiverruzKThe shard_fn should be ShardingStage1 when stage1 tensor fusion is enabled.z7Sharding tensor fusion only support ShardingStage1 now.)rrr.FLAGS_enable_pir_api)1_DistModel__convert_strategy_inner_strategyrr7r0_structured_to_parameter_name_parameter_to_structured_namergetenvrutils log_utils get_loggerloggingINFOinforinitrcrrrsrorrrrur@warningr_engine_mode_feed_name_list _inputs_spec _labels_specShardDataloader"_prepare_data_spec_from_dataloader batch_sampler batch_size_prepare_data_specdatasetrr r get_flags _in_pir_moder\r]r^) rrZloaderrryrmetrics input_specrrrk inner_optrs rtrDistModel.__init__ s $66x@"'"2"2"4":":"<. "<$!AvvI"<. *"??EEG. GTQADG. * 99Z JJ + +GLL 9 > >B  JJOO$O / !!669o66 ;// <00 i11>BB 8 B)--~>>$..%00 3)=)= M d6J6J   !  !(21 DLL %(21 DLL %  0 0 ??G  ) ) --88J //j  ) )#KK11;; " "!!$)9 !  "G. . s N6NcURRS(dURRSSS9 SUlURR S5 [ R "5 g)z Set the DistModel to "train" mode. In "train" mode, executing ``__call__`` method will update the parameters of the model and return the loss. r\Fmodeinit_parametersNr _has_prepared_prepare_programrto_moderdisable_staticrs rtr\DistModel.train sR ||))'2 LL ) )w ) N  W%rscURRS(dURRSSS9 SUlURR S5 [ R "5 g)zc Set the mode of DistModel to "eval". In "eval" mode, executing ``__call__`` will return the loss. r]FrNrrs rtr]DistModel.eval sR ||))&1 LL ) )vu ) M  V$rsc6URRS(dDURR[R"URR 5SSSS9 SUlURRS5 [R"5 g)z Set the mode of DistModel to "predict". In "predict" mode, executing ``__call__`` returns a dict that contains the outputs of the model. r^NFr) rrpreparerrrrrrrrs rtr^DistModel.predict ss ||)))4 LL  dll778 % !   Y'rsc|UcURc [S5eUc URnUS;a [S5eU$)Nz;Please set the mode or call train()/eval()/predict() first.r[z.mode can only be 'train', 'eval' or 'predict'.)rrrrs rt__validate_modeDistModel.__validate_mode sJ ',}}!"&*5'?U$V!"(-'9'9-2-=-= #-#7+0+=+=+?*AC(7E&+&C(+?';q'@154C,-5**8.)2?N,-?**8.)<2748 >B(;*:(-(:(:(<363G/52; -,h s,A0L EL?+L+ L(cURURRS9n[5(a [ U5nO,[ X RR UR5nSn0n[ RRR5 UR5H%upgXc;dSUSUS35eU"XsU5XV'M' SSS5 U$!,(df  U$=f)zW Args: local_state_dict(Dict[str, libpaddle.Tensor]): The state dict from program. rc[U[R[R[R R45(de[U[R5(d[R"U5n[U[R5(dSUS[ U5S35e[UR5[US5:XdSURSUSS35eURn[[R"US5RUS 5US S 9n[USU5n[XU5nUR5RUR:Xd/SUR5RS UR35e[R"XR55 U$) Nz local tensor:z type z is not paddle.Tensor.rzlocal tensor shape z! not equal to dims_mapping shape rr process_shaper)rz! not equal to local_tensor.shape:)rrrFrndarrayr rrrrrrr=r!r|r1)rr?rrrrs rtbuild_distributed_tensorIDistModel._build_distributed_state_dict..build_distributed_tensor[ sv}}bjj&++:L:LM lFMM::%}}\: lFMM::  ~VD4F3GG]^ :|))*c)N2K.LL %l&8&8%99Z[des[tZuuvw L(--L?34<<o.$K0 D 'y'@$GJ,\LK++-33|7I7II %k&>&>&@&F&F%GGhiui{i{h|} I MM,(@(@(B C rszvar z not in dist attrs:rN) rrrrr#_dist_contextsrr rrr7)rrr dist_attrsrglobal_state_dictvar_namer~s rtr'DistModel._build_distributed_state_dictM s !22 8J8J2K ==&'89J'!<<#>#>tzz#JJ 6 [[ & & ($4$:$:$< -8*$7 |1E-/Gx0/!+ %=)! ) (! s :C C+c 0nURURRS9nURSS9nSnUR(a URR R OSnURRb U(aSnUR5HupxUR5(dSUSUS35eXt;aXGn URUUR:Xda[URU R5(d<SURS U RS URS U RS 3 5eXpR;aURUOUn [UR55X*'M URR Gb["R$R&R)5 URR R5GHupU GHzn U R5uup/nUR5H4upx[+US R,U5nUcM#UR/U5 M6 [1U5S :XaMsUGHn/nUH_nUR,U-U;a([2R4"UR,U-S 35 O%UR/UUR,U-5 Ma [1U5[1U5:XdMSU;aUS nOB[1U5S:Xa!SnUS R6nUSR6nOSnSnSn[9UUUUS9n[U5X.U-'UHnUR;UU-5 M GM GM} GM SSS5 [=5(a0UR?U["R@RC5U5 gUR?U["R@RC55 g!,(df  N|=f)NrF)rTrrrz process_mesh: != z or placements:z not matchrz is not in state_dict.rrr.)rrr)"rrrrrvrcrrr7ryrr9rrwrr|rrr rrrr0rrwarningswarnrr"rrset_state_dictrr)rrrrcur_state_dict copy_tensorrrrcur_v param_namer[rrrr suffix_namesrconcat_tensorsori_pfused_wrrrs rtrDistModel.set_state_dict s 22 8J8J2Ke< ##  ) ) > >  << " " .3GK$$&DA99;; J$qc3I J J;"&)~~*,"8LL%"2"2## $ANN#348J8J7K?[\[g[gZhhlmrm}m}l~IJ :::2215  ,99I+J  (!'& << % % 1$$**,%)\\%?%?%E%E%GMC&. N#N$,MM+0::+>*??U(V%&%*$2$9$9(8f9L(M%&*9 #>2c/6JJ#-#7.B(;.=(6/52;    , , &--"<"<"> {-,sA6O/B)O//B O// O=cURRnUcU$[U[RR R R5(a URn[U[5(dS5eU$)NzUThe optimizer should be ShardingOptimizerStage1 when stage1 tensor fusion is enabled.) rrrrrrg decoratorOptimizerWithMixedPrecisionr@)rrys rt_get_shard_stage1_optimizer%DistModel._get_shard_stage1_optimizer syLL++      MM   ' ' C C  ",,I)%<== c =rsc UR(a URRROSnU(dS5eUR5nUcS5eUR 5Vs/sH#nXPR ;aUR UOUPM% nn[ [U[UR5555nU"XA5 UR 5Vs/sH#nXPR;aURUOUPM% nn[ [U[UR5555nU$s snfs snf)NFz:Can only convert state_dict when tensor fusion is enabled.z!The optimizer should not be None.) rvrcrrrrwrrorr6rx)rroptimizer_functionrryrparameter_namesstructured_namess rt!_convert_state_dict_tensor_fusion+DistModel._convert_state_dict_tensor_fusion sZ##  ) ) > >  $ H #446 $I&II$ __&  ':::2215'  #otJ4E4E4G/HIJ 91 __&  ':::2215'  #.Z5F5F5H0IJK -  s .*D: *D?c*SnURX5$)Nc&URU5 grw)(convert_state_dict_with_rank_unique_nameryrs rtrODistModel._convert_state_dict_with_rank_unique_name..optimizer_functions  > >z Jrsrrrrs rt)_convert_state_dict_with_rank_unique_name3DistModel._convert_state_dict_with_rank_unique_names  K55   rsc*SnURX5$)Nc&URU5 grw).convert_state_dict_without_tensor_fusion_paramrs rtrUDistModel._convert_state_dict_without_tensor_fusion_param..optimizer_function's  D DZ Prsrrs rt/_convert_state_dict_without_tensor_fusion_param9DistModel._convert_state_dict_without_tensor_fusion_param&s  Q55   rsc*SnURX5$)Nc&URU5 grw)+convert_state_dict_with_tensor_fusion_paramrs rtrRDistModel._convert_state_dict_with_tensor_fusion_param..optimizer_function/s  A A* Mrsrrs rt,_convert_state_dict_with_tensor_fusion_param6DistModel._convert_state_dict_with_tensor_fusion_param.s  N55   rsc*SnURX5$)Nc&URU5 grw)#convert_state_dict_with_origin_namers rtrJDistModel._convert_state_dict_with_origin_name..optimizer_function7s  9 9* Ersrrs rt$_convert_state_dict_with_origin_name.DistModel._convert_state_dict_with_origin_name6s  F55   rs)rrrrvrrxrwrr)rZrQrzShardDataloader | DataLoaderr!Layer | Callable[..., Any] | NoneryOptimizer | NonerStrategy | Nonerzlist[Metric] | Noner'list[list[DistributedInputSpec]] | Nonerfrg)rfrgrw)rz _Mode | NonerfrL)r(zSequence[Any] | Tensorrfr)rT)rzLiteral['opt', 'param', 'all']rrrfdict[str, Tensor])rr8rfrg)rmrnrorprrr\r]r^rrrrrrrurrrrrrrrr r&r,r2rrrlrsrtrprpf s)^37&*$('+>BUU-U0 U $ U " U%U<U Un     $ 8(;&:&=$>B+Z$$L @"05!\,\\  \|2!heN"#J    rsrpc v[U[5(Gac[5(GdSURnURnUR nUGb+Uc[ R"5OUnUcS5e[U[5(a3SURl SURl XtRl O[U[5(a3SURl SURl XtRl Ow[U[5(aWSURl SURl XtRl URHnUR!U5 M O [#S5eUbUR$(a ['XX#XES9n U $[(R*R-USS 9nU$) a Converts the ``layer`` with distributed tensor (constructed from ``paddle.distributed.shard_tensor``) to a static graph. ``to_static`` returns a DistModel instance containing the static graph for distributed training, evaluation and prediction. Args: layer(paddle.nn.Layer): The layer in dygraph mode, the parameters or its inputs can be distributed tensors. loader(ShardDataloader|paddle.io.DataLoader): The data loader used in dygraph mode, used to infer inputs_spec and labels_spec. loss(Loss|Callable|None, optional): The loss function for training or evaluating the model. Can be a `paddle.nn.Layer` instance or any callable function. Default: None. optimizer(paddle.optimizer.Optimizer|_ShardOptimizer|None, optional): The optimizer for training. It can `paddle.optimizer.Optimizer` or `_ShardOptimizer` wrapped by `shard_optimizer`. Default: None. strategy(paddle.distributed.Strategy|None, optional): Configs for parallel strategies and optimization settings (e.g. sharding, pipeline parallelism). Default: None. input_spec(list[list[paddle.distributed.DistributedInputSpec]]|None, optional): The custom input specs specify the shape, dtype, and name information of model inputs and labels. If it is not None, the input specs and label specs will be inferred from the custom input specs. The custom input specs should be a list containing two sublists: the first sublist represents theinput specs, and the second sublist represents the label specs. Default: None. Returns: DistModel: A ``DistModel`` instance converted the input ``layer``. Examples: .. code-block:: python >>> import numpy as np >>> import paddle >>> import paddle.distributed as dist >>> from paddle import nn >>> from paddle.distributed import Replicate, Shard >>> BATCH_SIZE = 4 >>> BATCH_NUM = 4 >>> IMAGE_SIZE = 16 >>> CLASS_NUM = 8 >>> class RandomDataset(paddle.io.Dataset): # type: ignore[type-arg] ... def __init__(self, images, labels, num_samples): ... self.images = images ... self.labels = labels ... self.num_samples = num_samples ... def __getitem__(self, idx): ... return self.images[idx], self.labels[idx] ... def __len__(self): ... return self.num_samples >>> class DemoNet(nn.Layer): ... def __init__(self, mesh): ... super().__init__() ... self._mesh = mesh ... self.linear_0 = nn.Linear(IMAGE_SIZE, IMAGE_SIZE) ... self.linear_1 = nn.Linear(IMAGE_SIZE, CLASS_NUM) ... self.relu = nn.ReLU() ... # shard the weights of this layer ... self.linear_0.weight = dist.shard_tensor( ... self.linear_0.weight, ... self._mesh, ... [Shard(1)], ... stop_gradient=False, ... ) ... self.linear_1.weight = dist.shard_tensor( ... self.linear_1.weight, ... self._mesh, ... [Shard(0)], ... stop_gradient=False, ... ) ... def forward(self, x): ... out = self.linear_0(x) ... out = self.relu(out) ... out = self.linear_1(out) ... return out >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> images = np.random.rand(BATCH_SIZE, IMAGE_SIZE).astype('float32') >>> labels = np.random.rand(BATCH_SIZE, CLASS_NUM).astype('float32') >>> dataset = RandomDataset(images, labels, BATCH_SIZE) >>> loader = paddle.io.DataLoader(dataset, batch_size=BATCH_SIZE) >>> mesh = dist.ProcessMesh([0, 1], dim_names=["x"]) >>> layer = DemoNet(mesh) >>> opt = paddle.optimizer.SGD( ... learning_rate=0.1, parameters=layer.parameters() ... ) >>> loss_fn = nn.MSELoss() >>> dist_loader = dist.shard_dataloader(loader, meshes=[mesh]) >>> dist_model = dist.to_static( ... layer, dist_loader, loss_fn, opt ... ) >>> # training >>> dist_model.train() >>> for batch_id, (image, label) in enumerate(dist_loader()): ... # in train mode, executing the __call__ method will ... # update the parameters of the model and return the ... # loss ... loss = dist_model(image, label) >>> # evaluation >>> dist_model.eval() >>> for batch_id, (image, label) in enumerate(dist_loader()): ... # in eval mode, executing the __call__ method will ... # return the loss ... loss = dist_model(image, label) >>> # prediction >>> dist_model.predict() >>> for batch_id, (image, label) in enumerate(dist_loader()): ... # in predict mode, executing the __call__ method will ... # return a dict that contains the outputs of the model, ... # where the value of "out0" is the first output. ... outs = dist_model(image) >>> # This case need to be executed in multi-card environment >>> # export CUDA_VISIBLE_DEVICES=0,1 >>> # python -m paddle.distributed.launch {test_case}.py z Sharding degree can not be None.Tr.rzdOnly sharding stage 1, 2 and 3 can to_static for now. User-defined shard_fn will be supported later.)rF)r7)rrsrrrrurrrrcr/stagedegreerrrrrr7rprjitr) rZrrryrrrkrAr dist_models rtrr?sF)_--kmm&&#44((  *2*:t}}H #. 2 .(N33+/!!(*+!!'+:!!(Hn55+/!!(*+!!'+:!!(Hn55+/!!(*+!!'+:!!(&66E//67*z8.. 4H  $$Uu$= rsc[R"5(aUR5SLa [S5eURnUR n[ R"5/[U5-n[XU5n[U[5(a/[R"UR540URD6$[R"UR55$[R R#5(aK[R$R'UR)5UR*5nUR-U5 U$[/S5e)a Converts a distributed tensor to a dense tensor. ``unshard_dtensor`` first make the ``dist_tensor`` be ``Replicate`` state on all processes and then converts it to a dense ``paddle.Tensor``. It can be treated as a reverse operation of ``shard_tensor``. Args: dist_tensor (paddle.Tensor): The distributed tensor which is constructed from a dense tensor with ``shard_tensor``. Returns: paddle.Tensor: The original dense tensor of the input ``dist_tensor``. Examples: .. code-block:: python >>> import paddle >>> import paddle.distributed as dist >>> from paddle.distributed import Replicate, Shard >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> mesh = dist.ProcessMesh([0, 1], dim_names=["x"]) >>> original_tensor = paddle.rand([4, 1024, 512]) >>> dist_tensor = dist.shard_tensor(original_tensor, mesh, [Shard(0)]) >>> # dense_tensor's shape is the same as original_tensor >>> dense_tensor = dist.unshard_dtensor(dist_tensor) Fr$z;`unshard_dtensor()` only supported in dynamic and pir mode.)rrryrrrrrrr rrrr|rrFrrr create_shaped_typerrrr)rrrreplicate_placements r_dist_tensordense_tensor_types rtunshard_dtensorrDs)8    E )HI I'' ++ $ 01C OC 3GH k> 2 2!--**,&&  ==!;!;!=> >    % % ' '"JJ99     1 1  ./" I  rsc\rSrSrSrSSSjjrSrSrSrSr S r S r S r SS jr S rSrSrSrg)ri+a ShardDataloader converts a dataloader to a new dataloader which provided two capabilities: 1. split dataloader by shard_dim to do data parallel. 2. reshard the output of dataloader to distributed tensor. if is_dataset_splitted is True, just need to do reshard. Args: dataloader (paddle.io.DataLoader): The dataloader to be sharded. meshes (ProcessMesh|list[ProcessMesh]|tuple[ProcessMesh]): The mesh list of the dataloader. Identify which mesh the input is on. if len(meshes) == 1 or type(meshes) == ProcessMesh, all the inputs are on the same mesh. input_keys (list[str]|tuple[str]): if the iteration result of dataloader is a dict of tensors, input_keys is the keys of this dict, identify which tensor is located on which mesh, one-to-one correspondence with meshes. i.e. dict[input_keys[i]] is on meshes[i]. Default: None, which means the outputs is a list, and the i'th input is on meshes[i]. shard_dims (list|tuple|str|int]): The mesh dimension to shard the dataloader. Users can specify the shard_dim of each mesh or specify a single shard_dim for all meshes. Default: None, which means the data loader will not be split, i.e. mp. is_dataset_splitted (bool): Whether the dataset has been split. dense_tensor_idx (list): A paired 2D list specifies the index of the dense_tensor in the output of dataloader. It allows users to identify which elements within each output batch are dense_tensor. first dense_tensor: the dense_tensor return by dataloader. second dense_tensor: num_or_sections specifies how to split first tensor: evenly (if a number) or unevenly (if a list). Default: None, meaning all outputs are dist_tensors. Note: For dense_tensor_idx settings, the idx must be paired. NcUSLaUc [S5e[U5UlURb[UR5S:Xa [S5e[R "5nUR U5(a[SUSUR35e[UR5S:HUlX0lURU5Ul URU5upUcJ[URS5Sn[URS5Sn Sn URU 5n O"URX5n URU 5n USLdUc#XlURR UlGO[#UR[$5(a4URR UlURUlXlGOs['URR U - 5Ul[#UR[(5(aSn Sn O,URR*n URR,n [%UR.UR U U U U S 9UlURR0URl[2R4R7UR.URUR8UR:UR<UR>UR@URBURDURFURHURJURLS 9 UlSURl'SUl(X`l)g) NTz7shard_dims must be set when is_dataset_splitted is Truerzmeshes must be setz process_id z* is in more than one mesh, the meshes are r.F)rr num_replicasrshuffle drop_last) rrrplaces return_list collate_fn num_workersuse_buffer_readerprefetch_factoruse_shared_memorytimeoutworker_init_fnpersistent_workers)*rr&_meshesrrr_process_id_in_multi_meshes_all_inputs_in_one_mesh _input_keys_process_shard_dims _shard_dims_get_mesh_and_shard_dimrget_rank_by_dim_and_process_id _dataloaderrrrr+intr,rHrIr _acc_stepsriorOrrJrKrLrMrNrOrPrQrR_persistent_workers pin_memoryr5dense_tensor_idx)r dataloadermeshes input_keys shard_dimsis_dataset_splittedrb process_idrrdp_rank dp_world_sizerHrIs rtrShardDataloader.__init__Gs $ &:+=I v << 3t||#4#912 2]]_  + +J 7 7j\)STXT`T`Sab (+4<<'8A'=$%33J?66zB <4<<?+A.D 0 0 34Q7IG --i8M99)PG --i8M $ &**<) (66AADO  002I J J(66AADO!+!9!9D ) !((33mCDO*224LMM! $22::&44>> !8"**??*# "D -7,D,D,O,OD   )%yy33"**"00$..!((&22%00&22",">"> * : :",">">"**)88#-#A#A 4 D  ',# 0rsc[U[[45(dUc/n[[ UR 55Hin[UR U[ [45(a-URU/[ UR U5-5 MXURU5 Mk U$[ U5[ UR 5:wa-[S[ U5S[ UR 535eU$)Nz6shard_dims must be the same length as meshes, but got r) rr]rrUrrTrrrr)rrfresrs rtrX#ShardDataloader._process_shard_dimss j3* - -1CC3t||,-dll1oe}==JJ |c$,,q/.BBCJJz* . J:#dll"33 LSQ[_L]]abefjfrfrbsatu rsc[[UR55Hn[URU[[ 45(aq[[URU55HJnXRUUR ;dM$URUUURUU4s s $ MXRUR ;dMURUURU4s $ g)N)NN)rUrrTrrr _process_idsrY)rrhrjs rtrZ'ShardDataloader._get_mesh_and_shard_dimss4<<()A$,,q/D%=99s4<<?34A!\\!_Q%7%D%DD#||Aq143C3CA3Fq3III5a!=!==<<?D,<, D%j C C;D 99 ((113DI$))_ z**rscSUlU$rwr~rs rtrShardDataloader.__call__jrrs) rVr\rWrTrYrrrbr5NNFN) rczpaddle.io.DataLoaderrdz4ProcessMesh | list[ProcessMesh] | tuple[ProcessMesh]rezlist[str] | tuple[str] | Nonerfzlist | tuple | str | int | Nonergrrblist[list[int]] | Nonerw)rmrnrorprrrXrZrUr{rrrrrrrrrrlrsrtrr+s>596:$)37S1(S1ES12 S1 4 S1 " S11S1j  %  &"8BF"TPl+ rsrc"[UUUUUU5$)a! Convert the dataloader to a ShardDataloader which provided two capabilities: 1. split dataloader by shard_dim to do data parallel if it it not None. 2. reshard the output of dataloader to distributed tensor. if is_dataset_splitted is True, it means that the dataset has been split by users, and just need to do reshard. only if is_dataset_splitted is False and shard_dims is not None, it will do split. Args: dataloader (paddle.io.DataLoader): The dataloader to be sharded. the output of dataloader must be a list or dict of paddle.Tensor with 2 elements, i.e. [input_data, label] or {"input_data": input_data, "label": label}, input_data and label can be a list to support multiple inputs. meshes (ProcessMesh|list[ProcessMesh]|tuple[ProcessMesh]): The mesh list of the dataloader. Identify which mesh the input is on. if len(meshes) == 1 or type(meshes) == ProcessMesh, all the inputs are on the same mesh. input_keys (list[str]|tuple[str]): if the iteration result of dataloader is a dict of tensors, input_keys is the keys of this dict, identify which tensor is located on which mesh, one-to-one correspondence with meshes. i.e. dict[input_keys[i]] is on meshes[i]. Default: None, which means the outputs is a list, and the i'th input is on meshes[i]. shard_dims (list(str)|tuple(str)|list(int)|tuple(int)|str|int]): The mesh dimension to shard the dataloader. Users can specify the shard_dim of each mesh or specify a single shard_dim for all meshes. Default: None, which means the data loader will not be split, i.e. mp. is_dataset_splitted (bool): Whether the dataset has been split, Default: False. dense_tensor_idx (list): A paired 2D list specifies the index of the dense_tensor in the output of dataloader. It allows users to identify which elements within each output batch are dense_tensor. first dense_tensor: the dense_tensor return by dataloader. second dense_tensor: num_or_sections specifies how to split first tensor: evenly (if a number) or unevenly (if a list). Default: None, meaning all outputs are dist_tensors. Note: For dense_tensor_idx settings, the idx must be paired. Returns: ShardDataloader: The sharded dataloader. Examples: .. code-block:: python :name: example-1 >>> import os >>> import numpy as np >>> import paddle >>> import paddle.distributed as dist >>> from paddle.io import BatchSampler, DataLoader, Dataset >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> mesh0 = dist.ProcessMesh([[0, 1], [2, 3]], dim_names=['x', 'y']) >>> mesh1 = dist.ProcessMesh([[4, 5], [6, 7]], dim_names=['x', 'y']) >>> paddle.seed(1024) >>> np.random.seed(1024) >>> class RandomDataset(Dataset): # type: ignore[type-arg] >>> def __init__(self, seq_len, hidden, num_samples=8): ... super().__init__() ... self.seq_len = seq_len ... self.hidden = hidden ... self.num_samples = num_samples ... self.inputs = [np.random.uniform(size=[self.seq_len, self.hidden]).astype("float32") for _ in range(num_samples)] ... self.labels = [np.array(index, dtype="float32") for index in range(num_samples)] ... def __getitem__(self, index): ... return self.inputs[index], self.labels[index] ... def __len__(self): ... return self.num_samples >>> class MlpModel(paddle.nn.Layer): ... def __init__(self): ... super(MlpModel, self).__init__() ... self.w0 = dist.shard_tensor( ... self.create_parameter(shape=[8, 8]), ... mesh0, [dist.Replicate(), dist.Shard(1)]) ... self.w1 = dist.shard_tensor( ... self.create_parameter(shape=[8, 8]), ... mesh1, [dist.Replicate(), dist.Shard(0)]) ... def forward(self, x): ... y = paddle.matmul(x, self.w0) ... y = dist.reshard(y, mesh1, [dist.Shard(0), dist.Shard(2)]) ... z = paddle.matmul(y, self.w1) ... return z >>> model = MlpModel() >>> dataset = RandomDataset(4, 8) >>> sampler = BatchSampler( ... dataset, ... batch_size=2, ... ) >>> dataloader = DataLoader( ... dataset, ... batch_sampler=sampler, ... ) >>> dist_dataloader = dist.shard_dataloader( ... dataloader=dataloader, ... meshes=[mesh0, mesh1], ... shard_dims="x" ... ) >>> opt = paddle.optimizer.AdamW(learning_rate=0.001, parameters=model.parameters()) >>> dist_opt = dist.shard_optimizer(opt) >>> def loss_fn(logits, label): ... # logits: [bs, seq_len, hidden], label: [bs] ... loss = paddle.nn.MSELoss(reduction="sum") ... logits = paddle.sum(logits, axis=[1, 2]) ... return loss(logits, label) >>> RUN_STATIC = eval(os.environ['RUN_STATIC']) >>> def run_dynamic(): ... for step, (input, label) in enumerate(dist_dataloader()): ... logits = model(input) ... loss = loss_fn(logits, label) ... print("step:{}, loss:{}".format(step, loss)) ... loss.backward() ... dist_opt.step() ... dist_opt.clear_grad() >>> def run_static(): ... dist_model = dist.to_static( ... model, dist_dataloader, loss_fn, opt ... ) ... dist_model.train() ... for step, (input, label) in enumerate(dist_dataloader()): ... print("label:", label) ... loss = dist_model(input, label) ... print("step:{}, loss:{}".format(step, loss)) >>> if RUN_STATIC == 0: ... run_dynamic() ... else: ... run_static() >>> # This case need to be executed in multi-card environment >>> # export CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7 >>> # RUN_STATIC=1 python -u -m paddle.distributed.launch --gpus "0,1,2,3,4,5,6,7" {test_case}.py >>> # RUN_STATIC=0 python -u -m paddle.distributed.launch --gpus "0,1,2,3,4,5,6,7" {test_case}.py .. code-block:: python :name: example-2 >>> import paddle >>> import paddle.distributed as dist >>> from paddle.io import BatchSampler, DataLoader, Dataset >>> import numpy as np >>> mesh0 = dist.ProcessMesh([[0, 1], [2, 3]], dim_names=['dp', 'mp']) >>> mesh1 = dist.ProcessMesh([[4, 5], [6, 7]], dim_names=['dp', 'mp']) >>> class RandomDataset(Dataset): # type: ignore[type-arg] ... def __init__(self, seq_len, hidden, num_samples=8): ... super().__init__() ... self.seq_len = seq_len ... self.hidden = hidden ... self.num_samples = num_samples ... self.inputs1 = [ ... np.random.uniform(size=[self.seq_len, self.hidden]).astype( ... "float32" ... ) ... for _ in range(num_samples) ... ] ... self.inputs2 = [ ... np.random.uniform(size=[self.seq_len, self.hidden]).astype( ... "float32" ... ) ... for _ in range(num_samples) ... ] ... self.labels = [ ... np.array(index, dtype="float32") for index in range(num_samples) ... ] ... def __getitem__(self, index): ... return { ... "inputs": [self.inputs1[index], self.inputs2[index]], ... "label": self.labels[index], ... } ... def __len__(self): ... return self.num_samples >>> dataset = RandomDataset(4, 8) >>> sampler = BatchSampler( ... dataset, ... batch_size=2, ... ) >>> dataloader = DataLoader( ... dataset, ... batch_sampler=sampler, ... ) >>> dist_dataloader = dist.shard_dataloader( ... dataloader=dataloader, ... meshes=[mesh0, mesh1], # or [[mesh0, mesh0], mesh1] ... shard_dims="dp", ... input_keys=["inputs", "label"], ... ) )r)rcrdrerfrgrbs rtshard_dataloaderrps$F   rscZ[RRRS5S$)N%FLAGS_enable_auto_parallel_align_mode)rr rrrlrsrtrr=s* ;; * */ - //rsc[5 g)a< Enables an automated Data Parallel (DP) setup for auto-parallel training. This function simplifies the process of implementing vanilla (standard) Data Parallelism within the auto-parallel framework. By calling ``enable_auto_dp()``, users can achieve data parallel training without needing to manually configure ``paddle.distributed.shard_dataloader`` (or a similar distributed dataloader interface) for DP-specific data sharding or distribution. This mode automates the setup required for DP communication and data handling. The function works by setting the related environment variable to ``1``. This signals to the auto-parallel system that it should automatically manage the data parallelism aspects of the training process according to a predefined strategy. A significant advantage of this automated DP mode is its inherent robustness and ability to handle scenarios that can be challenging for manual or other standard DP configurations. For instance, it is particularly effective for: - Training models where input data may have non-uniform shapes across different data parallel ranks (e.g., certain video generation models like Wanx). In such cases, where traditional DP might lead to program hangs due to shape mismatches during communication, this automated mode employs strategies (like adjusting data representation and gradient synchronization) to ensure smooth training. In essence, ``enable_auto_dp()`` provides two key benefits: 1. **Simplified DP Setup:** Automates the configuration for basic data parallelism, reducing manual setup effort (e.g., no need for manual ``shard_dataloader`` DP configuration). 2. **Robustness for Complex Cases:** Effectively handles advanced scenarios like non-uniform input shapes. Note: This function should typically be called at the very beginning of your training script, prior to initializing Paddle's distributed environment or any auto-parallel components. The underlying auto-parallel framework, including its data loading and optimizer components, must be designed to recognize and act upon the environment variable. Examples: .. code-block:: python >>> import numpy as np >>> import paddle >>> from paddle import nn >>> import paddle.distributed as dist >>> from paddle.io import Dataset, DataLoader >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> dist.enable_auto_dp() >>> BATCH_SIZE = 32 >>> CLASS_NUM = 10 >>> INPUT_DIM = 256 >>> STEPS = 100 >>> class RandomDataset(Dataset): # type: ignore[type-arg] ... def __init__(self, num_samples): ... rank = dist.get_rank() if dist.get_world_size() > 1 else 0 ... np.random.seed(42 + rank) ... self.num_samples = num_samples ... def __getitem__(self, idx): ... x = np.random.rand(INPUT_DIM).astype('float32') ... y = np.random.randint(0, CLASS_NUM, (1,)).astype('int64') ... return x, y ... def __len__(self): ... return self.num_samples >>> class SimpleNet(nn.Layer): ... def __init__(self): ... super().__init__() ... self.net = nn.Sequential( ... nn.Linear(INPUT_DIM, 102400), ... nn.Linear(102400, INPUT_DIM), ... nn.Linear(INPUT_DIM, CLASS_NUM), ... ) ... def forward(self, x): ... return self.net(x) >>> model = SimpleNet() >>> optimizer = paddle.optimizer.AdamW(learning_rate=1e-3, parameters=model.parameters()) >>> dataset = RandomDataset(num_samples=STEPS * BATCH_SIZE) >>> loader = DataLoader(dataset, batch_size=BATCH_SIZE, shuffle=False, drop_last=True) >>> model.train() >>> for step, (x, y) in enumerate(loader): ... y.stop_gradient = True ... loss = paddle.mean(model(x)) ... loss.backward() ... optimizer.step() ... model.clear_gradients() ... if step % 5 == 0: ... print(f"[step {step}] loss: {loss.item():.4f}") >>> # This case need to be executed in multi-card environment >>> # export CUDA_VISIBLE_DEVICES=0,1 >>> # python -m paddle.distributed.launch {test_case}.py N)r/rlrsrtenable_auto_dprCs Lrs)r~z paddle.Tensor)NNN)rz+Tensor | TensorLike | NestedNumericSequencerrrSequence[Placement]rzDTypeLike | NonerzPlaceLike | Nonerz bool | NonerfrF)r) r'zCallable[..., Tensor]rrrrr(rr)rrfrF)rrFrrrrrfrF) rZrQrrrkz0Callable[[str, Layer, ProcessMesh], None] | Nonera1Callable[[Any, ProcessMesh], list[Tensor]] | NonererrfrQrnr>)ryr-rkz.Callable[[str, Tensor, Tensor], Tensor] | Nonerr]rfrs)r+rKrfrKr)rZrQrz#ShardDataloader | DataLoader | Nonerr4ryr5rr6rr7rfrp)rrFrfrFr)rcrOrdz#ProcessMesh | Sequence[ProcessMesh]rezSequence[str] | Nonerfz0Sequence[str] | Sequence[int] | str | int | Nonergrrbrrfr) __future__rrr}rr collectionsrtypesrtypingrrrr numpyrrpaddle.distributedrVrr r r paddle.amp.auto_castr paddle.amp.grad_scalerrpaddle.autogradr paddle.baserpaddle.base.dygraph.baserpaddle.base.frameworkrrrrrrr paddle.distributed.auto_parallelrrr9*paddle.distributed.auto_parallel.interfacerr-paddle.distributed.auto_parallel.process_meshr2paddle.distributed.auto_parallel.static.completionr4paddle.distributed.auto_parallel.static.dist_contextr/paddle.distributed.auto_parallel.static.dist_opr -paddle.distributed.auto_parallel.static.utilsr!r"r#r$r%r&3paddle.distributed.fleet.utils.tensor_fusion_helperr'r(r)paddle.frameworkr*"paddle.io.dataloader.batch_samplerr+r,paddle.optimizerr- auto_dp_utilsr/r0r1rr2r3r4r5r6r7r8placement_typer9r:r;r<r=randomr>r?rcr@rArBcollections.abcrCrDtyping_extensionsrErFpaddle._typingrGrHrIrJ paddle.amprKrLrM7paddle.distributed.auto_parallel.static.dist_input_specrN paddle.iorO paddle.metricrP paddle.nnrQr:rRrSrTrUrVrWrXrYrZr_rqrarrrrrrrrrrr!r%r*r rmrqrsrrrrrrr,r. BaseConfigrrprrDrrrrrlrsrtrsV#  #99 !""11##;%NFP "' / 2+ &-,%$   9:E9: /)5 /& >$t"">$P#"!% OA 5OA OA$OA  OA  OA  OA OAdagaH+" 9;G TxGxxKO# L $  "2"2 "2$"2  "2  "2  "2Pww*w8Kw wzBFBFCG T T T ?T @ T A T  T n *c 5ic 5Ls/s/l ( =@'=@@?'?DY@'Y@|@D'(0M0M<0M"%0M 0MfL`AA*_}''_D V  V  v37.2"& $:> k k /k ,k k  k 8 kk\; |BBP (,CG %/3 JJ /J%JA J  J - JJZ/ frs