IOBoundAlgSchedulerSvc.cpp

Go to the documentation of this file.

#include <algorithm>
#include <map>
#include <queue>
#include <sstream>
#include <thread>
#include <unordered_set>

// Framework includes
#include "GaudiAlg/GaudiAlgorithm.h"
#include "GaudiKernel/IAlgorithm.h"
#include "GaudiKernel/IDataManagerSvc.h"
#include <Gaudi/Algorithm.h> // will be IAlgorithm if context getter promoted to interface

// Local
#include "IOBoundAlgSchedulerSvc.h"

// Instantiation of a static factory class used by clients to create instances of this service
DECLARE_COMPONENT( IOBoundAlgSchedulerSvc )


StatusCode IOBoundAlgSchedulerSvc::initialize() {

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

  // Activate the scheduler in another thread.
  info() << "Activating scheduler in a separate thread" << endmsg;
  m_thread = std::thread( std::bind( &IOBoundAlgSchedulerSvc::activate, this ) );

  return sc;
}
//---------------------------------------------------------------------------

StatusCode IOBoundAlgSchedulerSvc::finalize() {

  StatusCode sc( Service::finalize() );
  if ( !sc.isSuccess() ) warning() << "Base class could not be finalized" << endmsg;

  sc = deactivate();
  if ( !sc.isSuccess() ) warning() << "Scheduler could not be deactivated" << endmsg;

  info() << "Joining preemptive scheduler's thread" << endmsg;
  m_thread.join();

  return sc;
}
//---------------------------------------------------------------------------
void IOBoundAlgSchedulerSvc::activate() {

  // Now it's running
  m_isActive = true;

  // Wait for actions pushed into the queue by finishing tasks.
  action     thisAction;
  StatusCode sc( StatusCode::SUCCESS );

  // Continue to wait if the scheduler is running or there is something to do
  info() << "Start checking the queue of I/O-bound algorithm tasks.." << endmsg;
  while ( m_isActive or m_actionsQueue.size() > 0 ) {
    m_actionsQueue.pop( thisAction );
    std::thread th( thisAction );
    th.detach();
  }
}

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

StatusCode IOBoundAlgSchedulerSvc::deactivate() {

  if ( m_isActive ) {
    // Drain the scheduler
    // m_actionsQueue.push(std::bind(&IOBoundSchedulerSvc::m_drain,
    //                              this));
    // we set the flag in this thread, not in the last action, to avoid stall,
    // since we execute tasks asynchronously, in a detached thread, and it's possible that
    // the task doesn't complete by the time the last while-iteration is entered
    m_isActive = false;
    // This would be the last (empty) action, just to trigger one last while-iteration
    m_actionsQueue.push( []() -> StatusCode { return StatusCode::SUCCESS; } );
  }

  return StatusCode::SUCCESS;
}

StatusCode IOBoundAlgSchedulerSvc::push( IAlgTask& task ) {

  // the temporary lambda should be moved into the queue in here
  auto actionn = [&]() {
    debug() << " .. launching I/O-bound algo-closure .. " << endmsg;
    return task.execute();
  };

  m_actionsQueue.push( actionn );

  // until the queue to accelerator becomes limited
  return StatusCode::SUCCESS;
}

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