| 2493 | } |
| 2494 | |
| 2495 | bool AdmissionController::FindGroupToAdmitOrReject( |
| 2496 | ClusterMembershipMgr::SnapshotPtr& membership_snapshot, |
| 2497 | const TPoolConfig& pool_config, const TPoolConfig& root_cfg, bool admit_from_queue, |
| 2498 | PoolStats* pool_stats, QueueNode* queue_node, bool& coordinator_resource_limited, |
| 2499 | bool* is_trivial) { |
| 2500 | // Check for rejection based on current cluster size |
| 2501 | const string& pool_name = pool_stats->name(); |
| 2502 | string rejection_reason; |
| 2503 | if (RejectForCluster(pool_name, pool_config, admit_from_queue, &rejection_reason)) { |
| 2504 | DCHECK(!rejection_reason.empty()); |
| 2505 | queue_node->not_admitted_reason = rejection_reason; |
| 2506 | return false; |
| 2507 | } |
| 2508 | |
| 2509 | // Compute schedules |
| 2510 | Status ret = ComputeGroupScheduleStates(membership_snapshot, queue_node); |
| 2511 | if (!ret.ok()) { |
| 2512 | DCHECK(queue_node->not_admitted_reason.empty()); |
| 2513 | queue_node->not_admitted_reason = Substitute(REASON_SCHEDULER_ERROR, ret.GetDetail()); |
| 2514 | return false; |
| 2515 | } |
| 2516 | if (queue_node->group_states.empty()) { |
| 2517 | DCHECK(!queue_node->not_admitted_reason.empty()); |
| 2518 | if (IsReasonCoordinatorRemoved(queue_node->not_admitted_reason)) return false; |
| 2519 | return true; |
| 2520 | } |
| 2521 | |
| 2522 | // Get Coordinator Backend for the given admission request |
| 2523 | const AdmissionRequest& request = queue_node->admission_request; |
| 2524 | auto it = membership_snapshot->current_backends.find(PrintId(request.coord_id)); |
| 2525 | if (it == membership_snapshot->current_backends.end()) { |
| 2526 | queue_node->not_admitted_reason = REASON_COORDINATOR_NOT_FOUND; |
| 2527 | LOG(WARNING) << queue_node->not_admitted_reason; |
| 2528 | return true; |
| 2529 | } |
| 2530 | const BackendDescriptorPB& coord_desc = it->second; |
| 2531 | |
| 2532 | for (GroupScheduleState& group_state : queue_node->group_states) { |
| 2533 | const ExecutorGroup& executor_group = group_state.executor_group; |
| 2534 | ScheduleState* state = group_state.state.get(); |
| 2535 | state->UpdateMemoryRequirements(pool_config, coord_desc.admit_mem_limit(), |
| 2536 | executor_group.GetPerExecutorMemLimitForAdmission()); |
| 2537 | |
| 2538 | const string& group_name = executor_group.name(); |
| 2539 | int64_t group_size = executor_group.NumExecutors(); |
| 2540 | VLOG(3) << "Trying to admit query to pool " << pool_name << " in executor group " |
| 2541 | << group_name << " (" << group_size << " executors)"; |
| 2542 | |
| 2543 | const int64_t max_queued = GetMaxQueuedForPool(pool_config); |
| 2544 | const int64_t max_mem = GetMaxMemForPool(pool_config); |
| 2545 | const int64_t max_requests = GetMaxRequestsForPool(pool_config); |
| 2546 | VLOG_QUERY << "Trying to admit id=" << PrintId(state->query_id()) |
| 2547 | << " in pool_name=" << pool_name << " executor_group_name=" << group_name |
| 2548 | << " per_host_mem_estimate=" |
| 2549 | << PrintBytes(state->GetPerExecutorMemoryEstimate()) |
| 2550 | << " dedicated_coord_mem_estimate=" |
| 2551 | << PrintBytes(state->GetDedicatedCoordMemoryEstimate()) |
| 2552 | << " max_requests=" << max_requests << " max_queued=" << max_queued |
nothing calls this directly
no test coverage detected