ThreadPool.cpp

Go to the documentation of this file.

// Mantid Repository : https://github.com/mantidproject/mantid
//
// Copyright © 2018 ISIS Rutherford Appleton Laboratory UKRI,
//   NScD Oak Ridge National Laboratory, European Spallation Source,
//   Institut Laue - Langevin & CSNS, Institute of High Energy Physics, CAS
// SPDX - License - Identifier: GPL - 3.0 +
//----------------------------------------------------------------------
// Includes
//----------------------------------------------------------------------
#include "MantidKernel/ThreadPool.h"
 
#include "MantidKernel/ConfigService.h"
#include "MantidKernel/MultiThreaded.h"
#include "MantidKernel/ProgressBase.h"
#include "MantidKernel/Task.h"
#include "MantidKernel/ThreadPoolRunnable.h"
 
#include <Poco/Thread.h>
 
#include <algorithm>
#include <sstream>
#include <stdexcept>
// needed on windows and any place missing openmp
#if defined(_WIN32) || !defined(_OPENMP)
#include <Poco/Environment.h>
#endif
 
namespace Mantid::Kernel {
 
//--------------------------------------------------------------------------------
ThreadPool::ThreadPool(ThreadScheduler *scheduler, size_t numThreads, ProgressBase *prog)
    : m_scheduler(std::unique_ptr<ThreadScheduler>(scheduler)), m_started(false),
      m_prog(std::unique_ptr<ProgressBase>(prog)) {
  if (!m_scheduler)
    throw std::invalid_argument("NULL ThreadScheduler passed to ThreadPool constructor.");
 
  if (numThreads == 0) {
    // Uses Poco to find how many cores there are.
    m_numThreads = getNumPhysicalCores();
  } else
    m_numThreads = numThreads;
  // std::cout << m_numThreads << " m_numThreads \n";
}
 
//--------------------------------------------------------------------------------
ThreadPool::~ThreadPool() = default;
 
//--------------------------------------------------------------------------------
size_t ThreadPool::getNumPhysicalCores() {
// windows hangs with openmp for some reason
#if defined(_WIN32) || !defined(_OPENMP)
  int physicalCores = Poco::Environment::processorCount();
#else
  int physicalCores = PARALLEL_GET_MAX_THREADS;
#endif
 
  auto maxCores = Kernel::ConfigService::Instance().getValue<int>("MultiThreaded.MaxCores");
 
  if (!maxCores.is_initialized())
    return std::min(maxCores.get_value_or(0), physicalCores);
  else
    return physicalCores;
}
 
//--------------------------------------------------------------------------------
void ThreadPool::start(double waitSec) {
  if (m_started)
    throw std::runtime_error("Threads have already started.");
  // Now, launch that many threads and let them wait for new tasks.
  m_threads.clear();
  m_runnables.clear();
  for (size_t i = 0; i < m_numThreads; i++) {
    // Make a descriptive name
    std::ostringstream name;
    name << "Thread" << i;
    // Create the thread
    auto thread = std::make_unique<Poco::Thread>(name.str());
    // Make the runnable object and run it
    auto runnable = std::make_unique<ThreadPoolRunnable>(i, m_scheduler.get(), m_prog.get(), waitSec);
    thread->start(*runnable);
    m_threads.emplace_back(std::move(thread));
    m_runnables.emplace_back(std::move(runnable));
  }
  // Yep, all the threads are running.
  m_started = true;
}
 
//--------------------------------------------------------------------------------
void ThreadPool::schedule(const std::shared_ptr<Task> &task, bool start) {
  if (task) {
    m_scheduler->push(task);
    // Start all the threads if they were not already.
    if (start && !m_started)
      this->start();
  }
}
 
//--------------------------------------------------------------------------------
bool compareTasks(Task *lhs, Task *rhs) { return (lhs->cost() > rhs->cost()); }
 
//--------------------------------------------------------------------------------
void ThreadPool::joinAll() {
  // Are the threads REALLY started, or did they exit due to lack of tasks?
  if (m_started) {
    m_started = false;
    // If any of the threads are running, then YES, it is really started.
    for (auto &thread : m_threads)
      m_started = m_started || thread->isRunning();
  }
 
  // Start all the threads if they were not already.
  if (!m_started)
    this->start();
 
  // Clear any wait times so that the threads stop waiting for new tasks.
  for (auto &runnable : m_runnables)
    runnable->clearWait();
 
  // Sequentially join all the threads.
  for (auto &thread : m_threads) {
    thread->join();
  }
 
  // Clear the vectors (the threads are deleted now).
  m_threads.clear();
 
  // Get rid of the runnables too
  m_runnables.clear();
 
  // This will make threads restart
  m_started = false;
 
  // Did one of the threads abort or throw an exception?
  if (m_scheduler->getAborted()) {
    // Re-raise the error
    throw m_scheduler->getAbortException();
  }
}
 
} // namespace Mantid::Kernel