| 368 | } |
| 369 | |
| 370 | void HdfsScanNode::ScannerThread(bool first_thread, int64_t scanner_thread_reservation) { |
| 371 | SCOPED_THREAD_COUNTER_MEASUREMENT(thread_state_.thread_counters()); |
| 372 | SCOPED_THREAD_COUNTER_MEASUREMENT(runtime_state_->total_thread_statistics()); |
| 373 | // Make thread-local copy of filter contexts to prune scan ranges, and to pass to the |
| 374 | // scanner for finer-grained filtering. Use a thread-local MemPool for the filter |
| 375 | // contexts as the embedded expression evaluators may allocate from it and MemPool |
| 376 | // is not thread safe. |
| 377 | MemPool filter_mem_pool(expr_mem_tracker()); |
| 378 | MemPool expr_results_pool(expr_mem_tracker()); |
| 379 | vector<FilterContext> filter_ctxs; |
| 380 | Status filter_status = Status::OK(); |
| 381 | for (auto& filter_ctx: filter_ctxs_) { |
| 382 | FilterContext filter; |
| 383 | filter_status = filter.CloneFrom(filter_ctx, pool_, runtime_state_, &filter_mem_pool, |
| 384 | &expr_results_pool); |
| 385 | if (!filter_status.ok()) break; |
| 386 | filter_ctxs.push_back(filter); |
| 387 | } |
| 388 | |
| 389 | while (!done()) { |
| 390 | // Prevent memory accumulating across scan ranges. |
| 391 | expr_results_pool.Clear(); |
| 392 | // Check if we have enough thread tokens to keep using this optional thread. This |
| 393 | // check is racy: multiple threads may notice that the optional tokens are exceeded |
| 394 | // and shut themselves down. If we shut down too many and there are more optional |
| 395 | // tokens, ThreadAvailableCb() will be invoked again. |
| 396 | if (!first_thread && runtime_state_->resource_pool()->optional_exceeded()) break; |
| 397 | |
| 398 | bool unused = false; |
| 399 | // Wake up every SCANNER_THREAD_COUNTERS to yield scanner threads back if unused, or |
| 400 | // to return if there's an error. |
| 401 | ranges_issued_barrier_.Wait(SCANNER_THREAD_WAIT_TIME_MS, &unused); |
| 402 | |
| 403 | // Take a snapshot of remaining_scan_range_submissions before calling |
| 404 | // StartNextScanRange(). We don't want it to go to zero between the return from |
| 405 | // StartNextScanRange() and the check for when all ranges are complete. |
| 406 | int remaining_scan_range_submissions = shared_state_->RemainingScanRangeSubmissions(); |
| 407 | ScanRange* scan_range; |
| 408 | Status status = |
| 409 | StartNextScanRange(filter_ctxs, &scanner_thread_reservation, &scan_range); |
| 410 | if (!status.ok()) { |
| 411 | unique_lock<timed_mutex> l(lock_); |
| 412 | // If there was already an error, the main thread will do the cleanup |
| 413 | if (!status_.ok()) break; |
| 414 | |
| 415 | // Invoke SetDoneInternal() with the error status (which shuts down the |
| 416 | // RowBatchQueue) to ensure that GetNextInternal() notices the error. |
| 417 | SetDoneInternal(status); |
| 418 | break; |
| 419 | } |
| 420 | if (scan_range != nullptr) { |
| 421 | discard_result(DebugAction( |
| 422 | runtime_state_->query_options(), "HDFS_SCANNER_THREAD_OBTAINED_RANGE")); |
| 423 | // Got a scan range. Process the range end to end (in this thread). |
| 424 | ProcessSplit(filter_status.ok() ? filter_ctxs : vector<FilterContext>(), |
| 425 | &expr_results_pool, scan_range, &scanner_thread_reservation); |
| 426 | } |
| 427 |
no test coverage detected