| 2815 | } |
| 2816 | |
| 2817 | int64_t AdmissionController::GetMaxToDequeue( |
| 2818 | RequestQueue& queue, PoolStats* stats, const TPoolConfig& pool_config) { |
| 2819 | if (PoolLimitsRunningQueriesCount(pool_config)) { |
| 2820 | const int64_t max_requests = GetMaxRequestsForPool(pool_config); |
| 2821 | const int64_t total_available = max_requests - stats->agg_num_running(); |
| 2822 | if (total_available <= 0) { |
| 2823 | // There is a limit for the number of running queries, so we can |
| 2824 | // see that nothing can run in this pool. |
| 2825 | // This can happen in the case of over-admission. |
| 2826 | if (!queue.empty()) { |
| 2827 | LogDequeueFailed(queue.head(), |
| 2828 | Substitute(QUEUED_NUM_RUNNING, stats->agg_num_running(), max_requests, |
| 2829 | GetStalenessDetailLocked(" "))); |
| 2830 | } |
| 2831 | return 0; |
| 2832 | } |
| 2833 | |
| 2834 | // Use the ratio of locally queued requests to agg queued so that each impalad |
| 2835 | // can dequeue a proportional amount total_available. Note, this offers no |
| 2836 | // fairness between impalads. |
| 2837 | double queue_size_ratio = static_cast<double>(stats->local_stats().num_queued) |
| 2838 | / static_cast<double>(max<int64_t>(1, stats->agg_num_queued())); |
| 2839 | DCHECK(queue_size_ratio <= 1.0); |
| 2840 | return min(stats->local_stats().num_queued, |
| 2841 | max<int64_t>(1, queue_size_ratio * total_available)); |
| 2842 | } else { |
| 2843 | return stats->local_stats().num_queued; // No limit on num running requests |
| 2844 | } |
| 2845 | } |
| 2846 | |
| 2847 | void AdmissionController::LogDequeueFailed( |
| 2848 | QueueNode* node, const string& not_admitted_reason) { |