docs: def CallClient(self,
                   action_cls,
                   request=None,
                   next_state=None,
                   request_data=None,
                   **kwargs):
      """Calls the client asynchronously.
  
      This sends a message to the client to invoke an Action. The run
      action may send back many responses. These will be queued by the
      framework until a status message is sent by the client. The status
      message will cause the entire transaction to be committed to the
      specified state.
  
      Args:
         action_cls: The function to call on the client.
         request: The request to send to the client. If not specified (Or None) we
           create a new RDFValue using the kwargs.
         next_state: The state in this flow, that responses to this message should
           go to.
         request_data: A dict which will be available in the RequestState
           protobuf. The Responses object maintains a reference to this protobuf
           for use in the execution of the state method. (so you can access this
           data by responses.request). Valid values are strings, unicode and
           protobufs.
         **kwargs: These args will be used to construct the client action semantic
           protobuf.
  
      Raises:
         FlowRunnerError: If called on a flow that doesn't run on a single client.
         ValueError: The request passed to the client does not have the correct
                       type.
      """
      client_id = self.runner_args.client_id
      if client_id is None:
        raise FlowRunnerError("CallClient() is used on a flow which was not "
                              "started with a client.")
  
      if not isinstance(client_id, rdf_client.ClientURN):
        # Try turning it into a ClientURN
        client_id = rdf_client.ClientURN(client_id)
  
      if action_cls.in_rdfvalue is None:
        if request:
          raise ValueError(
              "Client action %s does not expect args." % action_cls.__name__)
      else:
        if request is None:
          # Create a new rdf request.
          request = action_cls.in_rdfvalue(**kwargs)
        else:
          # Verify that the request type matches the client action requirements.
          if not isinstance(request, action_cls.in_rdfvalue):
            raise ValueError("Client action expected %s but got %s" %
                             (action_cls.in_rdfvalue, type(request)))
  
      outbound_id = self.GetNextOutboundId()
  
      # Create a new request state
      state = rdf_flow_runner.RequestState(
          id=outbound_id,
          session_id=self.session_id,
          next_state=next_state,
          client_id=client_id)
  
      if request_data is not None:
        state.data = rdf_protodict.Dict(request_data)
  
      # Send the message with the request state
      msg = rdf_flows.GrrMessage(
          session_id=utils.SmartUnicode(self.session_id),
          name=action_cls.__name__,
          request_id=outbound_id,
          require_fastpoll=self.runner_args.require_fastpoll,
          queue=client_id.Queue(),
          payload=request,
          generate_task_id=True)
  
      cpu_usage = self.context.client_resources.cpu_usage
      if self.runner_args.cpu_limit:
        msg.cpu_limit = max(
            self.runner_args.cpu_limit - cpu_usage.user_cpu_time -
            cpu_usage.system_cpu_time, 0)
  
        if msg.cpu_limit == 0:
          raise FlowRunnerError("CPU limit exceeded.")
  
      if self.runner_args.network_bytes_limit:
        msg.network_bytes_limit = max(
            self.runner_args.network_bytes_limit -
            self.context.network_bytes_sent, 0)
        if msg.network_bytes_limit == 0:
          raise FlowRunnerError("Network limit exceeded.")
  
      state.request = msg
      self.QueueRequest(state)