/usr/local/lib64/python3.6/site-packages/torch/distributed/rpc/__pycache__
Edit: /usr/local/lib64/python3.6/site-packages/torch/distributed/rpc/__pycache__/api.cpython-36.pyc (26470B)
3
Eg@
@ s$ U d dl Z d dlZd dlZd dlZd dlZd dlZd dlmZmZm Z m
Z
m Z d dl Z d dl
mZ d dlmZmZmZmZmZmZmZmZmZmZmZmZm Z m Z m Z m Z m Z ddl!m"Z"m#Z#m$Z$m%Z% ddl&m'Z'm(Z( ej)e*Z+dZ,e$a-ej.d d
Z/d d Z0G d
d de1Z2e3 a4e
e 4ej5 Z6i Z7ee8e9f 7e j:e2Z;e j:;dd Z
ej? Z@ej.dd ZAe0de(fddZBe0dd ZCe0dd ZDe0dCd d ZEd d ZFe0dDd!d"ZGd#d$ ZHdEd%d&ZIe(dfd'd(ZJe d)ZKeeK ZLyG d*d+ d+eeeK ZMW nB eNk
rN G d,d- d-ejOeLjOZPG d.d+ d+eeLePd/ZMY nX d0d1 ZQxrejReD ]d\ZSZTeSjUd2reSd3krqdeVeTd4dZWeWdk stXd5eWjYd6d7ZWeQeSeWZZe[eMeSeZ qdW e0dde(fd8d9Z\dde(fd:d;Z]e0dde(fdd?Z_d@dA Z`dS )F N)DictGenericTypeVarSetAny)Future)PyRRefRemoteProfilerManager
WorkerInfoget_rpc_timeout_cleanup_python_rpc_handler)_delete_all_user_and_unforked_owner_rrefs_destroy_rref_context_get_current_rpc_agent_invoke_remote_builtin_invoke_remote_python_udf_invoke_remote_torchscript_invoke_rpc_builtin_invoke_rpc_python_udf_invoke_rpc_torchscript_is_current_rpc_agent_set_reset_current_rpc_agent_set_and_start_rpc_agent ) PythonUDF RPCExecMode_internal_rpc_pickler_build_rpc_profiling_key)DEFAULT_SHUTDOWN_TIMEOUTUNSET_RPC_TIMEOUTTc
c s | a z
dV W dta X dS )zX
rpc_pickler: (.internal._InternalRPCPickler) Overrides the default RPC pickler
N)_default_picklerr )Z rpc_pickler r! E/usr/local/lib64/python3.6/site-packages/torch/distributed/rpc/api.py_use_rpc_pickler8 s
r# c s t j fdd}|S )Nc s t std | |S )NzHRPC has not been initialized. Call torch.distributed.rpc.init_rpc first.)r RuntimeError)argskwargs)funcr! r" wrapperF s z%_require_initialized..wrapper) functoolswraps)r' r( r! )r' r" _require_initializedE s r+ c @ s e Zd Zdd ZdS )AllGatherStatesc C s i | _ tj | _d S )N)gathered_objects threadingEventproceed_signal)selfr! r! r" __init__S s
zAllGatherStates.__init__N)__name__
__module__ __qualname__r2 r! r! r! r" r, R s r, c C s( | j }dd |D at s$t| d S )Nc S s h | ]
}|j qS r! )name).0Z worker_infor! r! r" n s z#_init_rpc_states..)Zget_worker_infos_ALL_WORKER_NAMESr r )ZagentZ worker_infosr! r! r" _init_rpc_statesk s r: c C s~ t p |s$t}||ks$t| dt| }||jksJt| d| d||j|< |t|jj krp|jj W d Q R X d S )Nz is not expected by leader.z reported intent sequence id z twice. )_all_gather_dict_lockr9 AssertionError!_all_gather_sequence_id_to_statesr- setkeysr0 ) sequence_id worker_nameobj worker_namesstatesr! r! r" _gather_to_leaderu s
rE c
C sF t t| }W d Q R X |jj s2tdj| ||_|jj d S )Nz0Termination signal sequence id {} got set twice.)r; r= r0 is_setr< formatr- r> )r@ Z objects_maprD r! r! r" _broadcast_to_followers s rH c c s2 g t _z
dV W dztjjt j W dt `X X dS )aO
A context manager that collects all futures returned by ``rpc_async`` and
waits them on the context manager's exit; relieving the user of needing
to explicitly call wait.
Example::
>>> # On worker 0:
>>> import torch
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker0", rank=0, world_size=2)
>>> with rpc._wait_all():
>>> fut_1 = rpc.rpc_async(dst, torch.add, (torch.ones(2, 2), 1))
>>> fut_2 = rpc.rpc_async(dst, torch.add, (torch.ones(2, 2), 1))
>>> #fut_1 and fut_2 are waited on
N)_thread_local_var future_listtorchfuturesZwait_allr! r! r! r" _wait_all s
rM c C s |st dk stdt }t|d }t j j}t8 djt|}tj |d}|d t|< |t
| }W dQ R X ||k}|t krt }|rt
||| | nt|t
||| |f|d t t| } W dQ R X | jj |rt }
x0||h D ]"} t| t|| jf|d} | |
| < qW g }
xV|
j D ]J\} } y | j W n2 tk
rh } z|
j| |f W Y dd}~X nX q"W |
rtddd |
D d
|d d |
d d t tj|} W dQ R X | jS )
a
This is similar to torch.distributed.all_gather(), but is using RPC. It
picks the worker with the smallest name (alphabetic order) as the leader.
Then all followers send their data ``obj`` to the leader. After the leader
has received all, it will broadcast the results back to all followers. This
function blocks until all workers have received the gathered results.
Nz=`_ALL_WORKER_NAMES` is not initialized for `def _all_gather`.r r )r% timeoutz
Followers c S s g | ] }|d qS )r r! )r7 er! r! r"
s z _all_gather..z timed out in _all_gather after z.2fz! seconds. The first exception is )r9 r< sortedr get_worker_infor6 r; join_all_gather_sequence_idgetstrr r rE rpc_syncr= r0 waitdict rpc_asyncrH r- itemsr$ appendpop)rB rC rO Z leader_nameZ self_nameZ concat_namesZ sequence_numr@ Z is_leaderrD Z#worker_name_to_response_future_dictZ
follower_namefuterrorsexr! r! r" _all_gather sZ
&,rb c C sJ yt dt| W n2 tk
rD } ztjd| W Y dd}~X nX dS )a&
Synchronizes local and remote RPC processes.
This will block until all local and remote RPC processes specified under worker_names
reach this method to wait for all outstanding work to complete.
Args:
worker_names (List[str]): The set of workers to synchronize.
Nz&Failed to complete barrier, got error )rb r> r$ loggererror)rC ra r! r! r" _barrier s
re c C sL yt dtd W n6 tk
rF } ztjd| | W Y dd} ~ X nX dS )ag
Block until all local and remote RPC processes reach this method and wait
for all outstanding work to complete. Every RPC process must call this
method before exit to perform a graceful shutdown. This should be used to
terminate the RPC framework, and there is no guarantee that the RPC
framework will work after this method returns.
N)rO z;Failed to respond to 'Shutdown Proceed' in time, got error )rb r r$ rc rd )ra r! r! r" _wait_all_workers s rf c
C s8 | r.z t t t jdd W dt X nt dS )a
Perform a shutdown of the RPC agent, and then destroy the RPC agent. This
stops the local agent from accepting outstanding requests, and shuts
down the RPC framework by terminating all RPC threads. If ``graceful=True``,
this will block until all local and remote RPC processes reach this method
and wait for all outstanding work to complete. Otherwise, if
``graceful=False``, this is a local shutdown, and it does not wait for other
RPC processes to reach this method.
.. warning::
For :class:`~torch.futures.Future` objects returned by
:meth:`~torch.distributed.rpc.rpc_async`, ``future.wait()`` should not
be called after ``shutdown()``.
Args:
graceful (bool): Whether to do a graceful shutdown or not. If True,
this will 1) wait until there is no pending system
messages for ``UserRRefs`` and delete them; 2) block
until all local and remote RPC processes have reached
this method and wait for all outstanding work to
complete.
Example::
Make sure that ``MASTER_ADDR`` and ``MASTER_PORT`` are set properly
on both workers. Refer to :meth:`~torch.distributed.init_process_group`
API for more details. For example,
>>> export MASTER_ADDR=localhost
>>> export MASTER_PORT=5678
Then run the following code in two different processes:
>>> # On worker 0:
>>> import torch
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker0", rank=0, world_size=2)
>>> # do some work
>>> result = rpc.rpc_sync("worker1", torch.add, args=(torch.ones(1), 1))
>>> # ready to shutdown
>>> rpc.shutdown()
>>> # On worker 1:
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker1", rank=1, world_size=2)
>>> # wait for worker 0 to finish work, and then shutdown.
>>> rpc.shutdown()
T)shutdownN)rf r
r rT _finalize_shutdown)Zgracefulr! r! r" rg s 1
rg c
C s* z t t W d t j t t X d S )N)r _ignore_rref_leakr rg r r r! r! r! r" rh Y s
rh c C s" | dk rt j| S t j S dS )aI
Get :class:`~torch.distributed.rpc.WorkerInfo` of a given worker name.
Use this :class:`~torch.distributed.rpc.WorkerInfo` to avoid passing an
expensive string on every invocation.
Args:
worker_name (str): the string name of a worker. If ``None``, return the
the id of the current worker. (default ``None``)
Returns:
:class:`~torch.distributed.rpc.WorkerInfo` instance for the given
``worker_name`` or :class:`~torch.distributed.rpc.WorkerInfo` of the
current worker if ``worker_name`` is ``None``.
N)r rS )rA r! r! r" rS m s rS c C s< t | tr| S t | ts"t | tr*t| S tdj| d S )Nz"Cannot get WorkerInfo from name {})
isinstancer
rW intrS
ValueErrorrG )tor! r! r" _to_worker_info s
rn c C s0 t | j }|r|S tt }|j| |S d S )N)typeZ local_valuer
set_result)rrefblockingZ rref_typefuturer! r! r" _rref_typeof_on_owner s
rt c C s* t | j t| f|d}|r"|j S |S d S )N)r% rO )r[ ownerrt rY )rq rO rr r_ r! r! r" _rref_typeof_on_user s rv Tc @ s e Zd ZdS )RRefN)r3 r4 r5 r! r! r! r" rx s rx c @ s e Zd ZdS )RRefMetaN)r3 r4 r5 r! r! r! r" ry s ry c @ s e Zd ZdS )rx N)r3 r4 r5 r! r! r! r" rx s ) metaclassc s fdd}||_ |S )Nc s t tt| ||S )N)getattrsuperrx )r1 r% r& ) method_namer! r" method s z method_factory..method)__doc__)r} docstringr~ r! )r} r" method_factory s r ___str__r z4RRef user-facing methods should all have docstrings.z torch.distributed.rpc.PyRRefztorch.distributed.rpc.RRefc
C s2 t jjj|}t| }t jj }t|||tj |}|} |r@|nf }|rL|ni }t
|d}
|
rv|j } t | t jj
rv| }|dk rt|||f||} nVt |t jj
rt|jt jj|||
f||} n&tjt|||\}
}t||
|||
} |r$t jj st| dk s t| j| j }| j| W dQ R X | S )a4
Make a remote call to run ``func`` on worker ``to`` and return an
:class:`~torch.distributed.rpc.RRef` to the result value immediately.
Worker ``to`` will be the owner of the returned
:class:`~torch.distributed.rpc.RRef`, and the worker calling ``remote`` is
a user. The owner manages the global reference count of its
:class:`~torch.distributed.rpc.RRef`, and the owner
:class:`~torch.distributed.rpc.RRef` is only destructed when globally there
are no living references to it.
Args:
to (str or WorkerInfo or int): name/rank/``WorkerInfo`` of the destination worker.
func (callable): a callable function, such as Python callables, builtin
operators (e.g. :meth:`~torch.add`) and annotated
TorchScript functions.
args (tuple): the argument tuple for the ``func`` invocation.
kwargs (dict): is a dictionary of keyword arguments for the ``func``
invocation.
timeout (float, optional): timeout in seconds for this remote call. If the
creation of this
:class:`~torch.distributed.rpc.RRef` on worker
``to`` is not successfully processed on this
worker within this timeout, then the next time
there is an attempt to use the RRef (such as
``to_here()``), a timeout will be raised
indicating this failure. A value of 0 indicates
an infinite timeout, i.e. a timeout error will
never be raised. If not provided, the default
value set during initialization or with
``_set_rpc_timeout`` is used.
Returns:
A user :class:`~torch.distributed.rpc.RRef` instance to the result
value. Use the blocking API :meth:`torch.distributed.rpc.RRef.to_here`
to retrieve the result value locally.
.. warning ::
The ``remote`` API does not copy storages of argument tensors until
sending them over the wire, which could be done by a different thread
depending on the RPC backend type. The caller should make sure that the
contents of those tensors stay intact until the returned RRef is
confirmed by the owner, which can be checked using the
:meth:`torch.distributed.rpc.RRef.confirmed_by_owner` API.
.. warning ::
Errors such as timeouts for the ``remote`` API are handled on a
best-effort basis. This means that when remote calls initiated by
``remote`` fail, such as with a timeout error, we take a best-effort
approach to error handling. This means that errors are handled and set
on the resulting RRef on an asynchronous basis. If the RRef has not been
used by the application before this handling (such as ``to_here`` or
fork call), then future uses of the ``RRef`` will appropriately raise
errors. However, it is possible that the user application will use the
``RRef`` before the errors are handled. In this case, errors may not be
raised as they have not yet been handled.
Example::
Make sure that ``MASTER_ADDR`` and ``MASTER_PORT`` are set properly
on both workers. Refer to :meth:`~torch.distributed.init_process_group`
API for more details. For example,
>>> export MASTER_ADDR=localhost
>>> export MASTER_PORT=5678
Then run the following code in two different processes:
>>> # On worker 0:
>>> import torch
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker0", rank=0, world_size=2)
>>> rref1 = rpc.remote("worker1", torch.add, args=(torch.ones(2), 3))
>>> rref2 = rpc.remote("worker1", torch.add, args=(torch.ones(2), 1))
>>> x = rref1.to_here() + rref2.to_here()
>>> rpc.shutdown()
>>> # On worker 1:
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker1", rank=1, world_size=2)
>>> rpc.shutdown()
Below is an example of running a TorchScript function using RPC.
>>> # On both workers:
>>> @torch.jit.script
>>> def my_script_add(t1, t2):
>>> return torch.add(t1, t2)
>>> # On worker 0:
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker0", rank=0, world_size=2)
>>> rref = rpc.remote("worker1", my_script_add, args=(torch.ones(2), 3))
>>> rref.to_here()
>>> rpc.shutdown()
>>> # On worker 1:
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker1", rank=1, world_size=2)
>>> rpc.shutdown()
_wrapped_async_rpc_functionN)rK jit _builtins
_find_builtinrn autograd_profiler_enabled_enable_rpc_profilerr ZREMOTEhasattrr rj ScriptFunctionr r r6
_jit_internal_qualified_namer serializer r r< _call_end_callbacks_on_futureZ _get_futureZ_set_profiling_future)rm r' r% r& rO qualified_namedst_worker_infoshould_profile ctx_managerrf
is_async_execwrappedrq pickled_python_udftensorsr_ r! r! r" remote sH f
r c C s0 t |stdtjjj|}t| }tjj }t |||||} | }
|rN|nf }|rZ|ni }t
|d} | r|j } t | tjj
r| }|d k rt|||f||}
nRt |tjj
rt|jtjj||||| }
n&tjt|||\}}t||||| }
|r"tjj s
t|
d k st|
j|
}
W d Q R X |
S )Nz function should be callable.r )callable TypeErrorrK r r r rn r r r r r rj r r r r6 r r r r r r r< r )rm r' rpc_typer% r& Z rpc_timeoutr r r r r r r r_ r r r! r! r" _invoke_rpcz sT
r c C s t | |tj|||}|j S )a
Make a blocking RPC call to run function ``func`` on worker ``to``. RPC
messages are sent and received in parallel to execution of Python code. This
method is thread-safe.
Args:
to (str or WorkerInfo or int): name/rank/``WorkerInfo`` of the destination worker.
func (callable): a callable function, such as Python callables, builtin
operators (e.g. :meth:`~torch.add`) and annotated
TorchScript functions.
args (tuple): the argument tuple for the ``func`` invocation.
kwargs (dict): is a dictionary of keyword arguments for the ``func``
invocation.
timeout (float, 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 of 0
indicates an infinite timeout, i.e. a timeout
error will never be raised. If not provided,
the default value set during initialization
or with ``_set_rpc_timeout`` is used.
Returns:
Returns the result of running ``func`` with ``args`` and ``kwargs``.
Example::
Make sure that ``MASTER_ADDR`` and ``MASTER_PORT`` are set properly
on both workers. Refer to :meth:`~torch.distributed.init_process_group`
API for more details. For example,
>>> export MASTER_ADDR=localhost
>>> export MASTER_PORT=5678
Then run the following code in two different processes:
>>> # On worker 0:
>>> import torch
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker0", rank=0, world_size=2)
>>> ret = rpc.rpc_sync("worker1", torch.add, args=(torch.ones(2), 3))
>>> rpc.shutdown()
>>> # On worker 1:
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker1", rank=1, world_size=2)
>>> rpc.shutdown()
Below is an example of running a TorchScript function using RPC.
>>> # On both workers:
>>> @torch.jit.script
>>> def my_script_add(t1, t2):
>>> return torch.add(t1, t2)
>>> # On worker 0:
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker0", rank=0, world_size=2)
>>> ret = rpc.rpc_sync("worker1", my_script_add, args=(torch.ones(2), 3))
>>> rpc.shutdown()
>>> # On worker 1:
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker1", rank=1, world_size=2)
>>> rpc.shutdown()
)r r ZSYNCrY )rm r' r% r& rO r_ r! r! r" rX s DrX c C s. t | |tj|||}ttdr*tjj| |S )a
Make a non-blocking RPC call to run function ``func`` on worker ``to``. RPC
messages are sent and received in parallel to execution of Python code. This
method is thread-safe. This method will immediately return a
:class:`~torch.futures.Future` that can be awaited on.
Args:
to (str or WorkerInfo or int): name/rank/``WorkerInfo`` of the destination worker.
func (callable): a callable function, such as Python callables, builtin
operators (e.g. :meth:`~torch.add`) and annotated
TorchScript functions.
args (tuple): the argument tuple for the ``func`` invocation.
kwargs (dict): is a dictionary of keyword arguments for the ``func``
invocation.
timeout (float, 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 of 0
indicates an infinite timeout, i.e. a timeout
error will never be raised. If not provided,
the default value set during initialization
or with ``_set_rpc_timeout`` is used.
Returns:
Returns a :class:`~torch.futures.Future` object that can be waited
on. When completed, the return value of ``func`` on ``args`` and
``kwargs`` can be retrieved from the :class:`~torch.futures.Future`
object.
.. warning ::
Using GPU tensors as arguments or return values of ``func`` is not
supported since we don't support sending GPU tensors over the wire. You
need to explicitly copy GPU tensors to CPU before using them as
arguments or return values of ``func``.
.. warning ::
The ``rpc_async`` API does not copy storages of argument tensors until
sending them over the wire, which could be done by a different thread
depending on the RPC backend type. The caller should make sure that the
contents of those tensors stay intact until the returned
:class:`~torch.futures.Future` completes.
Example::
Make sure that ``MASTER_ADDR`` and ``MASTER_PORT`` are set properly
on both workers. Refer to :meth:`~torch.distributed.init_process_group`
API for more details. For example,
>>> export MASTER_ADDR=localhost
>>> export MASTER_PORT=5678
Then run the following code in two different processes:
>>> # On worker 0:
>>> import torch
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker0", rank=0, world_size=2)
>>> fut1 = rpc.rpc_async("worker1", torch.add, args=(torch.ones(2), 3))
>>> fut2 = rpc.rpc_async("worker1", min, args=(1, 2))
>>> result = fut1.wait() + fut2.wait()
>>> rpc.shutdown()
>>> # On worker 1:
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker1", rank=1, world_size=2)
>>> rpc.shutdown()
Below is an example of running a TorchScript function using RPC.
>>> # On both workers:
>>> @torch.jit.script
>>> def my_script_add(t1, t2):
>>> return torch.add(t1, t2)
>>> # On worker 0:
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker0", rank=0, world_size=2)
>>> fut = rpc.rpc_async("worker1", my_script_add, args=(torch.ones(2), 3))
>>> ret = fut.wait()
>>> rpc.shutdown()
>>> # On worker 1:
>>> import torch.distributed.rpc as rpc
>>> rpc.init_rpc("worker1", rank=1, world_size=2)
>>> rpc.shutdown()
rJ )r r ASYNCr rI rJ r] )rm r' r% r& rO r_ r! r! r" r[ s X
r[ c C sj t j }| rf|d kr6t|tjjr.tjj|n|j}n|}t ||t
j |j }t j
| tjjj|}|S )N)
contextlibsuppressrj rK r r r r r5 r rS r6 r Zset_current_profiling_keyr ZprofilerZrecord_function)r r r' r r r func_nameZrpc_profiling_keyr! r! r" r ^ s
r )N)T)N)T)a collectionsr r) inspectloggingr. typingr r r r r rK Z
torch.futuresr Ztorch._C._distributed_rpcr r r
r r r
r r r r r r r r r r r Zinternalr r r r constantsr r getLoggerr3 rc ri r contextmanagerr# r+ objectr, r> r9 RLockr; rU rW rk defaultdictr= r: rE rH localrI rM rb re rf rg rh rS rn rt rv rw ZGenericWithOneTypeVarrx r __class__ry r
getmembersr} r~
startswithr{ r r< replaceZ
new_methodsetattrr r rX r[ r r! r! r! r" s L
K ;
>G]