\;j3ddlZddlZddlZddlZddlmZddlmZddlm Z ddl m Z m Z ddl mZedgdZd Zd Zd Zdadad Zd ZdZdZdZddZddefdZddefdZdZdZdZdZ dZ!dZ"y)N) namedtuple)core)Node) PythonFunc _serialize)logger WorkerInfo)namerankipportiic|ayN_barrier_store)stores cG:\00. PROJECTS\API\Inventory\templateJSON\kerjaOCR\Lib\site-packages\paddle/distributed/rpc/rpc.py_set_barrier_storer&sNcbyrrrr_del_barrier_storer+srctjt||||}tj t ||yr)pickledumpsr rsetstr)r r r r self_infos r_set_self_infor 0s/ ZdB=>Is4y),rc"g}t}t|D]t}tjtj t |}|j|vsJd|j|j|j|v|S)Nz:The Worker name must be unique, but name `{}` is repeated.) rrangerloadsrgetrr addappend) world_size all_infossr infos r_exchange_all_service_infosr+5s~I Aj!||N..s4y9: IIQ  H G H  dii " rcdt}|j}|j}|d|S)N:)r get_host_ip get_free_port)noder free_ports r _gen_endpointr2Bs6 6D   B""$IT9+ rc|ttjdn|}|ttjdn|}tjdd}| t }t j d|d|||ntjd}|jd\}}t|}ttjd d }tj|||d k(|| }t||jd\} } t| } t||| | t|} g} | D]S} tj| j| j| j | j"}| j%|Utj&|| tj(t+||tj,t j d|d y)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() NPADDLE_TRAINER_IDPADDLE_TRAINERS_NUMPADDLE_WORKER_ENDPOINTTrainer z: worker endpoint: PADDLE_MASTER_ENDPOINTr-FLAGS_stop_check_timeout900r)timeoutz: Init RPC done!)intosenvirongetenvr2rr*splitrTCPStorerr r+r r r r r r&init_and_set_agent_instancerpc_start_worker_barrier_never_timeoutrpc_start_client)r r r'master_endpointworker_endpoint master_addr master_portstop_check_timeoutrr r r(c_infos node_infor*s rinit_rpcrMIs448<3rzz-. /TD   BJJ, -.  ii 8$?O'/ KK(4& 3O3DEF  &  ZZ0 1 /44S9Kk"KRYY'A5IJ MM  "  Eu$$S)HB t9D4r4(+J7IG  NNINNILL)..  t   $$T734, KK(4& 012rc@t|||||}|jS)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;futs rrpc_syncrVs#H b"dFG 4C 88:rc t|||||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() )rO)rQrRrSrTr;s r rpc_asyncrXsR r2tVW 55rc|r|nd}|r|ni}tt|||}|dz}|dkrtn|}tj|||}|S)Nrir)rr_MAX_RPC_TIMEOUT_MSr invoke_rpc)rQrRrSrTr; serial_obj timeout_msfutures rrOrOsU4RDV2FJr489J4J(2a$ZJ __RZ 8F Mrctjt|dkrydttzdz}tdz adk(}fd}|rPt d|Dcgc]}|t|z}}t j|tdzd||y|tdzg}||t j|tzdycc}w)N)dayszBarrier//rcVtj}t|dkDrtjdtj|z }tj|kDrt dj |ttd|}t|dkDryy)Nrg?)secondsz4Keys {} are not ready sinck rank {} is waiting them.cDttj|dk7S)Nrc)r<rr$)keys rzC_barrier_never_timeout.._check_keys_ready..s3~'9'9#'>#?1#Dr) timelensleepdatetime timedelta RuntimeErrorformatlistfilter) wait_keys start_time elapse_time global_rankr;s r_check_keys_readyz1_barrier_never_timeout.._check_keys_readysYY[ )nq JJsO))+ 2K!!+6@"JQQ!; DiPI)nq r)rlrm_BARRIER_TIMEOUT_MAX_DAYSr_barrier_countr"rr%)ruglobal_world_sizebarrier_prefix is_masterrvr rrr;s` @rrDrDs  &?@G1 #n"55;NaNq I 49CF2A6)$#c!f,- )$>C ,<>> # 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() r7z: rpc shutdown!N) get_current_worker_infor rjget_all_worker_infosrDrrpc_stop_workerrrr*)r*r r's rshutdownrsT* # $D 99D)+,J4, KK(4&01rc,tj|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() )rrpc_get_worker_info)r s rget_worker_infor5s6  # #D ))rc*tjS)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() )rrpc_get_all_worker_infosrrrr~r~Ss0  ( ( **rc*tjS)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() )rrpc_get_current_worker_inforrrr}r}ns0  + + --r)NNN)#rlr=rri collectionsr paddle.baser!paddle.distributed.launch.contextrpaddle.distributed.rpc.internalrr%paddle.distributed.utils.launch_utilsrr _DEFAULT_RPC_TIMEOUTrZrwrrxrrr r+r2rMrVrXrOrDrrr~r}rrrrs "2B8  &D E  $  - C3Lt5I%P 6J)6X&AR2>*<+6.r