o Me.@s8ddlZddlZddlZddlmZddlZddlmZddl m Z ddl m Z m Z mZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddl m!Z!ddlm"Z"m#Z#ddl m$Z$ddl%m&Z&m'Z'ddl(Z(gZ)ddZ*ddZ+ddZ,ee+Z-ee,Z.Gddde/Z0dS)N) MethodType) _global_flags)compiler)UserDefinedRoleMakerPaddleCloudRoleMaker RoleMakerBase)StrategyCompiler)DistributedStrategy)MetaOptimizerFactory)RuntimeFactory)wrap_decorator)parallel_helper)apply_build_strategy)topology)model_parallel_random_seed)_C_ops _legacy_C_ops)core)logger set_log_levelcCsv|jj}tds |St|di}|r|d}|jd}d|ji}|jj}|r4|jr4t dd|_t ||||S)NZFLAGS_apply_pass_to_program _pipeline_optZsection_programstartup_programZuse_cudazCurrently, the fuse_all_optimizer_ops pass has conflict with fuse_all_reduce_ops pass. Disable the fuse_all_optimizer_ops pass temporarily.F) _user_defined_strategybuild_strategyZ_copyrgetattrr_is_collectiveZfuse_all_reduce_opsZfuse_all_optimizer_opsrwarningr) main_programrconfigrZ pipeline_optZ pass_attrsZfuse_all_reducer ND:\Projects\ConvertPro\env\Lib\site-packages\paddle/distributed/fleet/fleet.pyapply_ir_passes,s"      r"cfdd}|S)Ncs(|d}|jdur td|i|S)Nrz+Fleet can not find suitable runtime handler)_runtime_handle ValueErrorargskwargsclsfuncr r!__impl__Is z*_inited_runtime_handler_..__impl__r r+r,r r*r!_inited_runtime_handler_Hs r.cr#)NcsB|d}|jdur|jdurtdjdS|i|S)NrTz:%s() function doesn't work when use non_distributed fleet.) _role_maker_is_non_distributedrr__name__r&r*r r!r,Us z,_is_non_distributed_check_..__impl__r r-r r*r!_is_non_distributed_check_Ts r2c@sNeZdZdZddZ    dgddZd d Zd d Zd dZddZ ddZ ddZ ddZ ddZ ddZddZddZdd Zdhd!d"Zd#d$Zd%d&Zdhd'd(Zd)d*Zd+d,Zeedid-d.Zeedid/d0Zd1d2Zeed3d4Zeed5d6Zeed7d8Zeed9d:Z eed;d<Z!eed=d>Z"eed?d@Z#eeggfdAdBZ$ee  C DdjdEdFZ%eedkdGdHZ&eedIdJZ'eedKdLZ(eedMdNZ)ee didOdPZ*didQdRZ+didSdTZ,dUdVZ-dWdXZ. dldYdZZ/d[d\Z0d]d^Z1d_d`Z2 dmdadbZ3 dmdcddZ4   dmdedfZ5dS)nFleeta Unified API for distributed training of PaddlePaddle Please reference the https://github.com/PaddlePaddle/PaddleFleetX for details Returns: Fleet: A Fleet instance Example for collective training: .. code-block:: python import paddle paddle.enable_static() import paddle.distributed.fleet as fleet fleet.init(is_collective=True) strategy = fleet.DistributedStrategy() optimizer = paddle.optimizer.SGD(learning_rate=0.001) optimizer = fleet.distributed_optimizer(optimizer, strategy=strategy) # do distributed training Example for parameter server training: .. code-block:: python import paddle paddle.enable_static() import paddle.distributed.fleet as fleet strategy = fleet.DistributedStrategy() fleet.init(strategy=strategy) optimizer = paddle.optimizer.SGD(learning_rate=0.001) optimizer = fleet.distributed_optimizer(optimizer) if fleet.is_first_worker(): print("this is first worker") print("current node index: {}".format(fleet.worker_index())) print("total number of worker num: {}".format(fleet.worker_num())) if fleet.is_worker(): print("this is worker") print("worker endpoints: {}".format(fleet.worker_endpoints(to_string=True))) print("server num: {}".format(fleet.server_num())) print("server endpoints: {}".format(fleet.server_endpoints(to_string=True))) if fleet.is_server(): print("this is server") fleet.stop_worker() cCs6d|_d|_d|_d|_d|_i|_tjd|_ dS)NFg) r/strategy_compilerrr$Z_util_contextpaddle optimizerZ Optimizeruser_defined_optimizerselfr r r!__init__szFleet.__init__NFINFOcst||dur t}t||_|dur.t|tr%||_t|jd|_ nt d t |t|t r;||_ |j|_n t d t ||j ddlmm}|j|j t|_|j ry|jrytjjrytjj}|dkryt dtjjr|dkrt |_!t"|j!|_#dSt$%rt&'dnd t(j)vrt&'d n t*|jj+t(j)d <tj,|jj-d krtj.dur|/|St&'d |S|jro|jj0}|1}|} |rd nd} t2t3| } tj.durt4tj.} | |_#| 5d|| | | |jj6} |p | }|d urdSd}d}|r#|jj7}t8|d}| r0|jj9}t8|d}|r=| r=||ks=J|rB|n|dkro| dksRJd}|}|fdd| D}| 5d||||S)aV Initialize role_maker in Fleet. This function is responsible for the distributed architecture what you want to run your code behind. Args: role_maker (RoleMakerBase, optional): A ``RoleMakerBase`` containing the configuration of environment variables related to distributed training.If you did not initialize the rolemaker by yourself, it will be automatically initialized to PaddleRoleMaker. The default value is None. is_collective (Boolean, optional): A ``Boolean`` variable determines whether the program runs on the CPU or GPU. False means set distributed training using CPU, and True means GPU.The default value is False.The default value is False. strategy (DistributedStrategy): Extra properties for distributed training. For details, please refer to paddle.distributed.fleet.DistributedStrategy. Default: None. log_level (Integer, String, optional): A ``Integer`` or ``String`` Variable determining how hight the logging level is. Default is "INFO". Returns: None Examples1: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() Examples2: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init(is_collective=True) Examples3: .. code-block:: python import paddle.distributed.fleet as fleet role = fleet.PaddleCloudRoleMaker() fleet.init(role) Examples4: .. code-block:: python import paddle.distributed.fleet as fleet strategy = fleet.DistributedStrategy() fleet.init(strategy=strategy) Examples5: .. code-block:: python import paddle.distributed.fleet as fleet strategy = fleet.DistributedStrategy() fleet.init(log_level = "DEBUG") N) is_collectivez8`is_collective` should be instance of `bool`, but got {}z>`role_maker` should be subclass of `RoleMakerBase`, but got {}rrz[CUDA_VISIBLE_DEVICES shoule be set only 1 card if you use `python` to launch fleet program.z6The dygraph parallel environment has been initialized.ZFLAGS_nccl_nringszYou have set the environment variable FLAGS_nccl_nrings outside the program, so the nccl_comm_num in DistributedStrategy will not take effect here.Fz=The dygraph hybrid parallel environment has been initialized.global mp_degreeZtensor_parallel_degreecsg|] }|kr|qSr r ).0idxr@Z mp_group_idr r! |s  zFleet.init..model):rr copydeepcopyr isinstanceboolrrr/r%formattyperZ_generate_rolepaddle.distributed.fleet distributedfleetutilZ_set_role_makerr r4r0r6fluidrZis_compiled_with_cudaZget_cuda_device_count framework_non_static_mode worker_numtpCommunicateTopology _topologyHybridCommunicateGroup_hcgrZ_is_parallel_ctx_initializedrrosenvironstrZ nccl_comm_numZinit_parallel_envZheter_ccl_modeZ_HYBRID_PARALLEL_GROUP_init_hybrid_parallel_envsharding worker_indexlistrangeZ_CommunicateGroupZset_comm_groupZtensor_parallelsharding_configsinttensor_parallel_configs)r: role_makerr=strategy log_levelrNZgpus_numZ use_shardingZ global_rankZglobal_world_sizeZglobal_ring_idZ global_ranksZcgZuse_tensor_parallelZuse_mpZmp_degree_shardingZmp_degree_tensor_parallelrarcZ mp_ring_idZmp_rankZmp_group_ranksr rCr!initsF               GE          z Fleet.initcCs.|jj|_|jd|_|jd|_|jd|_|jd|_|jdks&Jd|jdks/Jd|jdks8Jdt|jd |_t|jd |_|jdkrYtj }||j|j|_t|jd |_t j gd |j|j|j|jgd |_ t |j |_|jd kr|jj}|d }|d krtdSt|dSdS)z!initialize the hybrid environment dp_degreer@ pp_degreesharding_degreerz)mp_degree should be greater or equal to 0z)pp_degree should be greater or equal to 0z/sharding_degree should be greater or equal to 0r)datapiper]rE)Zhybrid_group_namesdimstensor_init_seedN)rZhybrid_configsrhr@rirjmaxr6rMZget_world_sizerTrUrVrWrXrcr)r:Znranksrcrnr r r!r\sB           zFleet._init_hybrid_parallel_envcC|jdusJ|jSN)rXr9r r r!get_hybrid_communicate_groupz"Fleet.get_hybrid_communicate_groupcCrqrr)rVr9r r r!get_hybrid_parallel_topologyrtz"Fleet.get_hybrid_parallel_topologycC |jS)an Check whether the node is the first instance of worker. Returns: bool: True if this is the first node of worker, False if not. Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() fleet.is_first_worker() )r/Z_is_first_workerr9r r r!is_first_worker zFleet.is_first_workercCrv)a  Get current worker index. Returns: int: node id Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() fleet.worker_index() )r/Z _worker_indexr9r r r!r^ zFleet.worker_indexcCrv)a Get current total worker number. Returns: int: worker numbers Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() fleet.worker_num() )r/Z _worker_numr9r r r!rSryzFleet.worker_numcCrvrr)r/Z _get_node_numr9r r r!node_num zFleet.node_numcCrvrr)r/Z_get_local_rankr9r r r! local_rankr{zFleet.local_rankcCrvrr)r/Z_get_local_device_idsr9r r r!local_device_idsr{zFleet.local_device_idscCrvrr)r/Z_get_world_device_idsr9r r r!world_device_idsr{zFleet.world_device_idscCrv)aY Check whether the node is an instance of worker. Returns: bool: True if this is a node of worker, False if not. Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() fleet.is_worker() )r/Z _is_workerr9r r r! is_workerrxzFleet.is_workercCrvrr)r/Z_is_coordinatorr9r r r!is_coordinatorr{zFleet.is_coordinatorcC|r d|jS|jS)aQ Get current worker endpoints, such as ["127.0.0.1:1001", "127.0.0.1:1002"]. Returns: list/string: server endpoints Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() fleet.worker_endpoints() ,)joinr/Z_get_trainer_endpointsr:Z to_stringr r r!worker_endpointss zFleet.worker_endpointscCst|jS)a Get current total worker number. Returns: int: server number Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() fleet.server_num() )lenr/_get_pserver_endpointsr9r r r! server_num)szFleet.server_numcCrv)a  Get current server index. Returns: int: node id Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() fleet.server_index() )r/Z _server_indexr9r r r! server_index:ryzFleet.server_indexcCr)aQ Get current server endpoints, such as ["127.0.0.1:1001", "127.0.0.1:1002"]. Returns: list/string: server endpoints Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() fleet.server_endpoints() r)rr/rrr r r!server_endpointsLs zFleet.server_endpointscCrv)aY Check whether the node is an instance of server. Returns: bool: True if this is a node of server, False if not. Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() fleet.is_server() )r/Z _is_serverr9r r r! is_serverbrxzFleet.is_servercCs|jddS)zH barrier all workers Returns: None ZworkerN)r/Z_barrierr9r r r!barrier_workeruszFleet.barrier_workercC|j|dS)ar initialize `Communicator` for parameter server training. Returns: None Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() # build net # fleet.distributed_optimizer(...) fleet.init_worker() N)r$Z _init_workerr:Zscopesr r r! init_worker~szFleet.init_workercCr)z- initialize coordinator node N)r$Z_init_coordinatorrr r r!init_coordinatorszFleet.init_coordinatorcCs|jdSrr)r$Z_make_fl_strategyr9r r r!make_fl_strategyszFleet.make_fl_strategycCs|jjS)z/ get worker(training node) ptr )r$Z_workerr9r r r! get_fl_clientszFleet.get_fl_clientcOs|jj|i|dS)a init_server executor to initialize startup program, if the `args` is not empty, it will run load_persistables for increment training. Returns: None Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() # build net # fleet.distributed_optimizer(...) fleet.init_server() N)r$Z _init_server)r:r'r(r r r! init_serverszFleet.init_servercC|j||dS)aa load fleet model from path Returns: None Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() # build net # fleet.distributed_optimizer(...) fleet.load_model("path", mode=0) N)r$Z_load_persistablesr:pathmoder r r! load_modelzFleet.load_modelcC|j|||dS)al load fleet one table from path Returns: None Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() # build net # fleet.distributed_optimizer(...) fleet.load_one_table(0, "path", mode=0) N)r$Z_load_one_tabler:Ztable_idrrr r r!load_one_tablezFleet.load_one_tablecCr)au load fleet inference model from path Returns: None Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() # build net # fleet.distributed_optimizer(...) fleet.load_inference_model("path", mode=1) N)r$Z_load_inference_modelrr r r!load_inference_modelrzFleet.load_inference_modelcC|jdS)a run server will run pserver main program with executor. Returns: None Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() # build net # fleet.distributed_optimizer(...) if fleet.is_server(): fleet.init_server() N)r$Z _run_serverr9r r r! run_serverszFleet.run_servercCr)a stop `Communicator` and give training complete notice to parameter server. Returns: None Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() # build net # fleet.distributed_optimizer(...) fleet.init_server() N)r$Z _stop_workerr9r r r! stop_worker(szFleet.stop_workerc Ksd}|s|sd}t}tj|}|rog}g} |D]} t| tr'|| qt| tjjr5|| jqt d|D]} t| trH| | q;t| tjjrV| | jq;t ddd| D} |j |||| ddddSd} d|vr{t |d} |j j ||d| ddS) NTFzfeed must be [str|Variable]cSs g|] }tj|qSr )r6staticdefault_main_programZ global_blockvar)rAnamer r r!rD_szFleet.save..rr)rr)r6ZCPUPlacerExecutorrHr[appendVariablerr%r$_save_inference_modelrb_save_persistables) r:dirnamefeedfetchconfigsZ inferenceplaceexecutorfeeded_var_namesZfetch_var_namesrZ fetch_varsZincrement_moder r r!save@s@       z Fleet.saveTrc Cs|j|||||||dS)a\ save inference model for inference. Returns: None Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() # build net # fleet.distributed_optimizer(...) fleet.init_server() N)r$r)r:rrrZ target_varsrZexport_for_deploymentrr r r!save_inference_modelos zFleet.save_inference_modelcCs|j||||dS)aX saves all persistable tensors from :code:`main_program` to the folder :code:`dirname`. You can refer to The :code:`dirname` is used to specify the folder where persistable tensors are going to be saved. If you would like to save tensors in separate files, set :code:`filename` None. Args: executor(Executor): The executor to run for saving persistable tensors. You can refer to :ref:`api_guide_executor_en` for more details. dirname(str, optional): The saving directory path. When you need to save the parameter to the memory, set it to None. main_program(Program, optional): The program whose persistbale tensors will be saved. Default: None. Returns: None Examples: .. code-block:: text import paddle paddle.enable_static() import paddle.distributed.fleet as fleet fleet.init() # build net # fleet.distributed_optimizer(...) exe = paddle.static.Executor(paddle.CPUPlace()) fleet.save_persistables(exe, "dirname", paddle.static.default_main_program()) N)r$r)r:rrrrr r r!save_persistabless+zFleet.save_persistablescKs|jj|fi|Srr)r$Z_save_cache_model)r:rrr r r!save_cache_modelszFleet.save_cache_modelcCrvrr)r$Z_check_save_pre_patch_doner9r r r!check_save_pre_patch_dones zFleet.check_save_pre_patch_donecCr)al save fleet one table from path Returns: None Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() # build net # fleet.distributed_optimizer(...) fleet.save_one_table(0, "path", mode=0) N)r$Z_save_one_tablerr r r!save_one_tablerzFleet.save_one_tablecCs|j|||||dS)a< save fleet one table from path Returns: None Examples: .. code-block:: python import paddle.distributed.fleet as fleet fleet.init() import paddle place = paddle.fluid.CPUPlace() exe = paddle.fluid.Executor(place) # build net # fleet.distributed_optimizer(...) fleet.save_dense_params(exe, "path", scope=paddle.static.global_scope(), program=paddle.static.default_main_program()) N)r$Z_save_dense_params)r:rrscopeprogramZ var_namesr r r!save_dense_paramss zFleet.save_dense_paramscCs|j|dSrr)r$Z_shrink)r: thresholdr r r!shrink sz Fleet.shrinkcCs4||_|dur|jrtdt||_i|_|S)a Optimizer for distributed training. For the distributed training, this method would rebuild a new instance of DistributedOptimizer. Which has basic Optimizer function and special features for distributed training. Args: optimizer(Optimizer): The executor to run for init server. strategy(DistributedStrategy): Extra properties for distributed optimizer. It is recommended to use DistributedStrategy in fleet.init(). The strategy here is for compatibility. If the strategy in fleet.distributed_optimizer() is not None, then it will overwrite the DistributedStrategy in fleet.init(), which will take effect in distributed training. Returns: Fleet: instance of fleet. Examples: .. code-block:: python import paddle import paddle.distributed.fleet as fleet fleet.init(is_collective=True) strategy = fleet.DistributedStrategy() optimizer = paddle.optimizer.SGD(learning_rate=0.001) optimizer = fleet.distributed_optimizer(optimizer, strategy=strategy) NaIt is recommended to use DistributedStrategy in fleet.init(). The strategy here is only for compatibility. If the strategy in fleet.distributed_optimizer() is not None, then it will overwrite the DistributedStrategy in fleet.init(), which will take effect in distributed training.)r8rrrrFrGrr5)r:r7rer r r!distributed_optimizers zFleet.distributed_optimizercCsTd}|jD] }t|dr|}nq|dur t|jdr |j}|dus(Jd|S)Namp_initzSamp_init can only be used when the amp(auto mixed precision) strategy is turned on.)r4Z_get_applied_meta_optimizerhasattrr8)r: amp_optimizerr7r r r!_get_amp_optimizer=s   zFleet._get_amp_optimizercCs|}|S)z)Return the real-time loss scaling factor.)rget_loss_scaling)r:rr r r!rNszFleet.get_loss_scalingcCs|}|||||S)a Init the amp training, such as cast fp32 parameters to fp16 type. Args: place(CUDAPlace): place is used to initialize fp16 parameters with fp32 values. scope(Scope): The scope is used to find fp32 parameters. test_program(Program): The program is used for testing. use_fp16_test(bool): Whether to use fp16 testing. Examples: .. code-block:: python import paddle import paddle.nn.functional as F paddle.enable_static() def run_example_code(): place = paddle.CUDAPlace(0) exe = paddle.static.Executor(place) data = paddle.static.data(name='X', shape=[None, 1, 28, 28], dtype='float32') conv2d = paddle.static.nn.conv2d(input=data, num_filters=6, filter_size=3) # 1) Use fp16_guard to control the range of fp16 kernels used. with paddle.static.amp.fp16_guard(): bn = paddle.static.nn.batch_norm(input=conv2d, act="relu") pool = F.max_pool2d(bn, kernel_size=2, stride=2) hidden = paddle.static.nn.fc(pool, size=10) loss = paddle.mean(hidden) # 2) Create the optimizer and set `multi_precision` to True. # Setting `multi_precision` to True can avoid the poor accuracy # or the slow convergence in a way. optimizer = paddle.optimizer.Momentum(learning_rate=0.01, multi_precision=True) # 3) These ops in `custom_black_list` will keep in the float32 computation type. amp_list = paddle.static.amp.CustomOpLists( custom_black_list=['pool2d']) # 4) The entry of Paddle AMP. # Enable pure fp16 training by setting `use_pure_fp16` to True. optimizer = paddle.static.amp.decorate( optimizer, amp_list, init_loss_scaling=128.0, use_dynamic_loss_scaling=True, use_pure_fp16=True) # If you don't use the default_startup_program(), you sholud pass # your defined `startup_program` into `minimize`. optimizer.minimize(loss) exe.run(paddle.static.default_startup_program()) # 5) Use `amp_init` after FP32 parameters initialization(such as `exe.run(startup_program)`). # If you want to perform the testing process, you should pass `test_program` into `amp_init`. optimizer.amp_init(place, scope=paddle.static.global_scope()) if paddle.is_compiled_with_cuda() and len(paddle.static.cuda_places()) > 0: run_example_code() )rr)r:rrZ test_programZ use_fp16_testrr r r!rSs9zFleet.amp_initcCs d|jvr tdiS|jdS)Nvalid_strategyzNWARNING: You may need to call minimize function before this function is calledr5printr9r r r!_final_strategy  zFleet._final_strategycC d|jvr tdgS|jdS)Napplied_meta_listzTWARNING: You may need to call minimize function before _get_applied_meta_list calledrr9r r r!_get_applied_meta_listrzFleet._get_applied_meta_listcCr)Napplied_graph_listzUWARNING: You may need to call minimize function before _get_applied_graph_list calledrr9r r r!_get_applied_graph_listrzFleet._get_applied_graph_listcCsNt|ts |||||Stjjs|js|j rt d| ||||S)a Add distributed operations to minimize ``loss`` by updating ``parameter_list``. Args: loss (Tensor): A ``Tensor`` containing the value to minimize. startup_program (Program, optional): :ref:`api_fluid_Program` for initializing parameters in ``parameter_list``. The default value is None, at this time :ref:`api_fluid_default_startup_program` will be used. parameter_list (Iterable, optional): Iterable of ``Tensor`` or ``Tensor.name`` to update to minimize ``loss``. The default value is None, at this time all parameters will be updated. no_grad_set (set, optional): Set of ``Tensor`` or ``Tensor.name`` that don't need to be updated. The default value is None. Returns: tuple: tuple (optimize_ops, params_grads), A list of operators appended by minimize and a list of (param, grad) tensor pairs, param is ``Parameter``, grad is the gradient value corresponding to the parameter. The returned tuple can be passed to ``fetch_list`` in ``Executor.run()`` to indicate program pruning. If so, the program will be pruned by ``feed`` and ``fetch_list`` before run, see details in ``Executor``. Examples: .. code-block:: python import paddle paddle.enable_static() import paddle.distributed.fleet as fleet import paddle.nn.functional as F hid_dim = 10 label_dim = 2 input_x = paddle.static.data(name='x', shape=[None, 13], dtype='float32') input_y = paddle.static.data(name='y', shape=[None, 1], dtype='int64') fc_1 = paddle.static.nn.fc(x=input_x, size=hid_dim, activation='tanh') fc_2 = paddle.static.nn.fc(x=fc_1, size=hid_dim, activation='tanh') prediction = paddle.static.nn.fc(x=[fc_2], size=label_dim, activation='softmax') cost = F.cross_entropy(input=prediction, label=input_y) avg_cost = paddle.mean(x=cost) fleet.init(is_collective=True) strategy = fleet.DistributedStrategy() optimizer = paddle.optimizer.SGD(learning_rate=0.001) optimizer = fleet.distributed_optimizer(optimizer, strategy=strategy) optimizer.minimize(avg_cost) # for more examples, please reference https://github.com/PaddlePaddle/PaddleFleetX z loss can be list only in PS mode) rHr__minimize_implr6rPrQrRr/r0rr%_minimize_losses_impl)r:lossrparameter_list no_grad_setr r r!minimizes 5 zFleet.minimizecCsi}t|j|d<tjjr|j}||_| |S|j j |_ t |j dsVt|j dt|jjd|j jd<|jjd|j jd<|jjd|j jd<|jjd|j jd<|j |d<|j g|d<||d <|dkrytjjd d |_tj}n|jd d |_||d <|g|d <|j|d<|jjs|jjrddlm}||}|||||\} } } } | | | | fSt|j} t|j|d<t|j}|r| D]}|||qg}g}g}| D]/}| ||j|j||!r|"s|#|q|!r|"r|#|q|#|q|j$%||j|j|||\}}|j$&||}t||d<t'(dt)|dt'(dt)|d|j$*}|j$+}||d<||d<||_||_,|j,-g} g} |j.r|j/s|j0duryt12||_0t34|j j5|j6dd}||j j _7|jj ||||dS|rt'(dt)t8|j j |j ||||d\} } t'(dt)t8|j j tj9}t'(dt)t8|t8|t8|j j krtjj:|j j t'(dt)t8|n |jj ||||d\} } | |d<| |d<|r$t'(dt)t8|j j |j ||||d\} } | |d<| |d <nt;|j j |||jj<shtj9}|j=dur>in|j=}|>|d!<|?|d"<|jj@AD]\}}|s_||vrc|||<qS||_=|j0durut12||_0d#dlBmCmD}|jEF|d| | fS)$Nuser_defined_strategydistributed_info_rhr@rirjorigin_main_programorigin_main_programsrFZfor_testorigin_startup_programorigin_startup_programsrd)AutoParallelizerrzvalid_strategy: zuser_defined_strategy: rr)Z loss_nameZshare_vars_fromrzbefore minimize program id: zafter minimize program id: zdefault program id: z!default program id after switch: program_optimize_opsprogram_params_gradsz"before graph minimize program id: Zgraph_optimize_opsZgraph_optimize_gradsmpi_sizempi_rankr)GrFrGrr6rPrQrRr8r5rblockrrrsetattrdictrarrdefault_startup_programclonerr/Z semi_autoZ auto_searchZauto_parallel.parallelizerrZ parallelizer Z_get_valid_meta_optimizersZ_is_strict_autoZ_enable_strategy_set_basic_infoZ _can_applyZ _is_graph_outrr4Zgenerate_optimizerZ_get_valid_strategyrdebugr[rrr _enable_envr0rr$r _create_runtimerZCompiledProgramZwith_data_parallelrZ_graphidrZswitch_main_programr"Z_is_heter_parameter_server_mode _fleet_optrSr^trainer_desc_configsitemsrLrMrNrO _set_strategy)r:rrrrcontextZ target_optrZauto_parallelizer optimize_ops params_gradsZdist_startup_progZdist_main_progZdistributed_optimizer_listZcopy_user_defined_strategyoptZvalid_optimizer_listZvalid_graph_optimizer_listZcan_not_apply_optimizer_listZmeta_optimizerZgraph_optimizerrrrZcompiled_programZdefault_programropt_infokvrNr r r!rsV                                    zFleet._minimize_implcCsi}|djj|_|j|d<g|d<|D] }|d|jjq||d<|dur9t|dkr5tjg}ntd|dj dd|_ |d|d <g|d <|D] }|d |qN|j |d <t |j|d <t |j|d <||_|d |_|jg}g} ddlm} | |j} | ||j |j|j| j||||d\}} ||d<| |d<|D]D}|jj}|jdurin|j} || d<|| d<|jjD]\} }|s| | vr|| | <q| |_tdtt|t|jq|j durt!"||_ ddl#m$m%}|j&'|d || fS)Nrrrrrz0startup_program can't be None when loss is list.Frrrrdrr)ParameterServerOptimizerrrrrrzfleet base opt info: )(rrrrrr6rrr%rrr/rFrGrr5rrZmeta_optimizersrr8rZminimize_losses_implrrSr^rrrrr[rr$r rrLrMrNrOr)r:ZlossesZstartup_programsrrrrrrrrZ ps_optimizerrrrrNr r r!rs                zFleet._minimize_losses_impl)NFNr<)Frr)NTr)Nr)NNF)NNN)6r1 __module__ __qualname____doc__r;rgr\rsrurwr^rSrzr|r}r~rrrrrrrris_non_distributed_checkinited_runtime_handlerrrrrrrrrrrrrrrrrrrrrrrrrrrrrr r r r!r3ks:  X-          -(-     / <   E _r3)1rFr6rYtypesrnumpynpZpaddle.fluid.frameworkrZ paddle.fluidrZbase.role_makerrrrZbase.strategy_compilerr Zbase.distributed_strategyr Zbase.meta_optimizer_factoryr Zbase.runtime_factoryr Zpaddle.fluid.wrapped_decoratorr Zpaddle.fluid.dygraphrZpaddle.fluid.irrbaserrTZ meta_parallelrrrrZutils.log_utilrrlogging__all__r"r.r2rrobjectr3r r r r!s8