| 2682 | } |
| 2683 | |
| 2684 | void AdmissionController::TryDequeue() { |
| 2685 | ClusterMembershipMgr::SnapshotPtr membership_snapshot = |
| 2686 | cluster_membership_mgr_->GetSnapshot(); |
| 2687 | // If a query was queued while the cluster is still starting up but the client facing |
| 2688 | // services have already started to accept connections, the whole membership can still |
| 2689 | // be empty. |
| 2690 | if (membership_snapshot->executor_groups.empty()) { |
| 2691 | return; |
| 2692 | } |
| 2693 | |
| 2694 | for (const PoolConfigMap::value_type& entry : pool_config_map_) { |
| 2695 | const string& pool_name = entry.first; |
| 2696 | const TPoolConfig& pool_config = entry.second; |
| 2697 | PoolStats* stats = GetPoolStats(pool_name, /* dcheck_exists=*/true); |
| 2698 | |
| 2699 | if (stats->local_stats().num_queued == 0) continue; // Nothing to dequeue |
| 2700 | DCHECK_GE(stats->agg_num_queued(), stats->local_stats().num_queued); |
| 2701 | |
| 2702 | RequestQueue& queue = request_queue_map_[pool_name]; |
| 2703 | int64_t max_to_dequeue = GetMaxToDequeue(queue, stats, pool_config); |
| 2704 | VLOG_RPC << "Dequeue thread will try to admit " << max_to_dequeue << " requests" |
| 2705 | << ", pool=" << pool_name |
| 2706 | << ", num_queued=" << stats->local_stats().num_queued |
| 2707 | << " cluster_size=" << GetClusterSize(*membership_snapshot); |
| 2708 | if (max_to_dequeue == 0) continue; // to next pool. |
| 2709 | |
| 2710 | while (max_to_dequeue > 0 && !queue.empty()) { |
| 2711 | QueueNode* queue_node = queue.head(); |
| 2712 | DCHECK(queue_node != nullptr); |
| 2713 | // Find a group that can admit the query |
| 2714 | bool is_cancelled = queue_node->admit_outcome->IsSet() |
| 2715 | && queue_node->admit_outcome->Get() == AdmissionOutcome::CANCELLED; |
| 2716 | |
| 2717 | bool coordinator_resource_limited = false; |
| 2718 | bool is_trivial = false; |
| 2719 | bool is_rejected = false; |
| 2720 | const UniqueIdPB& query_id = queue_node->admission_request.query_id; |
| 2721 | const TQueryExecRequest* exec_req = nullptr; |
| 2722 | if (!is_cancelled) { |
| 2723 | Status status = |
| 2724 | queue_node->admission_request.request.GetQueryExecRequest(&exec_req); |
| 2725 | if (!status.ok()) { |
| 2726 | LOG(WARNING) << "Failed to get query execution request for dequeued query " |
| 2727 | << PrintId(query_id) << ": " << status.GetDetail(); |
| 2728 | // Fall to rejection handling. |
| 2729 | queue_node->not_admitted_reason = |
| 2730 | Substitute("Failed to get query execution request: $0", status.GetDetail()); |
| 2731 | is_rejected = true; |
| 2732 | } |
| 2733 | if (!is_rejected) { |
| 2734 | is_rejected = !FindGroupToAdmitOrReject(membership_snapshot, pool_config, |
| 2735 | queue_node->root_cfg, |
| 2736 | /* admit_from_queue=*/true, stats, queue_node, coordinator_resource_limited, |
| 2737 | &is_trivial); |
| 2738 | } |
| 2739 | } |
| 2740 | |
| 2741 | if (!is_cancelled && !is_rejected |