| 832 | } |
| 833 | |
| 834 | Status AdmissionController::Init() { |
| 835 | RETURN_IF_ERROR(Thread::Create("scheduling", "admission-thread", |
| 836 | &AdmissionController::DequeueLoop, this, &dequeue_thread_)); |
| 837 | auto cb = [this](const StatestoreSubscriber::TopicDeltaMap& state, |
| 838 | vector<TTopicDelta>* topic_updates) { |
| 839 | UpdatePoolStats(state, topic_updates); |
| 840 | }; |
| 841 | // The executor only needs to read the entry with the key prefix "POOL:" from the topic. |
| 842 | // This can effectively reduce the network load of the statestore. |
| 843 | string filter_prefix = |
| 844 | FLAGS_is_executor && !FLAGS_is_coordinator ? TOPIC_KEY_POOL_PREFIX : ""; |
| 845 | Status status = subscriber_->AddTopic(request_queue_topic_name_, |
| 846 | /* is_transient=*/true, /* populate_min_subscriber_topic_version=*/false, |
| 847 | filter_prefix, cb); |
| 848 | if (!status.ok()) { |
| 849 | status.AddDetail("AdmissionController failed to register request queue topic"); |
| 850 | } |
| 851 | if (AdmissiondEnv::GetInstance() != nullptr) { |
| 852 | DCHECK(AdmissiondEnv::GetInstance()->process_mem_tracker() != nullptr); |
| 853 | pending_decompression_mem_tracker_.reset(new MemTracker(-1, "Pending Decompression", |
| 854 | AdmissiondEnv::GetInstance()->process_mem_tracker())); |
| 855 | } |
| 856 | return status; |
| 857 | } |
| 858 | |
| 859 | void AdmissionController::PoolStats::AdmitQueryAndMemory( |
| 860 | const ScheduleState& state, bool is_trivial, PerUserTracking& per_user_tracking) { |