| 133 | } |
| 134 | |
| 135 | void KuduScanNode::ThreadAvailableCb(ThreadResourcePool* pool) { |
| 136 | ScannerMemLimiter* mem_limiter = runtime_state_->query_state()->scanner_mem_limiter(); |
| 137 | while (true) { |
| 138 | unique_lock<mutex> lock(lock_); |
| 139 | // All done or all tokens are assigned. |
| 140 | if (done_.Load() || !HasScanToken()) break; |
| 141 | bool first_thread = thread_state_.GetNumActive() == 0; |
| 142 | |
| 143 | // * Don't start up a ScannerThread if the row batch queue is full since |
| 144 | // we are not scanner bound. |
| 145 | // * Don't start up a thread if there is not enough memory available for the |
| 146 | // estimated memory consumption (include reservation and non-reserved memory). |
| 147 | if (!first_thread) { |
| 148 | if (thread_state_.batch_queue()->IsFull()) break; |
| 149 | if (!mem_limiter->ClaimMemoryForScannerThread( |
| 150 | this, EstimateScannerThreadMemConsumption())) { |
| 151 | COUNTER_ADD(thread_state_.scanner_thread_mem_unavailable_counter(), 1); |
| 152 | break; |
| 153 | } |
| 154 | } |
| 155 | |
| 156 | // Check if we can get a token. We need at least one thread to run. |
| 157 | if (first_thread) { |
| 158 | pool->AcquireThreadToken(); |
| 159 | } else if (thread_state_.GetNumActive() >= thread_state_.max_num_scanner_threads() |
| 160 | || !pool->TryAcquireThreadToken()) { |
| 161 | mem_limiter->ReleaseMemoryForScannerThread( |
| 162 | this, EstimateScannerThreadMemConsumption()); |
| 163 | break; |
| 164 | } |
| 165 | |
| 166 | string name = Substitute( |
| 167 | "kudu-scanner-thread (finst:$0, plan-node-id:$1, thread-idx:$2)", |
| 168 | PrintId(runtime_state_->fragment_instance_id()), id(), |
| 169 | thread_state_.GetNumStarted()); |
| 170 | |
| 171 | // Reserve the first token so no other thread picks it up. |
| 172 | const string* token = GetNextScanToken(); |
| 173 | auto fn = [this, first_thread, token, name]() { |
| 174 | RuntimeState* state = this->runtime_state(); |
| 175 | GetThreadDebugInfo()->SetInstanceId(state->fragment_instance_id()); |
| 176 | this->RunScannerThread(first_thread, name, token); |
| 177 | }; |
| 178 | std::unique_ptr<Thread> t; |
| 179 | Status status = |
| 180 | Thread::Create(FragmentInstanceState::FINST_THREAD_GROUP_NAME, name, fn, &t, true); |
| 181 | if (!status.ok()) { |
| 182 | // Release the token and skip running callbacks to find a replacement. Skipping |
| 183 | // serves two purposes. First, it prevents a mutual recursion between this function |
| 184 | // and ReleaseThreadToken()->InvokeCallbacks(). Second, Thread::Create() failed and |
| 185 | // is likely to continue failing for future callbacks. |
| 186 | pool->ReleaseThreadToken(first_thread, true); |
| 187 | if (!first_thread) { |
| 188 | mem_limiter->ReleaseMemoryForScannerThread( |
| 189 | this, EstimateScannerThreadMemConsumption()); |
| 190 | } |
| 191 | |
| 192 | // Abort the query. This is still holding the lock_, so done_ is known to be |
nothing calls this directly
no test coverage detected