code: def _blocking_send(
      self, req: ray_client_pb2.DataRequest
  ) -> ray_client_pb2.DataResponse:
      with self.lock:
          self._check_shutdown()
          req_id = self._next_id()
          req.req_id = req_id
          self.request_queue.put(req)
          self.outstanding_requests[req_id] = req
  
          self.cv.wait_for(
              lambda: req_id in self.ready_data or self._in_shutdown
          )
          self._check_shutdown()
  
          data = self.ready_data[req_id]
          del self.ready_data[req_id]
          del self.outstanding_requests[req_id]
          self._acknowledge(req_id)
  
      return data
  
imports: import math
  import logging
  import queue
  import threading
  import warnings
  import grpc
  from collections import OrderedDict
  from typing import Any, Callable, Dict, TYPE_CHECKING, Optional, Union
  import ray.core.generated.ray_client_pb2
  import ray.core.generated.ray_client_pb2_grpc
  from ray.util.client.common import INT32_MAX, OBJECT_TRANSFER_CHUNK_SIZE, OBJECT_TRANSFER_WARNING_SIZE
  from ray.util.debug import log_once
label_desc: Use image node_id 13 for calling the DataClient obj's underlying member method code with example usage: obj._blocking_send(req) and returns: data