U ʹj<+@sddlZddlZddlZddlZddlZddlZddlZddlmZddl Z ddl m Z ddl mZdddddgZeaejZejZGd ddeZGd d d ZeZd dZd dZddZddZddZddZeddddgZ edddgZ!dS)N)Enum)_get_current_rpc_agent RPCExecMode serialize deserialize PythonUDFRemoteExceptionc@seZdZdZdZdZdZdS)rsyncasyncZ async_jitremoteN)__name__ __module__ __qualname__ZSYNCASYNCZ ASYNC_JITZREMOTErr]/var/www/html/bwave/python/venv/lib/python3.8/site-packages/torch/distributed/rpc/internal.pyrsc@steZdZdZddZddZeddZdd Zed d Z d d Z ddZ eddZ ddZ ddZddZdS)_InternalRPCPicklera  This class provides serialize() and deserialize() interfaces to serialize data to be "binary string + tensor table" format So for RPC python UDF function and args, non tensor data will be serialized into regular binary string, tensor data will be put into thread local tensor tables, this serialization format is consistent with builtin operator and args using JIT pickler. This format will make tensor handling in C++ much easier, e.g. attach tensor to distributed autograd graph in C++ cCs$tj|_|j|jtj<i|_dSN)copyregdispatch_tablecopy_dispatch_table_tensor_reducertorchZTensor_class_reducer_dict)selfrrr__init__)s z_InternalRPCPickler.__init__cCs||jkr||j|<dSr)r)rZ obj_classZreducerrrr_register_reducer0s z%_InternalRPCPickler._register_reducercCs tj|Sr)_thread_local_tensor_tables recv_tables)cls tensor_indexrrr_tensor_receiver5sz$_InternalRPCPickler._tensor_receivercCs&tj|ttjd}tj|ffS)N)r send_tablesappendlenrr")rZtensorr!rrrr:s z#_InternalRPCPickler._tensor_reducercCstjj|Sr)distrpcPyRRefZ _deserialize)r rref_fork_datarrr_py_rref_receiver@sz%_InternalRPCPickler._py_rref_receivercCs|}tj|ffSr)Z _serializerr+)rZpy_rrefr*rrr_py_rref_reducerDsz$_InternalRPCPickler._py_rref_reducercCs ||Sr)r,)rZrrefrrr _rref_reducerHsz!_InternalRPCPickler._rref_reducercCst|}tj|}|S)z Given a serialized representation of a ScriptModule created with torch.jit.save, loads and returns the ScriptModule. )ioBytesIOrjitload)r Zscript_module_serializedfmrrr_script_module_receiverKs  z+_InternalRPCPickler._script_module_receivercCs&t}tj||tj|ffS)z, Serializes a ScriptModule. )r.r/rr0saverr4getvalue)rZ script_moduler2rrr_script_module_reducerUsz*_InternalRPCPickler._script_module_reducercCst}t|}|j|_|j|jtjj<|j |jtjj <t |t j jrT|j|j|j<|jD]}|j||j|<q^ttdrtj}nd}gt_||tj}|dk r|t_nt`||fS)ze Serialize non tensor data into binary string, tensor data into tensor table r$N)r.r/_picklerrrr,r'r(r)r-ZRRef isinstancerr0Z ScriptModuler7 __class__rkeyshasattrrr$dumpr6)robjr2p class_nameZold_send_tablesZtensorsrrrr]s&   z_InternalRPCPickler.serializec Csttdrtj}nd}|t_ztt|}|}Wn<tk rr}zt|d}t|}||_ W5d}~XYnX|dk r|t_nt`|S)zJ Deserialize binary string + tensor table to original obj rNz Default RPC pickler does not serialize function code. Ensure that UDFs are defined on both caller and callee modules.) r<rr _unpicklerr.r/r1AttributeErrorstr __cause__)r binary_data tensor_tableZold_recv_tablesZ unpicklerrete except_strrrrrs$  z_InternalRPCPickler.deserializeN)r r r__doc__rr classmethodr"rr+r,r-r4r7rrrrrrrs    5rcCs t|Sr)_internal_rpc_picklerr)r>rrrrscCs t||Sr)rLr)rErFrrrrsc Csz"t|tr||j|j|j}Wnbtk r}zDdtdt|dt }t |t j dt|t|}W5d}~XYnX|S)z This function is exclusively called from C++. See ``torch/csrc/distributed/rpc/python_rpc_handler.cpp``. Runs a Python UDF and returns its return value. Wraps any exception in ``RemoteException`` if the function raises. zOn z:  )fileN)r9rBfuncargskwargs ExceptionrZget_worker_inforepr traceback format_excprintsysstderrrtype)Z python_udfresultrHrIrrr _run_functions " r[c Cs|t|trx|jdd}d}z||}Wn<tk rj}ztdt|d||W5d}~XYnX|dk rx|dS)Nzutf-8Zunicode_escapez8Failed to create original exception type. Error msg was z' Original exception on remote side was ) r9rmsgencodedecodeexception_type BaseException RuntimeErrorrC)rZZ exception_msgexcrHrrr_handle_exceptions rcc Cs$d|jd|d|d|d }|S)a Builds the key that RPC calls are profiled with using the autograd profiler. This will be the name of the corresponding Event recorded in the profiler. Args: exec_type (RPCExecMode): Type of RPC/RRef call func_name (str): Name of function being profiled. current_worker_name (str): Name of current worker. dst_worker_name (str): Name of the destination worker. Returns: String representing profiling key rpc_#( -> ))value) exec_type func_namecurrent_worker_nameZdst_worker_name profile_keyrrr_build_rpc_profiling_keys rnc CsRtjstdd|jdt|d|d|d }tj}tj|||S)ar This function should be called from RPC/RRef functions to create a RecordFunction object for profiling. This function also runs the before callbacks that start the profiling, though the user is responsible for running the appropriate callbacks when the function to be profiled finishes. Args: exec_type (RPCExecMode): Type of RPC/RRef call func_name (str): Name of function being profiled. current_worker_name (str): Name of current worker. dest_worker_name (str): Name of the destination worker. Returns: An instance of `torch.autograd._RecordFunction`. z$Autograd profiler should be enabled.rdrerfrgrh)rZautogradZ_profiler_enabledAssertionErrorrirCZ_RecordFunctionZ_run_before_callbacks)rjrkrlZdest_worker_namermrfrrr_start_record_functions $ rqrOrPrQr\r_)" collectionsrr.picklerW threadingrTenumrrZtorch.distributed distributedr'Ztorch._C._distributed_rpcr__all__localrPicklerr8 UnpicklerrArrrLrrr[rcrnrq namedtuplerrrrrrs4