Method QueueInboundCall

be/src/kudu/rpc/service_pool.cc:151–199

}

Status ServicePool::QueueInboundCall(unique_ptr<InboundCall> call) {
  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 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");

  // Queue message on service queue
  std::optional<InboundCall*> evicted;
  auto queue_status = service_queue_.Put(c, &evicted);
  if (queue_status == QUEUE_FULL) {
    RejectTooBusy(c);
    return Status::OK();
  }

  if (PREDICT_TRUE(evicted)) {
    RejectTooBusy(*evicted);
  }

  if (PREDICT_TRUE(queue_status == QUEUE_SUCCESS)) {
    // NB: do not do anything with 'c' after it is successfully queued --
    // a service thread may have already dequeued it, processed it, and
    // responded by this point, in which case the pointer would be invalid.
    return Status::OK();
  }

  Status status = Status::OK();
  if (queue_status == QUEUE_SHUTDOWN) {
    status = Status::ServiceUnavailable("Service is shutting down");
    c->RespondFailure(ErrorStatusPB::FATAL_SERVER_SHUTTING_DOWN, status);
  } else {
    status = Status::RuntimeError(Substitute("Unknown error from BlockingQueue: $0", queue_status));
    c->RespondFailure(ErrorStatusPB::FATAL_UNKNOWN, status);
  }
  return status;
}

void ServicePool::RunThread() {
  while (true) {