| 75 | } |
| 76 | |
| 77 | void ExecutorGroup::AddExecutor(const BackendDescriptorPB& be_desc) { |
| 78 | // be_desc.is_executor can be false for the local backend when scheduling queries to run |
| 79 | // on the coordinator host. |
| 80 | DCHECK(!be_desc.ip_address().empty()); |
| 81 | Executors& be_descs = executor_map_[be_desc.ip_address()]; |
| 82 | auto eq = [&be_desc](const BackendDescriptorPB& existing) { |
| 83 | // The IP addresses must already match, so it is sufficient to check the port. |
| 84 | DCHECK_EQ(existing.ip_address(), be_desc.ip_address()); |
| 85 | return existing.address().port() == be_desc.address().port(); |
| 86 | }; |
| 87 | if (find_if(be_descs.begin(), be_descs.end(), eq) != be_descs.end()) { |
| 88 | LOG(DFATAL) << "Tried to add existing backend to executor group: " |
| 89 | << be_desc.krpc_address(); |
| 90 | return; |
| 91 | } |
| 92 | if (!CheckConsistencyOrWarn(be_desc)) { |
| 93 | LOG(WARNING) << "Ignoring inconsistent backend for executor group: " |
| 94 | << be_desc.krpc_address(); |
| 95 | return; |
| 96 | } |
| 97 | if (be_descs.empty()) { |
| 98 | // Use the 'scheduling_seed' for hashing to allow different executor groups to |
| 99 | // schedule the same way. |
| 100 | executor_ip_hash_ring_.AddNode(be_desc.ip_address(), be_desc.scheduling_seed()); |
| 101 | } |
| 102 | be_descs.push_back(be_desc); |
| 103 | |
| 104 | // When computing ScanRange assignment, if there are multiple backends on a host, a |
| 105 | // round-robin approach is taken. That is, which backend a ScanRange assigned to the |
| 106 | // host will eventually be assigned to depends on the order of these backends in the |
| 107 | // vector, and the corresponding code is located in |
| 108 | // Scheduler::AssignmentCtx::SelectExecutorOnHost(). Since backend's remote data cache |
| 109 | // have data dump-load ability, so it is better to keep the order of backends before and |
| 110 | // after restarting consistent (here using port sorting), to ensure that ScanRange |
| 111 | // assignments do not change after certain backends or even entire clusters restart, to |
| 112 | // improve data cache hit rate. |
| 113 | auto cmp = [](const BackendDescriptorPB& a, const BackendDescriptorPB& b) { |
| 114 | return a.address().port() < b.address().port(); |
| 115 | }; |
| 116 | std::sort(be_descs.begin(), be_descs.end(), cmp); |
| 117 | |
| 118 | executor_ip_map_[be_desc.address().hostname()] = be_desc.ip_address(); |
| 119 | |
| 120 | DCHECK(be_desc.admit_mem_limit() > 0) << "Admit memory limit must be set for backends"; |
| 121 | if (per_executor_admit_mem_limit_ > 0) { |
| 122 | per_executor_admit_mem_limit_ = |
| 123 | std::min(be_desc.admit_mem_limit(), per_executor_admit_mem_limit_); |
| 124 | } else if (per_executor_admit_mem_limit_ == 0) { |
| 125 | per_executor_admit_mem_limit_ = be_desc.admit_mem_limit(); |
| 126 | } |
| 127 | |
| 128 | if (be_desc.ip_address() == "127.0.0.1") { |
| 129 | // Include localhost as an alias for filesystems that don't translate it. |
| 130 | LOG(INFO) << "Adding executor localhost alias for " |
| 131 | << be_desc.address().hostname() << " -> " << be_desc.ip_address(); |
| 132 | executor_ip_map_["localhost"] = be_desc.ip_address(); |
| 133 | } |
| 134 | } |