Method BlockingGet

be/src/kudu/rpc/service_queue.cc:42–73

}

bool LifoServiceQueue::BlockingGet(std::unique_ptr<InboundCall>* out) {
  auto consumer = tl_consumer_;
  if (PREDICT_FALSE(!consumer)) {
    consumer = tl_consumer_ = new ConsumerState(this);
    std::lock_guard<simple_spinlock> l(lock_);
    consumers_.emplace_back(consumer);
  }

  while (true) {
    {
      std::lock_guard<simple_spinlock> l(lock_);
      if (!queue_.empty()) {
        auto it = queue_.begin();
        out->reset(*it);
        queue_.erase(it);
        return true;
      }
      if (PREDICT_FALSE(shutdown_)) {
        return false;
      }
      consumer->DCheckBoundInstance(this);
      waiting_consumers_.push_back(consumer);
    }
    InboundCall* call = consumer->Wait();
    if (call != nullptr) {
      out->reset(call);
      return true;
    }
    // if call == nullptr, this means we are shutting down the queue.
    // Loop back around and re-check 'shutdown_'.
  }
}

QueueStatus LifoServiceQueue::Put(InboundCall* call,
                                  std::optional<InboundCall*>* evicted) {