Method ThreadTokenAvailableCb

be/src/exec/hdfs-scan-node.cc:255–368

}

void HdfsScanNode::ThreadTokenAvailableCb(ThreadResourcePool* pool) {
  // This is called to start up new scanner threads. It's not a big deal if we
  // spin up more than strictly necessary since they will go through and terminate
  // promptly. However, we want to minimize that by checking a conditions.
  //  1. Don't start up if the ScanNode is done
  //  2. Don't start up if all the ranges have been taken by another thread.
  //  3. Don't start up if the number of ranges left is less than the number of
  //     active scanner threads.
  //  4. Don't start up if no initial ranges have been issued (see IMPALA-1722).
  //  5. Don't start up a ScannerThread if the row batch queue is full since
  //     we are not scanner bound.
  //  6. Don't start up a thread if there is not enough memory available for the
  //     estimated memory consumption (include reservation and non-reserved memory).
  //  7. Don't start up a thread if it is an extra thread and we can't reserve another
  //     minimum reservation's worth of memory for the thread.
  //  8. Don't start up more than maximum number of scanner threads configured.
  //  9. Don't start up if there are no thread tokens.

  // Case 4. We have not issued the initial ranges so don't start a scanner thread.
  // Issuing ranges will call this function and we'll start the scanner threads then.
  // TODO: It would be good to have a test case for that.
  if (!initial_ranges_issued_.Load()) return;

  ScannerMemLimiter* scanner_mem_limiter =
      runtime_state_->query_state()->scanner_mem_limiter();
  Status status = Status::OK();
  while (true) {
    // The lock must be given up between loops in order to give writers to done_,
    // all_ranges_started_ etc. a chance to grab the lock.
    // IMPALA-8322: Another thread can hold this lock for significant periods of time
    // if the scan node is being cancelled. Since this function can be called while
    // holding other locks that can block other threads (e.g. ThreadResourceMgr::lock_),
    // avoid blocking unnecessarily. There are two remedies. First, we do a check of
    // done() to try to avoid acquiring the lock_, as there is nothing to do if
    // the scan node is done. Second, this uses a timeout of 10 milliseconds when
    // acquiring the lock_ to allow a periodic check of done(). The 10 millisecond
    // timeout is arbitrary.
    // TODO: This still leans heavily on starvation-free locks, come up with a more
    // correct way to communicate between this method and ScannerThread().
    if (done()) break;
    unique_lock<timed_mutex> lock(lock_, std::chrono::milliseconds(10));
    if (!lock.owns_lock()) {
      continue;
    }

    const int64_t num_active_scanner_threads = thread_state_.GetNumActive();
    const bool first_thread = num_active_scanner_threads == 0;
    const int64_t est_mem = thread_state_.estimated_per_thread_mem();
    const int64_t scanner_thread_reservation = resource_profile_.min_reservation;
    // Cases 1, 2, 3.
    if (done() || all_ranges_started_ ||
        num_active_scanner_threads >= shared_state_->progress().remaining()) {
      break;
    }

    if (!first_thread) {
      // Cases 5, 6 and 7.
      if (thread_state_.batch_queue()->IsFull()) break;