| 1569 | } |
| 1570 | |
| 1571 | void ImpalaHttpHandler::AdmissionStateHandler( |
| 1572 | const Webserver::WebRequest& req, Document* document) { |
| 1573 | const auto& args = req.parsed_args; |
| 1574 | Webserver::ArgumentMap::const_iterator pool_name_arg = args.find("pool_name"); |
| 1575 | bool get_all_pools = (pool_name_arg == args.end()); |
| 1576 | Value resource_pools(kArrayType); |
| 1577 | if (get_all_pools) { |
| 1578 | admission_controller_->AllPoolsToJson(&resource_pools, document); |
| 1579 | } else { |
| 1580 | admission_controller_->PoolToJson(pool_name_arg->second, &resource_pools, document); |
| 1581 | } |
| 1582 | |
| 1583 | // Now get running queries from CRS map. |
| 1584 | struct QueryInfo { |
| 1585 | TUniqueId query_id; |
| 1586 | int64_t executor_mem_limit; |
| 1587 | int64_t executor_mem_to_admit; |
| 1588 | int64_t coord_mem_limit; |
| 1589 | int64_t coord_mem_to_admit; |
| 1590 | unsigned long num_backends; |
| 1591 | }; |
| 1592 | unordered_map<string, vector<QueryInfo>> running_queries; |
| 1593 | if (is_admissiond_) { |
| 1594 | AdmissiondEnv::GetInstance() |
| 1595 | ->admission_control_service() |
| 1596 | ->admission_state_map_.DoFuncForAllEntries( |
| 1597 | [&running_queries]( |
| 1598 | const std::shared_ptr<AdmissionControlService::AdmissionState>& |
| 1599 | query_info) { |
| 1600 | lock_guard<mutex> l(query_info->lock); |
| 1601 | if (query_info->schedule.get() != nullptr) { |
| 1602 | TUniqueId query_id; |
| 1603 | UniqueIdPBToTUniqueId(query_info->query_id, &query_id); |
| 1604 | running_queries[query_info->request_pool].push_back( |
| 1605 | {query_id, query_info->schedule->per_backend_mem_limit(), |
| 1606 | query_info->schedule->per_backend_mem_to_admit(), |
| 1607 | query_info->schedule->coord_backend_mem_limit(), |
| 1608 | query_info->schedule->coord_backend_mem_to_admit(), |
| 1609 | static_cast<unsigned long>( |
| 1610 | query_info->schedule->backend_exec_params().size())}); |
| 1611 | }; |
| 1612 | }); |
| 1613 | } else { |
| 1614 | server_->query_driver_map_.DoFuncForAllEntries( |
| 1615 | [&running_queries](const std::shared_ptr<QueryDriver>& query_driver) { |
| 1616 | // Make sure only queries past admission control are added. |
| 1617 | ClientRequestState* request_state = query_driver->GetActiveClientRequestState(); |
| 1618 | auto query_state = request_state->exec_state(); |
| 1619 | if (query_state != ClientRequestState::ExecState::INITIALIZED |
| 1620 | && query_state != ClientRequestState::ExecState::PENDING |
| 1621 | && request_state->schedule() != nullptr) |
| 1622 | running_queries[request_state->request_pool()].push_back( |
| 1623 | {request_state->query_id(), |
| 1624 | request_state->schedule()->per_backend_mem_limit(), |
| 1625 | request_state->schedule()->per_backend_mem_to_admit(), |
| 1626 | request_state->schedule()->coord_backend_mem_limit(), |
| 1627 | request_state->schedule()->coord_backend_mem_to_admit(), |
| 1628 | static_cast<unsigned long>( |
nothing calls this directly
no test coverage detected