code: def test_client_timeout(self):
      port = portpicker.pick_unused_port()
      address = "localhost:{}".format(port)
  
      @eager_def_function.function(
          input_signature=[
              tensor_spec.TensorSpec([], dtypes.int32),
              tensor_spec.TensorSpec([], dtypes.int32),
          ]
      )
      def add(a, b):
          return math_ops.add(a, b)
  
      server = rpc_ops.GrpcServer(address)
  
      def start_server():
          # Delay server start to simulate deadline exceeded for 1st RPC call
          # response. Client waits till server is started, thus it can trigger
          # deadline exceeded.
          time.sleep(1)
          server.register("add", add)
          server.start()
  
      t = threading.Thread(target=start_server)
      t.start()
  
      def ensure_server_is_ready(client):
          server_ready = False
          while not server_ready:
              result_or = client.call(
                  "add",
                  [constant_op.constant(20), constant_op.constant(30)],
              )
              if result_or.is_ok():
                  server_ready = True
              else:
                  error_code, _ = result_or.get_error()
                  if error_code == errors.UNAVAILABLE:
                      server_ready = False
                  else:
                      server_ready = True
          return
  
      # Create client with list_registered_methods fails before server is started.
      with self.assertRaises(errors.DeadlineExceededError):
          rpc_ops.GrpcClient(
              address,
              name="client1",
              list_registered_methods=True,
              timeout_in_ms=1,
          )
  
      # Create same client again should succeed with
      # list_registered_methods=False. Default timeout for client is 1 ms.
      client = rpc_ops.GrpcClient(
          address,
          name="client1",
          list_registered_methods=False,
          timeout_in_ms=1,
      )
  
      ensure_server_is_ready(client)
      # Make explicit RPC call, the timeout of 1 ms should lead to
      # deadline exceeded error.
  
      result_or = client.call(
          "add",
          [constant_op.constant(20), constant_op.constant(30)],
          timeout_in_ms=1,
      )
      self.assertAllEqual(result_or.is_ok(), False)
      error_code, error_message = result_or.get_error()
      self.assertAllEqual(
          error_code, errors.DEADLINE_EXCEEDED, error_message
      )
  
      # Specifying reasonable timeout for call should succeed.
      result_or = client.call(
          "add",
          [constant_op.constant(20), constant_op.constant(30)],
          timeout_in_ms=5000,
      )
      self.assertAllEqual(result_or.is_ok(), True)
      error_code, _ = result_or.get_error()
  
      # Test timeouts for convenience methods
  
      # Restart server again with delay to simulate deadline exceeded.
      del server
      server = rpc_ops.GrpcServer(address)
      t = threading.Thread(target=start_server)
      t.start()
  
      # Client with no default timeout.
      client = rpc_ops.GrpcClient(
          address, name="client2", list_registered_methods=True
      )
  
      # Succeeds with reasonable timeout.
      result_or = client.add(
          constant_op.constant(20),
          constant_op.constant(30),
          timeout_in_ms=5000,
      )
      self.assertAllEqual(result_or.is_ok(), True)
  
imports: import threading
  import time
  import numpy
  import portpicker
  from tensorflow.python.distribute.experimental.rpc import rpc_ops
  from tensorflow.python.eager import context
  from tensorflow.python.eager import def_function
  from tensorflow.python.framework import config
  from tensorflow.python.framework import constant_op
  from tensorflow.python.framework import dtypes
  from tensorflow.python.framework import errors
  from tensorflow.python.framework import ops
  from tensorflow.python.framework import tensor_spec
  from tensorflow.python.framework import test_util
  from tensorflow.python.ops import data_flow_ops
  from tensorflow.python.ops import math_ops
  from tensorflow.python.ops import resource_variable_ops
  from tensorflow.python.ops import variables
  from tensorflow.python.platform import test
  from tensorflow.python.util import nest
label_desc: Use image node_id 10 for calling the RpcOpsTest obj's underlying member method code with example usage: obj.test_client_timeout() and returns: math_ops