Method serve

be/src/rpc/TAcceptQueueServer.cpp:332–425

}

void TAcceptQueueServer::serve() {
  // Start the server listening
  serverTransport_->listen();

  // Run the preServe event
  if (eventHandler_ != nullptr) {
    eventHandler_->preServe();
  }

  if (FLAGS_accepted_cnxn_setup_thread_pool_size > 1) {
    LOG(INFO) << "connection_setup_thread_pool_size is set to "
              << FLAGS_accepted_cnxn_setup_thread_pool_size;
  }

  // New - this is the thread pool used to process the internal accept queue.
  ThreadPool<shared_ptr<TAcceptQueueEntry>> connection_setup_pool("setup-server",
      "setup-worker", FLAGS_accepted_cnxn_setup_thread_pool_size,
      FLAGS_accepted_cnxn_queue_depth,
      [this](int tid, const shared_ptr<TAcceptQueueEntry>& item) {
        this->SetupConnection(item.get());
      });
  // Initialize the thread pool
  Status status = connection_setup_pool.Init();
  if (!status.ok()) {
    status.AddDetail("TAcceptQueueServer: thread pool could not start.");
    string errStr = status.GetDetail();
    GlobalOutput(errStr.c_str());
    stop_ = true;
  }

  while (!stop_) {
    try {
      // Fetch client from server
      shared_ptr<TTransport> client = serverTransport_->accept();

      TSocket* socket = reinterpret_cast<TSocket*>(client.get());
      VLOG(1) << Substitute("New connection to server $0 from client $1",
          name_, socket->getSocketInfo());

      shared_ptr<TAcceptQueueEntry> entry{new TAcceptQueueEntry};
      entry->client_ = client;
      if (queue_timeout_ms_ > 0) {
        entry->expiration_time_ = MonotonicMillis() + queue_timeout_ms_;
      }

      // New - the work done to set up the connection has been moved to SetupConnection.
      // Note that we move() entry so it's owned by SetupConnection thread.
      if (!connection_setup_pool.Offer(std::move(entry))) {
        string errStr = string("TAcceptQueueServer: thread pool unexpectedly shut down.");
        GlobalOutput(errStr.c_str());
        stop_ = true;
        break;
      }
      if (metrics_enabled_) queue_size_metric_->Increment(1);
    } catch (const TTransportException& ttx) {
      if (!stop_ || ttx.getType() != TTransportException::INTERRUPTED) {
        string errStr =
            string("TAcceptQueueServer: TServerTransport died on accept: ") + ttx.what();