IOBoundAlgSchedulerSvc.cpp

Go to the documentation of this file.

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

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

// Local
#include "IOBoundAlgSchedulerSvc.h"

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

//===========================================================================
// Infrastructure methods

IOBoundAlgSchedulerSvc::IOBoundAlgSchedulerSvc( const std::string& name, ISvcLocator* svcLoc ):
 base_class(name,svcLoc),
 m_isActive(false)
{}

//---------------------------------------------------------------------------
IOBoundAlgSchedulerSvc::~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([this]() -> 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;
}

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