BVhJdZddlmZmZmZddlmcmcmcm cm Z ddl m Z ddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZddlmZ ddl!m"Z"ddl#m$Z$dejJfdZ&dejJfdZ'e$dgGdde(Z)e$dgGdde(Z*Gdde)Z+Gdd e*Z,Gd!d"e(Z-y)#z(Module to expose RPC APIs in tensorflow.)OptionalSequenceUnionN)tf_rpc_service_pb2)context) def_function)function) constant_op)dtypes)errors) none_tensor) type_spec)math_ops)resource_variable_ops)nested_structure_coder)core)nest) tf_exportfuncctjtj|j}t j |}|jSN)r map_structurertype_spec_from_valuestructured_outputsrencode_structureSerializeToString)r output_specsoutput_specs_protos e/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/distribute/experimental/rpc/rpc_ops.pyget_output_specs_from_functionr %sC##I$B$B$($;$;=,->>|L  - - //cj|j\}}tj|}|jSr)structured_input_signaturerrr)r arg_specs_arg_specs_protos rget_input_specs_from_functionr',s000,)Q*;;IF/  * * ,,r!z"distribute.experimental.rpc.Server)v1cheZdZdZedZdedeeje jffdZ dZ y)ServerzA Server base class for accepting RPCs for registered tf.functions. Functions can be registered on the server and are exposed via RPCs. c:|dk7r tdt|S)aCreate TF RPC server at given address. Args: rpc_layer: Communication layer between client and server. Only "grpc" rpc layer is supported at the moment. address: Address where RPC server is hosted. Returns: An instance of `tf.distribute.experimental.rpc.Server` class. Raises: A ValueError if rpc_layer other than "grpc" is used. Only GRPC is supported at the moment. Example usage: >>> import portpicker >>> @tf.function(input_signature=[ ... tf.TensorSpec([], tf.int32), ... tf.TensorSpec([], tf.int32)]) ... def remote_fn(a, b): ... return tf.add(a, b) >>> port = portpicker.pick_unused_port() >>> address = "localhost:{}".format(port) >>> server = tf.distribute.experimental.rpc.Server.create("grpc", address) >>> server.register("addition", remote_fn) >>> server.start() grpc-Only GRPC backend is supported at the moment.)address) ValueError GrpcServer) rpc_layerr.s rcreatez Server.create9s$@F F GG g &&r! method_namerctd)aMethod for registering tf.function on server. Registered methods can be invoked remotely from clients. Args: method_name: Name of the tf.function. Clients use this method_name to make RPCs. func: A `tf.function` or ConcreteFunction to register. GPlease use create_server method to create aconcrete subclass of Server.NotImplementedError)selfr3rs rregisterzServer.register]s = >>r!ctd)zvStarts the RPC server on provided address. Server listens for new requests from client, once it is started. r5r6r8s rstartz Server.startls = >>r!N)__name__ __module__ __qualname____doc__ staticmethodr2strrrFunction tf_functionConcreteFunctionr9r<r!rr*r*2sP !'!'F ># ><00&7789 >>r!r*z"distribute.experimental.rpc.ClientcZeZdZdZeddZ ddedeee jfdZ y) Clientz-Client class for invoking RPCs to the server.cr|dk7r tdtjrd}nd}t||||S)a Create TF RPC client to connect to the given address. Args: rpc_layer: Communication layer between client and server. Only "grpc" rpc layer is supported at the moment. address: Address of the server to connect the RPC client to. name: Name of the RPC Client. You can create multiple clients connecting to same server and distinguish them using different names. timeout_in_ms: The default timeout to use for outgoing RPCs from client. 0 indicates no timeout. Exceeding timeout during RPC will raise DeadlineExceeded error. Returns: An instance of `tf.distribute.experimental.rpc.Client` with the following dynamically added methods for eagerly created clients: * `Registered methods` e.g. multiply(**args): If Client is created when executing eagerly, client will request the list of registered methods from server during client creation. The convenience methods for RPCs will be dynamically added to the created Client instance. For example, when a server has method "multiply" registered, the client object created in eager mode will have 'multiply' method available. Users can use client.multiply(..) to make RPC, instead of client.call("multiply", ...) Both "call" and "multiply" methods are non-blocking i.e. they return a StatusOrResult object which should be used to wait for getting value or error. Along with the above, blocking versions of the registered methods are also dynamically added to client instance. e.g. multiply_blocking(**args). These methods block till the RPC is finished and return response for successful RPC. Otherwise raise exception. These methods are not available when Client is created inside a tf.function. Raises: A ValueError if rpc_layer other than "grpc" is used. Only GRPC is supported at the moment. A DeadlineExceeded exception in eager mode if timeout exceeds while creating and listing client methods. Example usage: >>> # Have server already started. >>> import portpicker >>> @tf.function(input_signature=[ ... tf.TensorSpec([], tf.int32), ... tf.TensorSpec([], tf.int32)]) ... def remote_fn(a, b): ... return tf.add(a, b) >>> port = portpicker.pick_unused_port() >>> address = "localhost:{}".format(port) >>> server = tf.distribute.experimental.rpc.Server.create("grpc", address) >>> server.register("addition", remote_fn) >>> server.start() >>> # Start client >>> client = tf.distribute.experimental.rpc.Client.create("grpc", ... address=address, name="test_client") >>> a = tf.constant(2, dtype=tf.int32) >>> b = tf.constant(3, dtype=tf.int32) >>> result = client.call( ... args=[a, b], ... method_name="addition", ... output_specs=tf.TensorSpec((), tf.int32)) >>> if result.is_ok(): ... result.get_value() >>> result = client.addition(a, b) >>> if result.is_ok(): ... result.get_value() >>> value = client.addition_blocking(a, b) r,r-TF)r.namelist_registered_methods timeout_in_ms)r/rexecuting_eagerly GrpcClient)r1r.rJrLrKs rr2z Client.createysJhF F GG  " $ %   7#  %%r!Nr3argsctd)aMethod for making RPC calls to remote server. This invokes RPC to the server, executing the registered method_name remotely. Args: method_name: Remote registered method to invoke args: List of arguments for the registered method. output_specs: Output specs for the output from method. For example, if tf.function is: @tf.function(input_signature=[ tf.TensorSpec([], tf.int32), tf.TensorSpec([], tf.int32) ]) def multiply_fn(a, b): return tf.math.multiply(a, b) output_spec is: tf.TensorSpec((), tf.int32) If you have access to TF Function, the output specs can be generated from tf.function by calling: output_specs = tf.nest.map_structure(tf.type_spec_from_value, tf_function.get_concrete_function().structured_outputs If output_specs are not provided, flattened list of tensors will be returned in response. timeout_in_ms: Timeout for this call. If 0, default client timeout will be used. Returns: An instance of `StatusOrResult` class with the following available methods. * `is_ok()`: Returns True of RPC was successful. * `get_error()`: Returns TF error_code and error message for the RPC. * `get_value()`: Returns the returned value from remote TF function execution when RPC is successful. Calling any of the above methods will block till RPC is completed and result is available. z)Must be implemented in inherited classes.r6)r8r3rOrrLs rcallz Client.callsP I JJr!)rNNr) r=r>r?r@rAr2rBrr core_tf_typesTensorrQrFr!rrHrHusN5]%]%B=A (K(K(=#7#789(Kr!rHcdeZdZdZdefdZdedeeje jffdZ dZ y) r0a6GrpcServer object encapsulates a resource with GRPC server. Functions can be registered locally and are exposed via RPCs. Example: ``` server = rpc_ops.GrpcServer("host:port") @tf.function def add(a, b): return a + b server.register("add", add) server.start() ``` r.ctj||_tjr;t j |j|jj|_ytd)Nhandle handle_devicez-Please create the server outside tf.function.) gen_rpc_ops rpc_server_server_handlerrMrEagerResourceDeleterdevice_handle_deleterr7)r8r.s r__init__zGrpcServer.__init__sZ%009D  "2GG$$D4G4G4N4NPd O PPr!r3rc t|tjr|jjr|j t d|j}tj|j||jt|t||yt|tjrBtj|j||jt|t||yt d)z!Method for registering functions.Nz/Input signature not specified for the function.)r3captured_inputs input_specsrfz4Only TF functions are supported with Register method) isinstancerrC function_spec arg_namesinput_signaturer/get_concrete_functionr[rpc_server_registerr]rcr'r rDrE)r8r3r concrete_fns rr9zGrpcServer.registers $ --.   % %    'LM M..0k%%   !%553K@5kB   D+66 7%%   !..3D95d;   M NNr!cBtj|jy)zStarts GRPC server.N)r[rpc_server_startr]r;s rr<zGrpcServer.start:s  !4!45r!N) r=r>r?r@rBrarrrCrDrEr9r<rFr!rr0r0sL QcQO#O<00&7789O<6r!r0cfeZdZdZ d dedefdZdZ d dedeee jfd Z y) rNa|Client wrapper to connect to remote RPC server using GRPC. If Client is created with (list_registered_methods=True): 1. Input and output specs for the methods till this point will be fetched from Server. 2. convenience methods are added to invoke registered methods directly from client. For example: For call a server method `add` client.add(a, b) or client.add_async(a, b) can be used instead of client.call(args=[a,b], output_specs=[..]) Prerequisite for using list_registered_methods=True: 1. Server should be already started with the registered methods. 2. Client must be created in Eager mode. r.rJctj||||\|_}tjr;t j |j|jj|_n td||_ i|_ |jD]}tj}|j|t!j"|j$}t!j"|j&} ||j|j(<d|j(zdz|z} |j+|j(|| |j| y)N) shared_nameserver_addressrKrLrXz0Client creation is supported only in eager mode.z RPC Call for z method to server )r[ rpc_client_client_handlerrMrr^r_r`r7_server_address_method_registrynumpyrpc_pb2RegisteredMethodParseFromStringr decode_protorrdmethod _add_method) r8r.rJrKrLmethodsr|mrrd doc_strings rrazGrpcClient.__init__Qs. $/#9#9 7# $% D   "2GG$$D4G4G4N4NPd < >>"DD--/ #  " " $a+88Hl*77 Fk(4dAHH%#QXX-0DDwNj qxx{D.validate_and_get_flat_inputsss4   ""45LL&k r!r)rLc\tj||\}}t||SNrOr3rL)r[rpc_callStatusOrResult)rLrO status_ordeleter client_handler3rrs r call_wrapperz,GrpcClient._add_method..call_wrapper{s:&// +T2!% 'i Iw ==r!c4tj ||\}}t||}|jr|j S|j \}}t j|jdd|jr) r[rris_ok get_value get_errorr exception_type_from_error_coderw) rLrOrr error_code error_msgrr3rrs rcall_blocking_wrapperz5GrpcClient._add_method..call_blocking_wrappers&// +T2!% 'i !G\Bi  ""$$ ) 3 3 5 IGf33J4D4D4FG $ )+ +r! _blockingN)setattrr@) r8r3rrdrrrrblocking_method_namers ```` @rr}zGrpcClient._add_methodos^+,>>45 + + D+|,%L&4 D&(=>$.!r!Nr3rOc|g}tj|jtj|||\}}t |||S)aMethod to invoke remote registered functions on the connected server. Server should be started before making an RPC Call. Args: method_name: Registered method to invoke on Server. args: Input arguments for the method. output_specs: Output specs for the output from method. timeout_in_ms: Timeout for this call. If 0, default client timeout will be used. Returns: StatusOrResult object. This function issues the RPC call to server, it does not block for the duration of RPC. Please call is_ok, get_error or get_value methods on the returned object to blocked till RPC finishes. r)r[rrtrrr)r8r3rOrrLrrs rrQzGrpcClient.callsO* | d$--  \\$ # %Iw )Wl ;;r!)rRFrrS) r=r>r?r@rBrar}rrrTrUrQrFr!rrNrN?s^&', ###<'/V=A <<(=#7#789>$**%..q E GGr!cR|j|j|jfS)zhReturns (TF Error Code, Error Message) from RPC Response. This call will block for RPC result. )rrrr;s rrzStatusOrResult.get_errors'     T00 00r!c|j|j$t|jtjrg}d}n8d}t j |jDcgc]}|j}}tj|j|}|ryt j|j|Scc}w)zReturns the returned response value from RPC Call when RPC is successful. The returned value is tensors in the output_specs format as returned from the RPC call This call will block for RPC result. NTF)Tout) rrrfr NoneTensorSpecrrrr[ rpc_get_valuerpack_sequence_as)r8flat_output_dtypes return_nonesresults rrzStatusOrResult.get_values  !Z0B0B0;0J0J&Lkk-1\\$:L:L-MNAGGNN  & &t=O PF   " "4#5#5v >> Os)Cr) r=r>r?r@rarrrrrrFr!rrrs$;. ;G1?r!r).r@typingrrr:tensorflow.distribute.experimental.rpc.kernels.gen_rpc_ops distribute experimentalrpckernelsr[,tensorflow.distribute.experimental.rpc.protorrxtensorflow.python.eagerrrr rDtensorflow.python.frameworkr r r r rtensorflow.python.opsrrtensorflow.python.saved_modelrtensorflow.python.typesrrTtensorflow.python.utilr tensorflow.python.util.tf_exportrrEr r'objectr*rHr0rNrrFr!rrs/,,PPPV+0;3..31*7@9'60)E)E0- (D(D-  /B7?>V?>8?>D /B7KKVKK8KK\8686vu<u