o "j2@sddlZddlZddlZddlZddlmZddlmZddlm Z ddl m Z m Z ddl mZedgdZd Zd Zd Zdadad d ZddZddZddZddZd(ddZddefddZddefddZddZddZd d!Zd"d#Z d$d%Z!d&d'Z"dS))N) namedtuple)core)Node) PythonFunc _serialize)logger WorkerInfo)namerankipportiicCs|adSN_barrier_store)storer[/var/www/html/Deteccion_Ine/venv/lib/python3.10/site-packages/paddle/distributed/rpc/rpc.py_set_barrier_store&srcCsbdSrrrrrr_del_barrier_store+srcCs(tt||||}tt||dSr)pickledumpsrrsetstr)r r r r Z self_inforrr_set_self_info0srcCsXg}t}t|D] }ttt|}|j|vsJd||j| |q |S)Nz:The Worker name must be unique, but name `{}` is repeated.) rrangerloadsrgetrr addappend) world_size all_infossr inforrr_exchange_all_service_infos5s    r$cCs$t}|}|}|d|S)N:)rZ get_host_ipZ get_free_port)noder Z free_portrrr _gen_endpointBsr'cCsX|dur ttjdn|}|durttjdn|}tdd}|dur't}td|d||dur8|ntjd}|d\}}t|}ttd d }tj |||d k||d }t ||d\} } t| } t ||| | t |} g} | D]} t | j| j| j| j}| |qzt|| tt||ttd|d dS)a init rpc. Args: name (str): worker name. rank (int, optional): worker id, default is None. world_size (int, optional): number of workers, default is None. master_endpoint (str, optional): id address of master, other nodes communicate with the master to get the information of all worker nodes, default is None. Returns: None. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle.distributed.rpc as rpc >>> rpc.init_rpc("worker0", rank=0, world_size=1, ... master_endpoint="127.0.0.1:8001") >>> rpc.shutdown() NZPADDLE_TRAINER_IDZPADDLE_TRAINERS_NUMZPADDLE_WORKER_ENDPOINTTrainer z: worker endpoint: ZPADDLE_MASTER_ENDPOINTr%ZFLAGS_stop_check_timeoutZ900r)timeoutz: Init RPC done!)intosenvirongetenvr'rr#splitrZTCPStorerrr$rr r r r rZinit_and_set_agent_instanceZrpc_start_worker_barrier_never_timeoutZrpc_start_client)r r r Zmaster_endpointZworker_endpointZ master_addrZ master_portZstop_check_timeoutrr r r!Zc_infosZ node_infor#rrrinit_rpcIsN    r0cCst|||||}|S)a Make a blocking RPC call to run function ``fn`` on worker ``to``. Attention: Users must use this API in a secure network environment. Args: to (str): name of the destination worker. fn (fn): a callable function, such as Python callables. args (tuple, optional): the argument tuple for the ``fn`` invocation, default is None. kwargs (dict, optional): is a dictionary of keyword arguments for the ``fn`` invocation, default is None. timeout (int, optional): timeout in seconds to use for this RPC. If the RPC does not complete in this amount of time, an exception indicating it has timed out will be raised. A value less than or equal to 0 indicates an infinite timeout, i.e. a timeout error will never be raised. The default value is -1. Returns: Returns the result of running ``fn`` with ``args`` and ``kwargs``. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle.distributed.rpc as rpc >>> def add(a, b): ... return a + b >>> rpc.init_rpc("worker0", rank=0, world_size=1, ... master_endpoint="127.0.0.1:8002") >>> ret = rpc.rpc_sync("worker0", add, args=(2, 3)) >>> rpc.shutdown() ) _invoke_rpcwait)tofnargskwargsr)Zfutrrrrpc_syncs$r7cCst|||||S)a Make a non-blocking RPC call to run function ``fn`` on worker ``to``. Attention: Users must use this API in a secure network environment. Args: to (str): name of the destination worker. fn (fn): a callable function, such as Python callables. args (tuple, optional): the argument tuple for the ``fn`` invocation, default is None. kwargs (dict, optional): is a dictionary of keyword arguments for the ``fn`` invocation, default is None. timeout (int, optional): timeout in seconds to use for this RPC. If the RPC does not complete in this amount of time, an exception indicating it has timed out will be raised. A value less than or equal to 0 indicates an infinite timeout, i.e. a timeout error will never be raised. The default value is -1. Returns: Returns a :class:`FutureWrapper` object that can be waited on. When completed, the return value of ``fn`` on ``args`` and ``kwargs`` can be got by `fut.wait()`. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle.distributed.rpc as rpc >>> def add(a, b): ... return a + b >>> rpc.init_rpc("worker0", rank=0, world_size=1, ... master_endpoint="127.0.0.1:8003") >>> fut = rpc.rpc_async("worker0", add, args=(2, 3)) >>> print(fut.wait()) 5 >>> rpc.shutdown() )r1)r3r4r5r6r)rrr rpc_asyncs)r8cCsR|r|nd}|r |ni}tt|||}|d}|dkrtn|}t|||}|S)Nrir)rr_MAX_RPC_TIMEOUT_MSrZ invoke_rpc)r3r4r5r6r)Z serial_objZ timeout_msfuturerrrr1s  r1cstjtd|dkr dSdttdtd7adk}fdd}|rAfd d td|D}ttdd||dStdg}||ttddS) N)dayszBarrier//rcspt}t|dkr6tdt|}tj|dkr%td|ttdd|}t|dks dSdS)Nrg?)secondsz4Keys {} are not ready sinck rank {} is waiting them.cSstt|dkS)Nr>)r*rr)keyrrrszC_barrier_never_timeout.._check_keys_ready..) timelensleepdatetime timedelta RuntimeErrorformatlistfilter) wait_keys start_timeZ elapse_time) global_rankr)rr_check_keys_readys    z1_barrier_never_timeout.._check_keys_readycsg|]}t|qSr)r).0r )barrier_prefixrr s z*_barrier_never_timeout..)rErF_BARRIER_TIMEOUT_MAX_DAYSr_barrier_countrrr)rMZglobal_world_sizeZ is_masterrNrKr)rPrMr)rr/s   r/cCsDt}|j}tt}t||ttt d|ddS)a+ Perform a shutdown of the RPC agent, stop the worker and destroy the agent. This will block until all local and remote RPC processes reach this method and wait for all outstanding work to complete. Returns: None. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle.distributed.rpc as rpc >>> rpc.init_rpc("worker0", rank=0, world_size=1, ... master_endpoint="127.0.0.1:8004") >>> rpc.shutdown() r(z: rpc shutdown!N) get_current_worker_infor rCget_all_worker_infosr/rZrpc_stop_workerrrr#)r#r r rrrshutdowns  rVcCs t|S)a Get worker information by worker name. Args: name (str): name of the worker. Returns: class `WorkerInfo` with attribute `name`, `rank`, `ip` and `port`. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle.distributed.rpc as rpc >>> import os >>> os.environ["PADDLE_WORKER_ENDPOINT"] = "127.0.0.1:9002" >>> rpc.init_rpc("worker0", rank=0, world_size=1, ... master_endpoint="127.0.0.1:8005") >>> print(rpc.get_worker_info("worker0")) {name: worker0, rank: 0, ip: 127.0.0.1, port: 9002} >>> rpc.shutdown() )rZrpc_get_worker_info)r rrrget_worker_info5s rWcCtS)aY Get all worker informations. Returns: List[WorkerInfo]. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle.distributed.rpc as rpc >>> import os >>> os.environ["PADDLE_WORKER_ENDPOINT"] = "127.0.0.1:9003" >>> rpc.init_rpc("worker0", rank=0, world_size=1, ... master_endpoint="127.0.0.1:8006") >>> print(rpc.get_all_worker_infos()) [{name: worker0, rank: 0, ip: 127.0.0.1, port: 9003}] >>> rpc.shutdown() )rZrpc_get_all_worker_infosrrrrrUSrUcCrX)a Get current worker information. Returns: class `WorkerInfo` with attribute `name`, `rank`, `ip` and `port`. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> import paddle.distributed.rpc as rpc >>> import os >>> os.environ["PADDLE_WORKER_ENDPOINT"] = "127.0.0.1:9004" >>> rpc.init_rpc("worker0", rank=0, world_size=1, ... master_endpoint="127.0.0.1:8007") >>> print(rpc.get_current_worker_info()) {name: worker0, rank: 0, ip: 127.0.0.1, port: 9004} >>> rpc.shutdown() )rZrpc_get_current_worker_inforrrrrTnrYrT)NNN)#rEr+rrB collectionsrZ paddle.baserZ!paddle.distributed.launch.contextrZpaddle.distributed.rpc.internalrrZ%paddle.distributed.utils.launch_utilsrrZ_DEFAULT_RPC_TIMEOUTr9rRrrSrrrr$r'r0r7r8r1r/rVrWrUrTrrrrs:     F(, )