o p i#@sfdZddlZddlZddlZddlZddlZddlmZddlZ ddl m Z ddl m Z ddlmZmZmZddlmZmZddlmZdd lmZmZmZdd lmZd d gZd dgZdZdZ ej!"Z#ej!$Z%Z&ej!j'j(Z)ej!j'j*Z+ej!j'j,Z-ej!j'j.Z/da0GdddZ1ddZ2GdddZ3ddZ4ddZ5GdddZ6GdddZ7Gd d!d!Z8dS)"a Steps to transpile trainer: 1. split variable to multiple blocks, aligned by product(dim[1:]) (width). 2. rename split grad variables to add trainer_id suffix ".trainer_%d". 3. modify trainer program add split_op to each grad variable. 4. append send_op to send split variables to server and 5. add recv_op to fetch params(split blocks or origin param) from server. 6. append concat_op to merge split blocks to update local weights. Steps to transpile pserver: 1. create new program for parameter server. 2. create params and grad variables that assigned to current server instance. 3. create a sub-block in the server side program 4. append ops that should run on current server instance. 5. add listen_and_serv op N)reduce) framework) grad_var_name)BlockProgramcore) PSDispatcher RoundRobin)Constant) Parameterdefault_main_programdefault_startup_program) unique_name lookup_tablelookup_table_v2Zlookup_table_gradZlookup_table_v2_gradZ op_namescopez@CLIPFc@seZdZdZdZdZdZdS)DistributedModerN)__name__ __module__ __qualname__SYNCASYNC HALF_ASYNCZGEOrrz/home/app/PaddleOCR-VL/.venv_paddleocr/lib/python3.10/site-packages/paddle/distributed/transpiler/distribute_transpiler.pyrGs rcGstrt|dSdSN) PRINT_LOGprint)argsrrrlogNs r!c@seZdZddZddZdS)VarBlockcCs||_||_||_dSrvarnameoffsetsize)selfr$r%r&rrr__init__Ts zVarBlock.__init__cCs|jd|jd|jS)N:r#r'rrr__str__ZszVarBlock.__str__N)rrrr(r+rrrrr"Ss r"cCs||kp ||dS)N.block) startswith)p_namevar_namerrrsame_or_split_var^sr0cCsg}|D]z}|}tdd|jd}tt|t|}|dkr"d}||kr(|}tt|t|}t|jdkrUtdd|jddd} || } | dkrU|| | 7}tt|t|}t|D]} t ||| |} t |j | | } | t | qdq|S)a  We may need to split dense tensor to one or more blocks and put them equally onto parameter server. One block is a sub-tensor aligned by dim[0] of the tensor. We need to have a minimal block size so that the calculations in the parameter server side can gain better performance. By default minimum block size 8K elements (maybe 16bit or 32bit or 64bit). Args: var_list (list): List of variables. slice_count (int): Numel of count that variables will be sliced, which could be the pserver services' count. min_block_size (int): Minimum split block size. Returns: blocks (list[(varname, block_id, current_block_size)]): A list of VarBlocks. Each VarBlock specifies a shard of the var. cS||Srrxyrrrxz slice_variable..rrrcSr1rrr2rrrr5r6N)rshapeintmathfloorfloatceillenrangeminr"nameappendstr)Zvar_listZ slice_countmin_block_sizeblocksvarZ split_countZ var_numelZmax_pserver_count block_sizeZdim1Zremainsblock_idZcurr_block_sizeblockrrrslice_variablebs0   rIc@seZdZdZdZdZdZdZdZdZ dZ dZ dZ dZ dZdZdZdZdZd ZdZd d Zed d Zejdd ZeddZejddZdS)DistributeTranspilerConfiga_ :api_attr: Static Graph A configuration class that provide support for transpiler distributed jobs. Some important parameters are explained as follows: .. py:attribute:: slice_var_up (bool) Whether to do Tensor slice for parameter servers, default is True. .. py:attribute:: split_method (PSDispatcher) Methods of dispatching parameters for server, `RoundRobin` or `HashName` (both from `paddle.incubate.distributed.fleet.parameter_server.ir.ps_dispatcher`) can be used and default is RoundRobin. Try to choose the best method to balance loads for parameter servers. .. py:attribute:: min_block_size (int) Minimum number of split elements in block, default is 8192. According to : https://github.com/PaddlePaddle/Paddle/issues/8638#issuecomment-369912156 We can use bandwidth efficiently when data size is larger than 2MB.If you want to change it, please be sure you have read the slice_variable function. You can find the definition of slice_variable in https://github.com/PaddlePaddle/Paddle/blob/develop/python/paddle/transpiler/distribute_transpiler.py . Examples: .. code-block:: python >>> from paddle.distributed.transpiler.distribute_transpiler import RoundRobin >>> import paddle.distributed.transpiler as transpiler >>> config = transpiler.DistributeTranspilerConfig() >>> config.slice_var_up = True >>> config.split_method = RoundRobin >>> config.min_block_size = 81920 TN FZpserverdrrcCsdSrrr*rrrr(sz#DistributeTranspilerConfig.__init__cC|jSr)4_DistributeTranspilerConfig__runtime_split_send_recvr*rrrruntime_split_send_recvz2DistributeTranspilerConfig.runtime_split_send_recvcC,|durtd|r|jrtd||_dS)Nz%runtime_split_send_recv can't be Nonezeif you want to set runtime_split_send_recv to be true, make ensure config.sync_mode is false at first) ValueError&_DistributeTranspilerConfig__sync_moderNr'valuerrrrO  cCrMr)rSr*rrr sync_moderPz$DistributeTranspilerConfig.sync_modecCrQ)Nzsync_mode can't be Nonezeif you want to set sync_mode to be true, make ensure config.runtime_split_send_recv is false at first)rRrNrSrTrrrrWrV)rrr__doc__ slice_var_up split_methodrCenable_dc_asgdmode print_log wait_portrNrS half_asynccompletely_not_asyncZ geo_sgd_modeZgeo_sgd_need_push_nums nccl_comm_numuse_hierarchical_allreduce#hierarchical_allreduce_inter_nrankscollective_moder(propertyrOsetterrWrrrrrJs8*   rJc@seZdZddZdS)ServerRuntimeConfigcCs:ttdd|_ttdd|_ttdd|_dS)NZFLAGS_rpc_send_thread_numZ12ZFLAGS_rpc_get_thread_numZFLAGS_rpc_prefetch_thread_num)r8osgetenv_rpc_send_thread_num_rpc_get_thread_num_rpc_prefetch_thread_numr*rrrr(s    zServerRuntimeConfig.__init__N)rrrr(rrrrrgs rgc@seZdZdZdiddZdiddZ  djdd Z   dkd d Zd d ZddZ ddZ      dlddZ ddZ ddZ ddZdmddZddZd d!Zd"d#Z dnd$d%Zd&d'Zd(d)Zd*d+Zd,d-Zd.d/Zd0d1Zd2d3Zd4d5Zd6d7Z 8dod9d:Zdmd;d<Zed=d>Z d?d@Z!dAdBZ"dCdDZ#dEdFZ$dGdHZ%dIdJZ&dKdLZ'dMdNZ(dOdPZ)dQdRZ*dSdTZ+dUdVZ,dWdXZ-dYdZZ.d[d\Z/d]d^Z0d_d`Z1dadbZ2dcddZ3dedfZ4dgdhZ5dS)pDistributeTranspilera :api_attr: Static Graph **DistributeTranspiler** Convert the base program to distributed data-parallelism programs. Supports two modes: parameter server(pserver) mode and nccl2 mode. In pserver mode, the main_program will be transformed to use a remote parameter server to do parameter optimization. And the optimization graph will be put into a parameter server program. In nccl2 mode, the transpiler will append a NCCL_ID broadcasting op in startup_program to share the NCCL_ID across the job nodes. After transpile_nccl2 called, you ***must*** pass trainer_id and num_trainers argument to ParallelExecutor to enable NCCL2 distributed mode. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle >>> import paddle.base as base >>> import paddle.distributed.transpiler as transpiler >>> paddle.enable_static() >>> x = paddle.static.data(name='x', shape=[1,13], dtype='float32') >>> y = paddle.static.data(name='y', shape=[1], dtype='float32') >>> y_predict = paddle.static.nn.fc(x, size=1, activation=None) >>> cost = paddle.nn.functional.square_error_cost(input=y_predict, label=y) >>> avg_loss = paddle.mean(cost) >>> sgd_optimizer = paddle.optimizer.SGD(learning_rate=0.001) >>> sgd_optimizer.minimize(avg_loss) >>> # for pserver mode >>> pserver_endpoints = "192.168.0.1:6174,192.168.0.2:6174" >>> trainer_endpoints = "192.168.0.1:6174,192.168.0.2:6174" >>> current_endpoint = "192.168.0.1:6174" >>> trainer_id = 0 >>> trainers = 4 >>> role = "PSERVER" >>> t = transpiler.DistributeTranspiler() >>> t.transpile( ... trainer_id, pservers=pserver_endpoints, trainers=trainers) >>> if role == "PSERVER": ... pserver_program = t.get_pserver_program(current_endpoint) ... pserver_startup_program = t.get_startup_program(current_endpoint, ... pserver_program) ... elif role == "TRAINER": ... trainer_program = t.get_trainer_program() >>> # for nccl2 mode >>> trainer_num = 2 >>> trainer_id = 0 >>> config = transpiler.DistributeTranspilerConfig() >>> config.mode = "nccl2" >>> trainer_endpoints = "192.168.0.1:6174,192.168.0.2:6174" >>> t = transpiler.DistributeTranspiler(config=config) >>> t.transpile(trainer_id=trainer_id, trainers=trainer_endpoints, current_endpoint="192.168.0.1:6174") >>> exe = paddle.static.ParallelExecutor( ... use_cuda=True, ... loss_name=avg_loss.name, ... num_trainers=trainer_num, ... trainer_id=trainer_id ... ) NcCs|dur||_nt|_||jjdurt|j_|jjs"|jjr'tj|_ n |jj r0tj |_ ntj |_ |jj r:da|jjdksBJ|jjjdtksMJd|_dS)NTrKr)configrJ_set_server_configrZr rWr`rrdistributed_moderOrrr]rrC __bases__r counter_var)r'rnrrrr(Ss     zDistributeTranspiler.__init__cCs0|dur t|_dSt|tr||_dStd)NzQIn DistributeTranspiler, server_config must be an instance of ServerRuntimeConfig)rg server_config isinstance TypeError)r'rsrrrroks   z'DistributeTranspiler._set_server_configTc Cs4ddlm}|s t}|dkr|d}|||dkr#|r#|||jddtjj j d}t d|j j D]} |jd| dtjj j dq7|j jrxt d|j j D]"} |jd | dtjj j d|jd | dtjj j dqU|jd id|i|d||j j |j j|j jd d |Std)Nrwait_server_ready,ZNCCLIDTr@ persistabletyperZNCCLID_ZHierarchical_inter_NCCLID_ZHierarchical_exter_NCCLID_Z gen_nccl_id)trainers trainer_idrarbrcr{inputsoutputsattrszmust set trainer_id > 0)5paddle.distributed.fleet.base.private_helper_functionrwr splitremove global_block create_varrVarDescVarTypeRAWr>rnrarb append_oprcrR) r'r}r|current_endpointstartup_programr^rwZworker_endpointsZ nccl_id_varirrr_transpile_nccl2usT      z%DistributeTranspiler._transpile_nccl2c Csddlm}t|tr|d} nt|tr|} n |dkr%tdt|t| dkr3|dkr3td|dur:t}|durAt }d} |dkrO| |j j } n|d kr[| |j j } n|dkrd|} ntd || j|||| ||d dS) Nr) collectiverxZsingle_process_multi_threadzinvalid trainers config: rz(invalid trainer number in distributed: 1Zgrad_allreduceZ local_sgdzinvalid collective_mode: )r main_programZrank endpointsrr^)Zpaddle.distributed.transpilerrrtrBrlistrRr=r r Z GradAllReducernraZLocalSGDZSingleProcessMultiThread transpile) r'rdr}r|rrrr^rrZ transpilerrrr_transpile_collectives<      z*DistributeTranspiler._transpile_collectivecCsBg}gd}|jD]}|j|vr|ddur||q |S)N)rncerZremote_prefetchT)ropsr{attrrA)r'rsparse_update_opssparse_update_op_typesoprrr _get_all_remote_sparse_update_ops  z5DistributeTranspiler._get_all_remote_sparse_update_opc s|D];\}}|j|}|d}|d}g}d} g} t|jD]/\} } || jvr<| dkr<| j} || | | q!|| jvrP| | jkrP|| | | q!| tvr@j fdd|D} fdd|D}j |d dd}|d d}fd d|D}| ddd D] }  | qd gt|}d gt|}tj D]N\} } tdt| jD]}| | j|}t|D] \}}|j|vr| ||<qqtdt| jD]}| | j|}t|D] \}}|j|vr| ||<qqqt|t|dkr+t|d}j|d ||d d |i|||||j| ddntd| ddd D] } |j| q6qdS)Nrrcsg|]}|qSr)index.0r)all_opsrr szHDistributeTranspiler._update_remote_sparse_update_op..c$g|]}j|ddqS)Idsr)rvarsinputrprogramrrr W padding_idxcr)Outr)rroutputrrrrrrdistributed_lookup_tablerrZOutputs) table_namesheight_sectionsrrr}Zlookup_table_versionrr{rrrzIsomething wrong with distribute_transpiler, submit a issue is recommended)itemssparse_param_to_height_sections enumeraterinput_arg_namesr{rALOOKUP_TABLE_TYPErrrrr _remove_opr=r> output_namesrr@ input_namesr?max _insert_opr}rRpop)r'rneed_sparse_update_params param_varnamerrrrrop_typeZused_opsidxrZop_idxsrwrrZ inputs_idxsZ outputs_idxsrZoutsin_idin_varZinsZout_idout_varZdistributed_idxr)rrr_update_remote_sparse_update_ops           z4DistributeTranspiler._update_remote_sparse_update_opcCs |jD] }||jvr dSqdSNTF)rr)r' param_namerrrr$_is_input_of_remote_sparse_update_op@s  z9DistributeTranspiler._is_input_of_remote_sparse_update_op127.0.0.1:6174rc6s ddlm}ddlm} m} d} t| tjd|durt}|dur%t }|_ |_ j _ jjdkrt|ts>J|dj _jjj _jjj _jjrtj j} jjd krgtj_| jjksyJd | d jj| jjdksJd | d jjd tjjj _j||||jjddSjjdkrj jj!|||||jjddS|_"|_#|_$|d} | _%| _&'\_(_)j*j%}|j _+j+du_,i_-i_.j)D]\}}|j/j-|j/<|j/j.|j/<q0j _1i_2g_3dj _4j%j _5|j _6j$dkj _7j+r0j+ndj _89|:g}t;j<=}jj>sXt?j@Aj jBt?j@C|i_D|D]\}}|E|}jj>stt|d kstJ|}t|d kr|dj/}| |F|dd}n.t|d kr|FjG|}| |F|dd}jjHsI|||||d 7}ntJd||djKtjLjMjNkrڈj.|}O|rdd|Dj2|<|FjPtQRd}|jD|<jjHr|FjG|g}S|}jjTrj"d krfdd|D}ndd|D}n|}g}g}|FjU|d dd|id|id|d|d|tVtWtXj.||gidtY|D] \}}|Z|qCq]|FjPtQRd} j,rl|FjPtQRdjDj+<t;jD[}!j#sȈ\}"t|"dkrȈj]r|FjPtQRd}#jjHrj]j/g}ng}g}|Fj^ddj]id|#id| d|d|d dd!d"tVtWtXj]j/j]j/gid#|!Z|#j#rg}$|Fj^d$dt;|!id| id%| d&j$d'd"tVtWid#|$Z| n$jjHrjj_r|Fj^d$dt;|!id| id%| d&j$d'dtVtWid#g}%tY|D] \}}|%Zj`|q|:|E|%}tY|D])\}&}'ja|'d(Z|%|&ja|'d)Z||&j&b|%|&j/}(|'|(_cq3i})g}*jd=D]\}+},g}-g}.|,D]}d*d|%De|j/}|-Z|||.Z|j/qpj#r| }/njDj-|+}/j-|+}0|0}1j<|0}2t|2d kr|2dj/}1|+j2vr|.D] }3j&b|3}(d+|(_fq|-|.f|)|+<qfg}4jjHr|FjG|+}5d,d|,D}4|5g},|*g|,|Fj^d-d|/gid|,id|-d.|4d&j$tVtWtX|+|1gid#qfh||)j#r,|Fj^d/d|$id|*id%| d&j$tVtWid#jd=D]4\}+},t|,d kr>q1|FjG|+}5|+j2vrdjjHsd|Fj^d0d|,id|5gid1dtVtiid#q1jj|%|d2j,r}k|| l|| mj&j _ndS)3aW Transpile the input program to distributed programs with config and arguments. Args: trainer_id (int): id for current trainer worker, if you have n workers, the id may range from 0 ~ n-1 program (Program|None): program to transpile, default is paddle.static.default_main_program(). startup_program (Program|None): startup_program to transpile, default is paddle.static.default_startup_program(). pservers (str): comma separated ip:port string for the pserver list. trainers (int|str): in pserver mode this is the number of trainers, in nccl2 mode this is a string of trainer endpoints. sync_mode (bool): Do sync training or not, default is True. startup_program (Program|None): startup_program to transpile, default is paddle.static.default_main_program(). current_endpoint (str): need pass current endpoint when transpile as nccl2 distributed mode. In pserver mode this argument is not used. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> t = paddle.distributed.transpiler.DistributeTranspiler() >>> t.transpile( ... trainer_id=0, ... pservers="127.0.0.1:7000,127.0.0.1:7001", ... trainers=2, ... sync_mode=False, ... current_endpoint="127.0.0.1:7000") r)find_distributed_lookup_table)VarsDistributedfind_op_by_output_argz API is deprecated since 2.0.0 Please use FleetAPI instead. WIKI: https://github.com/PaddlePaddle/Fleet/blob/develop/markdown_doc/transpiler )fileNnccl2rxrz trainers_num:z' < hierarchical_allreduce_inter_nranks:z) mod hierarchical_allreduce_inter_nranks:z != 0)rr^r)rdr}r|rrrr^T)reversez6Can not insert the send op by original variable name :cSsg|]}|jdqS)r)r7)r splited_varrrrrs z2DistributeTranspiler.transpile..r@csg|] }|jdjqS) .trainer_)r@r}rrEr*rrr+scSg|]}|jqSrrrrrrr0sendXrepmapsections send_varnamesrZ merge_addZuse_send_handlerFr~Z send_barrierrr}r_paramsgradscSrrrrvrrrrrZRemotePrefetchcSrrrrrrrrrrecv recv_varnames fetch_barrierconcataxis) recv_varseplist)oZ*paddle.distributed.distribute_lookup_tabler%paddle.distributed.transpiler.detailsrrrsysstderrr r origin_programrcloneZorigin_startup_programrnr\rtrBrZ_trainers_endpointsraZ_nccl_comm_numrbZ_use_hierarchical_allreducer=rcrZget_cuda_device_countr8Z$_hierarchical_allreduce_inter_nranksrr^rrd trainer_numrWr}pserver_endpoints vars_overview_get_optimize_pass optimize_ops params_gradsrZ table_namehas_distributed_lookup_tableparam_name_to_grad_namegrad_name_to_param_namer@rrrneed_delete_optimize_varsZ_is_distributedZ _endpointsZ _ps_endpointZ _is_chief_distributed_lookup_table_init_splited_varsresetrgrad_var_mappingrrYnprandomseed random_seedshufflegrad_name_to_send_dummy_outdispatchrrrO_insert_split_opAssertionErrorr{rr SELECTED_ROWSrrrgenerate_control_dev_var_name_get_splited_var_sectionsr`rRPC_OP_ROLE_ATTR_NAMERPC_OP_ROLE_ATTR_VALUEOP_ROLE_VAR_ATTR_NAMErrAvalues _get_lr_opsrrrr_grad_param_mappingparam_grad_ep_mappingZget_distributed_var_by_sliceendpointparam_var_mappingrvtypeextendrDIST_OP_ROLE_ATTR_VALUE_get_trainer_startup_program&_replace_lookup_table_op_with_prefetch#_split_table_grad_and_add_send_vars_get_distributed_optimizer_varsZ_parameters_on_pservers)6r'r}rZpserversr|rWrrrrrerr_msgZ trainers_numrZ ps_dispatcher param_vargrad_varZ send_varsZgrad_var_mapping_itemsZ grad_varname splited_varsrZsplited_grad_varnamerorig_varsparse_param_nameZ dummy_outputZsend_input_varsrr_rEZsend_barrier_outZ input_depslr_opsZdecay_dummy_outputZfetch_barrier_inputrrepZdistributed_varrZall_recv_outputsrrepsrZ recv_dep_inZorig_grad_nameZrecv_op_role_var_nameZsplited_trainer_gradrr orig_paramrr*rrFsh -                                                   zDistributeTranspiler.transpilecCsddg}g}|jjD]'}|j|vr$|ddur$||dd|jdkr3||ddq |jr=||jt t |S)Nrr is_sparseTrrr) rrrr{rrArrrrset)r'rsparse_table_namesrrrr_get_sparse_table_names s    z,DistributeTranspiler._get_sparse_table_namesc Csddlm}|D]N}|jj|}g}|jjD] }||jvr&||qt|}|dkr7t dt ||d}|jj did|id| did||j|qdS) Nr delete_opsrz&table init op num should be 1, now is Z fake_initrr7r~) rr'rrrroutput_arg_namesrAr=rRrBrr) r'r$r'r table_varZtable_param_init_oprZ init_op_numZ table_init_oprrr_fake_init_sparsetables,      z+DistributeTranspiler._fake_init_sparsetablec CsDddlm}g}g}g}|jD]}||j||dqtt|}tt|}|D] }||vr9||q.tt|}|rg} |D]}g} |j j D] }||j vr\| |qP| | qF||j | |D]}|j |r|j |qmdS||j |j|D]}|j |r|j |qdS)Nrr&Z op_role_var)rr'rrrrrr#rArrrr(has_varZ _remove_varr) r' is_startupr'Z optimize_varsZoptimize_op_role_varsZoptimize_need_delete_varsrrErZinit_opsZ param_init_oprrr_delete_trainer_optimizer7sF          z.DistributeTranspiler._delete_trainer_optimizercCsddlm}ddlm}|jdd|}|||}||j ||jdd|j |j |r>||j |jS)a Get transpiled trainer side program. The program on trainer side compared with origin program has following difference: - Delete optimizer related op, because parameter updated on Pserver - After the op which computed gradient of each parameter, add ``Send_op`` and ``Recv_op`` Args: wait_port(bool): Whether to wait for the parameter server to be ready before returning to program, default is True Returns: Program: trainer side program. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle.distributed.transpiler as transpiler >>> # this is an example, find available endpoints in your case >>> pserver_endpoints = "192.168.0.1:6174,192.168.0.2:6174" >>> trainer_id = 0 >>> trainers = 4 >>> t = transpiler.DistributeTranspiler() >>> t.transpile(trainer_id, trainers=trainers, pservers=pserver_endpoints) >>> trainer_program = t.get_trainer_program() rrvr&T)r,F) rrwrr'r-r%r*r rrr+rr)r'r^rwr'r$rrrrget_trainer_program]s       z(DistributeTranspiler.get_trainer_programc Cs|j}|}|jD]X\}}||vrq g}|D]}dd|D|j} ||| q|D]}||jr;q0|j |jd|j |j |j |j dq0|jddgid|id|d |jttid } q |j td } |jd id| id |jd |jttid |jD]I\}}||vrqt|dkrq||jvr|j|} n|jj|} |j || j| j | j | j d} |jdd|id| giddid q|S)a Get transpiled trainer side startup program. Args: recv_vars (list): Variable list to recv for current trainer_id eplist (list): A list of strings indicating Returns: Program: trainer side startup program. cSrrrrrrrrrzEDistributeTranspiler._get_trainer_startup_program..F)r@rzr{dtyper7 lod_levelrrrrr}r~rrrrr@rzr{r/r7rrr)rr%rrrr@rArr+rr{r/r7r0rr}rrrrrr=rrrz)r'rrrr$r$rr rErrZfetch_barrier_outr!origin_param_varrrrrs      z1DistributeTranspiler._get_trainer_startup_programc* stjdt}jj|_|jg}jdD] }| |qjdD]Y}|j d}|dkrC|j d|}n|j }| j |d|j |j|jd}jsajjrjd krtjD]}| j |d|d |j |j|jd} || qfq/||q/j} g} tjD]\} } | r| r| | qjjdurjd usJg_jdD])}tjD]!}|j d|d }| j ||j |j|jd }j||fqqg}gfd d}fdd}g}d}t|dkr4||j d }||t|D]\} } !|| }|||q|j"}g}|j d }t| D]q\}}||}|||#t$d}d}tjD]%\} } | #t$d } | #t$d|kr%|| |j}|rnq[|rtjD]%\} } | #t$d|kr| |vrt&d| j | j'||| ||||qq?t(t)|}|r||j d }!||!|D] }"||"|!|d|qg}#j*rj+,}$-|$|||}%||%.|$||%}&/||%j"}'j0|_1|#2|&t|dkr&t34dt5d|j d }||}(||(|j+,jj6||j7j8j7j9j7j:d })j*rH|'|)d<jjrQd|)d<t|#dkr\|#|)d<| j;dd|ii|)d|<|_=|S)a Get parameter server side program.The program on pserver side compared with origin program has following difference: - Only the following op is included: optimize-related op and communication-related op - NO.0 block only has variable definitions and ``listen_and_serv_op`` - Every variable which need to be updated has a unique block Args: endpoint (str): current parameter server endpoint. Returns: Program: the program for current parameter server to run. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle.distributed.transpiler as transpiler >>> # this is an example, find available endpoints in your case >>> pserver_endpoints = "192.168.0.1:6174,192.168.0.2:6174" >>> current_endpoint = "192.168.0.1:6174" >>> trainer_id = 0 >>> trainers = 4 >>> t = transpiler.DistributeTranspiler() >>> t.transpile( ... trainer_id, pservers=pserver_endpoints, trainers=trainers) >>> pserver_program = t.get_pserver_program(current_endpoint) zsget_pserver_program() is deprecated, call get_pserver_programs() to get pserver main and startup in a single call. rrrrNTr1rFZ_bakr@r{r7r/c sB|r|||j|dS||vr||dSdSr)_is_optimizer_op_append_pserver_opsr_append_pserver_non_opt_ops)rrHgrad_to_block_id merged_varr)rr'sparse_grad_to_paramrr__append_optimize_op__rs  zHDistributeTranspiler.get_pserver_program..__append_optimize_op__c s|dsdS|d}j|j}t|tsJ||j}|j D]}| |q#|j D]} |||}|||q.| d|dS)NZ sub_block)Zhas_attrrrrHidrtr _create_blockrr_clone_variabler _clone_lr_op _set_attr) rrZlr_blockZorigin_block_descZ origin_blockZ new_sub_blockrEZ origin_op cloned_op)__clone_lr_op_sub_block__r'rrrAs      zKDistributeTranspiler.get_pserver_program..__clone_lr_op_sub_block__rzappend opt op: z pserver [z] has no optimize block!!) optimize_blocksrZ pserver_idZFaninrpr7r9lr_decay_block_idZrpc_get_thread_numZrpc_send_thread_numZrpc_prefetch_thread_numZcheckpint_block_idZdc_asgdprefetch_var_name_to_block_idZlisten_and_servrr~)>rrwriterrrZ_copy_dist_param_info_fromr  _clone_varrr@findrr{r/r7rWrnr`rr>rA _create_ufindrrr4_is_opt_op_on_pserverr[param_bak_listr r=r<Z num_blocksr6rrr _append_pserver_grad_merge_opsr!rrr#rrr_create_table_optimize_block_create_prefetch_block_create_checkpoint_save_blockrrrloggingwarningrBrprsrkrjrlr_sync_with_cpppserver_program)*r'rrRZ recv_inputsrZsuff_idx orig_var_nameZsingle_trainer_varr}rEufindopt_op_on_pserverrrprZparam_bak_nametmpvarZ global_opsr:rrBrCZlr_decay_blockr@r7 pre_block_idxropt_opZ per_opt_blockZoptimize_target_param_namer8grad_varname_for_blockZopt_state_blockZglb_oprD pserver_indextable_opt_blockZ!lookup_table_var_name_to_block_idZcheckpoint_block_idZ empty_blockrr)rArr'r9rget_pserver_programsj%                           z(DistributeTranspiler.get_pserver_programcCs ||}|j||d}||fS)aq Get pserver side main program and startup program for distributed training. The ``main_program`` returned by this function is consistent with the return value of the function ``get_pserver_program`` . Args: endpoint (str): current pserver endpoint. Returns: tuple: (main_program, startup_program), of type "Program" Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle.distributed.transpiler as transpiler >>> # this is an example, find available endpoints in your case >>> pserver_endpoints = "192.168.0.1:6174,192.168.0.2:6174" >>> current_endpoint = "192.168.0.1:6174" >>> trainer_id = 0 >>> trainers = 4 >>> t = transpiler.DistributeTranspiler() >>> t.transpile( ... trainer_id, pservers=pserver_endpoints, trainers=trainers) >>> pserver_program, pserver_startup_program = t.get_pserver_programs(current_endpoint) )rR)r]get_startup_program)r'rZ pserver_progZpserver_startuprrrget_pserver_programs.s z)DistributeTranspiler.get_pserver_programscst}|j}|j|_|j|dfdd}|j}t}|D]\} } | | } | || j <q$|j D]e} t} d}| j dvry| j D].}|| |d\}} |rbd}||| |<qJ| |d|vrxd}|| |d| |<qJ|r||| }| j dvr| d t| d j|j| j || | d q:|jjr|jD]"\}}|j|j }|j|j }|jd d |id |idq|S)aF **Deprecated** Get startup program for current parameter server. Modify operator input variables if there are variables that were split to several blocks. Args: endpoint (str): current pserver endpoint. pserver_program (Program): deprecated, call get_pserver_program first. startup_program (Program): deprecated, should pass startup_program when initializing Returns: Program: parameter server side startup program. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> pserver_endpoints = "192.168.0.1:6174,192.168.0.2:6174" >>> trainer_endpoints = "192.168.0.1:6174,192.168.0.2:6174" >>> current_endpoint = "192.168.0.1:6174" >>> trainer_id = 0 >>> trainers = 4 >>> t = paddle.distributed.transpiler.DistributeTranspiler() >>> t.transpile(trainer_id, pservers=pserver_endpoints, trainers=trainers) >>> pserver_program = t.get_pserver_program(current_endpoint) >>> pserver_startup_program = t.get_startup_program(current_endpoint, ... pserver_program) rcs@tD]\}}|j}t||r||kr||jfSqdgfS)Nr)rr@r0r7)r$rZ splited_paramZpnamerrr_get_splited_name_and_shapezs zMDistributeTranspiler.get_startup_program.._get_splited_name_and_shapeF)rrrrT)Zgaussian_random fill_constantZuniform_randomZtruncated_gaussian_randomr7rr~Zassignrr{rr)rrrr rr collections OrderedDictrr=r@rr{rr_get_input_map_from_opr?rr7r all_attrsrnr[rJ)r'rrRrZs_progZ orig_s_prograZ pserver_varsZcreated_var_maprrErWrZ new_outputsZ op_on_pserverkeyZnewname new_inputsrVp_bakZstartup_param_varZstartup_tmpvarrr`rr^QsX$       z(DistributeTranspiler.get_startup_programc Csd}d}d}d}||j\}}}|s|||fSt||d}d} |j|} d} t| djdkrAtdd| djdd} | d|D] }| |jd7} qG| | }d}|||fS) NrHrFrrcSr1rrr2rrrr5r6z:DistributeTranspiler._get_slice_var_info..T)_get_varname_partsr@r8rrr=r7r) r' slice_varZ block_suffixZ block_idxr%is_slicerS block_namerZ skip_dim0Z slice_varsorig_dim1_flattenrrr_get_slice_var_infos(   z(DistributeTranspiler._get_slice_var_infocs$fdd}jD]}||q dS)Nc s8ddlm}g}tjD]\}}|r!||r!||q |D]u}d}|jD]}|dkrA||d}j ||}nq+|jD]S}|dvrLqEj j ||d} |j|| j|jj} | |jjkr|| j| | j| j| j| jd} j j| | |j|j|jd|dqEj j| | dddd|dqEq$dS) Nr) VarStructParamrrGrad LearningRateZ Beta1TensorZ Beta2Tensorr@r7r/r{r0rzZ Optimizer) origin_varrlrmrGr%rrF)rrqrrr4rIrArrrZ$get_distributed_var_by_origin_and_eprrr_get_optimizer_input_shaper{r7slicer@r/r0rzadd_distributed_varrmrGr%) rrqrUrrrYZdist_varrhrrw new_shaperr*rr_get_distributed_optimizer_varsp         z\DistributeTranspiler._get_distributed_optimizer_vars.._get_distributed_optimizer_var)r)r'r|rrr*rrs  B z4DistributeTranspiler._get_distributed_optimizer_varscsjjrQfdd|D}fdd|D}tfdd|D_jdjrAfddttjD_||fSfddttjD_||fS) Ncsg|] }|jjkr|qSrr@r)rparamr*rrr!szGDistributeTranspiler._update_dist_lookup_table_vars..cs g|] }|jtjkr|qSr)r@rr)rZgradr*rrr$s c3s$|] }|djjkr|VqdS)rNr})rZ param_gradr*rr )szFDistributeTranspiler._update_dist_lookup_table_vars..rcs<g|]}jjdjd|jjjdqSr .pserver_r3)rrr@r}r{r7r/rrrr'table_grad_varrrr0scs4g|]}jjd|jjjdqS)rr3rrr@r{r7r/r)rrrrr:s) rrnexttable_param_gradrWr>r=rtrainer_side_table_grad_list)r' param_list grad_listrrrr_update_dist_lookup_table_varss*       z3DistributeTranspiler._update_dist_lookup_table_varsc sg}g}t}jD]0\}}t|tkr|jdurq |j|vr*||||j|j|vr:||||jq ||j\}}j j r`t |t j j j}t |t j j j}nt |dj j}t |dj j}t |t |kszJj|_jD]%\}} j|} | D]} | \} } }jj| | | || ddqqjj|jdkd_t_t||D]'\}}|d\}}}|d\}}}j|t|jj|t|<qt_ fddj DdS) NFrrr)rwrlrGr%rmr)add_trainer_suffixr)cs"g|] }j|ggdiqS))rr)r update)rrr*rrrsz;DistributeTranspiler._init_splited_vars..)!r#rr{r Z trainabler@rAaddrrnrYrIr=rrC_create_vars_from_blocklistrrrrrErprrzrrrdrer ziprr8r )r'rrZparam_grad_setrVgZ grad_blocksZ param_blocksZ orig_namerrrrmrGr%Zg_nameZg_bidrr.Zp_bidrr*rrEs             z'DistributeTranspiler._init_splited_varscCsddlm}g|_g|_g|_g|_d}d}|rd}|j}|D][}|jt kr||j | ddkr|| ds;t dd}|dkrC|nt||}| d }|d } |j|d} |j| |j| d} |j| |||gnq!|stt|jD]H} |jtd t| |jdj|jdj|jdjd } |j| |jtd t| |jdj|jdj|jdjd } |j| q|j|dd |jid |jid|j|ddd|jid |jid|id|j|dd|j|j|jdd |jiddS)Nrr&rTFris_distributedz[lookup_table_op that lookup an distributed embedding tableshould set is_distributed to truerrZprefetch_compress_in_tmp_r3Zprefetch_compress_out_tmp_ split_ids)rr{rrrZprefetchrrrrZ merge_ids)rZRowsr)rr'Zall_in_ids_varsall_prefetch_input_varsall_prefetch_output_varsZall_out_emb_varsrrr{rrrr RuntimeErrorrrrrrAr>r=rrrBr7r/r)r'rrr'Zlookup_table_op_indexZcontinue_search_lookup_table_oprrZids_nameZout_nameZids_varrrrrrrrs         "        z;DistributeTranspiler._replace_lookup_table_op_with_prefetchcCs|j}t|j}|D]W}||jvrct||}|j|ddd|j|gid|j it t id|j|ddd|j id|j rL|j |jgngid |d |jt tt|j||giddSq dS) Nrrrrrrrrrr})rrrrr(rrrrrrrrWrr}rr r)r'rrrZtable_grad_namerZop_indexrrrrsD    z8DistributeTranspiler._split_table_grad_and_add_send_varsc Cs|j|j}g}||j}|j|}|j|j|j|j |j d}|j |} |j| j| j| j | j d} |j d||dd| iddddd| |jd t|j|S) Nr3Zlookup_sparse_tablerrTr)r"rrr~r))rrrr<rrrr@r{r7r/rrrArB) r'r[rRoptimize_blockr)rDZprefetch_blockZ trainer_idsZ pserver_idsZ trainer_outZ pserver_outrrrrM/s<    z+DistributeTranspiler._create_prefetch_blockcs|}tfddjD}jjj}tt |j dt t j }t|j } || d<j|j| |jtjjjdd} | jtjjjjjtj} jj|dd} jrjdfdd tjD} |jd d | id | giid n!| j}j j}|!|st"d|d| j#||} | g| g| gd}d| gi}t$%d|j&|jd||d|'| jdt(|j)|S)Nc3s2|]}d|jvr|ddjkr|VqdS)rrrN)rrrrr*rrrZs zDDistributeTranspiler._create_table_optimize_block..rT)r@r7r/r{rzrurcs:g|]}jjd|djjjdqSrrr)r[rRrrrrszEDistributeTranspiler._create_table_optimize_block..sumrrr~zorigin_grad_var: z grad_var:)rrrtruParamOutz\distribute lookup table only support sgd optimizer, change it's optimizer to sgd instead of sgdrcr))*r<rrrrrrr8r9r<r7r;r=rrrr@r/rrrrdescset_typer=rrrWrr>rrrr-rR _rename_varrOrPr{rArBr)r'r[rRrXr7r\Z table_opt_opr2Zzero_dimZ table_shaperrZlr_varZpserver_side_table_grad_listorigin_grad_nameZsplited_grad_namerrr)r[rRr'rrrLSs            z1DistributeTranspiler._create_table_optimize_blockcCsH|jddtjjjd||}|jdd|jgiiddid|j S) z? create a new block to handle save checkpoint. ZkLookupTablePathTrysaver file_pathnoner~) rrrrrrr<rrr)r'rRrXZcheckpoint_save_blockrrrrNs  z2DistributeTranspiler._create_checkpoint_save_blockFc Cst}t}|D]}|d\}}} ||vrg||<||t|t| fq |D]\}} ||} t| dkrl|j r`|r`| j d|j } | || || g||<n || j g||<q.g||<| j } d}t| dkrtdd| ddd}t| D]M\}}|d} | |}|g}t| dkr|| ddd} |j r|r|d |d|j } n|d |} |j| d | j| j|d }|||q|q.|S) af Create vars for each split. NOTE: only grads need to be named for different trainers, use add_trainer_suffix to rename the grad vars. Args: program (ProgramDesc): ProgramDesc which gradients belong. block_list (list[(varname, block_id, block_size)]): List of gradient blocks. add_trainer_suffix (Bool): Add trainer suffix to new variable's name if set True. Returns: var_mapping (collections.OrderedDict(varname->[new_varname_variable])):A dict mapping from original var name to each var split. r)rrrcSr1rrr2rrrr5r6zBDistributeTranspiler._create_vars_from_blocklist..Nrr,F)r@rzr/r{r7)rdrerrAr8rrrEr=rWr@r}rr7rrrrr/r{rQ)r'rZ block_listrZ block_mapZ var_mappingZ block_strr$r%r&rrZ new_var_name orig_shaperorrHrowsZ splited_shaperErrrrs\       z0DistributeTranspiler._create_vars_from_blocklistcCs |j|j|j|j|j|j|dS)Nrv)rr@r7r/r{r0)r'rHrErzrrrrF szDistributeTranspiler._clone_varcCs"g}|D] }||jdq|S)Nr)rAr7)rrrrrrr" sz.DistributeTranspiler._get_splited_var_sectionsc Cs||}|jtjjjkr5|j|j}||r||j |<| j |ddd|id|id|t t iddS|jtjjjkrU| j |ddd|id|id|t t iddStd dS) NrZsplit_selected_rowsrrrrZ split_byrefrzrr;)r'rrZrr7rr pserver_block grad_blockrZ orig_varnamernZ trainer_nameZmerged_var_namer8Z vars2mergerZper_trainer_namerrrrK sT   z3DistributeTranspiler._append_pserver_grad_merge_opscsjjdjjjdd}jdtjjjtjjj dgdd}g}|j D]\}}|jjkr6| |q'j d||dd |id fd d } | } j d |dd | id | } j d| |dd | id | } j d| |dd | id | } j d|| dd | id | S)Nz .local_bakFr@r7r{r/rzz @TRAINER_ID@r)r@r{r/r7rzZref_by_trainer_id)rZ TrainerIdrrccs jtdjjjddS)NZ tmp_dc_outputFr)rrgenerater7r{r/rrHrrr__create_temp_var__ szEDistributeTranspiler._append_dc_asgd_ops..__create_temp_var__Zelementwise_sub)rYZelementwise_mulZelementwise_add) rr@r7r{r/rrrrZINT64rJrAr)r'rHrrZlocal_param_bakZtrainer_id_varZ ref_inputsrVrjrZo1o2Zo3Zo4rrr_append_dc_asgd_ops sd     z(DistributeTranspiler._append_dc_asgd_opscs>|j}|} t} fdd} jjr!| |} || |} |jD]}|dkrTjjr3| | |<q$||d}t |vrO| |rO| |}|| |<q$|| |<q$|dkrr| |}|sadS| j |jd|j|jd}|| |<q$|dkr||d}|| jvr| j||d| |<q$|j|}| j |j|j|j|jd}|| |<q$|jD]5}d}|d vrqjj||d}| d} |j||j| j}| j |j|j|j|d}|| |<qjj|}| d|d <|j|j| ||d | djt jjjkr|t| djd t| djdSdS) Ncs<d}jdD]}t|j|ddr|}|Sq |S)Nrrrr)r r0r@r)rY param_blockrVrr'rr_get_param_block szBDistributeTranspiler._append_pserver_ops.._get_param_blockrtrrrT)r@rzr/r7rursrr~r))rrrdrernr[rrrrZkNewGradSuffixr+rErr@r/r7rrzrrxr{_get_output_map_from_oprrgrrrrArB)r'rrYrr7rr8r9rrrirrdcrhrZnew_gradrrWZ lr_varnamerwr{rErrrrr5 s                z(DistributeTranspiler._append_pserver_opscCspd}|D]/\}}||j||jkr5|jddkr5||j|jvs0||j|jvr5|}|Sq|S)aP Return pserver side grad/param variable, return None if the variable is not grad/param, e.g. a@GRAD -> a@GRAD.block0 a@GRAD -> a@GRAD (a is not split) fc_0.w_0 -> fc_0.w_0.block_0 fc_0.w_0 -> fc_0.w_0 (weight is not split) _generated_var_123 -> None Nrr)rrr@rGrr)r'rEZvar_dictrrrrrr_get_pserver_grad_param_varq s   z0DistributeTranspiler._get_pserver_grad_param_varc Cs||jj|}|D]\}}t|ts|g}|D]}||jvr*||qq||jj|}|D]\}}t|tsF|g}|D]}||jvrV||qHq:|j |j ||| dSNr~) rfrrrrrtrr=rrr{rg) r'rrHrrrhvarlistrErrrrr> s2      z!DistributeTranspiler._clone_lr_opc Cs|j}||jj|}|D]K\}}t|ts|g}tt |D]4}||}| ||j} | r9| ||<q#|j |jvrM| |} | ||<q#|j|j ||<q#|||<q| |jj|} | D]K\}}t|tsw|g}tt |D]4}||}| ||j} | r| ||<q}|j |jvr| |} | ||<q}|j|j ||<q}|| |<qk|j|j|| |dSr)rrfrrrrrtrr>r=rr@r=rrr{rg) r'rrYrrrhrrrErrWrrrrr6 sV            z0DistributeTranspiler._append_pserver_non_opt_opscCs@t|jt|j@st|jt|j@rdSdSr)r#rr(r)r'op1op2rrr_is_op_connected sz%DistributeTranspiler._is_op_connectedcCsfddlm}||}tt|D] }t|t|D]}||}||}|||r/|||qq|S)Nr UnionFind)rrr>r=runion)r'rrrTrjrrrrrrH s   z"DistributeTranspiler._create_ufindcCsd|jvr d|jvr dSdS)NrrruTF)r)r'rrrrr4 sz%DistributeTranspiler._is_optimizer_opcCsbdd|j|dD}|dd|vrdS|D]}|dd}t||r.||kr.dSqdS)NcSrrr)rrVrrrr sz>DistributeTranspiler._is_opt_op_on_pserver..rrrrTF)r rr0)r'rrZ param_namesnr~rrrrI s z*DistributeTranspiler._is_opt_op_on_pservercC\t}|jD]$}g}||D] }|||qt|dkr'|d||<q|||<q|S)z8Returns a dict from op input name to the vars in varmap.rr)rdrerrrAr=r'ZvarmaprZiomaprhrr$rrrrf    z+DistributeTranspiler._get_input_map_from_opcCr)z9Returns a dict from op output name to the vars in varmap.rr)rdrerrrAr=rrrrr rz,DistributeTranspiler._get_output_map_from_opc sg}j}t|jD]\}}t|t}|ttks'|ttttBkrj dur|j dkr؈ jj |} jj |}|D]}||qGfddtjD} tjjD]3\} }|j dkr|jD]&}t||dkr||djkrjj| dtd jqoqc| D]&} | jjd jkr| _jj| j| j | j| j| jtdd qtj } |!||j"|d d | i|| tid}|#|t$d|j q |S)NF incrementc s:g|]}jjjd|jjjjdqS)r)r@r{r7r/rz)rrrr@r{r7r/rz)rZid_rrr'rrr- s z4DistributeTranspiler._get_lr_ops..rbrrrUgr)r@r{r/r7rzZ initializerrrr~zappend lr op: )%rrrrr8rrLR_SCHED_OP_ROLE_ATTR_VALUEOPT_OP_ROLE_ATTR_VALUErWr{rfrrr>rrrr=rr@r?r;r}rrrr/r7rzr rop_proto_and_checker_makerkOpRoleAttrNamerrrAr!) r'rrHrrZrole_idrrrhZall_trainer_counter_inputsrrEop_role_attr_namerrrr  s|              z DistributeTranspiler._get_lr_opsc Csddlm}g}t}|jD]}||r||ddqg}|j}|j D]}t|j |@r8| |q*||j }|j D]"}|j D]} || krb| || rb||sb|| sb| || qFqA|j D]}|D]} ||| rz| |nqkqg|S)Nrrru)rrr#rr4rrrrrr(rArrZ is_connected) r'rrZlr_varsrZfind_opsrHrTrrrrr_get_lr_ops_deprecatedb sD              z+DistributeTranspiler._get_lr_ops_deprecatedcCsBtj}tjjj}||jvrt||t|krdSdSr)rrOpRoleOptimizerZ attr_namesr8rg)r'rZop_makerZ optimize_rolerrr_is_opt_role_op s z$DistributeTranspiler._is_opt_role_opc Cs|j}g}g}t}|jj}|jD]`}||rtt|vrAt| tvrA|j j dkrA|j j dkrA| dt tjjjq||| trs| td}| td}||vrs||td|||||||gq q|} | r|| }||fS)z Get optimizer operators, parameters and gradients from origin_program Returns: opt_ops (list): optimize operators. params_grads (dict): parameter->gradient. rrZop_rolerrzadding param_grad pair: )rrr#rrr OP_NAME_SCOPErgCLIP_OP_NAME_SCOPErrnr\r?r8rrrZBackwardrAr rr!_get_distribute_update_vars) r'rHZopt_opsrZoptimize_paramsorigin_var_dictrrZ grad_nameZspecial_distribute_update_varsrrrr sD            z'DistributeTranspiler._get_optimize_passc Cs|j}|jj}g}|jD]!}d}||vr1||r1||dD] }|||q'qtt |}g}|D] } || | gq<|S)a This Function is used for a special model, like PyramidDnn which has pyramid hash op. Some Parameters don't use optimizing op to update its value, but updated in its BP process. In these cases, Transpilse can't find these special vars by optimizing op information. So we add this function and add attr "distribute_update_vars" to tell transpiler these Parameter need to be updated in distribute training. We assume these special var send and receive the same var_name. Zdistribute_update_varsrx) rrrrrgrrrArr#) r'rHrrrZ special_attrrZ unique_paramsrrErrrr s     z0DistributeTranspiler._get_distribute_update_varsr)NT)NNT)NrrTNr)T)NN)F)6rrrrXr(rorrrrrrr%r*r-r.rr]r_r^rprrrrrrMrLrNrrF staticmethodrrrxrkrrKrr5rr>r6rrHr4rIrfrr rrrrrrrrrms J  C 2 Q  J &5`>$ fF+^b*$g H #&6Ar4     F' 3rm)9rXrdrOr9rhr functoolsrnumpyrZpaddlerZpaddle.base.frameworkrZpaddle.frameworkrrrZCpaddle.incubate.distributed.fleet.parameter_server.ir.ps_dispatcherrr Zpaddle.nn.initializerr Z paddle.staticr r r Z paddle.utilsrrZLOOKUP_TABLE_GRAD_TYPErrrZkOpRoleVarAttrNamer rrrrrrZRPCrZDistrZLRSchedrrrr!r"r0rIrJrgrmrrrrsF            /j