o * iY`@slddlmZddlZddlZddlZddlZddlZddlmZm Z m Z m Z ddl m Z ddlmZddlmZmZddlmZddlmZdd lmZmZmZdd lmZmZmZdd lm Z erzdd l!m"Z"m#Z#dd l$m%Z%m&Z&Gddde Z'gZ(GdddZ)ddZ*ddZ+ddZ,ddZ-ddZ.ddZ/ddZ0d d!Z1Gd"d#d#Z2 $ % & 'd6d7d4d5Z3dS)8) 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) NotRequiredUnpackc@s.eZdZUded<ded<ded<ded<dS) _SpawnOptionsz3NotRequired[Literal['spawn', 'fork', 'forkserver']] start_methodzNotRequired[str | None]gpusxpuszNotRequired[str]ipsN)__name__ __module__ __qualname____annotations__r!r!d/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/paddle/distributed/spawn.pyr3s  rc@seZdZddZdS)ParallelEnvArgscCs(d|_d|_d|_d|_d|_d|_dS)NT)cluster_node_ipsnode_ipr started_port print_configselected_devices)selfr!r!r"__init__>s  zParallelEnvArgs.__init__N)rrrr*r!r!r!r"r#=s r#cCsTgd}gd}|D]}||vr'||vrtd|dtq td|dq 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)optionsZsupported_optionsZdeprecated_optionskeyr!r!r"_options_valid_checkVs  r2cCsft}|tvrt|ddSd|vrtSd|vr#tSd|vr+tSt d|d)N:rgpuxpucpuI`paddle.distributed.spawn` does not support parallel training on device `` now.) rrget_available_custom_deviceget_custom_device_countsplitget_cuda_device_countget_xpu_device_countmultiprocessing cpu_count RuntimeErrorZdevicer!r!r"_get_default_nprocsus  rBcCsJt}|tvr dSd|vrdSd|vrdSd|vrdStd|d ) Nxcclr4ncclr5bkclr6gloor7r8)rrr9r@rAr!r!r"_get_default_backends  rGcCs>d}dd|dD}t|dkr|d}|St\}}|S)NcSsg|]}|qSr!)strip.0xr!r!r" sz _get_node_ip..,r)r;lenr)rr%Znode_ips_r!r!r" _get_node_ips  rQc sd|vs |ddkrt|d<t|dt|dg}t}|dd|_|jdur;|dd|_|jdur;d|_|ddkr|dd|_|jdurT|dd|_td d}|dusb|d krnd d t t Dn| d |jdurt |krtdt d|dd fdd t d|D|_n_|j d }t ||krtdt |d|d|D]}|vrtd|d qn0|ddkrb|dd|_|jdur|dd|_tdd}|dus|d krdd t t Dn| d |jdur1t |kr tdt d|dd fdd t d|D|_n|j d }t ||krKtdt |d|d|D]}|vr_td|d qMn|ddkrtd d!|_d|_|j|_|d"ddusJd#t |j d d$ksJd%td$ksJd&n]|dd'krd|_t d}td(|d)d}|dus|d krd*d t t |Dn| d t |krtdt d|d+|d,d fd-d t d|D|_|d.d|_|jdurt|j|_|d/d|_|d"d|_|jdur't|_|ddkr?tt d|}t|t j!|\} } nt"|\} } | j#D]} |$t%| | |dqH|d0d1|_&|j&rft'||S)2Nr+autorr$z 127.0.0.1rDrr(ZCUDA_VISIBLE_DEVICEScSg|]}t|qSr!strrIr!r!r"rLz,_get_subprocess_env_list..rMzthe 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.cg|]}t|qSr!rUrIZenv_devices_listr!r"rLrzThe 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 ({}).rErZXPU_VISIBLE_DEVICEScSrTr!rUrIr!r!r"rLrWz), please ensure that the correct `nprocs` argument is passed or the environment variable `XPU_VISIBLE_DEVICES` is correctly configured.crXr!rUrIrYr!r"rLrZzK), please ensure that the correct `nprocs` and `xpus` arguments are passed.zBThe selected xpu card {} cannot found in XPU_VISIBLE_DEVICES ({}).rFzYour 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 cloudrNzJCPUONLY spawn only support single trainer, that is len(ips)=1, but got %s.z+CPUONLY spawn doesn't support multi-trainerrCZFLAGS_selected_scSrTr!rUrIr!r!r"rL#szj), please ensure that the correct `nprocs` argument is passed or the environment variable `FLAGS_selected_zs` is correctly configured.crXr!rUrIrYr!r"rL2rZr%r&r'F)(rGrrr#getr$r(osgetenvrangerr<r;rOr@joinr/formatr=r,r-Zpaddle_cpuonlyrr Zget_all_custom_device_typer:r%rQr&r listr r ZCPUr Ztrainersappendrr'r) nprocsr0Zprocesses_env_listargsZ env_devicesZselected_device_listZcard_idZcustom_device_nameZdevices_per_procZclusterZpodZtrainerr!rYr"_get_subprocess_env_lists                                     rfcCs tjddtjdddS)NZ http_proxyZ https_proxy)r]environpopr!r!r!r"_remove_risky_envWsricCsR|dkr td|din|dkrtd|din |D] }||tj|<qdS)NrDZFLAGS_selected_gpusrEZFLAGS_selected_xpus)rr]rg)env_dictr+var_namer!r!r"_set_trainer_env^s rlcCspztt||||}||WdStyYdSty7ddl}||tdYdSw)NrrN) rirlputKeyboardInterrupt Exception traceback format_excsysexit)funcre error_queue return_queuerjr+resultrpr!r!r" _func_wrappervs   rxc@s&eZdZddZdddZddZdS) MultiprocessContextcCs*||_||_||_ddt|D|_dS)NcSsi|]\}}|j|qSr!)sentinel)rJindexprocessr!r!r" s z0MultiprocessContext.__init__..) error_queues return_queues processes enumerate sentinels)r)rr~rr!r!r"r*s  zMultiprocessContext.__init__NcCst|jdkr dStjj|j|d}d}|D]}|j|}|j|}||j dkr2|}nq|dur>t|jdkS|jD]}| rK| |qA| |dS)NrT)timeout) rOrr> connectionwaitkeysrhrr`exitcodeis_alive terminate_throw_exception)r)rready error_indexrzr{r|r!r!r"r`s*      zMultiprocessContext.joincCs|j|r.|j|j}|dkr#t| j}td|d|dtd|d|d|j|}d|d}||7}t|)NrzProcess z terminated with signal .z terminated with exit code z9 ---------------------------------------------- Process zV terminated with the following error: ---------------------------------------------- ) r~emptyrrsignalSignalsnameror\)r)rrrZoriginal_tracemsgr!r!r"rs" z$MultiprocessContext._throw_exception)N)rrrr*r`rr!r!r!r"rys ryr!TFrtCallable[..., None]re Iterable[Any]rdintr`booldaemonr0Unpack[_SpawnOptions]returnc Kst||dkr t}t||}|dd}|durd}t|}g} g} g} t|D]1} |} |}|jt ||| ||| |dfd}||_ | | | | || |q+t | | | }|sg|S|sp |rk|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') rrNspawnr+)targetre)r2rBrfr\r>Z get_contextr_ SimpleQueueProcessrxrstartrcryr`)rtrerdr`rr0Zprocs_env_listrmpr~rrirurvr|contextr!r!r"rsHy         r)r!rTF)rtrrerrdrr`rrrr0rrry)4 __future__rr>r]rrrr,typingrrrrZ paddle.baserZ paddle.devicerZpaddle.distributed.cloud_utilsr r Z$paddle.distributed.fleet.cloud_utilsr Zpaddle.distributed.fleet.launchr Z%paddle.distributed.fleet.launch_utilsr rrZ%paddle.distributed.utils.launch_utilsrrrZpaddle.frameworkrcollections.abcrrZtyping_extensionsrrr__all__r#r2rBrGrQrfrirlrxryrr!r!r!r"sH       9B