Method QueueInboundCall

be/src/rpc/impala-service-pool.cc:159–269

}

kudu::Status ImpalaServicePool::QueueInboundCall(
    std::unique_ptr<kudu::rpc::InboundCall> call) {
  kudu::rpc::InboundCall* c = call.release();

  vector<uint32_t> unsupported_features;
  for (uint32_t feature : c->GetRequiredFeatures()) {
    if (!service_->SupportsFeature(feature)) {
      unsupported_features.push_back(feature);
    }
  }

  if (!unsupported_features.empty()) {
    c->RespondUnsupportedFeature(unsupported_features);
    return kudu::Status::NotSupported(
        "call requires unsupported application feature flags",
        JoinMapped(unsupported_features,
        [] (uint32_t flag) { return std::to_string(flag); }, ", "));
  }

  TRACE_TO(c->trace(), "Inserting onto call queue"); // NOLINT(*)

  // Queue message on service queue.
  const int64_t transfer_size = c->GetTransferSize();
  {
    // Drops an incoming request if consumption already exceeded the limit. Note that
    // the current inbound call isn't counted towards the limit yet so adding this call
    // may cause the MemTracker's limit to be exceeded. This is done to ensure fairness
    // among all inbound calls, otherwise calls with larger payloads are more likely to
    // fail. The check and the consumption need to be atomic so as to bound the memory
    // usage.
    unique_lock<SpinLock> mem_tracker_lock(mem_tracker_lock_);
    if (UNLIKELY(service_mem_tracker_->AnyLimitExceeded(MemLimit::HARD))) {
      if (c->remote_method().method_name() == DataStreamService::END_DATA_STREAM) {
        // EndDataStream operations use very little memory and help free up other
        // resources, so ignore memory hard limit and always add them to the queue. This
        // can help complete queries earlier when under heavy load that would otherwise
        // drop the requests and require them to be retried.
        LOG(WARNING) << "Admitting " << c->remote_method().method_name()
            << " request on " << service_->service_name()
            << "from " << c->remote_address().ToString()
            << " to complete query despite exceeding memory limit; memory consumption is "
            << PrettyPrinter::Print(service_mem_tracker_->consumption(), TUnit::BYTES);
      } else {
        // Discards the transfer early so the transfer size drops to 0. This is to ensure
        // the MemTracker::Release() call in FailAndReleaseRpc() is correct as we haven't
        // called MemTracker::Consume() at this point.
        mem_tracker_lock.unlock();
        c->DiscardTransfer();
        RejectTooBusy(c);
        return kudu::Status::OK();
      }
    }
    if (UNLIKELY(!FLAGS_debug_actions.empty())) {
      Status status = DebugAction(FLAGS_debug_actions, "INBOUND_GETQUERYSTATUS_REJECT");
      if (!status.ok() && c->remote_method().method_name() == "GetQueryStatus") {
        mem_tracker_lock.unlock();
        c->DiscardTransfer();
        RejectTooBusy(c);