zjY`ddlmZddlZddlZddlZddlZddlZddlmZm Z m Z m Z ddl m Z ddlmZddlmZmZddlmZddlmZdd lmZmZmZdd lmZmZmZdd lm Z erdd l!m"Z"m#Z#dd l$m%Z%m&Z&Gdde Z'gZ(GddZ)dZ*dZ+dZ,dZ-dZ.dZ/dZ0dZ1GddZ2 d-d.d,Z3dS)/) annotationsN) TYPE_CHECKINGAnyLiteral TypedDict)core) get_device)_get_trainers_numget_cluster_and_pod)use_paddlecloud)get_cluster_from_args) DeviceModeblock_windows_and_macos check_backend)_prepare_trainer_env_print_argumentsget_host_name_ip) set_flags)CallableIterable) NotRequiredUnpackc8eZdZUded<ded<ded<ded<dS) _SpawnOptionsz3NotRequired[Literal['spawn', 'fork', 'forkserver']] start_methodzNotRequired[str | None]gpusxpuszNotRequired[str]ipsN)__name__ __module__ __qualname____annotations__h/lsinfo/ai/hellotax_ai/data_center/backend/venv/lib/python3.11/site-packages/paddle/distributed/spawn.pyrr3sBIIII%%%%%%%%r$rceZdZdZdS)ParallelEnvArgscZd|_d|_d|_d|_d|_d|_dS)NT)cluster_node_ipsnode_ipr started_port print_configselected_devices)selfs r%__init__zParallelEnvArgs.__init__>sA $  $!! !%r$N)rr r!r/r#r$r%r'r'=s#%%%%%r$r'cgd}gd}|D]<}||vr6||vrtjd|dt)td|d=dS)N)rrrrr,backend)r-r+r)r*r zThe config option (z|) of `paddle.distributed.spawn` is deprecated. Please use the latest config options stated in the `spawn` API documentation.z1) of `paddle.distributed.spawn` is not supported.)warningswarnDeprecationWarning ValueError)optionssupported_optionsdeprecated_optionskeys r%_options_valid_checkr:Vs   ' ' '((( d#ddd& !`#``` (  r$cRt}|tjvr-tj|ddSd|vrtjSd|vrtjSd|vrtjStd|d)N:rgpuxpucpuI`paddle.distributed.spawn` does not support parallel training on device `` now.) r rget_available_custom_deviceget_custom_device_countsplitget_cuda_device_countget_xpu_device_countmultiprocessing cpu_count RuntimeErrordevices r%_get_default_nprocsrLus \\F 13333+FLL,=,=a,@AAA &)+++ &(*** &(*** fX^ f f f   r$ct}|tjvrdSd|vrdSd|vrdSd|vrdStd|d ) Nxcclr=ncclr>bkclr?gloor@rA)r rrBrIrJs r%_get_default_backendrRso \\F 13333v &v &v &v fX^ f f f   r$cd}d|dD}t|dkr |d}nt\}}|S)Nc6g|]}|Sr#)strip.0xs r% z _get_node_ip..s 222a 222r$,r)rDlenr)rr*node_ips_s r% _get_node_ipr_sTG22399S>>222H 8}}1+%'' 7 Nr$c  d|vs |ddkrt|d<t|dt|dg}t}|dd|_|j)|dd|_|jd|_|ddkr|dd|_|j|dd|_tjd d}||d kr+d ttj D n| d  |jlt |kr#td t d|dd  fdtd|D|_n|j d }t||kr#t!dt|d|d|D]<}| vr6t!d|d  =nR|ddkr|dd|_|j|dd|_tjdd}||d kr+dttjD n| d  |jlt |kr#td t d|dd  fdtd|D|_n8|j d }t||kr#t!dt|d|d|D]<}| vr6t!d|d  =n|ddkrt'jdd |_d|_|j|_|d!d Jd"t|j d d#ks Jd$t/d#ks Jd%n|dd&krd|_tjd}tjd'|d(d}||d kr,d)ttj|D n| d  t |kr&td t d|d*|d+d  fd,td|D|_|d-d|_|jt7|j|_|d.d|_|d!d|_|jt;|_|ddkr.+   A   r$rZzthe number of visible devices(z-) is less than the number of spawn processes(z), please ensure that the correct `nprocs` argument is passed or the environment variable `CUDA_VISIBLE_DEVICES` is correctly configured.c:g|]}t|Sr#rerWrXenv_devices_lists r%rYz,_get_subprocess_env_list..'DDDa%a())DDDr$rzThe number of selected devices(z0) is not equal to the number of spawn processes(zK), please ensure that the correct `nprocs` and `gpus` arguments are passed.zCThe selected gpu card {} cannot found in CUDA_VISIBLE_DEVICES ({}).rPrXPU_VISIBLE_DEVICESc,g|]}t|Sr#rerVs r%rYz,_get_subprocess_env_list..rgr$z), please ensure that the correct `nprocs` argument is passed or the environment variable `XPU_VISIBLE_DEVICES` is correctly configured.c:g|]}t|Sr#reris r%rYz,_get_subprocess_env_list..rkr$zK), please ensure that the correct `nprocs` and `xpus` arguments are passed.zBThe selected xpu card {} cannot found in XPU_VISIBLE_DEVICES ({}).rQzYour model will be trained under CPUONLY mode by using GLOO,because CPUPlace is specified manually or your installed PaddlePaddle only support CPU Device.Tr z.CPUONLY spawn doesn't support use paddle cloudr[zJCPUONLY spawn only support single trainer, that is len(ips)=1, but got %s.z+CPUONLY spawn doesn't support multi-trainerrNFLAGS_selected_sc,g|]}t|Sr#rerVs r%rYz,_get_subprocess_env_list..#s.   A   r$zj), please ensure that the correct `nprocs` argument is passed or the environment variable `FLAGS_selected_zs` is correctly configured.c:g|]}t|Sr#reris r%rYz,_get_subprocess_env_list..2s' @ @ @!S!!$ % % @ @ @r$r*r+r,F)(rRrrr'getr)r-osgetenvrangerrErDr\rIjoinr5formatrFr2r3paddle_cpuonlyrr get_all_custom_device_typerCr*r_r+r listr rCPUr trainersappendrr,r) nprocsr6processes_env_listargs env_devicesselected_device_listcard_idcustom_device_namedevices_per_procclusterpodtrainerrjs @r%_get_subprocess_env_listrsU 79#5#?#?133 ')$%%%GI.///   D$KKt44D $ ' ,> E E  ($/D !yV## ' FD 9 9  ($+KK0BD$I$ID !i 6==  +"3"3   %d&@&B&B C C      +0055   (#$$v--"FS9I5J5JFF*0FFF %(HHDDDD5F3C3CDDD%%D ! !$(#8#>#>s#C#C '((F22 Hc:N6O6OHH5;HHH 0  "222$55;V#SXX.>%?%?663   v % % ' FD 9 9  ($+KK0BD$I$ID !i 5t<<  +"3"3   %d&?&A&A B B      +0055   (#$$v--"ES9I5J5JEE*0EEE %(HHDDDD5F3C3CDDD%%D ! !$(#8#>#>s#C#C '((F22 Hc:N6O6OHH5;HHH 0  "222$44:F#SXX.>%?%?553   v % %  m   # $({{,d33;; <<;;4(..s3344999 X:99!""a''' 9('''  v % % $!<>>qAi G2D G G GNN  +"3"3  t;z0MultiprocessContext.__init__..s.   (6wG e   r$) error_queues return_queues processes enumerate sentinels)r.rrrs r%r/zMultiprocessContext.__init__sG( +"  :CI:N:N   r$Nc6t|jdkrdStj|j|}d}|D]L}|j|}|j|}||j dkr|}nM|t|jdkS|jD]>}| r| |?| |dS)NrT)timeout) r\rrG connectionwaitkeysrrrwexitcodeis_alive terminate_throw_exception)r.rready error_indexrrrs r%rwzMultiprocessContext.joins* t~  ! # #4*// N   ! !70     HN&&x00EnU+G LLNNN1$$# %  t~&&!+ +~  G!! $!!### LLNNNN k*****r$cn|j|r^|j|j}|dkr0t j| j}td|d|dtd|d|d|j|}d|d}||z }t|)NrzProcess z terminated with signal .z terminated with exit code z9 ---------------------------------------------- Process zV terminated with the following error: ---------------------------------------------- ) remptyrrsignalSignalsnamerrs)r.rrroriginal_tracemsgs r%rz$MultiprocessContext._throw_exceptions  [ ) / / 1 1 ~k2;H!||~xi005K{KKDKKK R{RRxRRR*;7;;== A" A A A ~nnr$)N)rr r!r/rwrr#r$r%rrsF    ++++6r$rr#TFrCallable[..., None]r Iterable[Any]rintrwbooldaemonr6Unpack[_SpawnOptions]returnc t||dkrt}t||}|dd}|d}t j|}g} g} g} t |D]} |} |}|t||| ||| |df}||_ | | | | || |t| | | }|s|S|s ||S)a6 Start multiple processes with ``spawn`` method for parallel training. .. note:: ``spawn`` now only supports GPU or XPU collective mode. The collective mode of GPU and XPU cannot be started at the same time, so the option `gpus` and `xpus` cannot be configured at the same time. Args: func (function): The target function is called by spawned process. This function need to be able to pickled, so it must be defined at the top level of a module. args (list|tuple, optional): Arguments passed to ``func``. nprocs (int, optional): Number of processed to start. Default: -1. when nprocs is -1, the available device will be obtained from the environment variable when the model is executed: If use GPU, the currently available device ID is obtained from the environment variable CUDA_VISIBLE_DEVICES; If use XPU, the currently available device ID is obtained from the environment variable XPU_VISIBLE_DEVICES. join (bool, optional): Perform a blocking join on all spawned processes. Default: True. daemon (bool, optional): The spawned processes' daemon flag. Default: False. **options(dict, optional): Other initial parallel execution environment configuration options. The following options are currently supported: (1) start_method (string): the way to start a process. The start method can be ``spawn`` , ``fork`` , ``forkserver`` . Because the CUDA runtime does not support the ``fork`` start method, when use CUDA in subprocesses, we should start process by ``spawn`` or ``forkserver`` method. Default: "spawn" ; (2) gpus (string): The training process will run on the selected gpus, such as "0,1,2,3". Default: None; (3) xpus (string): The training process will run on the selected xpus, such as "0,1,2,3". Default: None; (5) ips (string): Paddle cluster nodes ips, such as "192.168.0.16,192.168.0.17". Default: "127.0.0.1" . Returns: ``MultiprocessContext`` object, it hold the spawned processes. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle >>> import paddle.nn as nn >>> import paddle.optimizer as opt >>> import paddle.distributed as dist >>> class LinearNet(nn.Layer): ... def __init__(self): ... super().__init__() ... self._linear1 = nn.Linear(10, 10) ... self._linear2 = nn.Linear(10, 1) ... def forward(self, x): ... return self._linear2(self._linear1(x)) >>> def train(print_result=False): ... # 1. initialize parallel environment ... group = dist.init_parallel_env() ... process_group = group.process_group if group else None ... # 2. create data parallel layer & optimizer ... layer = LinearNet() ... dp_layer = paddle.DataParallel(layer, group = process_group) # type: ignore[arg-type] ... loss_fn = nn.MSELoss() ... adam = opt.Adam( ... learning_rate=0.001, parameters=dp_layer.parameters()) ... # 3. run layer ... inputs = paddle.randn([10, 10], 'float32') ... outputs = dp_layer(inputs) ... labels = paddle.randn([10, 1], 'float32') ... loss = loss_fn(outputs, labels) ... if print_result is True: ... print("loss:", loss.numpy()) ... loss.backward() ... adam.step() ... adam.clear_grad() >>> # Usage 1: only pass function. >>> # If your training method no need any argument, and >>> # use all visible devices for parallel training. >>> if __name__ == '__main__': ... dist.spawn(train) >>> # Usage 2: pass function and arguments. >>> # If your training method need some arguments, and >>> # use all visible devices for parallel training. >>> if __name__ == '__main__': ... dist.spawn(train, args=(True,)) >>> # Usage 3: pass function, arguments and nprocs. >>> # If your training method need some arguments, and >>> # only use part of visible devices for parallel training. >>> # If your machine hold 8 cards {0,1,2,3,4,5,6,7}, >>> # this case will use cards {0,1}; If you set >>> # CUDA_VISIBLE_DEVICES=4,5,6,7, this case will use >>> # cards {4,5} >>> if __name__ == '__main__': ... dist.spawn(train, args=(True,), nprocs=2) >>> # Usage 4: pass function, arguments, nprocs and gpus. >>> # If your training method need some arguments, and >>> # only use part of visible devices for parallel training, >>> # but you can't set your machine's environment variable >>> # CUDA_VISIBLE_DEVICES, such as it is None or all cards >>> # {0,1,2,3,4,5,6,7}, you can pass `gpus` to >>> # select the GPU cards you want to use. For example, >>> # this case will use cards {4,5} if your machine hold 8 cards. >>> if __name__ == '__main__': ... dist.spawn(train, args=(True,), nprocs=2, gpus='4,5') rrNspawnr1)targetr)r:rLrrsrG get_contextrv SimpleQueueProcessrrstartr~rrw)rrrrwrr6procs_env_listrmprrrirrrcontexts r%rrsr!!!||$&& .fg>>N;;~t44L  $\ 2 2BLMI 6]]""nn&& ~~'' ** q! "      K(((\***!!!!!)\=IIG llnn  llnn  Nr$)r#rTF)rrrrrrrwrrrr6rrr)4 __future__rrGrtrrr2typingrrrr paddle.baser paddle.devicer paddle.distributed.cloud_utilsr r $paddle.distributed.fleet.cloud_utilsr paddle.distributed.fleet.launchr %paddle.distributed.fleet.launch_utilsrrr%paddle.distributed.utils.launch_utilsrrrpaddle.frameworkrcollections.abcrrtyping_extensionsrrr__all__r'r:rLrRr_rrrrrrr#r$r%rs#""""" 999999999999$$$$$$A@@@@@AAAAAA  '&&&&& 2222222255555555  %%%%%%%%2>       uuup(((2220$========D ooooooor$