Method manageWorkers

libminifi/src/utils/ThreadPool.cpp:115–174

template<typename T>
void ThreadPool<T>::manageWorkers() {
  for (int i = 0; i < max_worker_threads_; i++) {
    std::stringstream thread_name;
    thread_name << name_ << " #" << i;
    auto worker_thread = std::make_shared<WorkerThread>(thread_name.str());
    worker_thread->thread_ = createThread(std::bind(&ThreadPool::run_tasks, this, worker_thread));
    thread_queue_.push_back(worker_thread);
    current_workers_++;
  }

  if (daemon_threads_) {
    for (auto &thread : thread_queue_) {
      thread->thread_.detach();
    }
  }

  if (nullptr != thread_manager_) {
    while (running_) {
      auto waitperiod = std::chrono::milliseconds(500);
      {
        std::unique_lock<std::recursive_mutex> lock(manager_mutex_, std::try_to_lock);
        if (!lock.owns_lock()) {
          // Threadpool is being stopped/started or config is being changed, better wait a bit
          std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
        if (thread_manager_->isAboveMax(current_workers_)) {
          auto max = thread_manager_->getMaxConcurrentTasks();
          auto differential = current_workers_ - max;
          thread_reduction_count_ += differential;
        } else if (thread_manager_->shouldReduce()) {
          if (current_workers_ > 1)
            thread_reduction_count_++;
          thread_manager_->reduce();
        } else if (thread_manager_->canIncrease() && max_worker_threads_ > current_workers_) {  // increase slowly
          std::unique_lock<std::mutex> lock(worker_queue_mutex_);
          auto worker_thread = std::make_shared<WorkerThread>();
          worker_thread->thread_ = createThread(std::bind(&ThreadPool::run_tasks, this, worker_thread));
          if (daemon_threads_) {
            worker_thread->thread_.detach();
          }
          thread_queue_.push_back(worker_thread);
          current_workers_++;
        }
        std::shared_ptr<WorkerThread> thread_ref;
        while (deceased_thread_queue_.tryDequeue(thread_ref)) {
          std::unique_lock<std::mutex> lock(worker_queue_mutex_);
          if (thread_ref->thread_.joinable())
            thread_ref->thread_.join();
          thread_queue_.erase(std::remove(thread_queue_.begin(), thread_queue_.end(), thread_ref), thread_queue_.end());
        }
      }
      std::this_thread::sleep_for(waitperiod);
    }
  } else {
    for (auto &thread : thread_queue_) {
      if (thread->thread_.joinable())
        thread->thread_.join();
    }