| 1540 | } |
| 1541 | |
| 1542 | bool AdmissionController::RejectForSchedule( |
| 1543 | const ScheduleState& state, const TPoolConfig& pool_cfg, string* rejection_reason) { |
| 1544 | DCHECK(rejection_reason != nullptr && rejection_reason->empty()); |
| 1545 | bool default_group = |
| 1546 | state.executor_group() == ImpalaServer::DEFAULT_EXECUTOR_GROUP_NAME; |
| 1547 | |
| 1548 | // Compute the max (over all backends), the cluster totals (across all backends) for |
| 1549 | // min_mem_reservation_bytes, thread_reservation, the min admit_mem_limit |
| 1550 | // (over all executors) and the admit_mem_limit of the coordinator. |
| 1551 | pair<const NetworkAddressPB*, int64_t> largest_min_mem_reservation(nullptr, -1); |
| 1552 | int64_t cluster_min_mem_reservation_bytes = 0; |
| 1553 | pair<const NetworkAddressPB*, int64_t> max_thread_reservation(nullptr, 0); |
| 1554 | pair<const NetworkAddressPB*, int64_t> min_executor_admit_mem_limit( |
| 1555 | nullptr, std::numeric_limits<int64_t>::max()); |
| 1556 | pair<const NetworkAddressPB*, int64_t> coord_admit_mem_limit( |
| 1557 | nullptr, std::numeric_limits<int64_t>::max()); |
| 1558 | int64_t cluster_thread_reservation = 0; |
| 1559 | for (const auto& e : state.per_backend_schedule_states()) { |
| 1560 | const BackendScheduleState& be_state = e.second; |
| 1561 | // TODO(IMPALA-8757): Extend slot based admission to default executor group |
| 1562 | if (!default_group |
| 1563 | && be_state.exec_params->slots_to_use() > be_state.be_desc.admission_slots()) { |
| 1564 | *rejection_reason = Substitute(REASON_NOT_ENOUGH_SLOTS_ON_BACKEND, |
| 1565 | be_state.exec_params->slots_to_use(), |
| 1566 | NetworkAddressPBToString(be_state.be_desc.address()), |
| 1567 | be_state.be_desc.admission_slots()); |
| 1568 | return true; |
| 1569 | } |
| 1570 | |
| 1571 | cluster_min_mem_reservation_bytes += |
| 1572 | be_state.exec_params->min_mem_reservation_bytes(); |
| 1573 | if (be_state.exec_params->min_mem_reservation_bytes() |
| 1574 | > largest_min_mem_reservation.second) { |
| 1575 | largest_min_mem_reservation = |
| 1576 | make_pair(&e.first, be_state.exec_params->min_mem_reservation_bytes()); |
| 1577 | } |
| 1578 | cluster_thread_reservation += be_state.exec_params->thread_reservation(); |
| 1579 | if (be_state.exec_params->thread_reservation() > max_thread_reservation.second) { |
| 1580 | max_thread_reservation = |
| 1581 | make_pair(&e.first, be_state.exec_params->thread_reservation()); |
| 1582 | } |
| 1583 | if (!FLAGS_clamp_query_mem_limit_backend_mem_limit) { |
| 1584 | if (be_state.exec_params->is_coord_backend()) { |
| 1585 | coord_admit_mem_limit.first = &e.first; |
| 1586 | coord_admit_mem_limit.second = be_state.be_desc.admit_mem_limit(); |
| 1587 | } else if (be_state.be_desc.admit_mem_limit() |
| 1588 | < min_executor_admit_mem_limit.second) { |
| 1589 | min_executor_admit_mem_limit.first = &e.first; |
| 1590 | min_executor_admit_mem_limit.second = be_state.be_desc.admit_mem_limit(); |
| 1591 | } |
| 1592 | } |
| 1593 | } |
| 1594 | |
| 1595 | // Checks related to the min buffer reservation against configured query memory limits: |
| 1596 | const TQueryOptions& query_opts = state.query_options(); |
| 1597 | if (query_opts.__isset.buffer_pool_limit && query_opts.buffer_pool_limit > 0) { |
| 1598 | if (largest_min_mem_reservation.second > query_opts.buffer_pool_limit) { |
| 1599 | *rejection_reason = Substitute(REASON_BUFFER_LIMIT_TOO_LOW_FOR_RESERVATION, |