ThreadPoolSvc.cpp

Go to the documentation of this file.

#include "ThreadPoolSvc.h"

#include "GaudiKernel/SvcFactory.h"
#include "GaudiKernel/ConcurrencyFlags.h"
#include "ThreadInitTask.h"

#include "tbb/task_scheduler_init.h"
#include "tbb/task_scheduler_observer.h"
#include "tbb/task.h"
#include "tbb/tick_count.h"
#include "tbb/tbb_thread.h"

using namespace tbb;

DECLARE_SERVICE_FACTORY(ThreadPoolSvc)

//=============================================================================


ThreadPoolSvc::ThreadPoolSvc( const std::string& name, ISvcLocator* svcLoc ):
  base_class(name,svcLoc),
  m_threadInitTools(this),
  m_init(false),
  m_threadPoolSize(0),
  m_tbbSchedInit(nullptr),
  m_barrier(nullptr)
{

  declareProperty("ThreadInitTools", m_threadInitTools,
                  "ToolHandleArray of IThreadInitTools");

}

//-----------------------------------------------------------------------------

StatusCode
ThreadPoolSvc::initialize() {

  // Initialise mother class (read properties, ...)
  StatusCode sc(Service::initialize());
  if (!sc.isSuccess()) {
    warning () << "Base class could not be initialized" << endmsg;
    return StatusCode::FAILURE;
  }

  if (m_threadInitTools.retrieve().isFailure()) {
    error() << "Unable to retrieve ThreadInitTools Array" << endmsg;

    return StatusCode::FAILURE;
  } else {
    if (m_threadInitTools.size() != 0) {
      info() << "retrieved " << m_threadInitTools.size() << " thread init tools"
             << endmsg;
    } else {
      info() << "no thread init tools attached" << endmsg;
    }
  }

  return StatusCode::SUCCESS;

}

//-----------------------------------------------------------------------------

StatusCode
ThreadPoolSvc::finalize() {

  if (!m_init) {
    warning() << "Looks like the ThreadPoolSvc was created, but thread pool "
              << "was never initialized" << endmsg;
  }

  return StatusCode::SUCCESS;

}

//-----------------------------------------------------------------------------

StatusCode
ThreadPoolSvc::initPool(const int& poolSize) {

  tbb::spin_mutex::scoped_lock lock( m_initMutex );

  m_threadPoolSize = poolSize;

  if (msgLevel(MSG::DEBUG))
    debug() << "ThreadPoolSvc::initPool() poolSize = " << poolSize << endmsg;
  //There is a problem in the piece of the code below.  if
  // m_threadPoolSize is set to something negative which is < -1,
  // algorithm below might not behave as expected. For the time being
  // I've choosen to create the barrier with the default number of
  // threads created by the task scheduler init assuming that a
  // negative value will choose automatic thread creation which will
  // create default number of threads.
  // SK

  // -100 prevents the creation of the pool and the scheduler directly
  // executes the tasks.
  if (-100 != m_threadPoolSize) {
    if (msgLevel(MSG::DEBUG))
      debug() << "Initialising a thread pool of size "
              << m_threadPoolSize << endmsg;

    // Leave -1 in case selected, increment otherwise
    // - What?
    int thePoolSize = m_threadPoolSize;
    if (thePoolSize != -1) thePoolSize += 1;


    // Create the TBB task scheduler
    m_tbbSchedInit = std::unique_ptr<tbb::task_scheduler_init>( new tbb::task_scheduler_init(thePoolSize) );
    // Create the barrier for task synchronization
    if(m_threadPoolSize<=-1)thePoolSize=m_tbbSchedInit->default_num_threads();
    if (msgLevel(MSG::DEBUG)){
      debug() << "creating barrier of size " << thePoolSize << endmsg;
    }
    Gaudi::Concurrency::ConcurrencyFlags::setNumThreads(thePoolSize);

    m_barrier = std::unique_ptr<boost::barrier>( new boost::barrier(thePoolSize) );

  } else {
    Gaudi::Concurrency::ConcurrencyFlags::setNumThreads(1);
  }

  // Launch the init tool tasks
  const bool terminate = false;
  if (launchTasks(terminate).isFailure())
    return StatusCode::FAILURE;

  if (msgLevel(MSG::DEBUG))
    debug() << "Thread Pool initialization complete!" << endmsg;

  m_init = true;

  return StatusCode::SUCCESS;

}

//-----------------------------------------------------------------------------

StatusCode
ThreadPoolSvc::terminatePool() {
  tbb::spin_mutex::scoped_lock lock( m_initMutex );
  if (msgLevel(MSG::DEBUG))
    debug() << "ThreadPoolSvc::terminatePool()" << endmsg;

  if (!m_init) {
    error() << "Trying to terminate uninitialized thread pool!" << endmsg;
    return StatusCode::FAILURE;
  }

  // Launch the termination tasks
  const bool terminate = true;
  if (launchTasks(terminate).isFailure())
    return StatusCode::FAILURE;

  if (msgLevel(MSG::DEBUG))
    debug() << "Thread pool termination complete!" << endmsg;

  return StatusCode::SUCCESS;
}

//-----------------------------------------------------------------------------

StatusCode
ThreadPoolSvc::launchTasks(bool terminate) {

  if (m_threadInitTools.empty()) return StatusCode::SUCCESS;

  const std::string taskType = terminate ? "termination" : "initialization";

  // If we have a thread pool (via a scheduler), then we want to queue
  // the tasks in TBB to execute on each thread.
  if(m_tbbSchedInit) {

    // Create one task for each worker thread in the pool
    for (int i = 0; i < m_threadPoolSize; ++i) {
      if (msgLevel(MSG::DEBUG))
        debug() << "creating ThreadInitTask " << i << endmsg;
      tbb::task* t = new(tbb::task::allocate_root())
        ThreadInitTask( m_threadInitTools, m_barrier.get(), serviceLocator(), terminate );

      // Queue the task
      tbb::task::enqueue( *t );
      this_tbb_thread::sleep(tbb::tick_count::interval_t(.1));
    }

    // Now wait for all the workers to reach the barrier
    if (msgLevel(MSG::DEBUG))
      debug() << "waiting at barrier for all ThreadInitTool to finish executing" << endmsg;
    m_barrier->wait();

    // Check to make sure all Tools were invoked.
    // I'm not sure this mechanism is worthwhile.
    for (auto& t : m_threadInitTools) {
      // Number of threads initialized but not terminated.
      int numInit = t->nInit();
      // Expected number based on the type of task.
      int expectedNumInit = terminate? 0 : m_threadPoolSize;
      if (numInit != expectedNumInit) {
        error() << "not all threads " << (terminate? "terminated" : "initialized")
                << " for tool " << t << " : "
                << t->nInit() << " out of " << m_threadPoolSize
                << " are currently active" << endmsg;
        return StatusCode::FAILURE;
      }
    }

  }

  // In single-threaded mode, there is no scheduler, so we simply call
  // the task wrapper directly in this thread.
  else {
    if (msgLevel(MSG::DEBUG))
      debug() << "launching ThreadInitTask " << taskType << "in this thread." << endmsg;
    boost::barrier* noBarrier = nullptr;
    ThreadInitTask theTask(m_threadInitTools, noBarrier, serviceLocator(), terminate);
    theTask.execute();
  }

  // Now, we do some error checking
  if (ThreadInitTask::execFailed()) {
    error() << "a ThreadInitTask failed to execute successfully" << endmsg;
    return StatusCode::FAILURE;
  }

  return StatusCode::SUCCESS;
}