Method RunLoop

be/src/kudu/util/threadpool.cc:140–174

}

void SchedulerThread::RunLoop() {
  while (!shutdown_.WaitFor(MonoDelta::FromMilliseconds(schedule_period_ms_))) {
    MonoTime now = MonoTime::Now();
    vector<SchedulerTask> pending_tasks;
    {
      MutexLock auto_lock(mutex_);
      auto upper_it = future_tasks_.upper_bound(now);
      for (auto it = future_tasks_.begin(); it != upper_it; it++) {
        pending_tasks.emplace_back(std::move(it->second));
      }
      future_tasks_.erase(future_tasks_.begin(), upper_it);
    }

    for (const auto& task : pending_tasks) {
      ThreadPoolToken* token = task.thread_pool_token();
      while (token != nullptr) {
        Status s = token->Submit(task.func());
        if (s.ok()) {
          break;
        }
        DCHECK(s.IsServiceUnavailable())
            << Substitute("threadpool token Submit status: $0", s.ToString());

        if (!token->MaySubmitNewTasks()) {
          // threadpool token is Shutdown, skip the task.
          break;
        }
        // If developers use ThreadPoolToken::Schedule(...) too frequent, blocking queue's
        // capacity will be full, then retry submit the task again.
        VLOG(1) << Substitute("threadpool token Submit status: $0, retry the task", s.ToString());
        SleepFor(MonoDelta::FromMilliseconds(1));
      }
    }
  }
}

////////////////////////////////////////////////////////
// ThreadPoolToken