Method RunThread

be/src/rpc/impala-service-pool.cc:271–303

}

void ImpalaServicePool::RunThread() {
  while (true) {
    std::unique_ptr<kudu::rpc::InboundCall> incoming;
    if (!service_queue_.BlockingGet(&incoming)) {
      VLOG(1) << "ImpalaServicePool: messenger shutting down.";
      return;
    }

    // We need to call RecordHandlingStarted() to update the InboundCall timing.
    incoming->RecordHandlingStarted(incoming_queue_time_.get());
    ADOPT_TRACE(incoming->trace());

    if (UNLIKELY(incoming->ClientTimedOut())) {
      TRACE_TO(incoming->trace(), "Skipping call since client already timed out"); // NOLINT(*)

      // Respond as a failure, even though the client will probably ignore
      // the response anyway.
      FailAndReleaseRpc(kudu::rpc::ErrorStatusPB::ERROR_SERVER_TOO_BUSY,
          kudu::Status::TimedOut("Call waited in the queue past client deadline"),
          incoming.release());
      continue;
    }

    const string& method_name = incoming->remote_method().method_name();
    int64_t transfer_size = incoming->GetTransferSize();
    payload_size_histograms_[method_name]->Update(transfer_size);

    TRACE_TO(incoming->trace(), "Handling call"); // NOLINT(*)
    // Release the InboundCall pointer -- when the call is responded to, it will get
    // deleted at that point.
    service_->Handle(incoming.release());
  }
}

const string ImpalaServicePool::service_name() const {
  return service_->service_name();