ForwardSchedulerSvc.cpp

Go to the documentation of this file.

// 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>
#include "tbb/task.h"

// C++
#include <unordered_set>
#include <algorithm>
#include <map>
#include <sstream>
#include <queue>

// Local
#include "ForwardSchedulerSvc.h"
#include "AlgoExecutionTask.h"
#include "AlgResourcePool.h"
#include "EFGraphVisitors.h"

// External libs
// DP waiting for the TBB service
#include "tbb/task_scheduler_init.h"

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

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

ForwardSchedulerSvc::ForwardSchedulerSvc( const std::string& name, ISvcLocator* svcLoc ):
 base_class(name,svcLoc),
 m_isActive(false),
 m_algosInFlight(0),
 m_updateNeeded(true)
{
  declareProperty("MaxEventsInFlight", m_maxEventsInFlight = 0 );
  declareProperty("ThreadPoolSize", m_threadPoolSize = -1 );
  declareProperty("WhiteboardSvc", m_whiteboardSvcName = "EventDataSvc" );
  // Will disappear when dependencies are properly propagated into the C++ code of the algos
  declareProperty("AlgosDependencies", m_algosDependencies);
  declareProperty("MaxAlgosInFlight", m_maxAlgosInFlight = 0, "Taken from the whiteboard. Deprecated" );
  // XXX: CF tests. Temporary property to switch between ControlFlow implementations
  declareProperty("useGraphFlowManagement", m_CFNext = false );
  declareProperty("DataFlowManagerNext", m_DFNext = false );
  declareProperty("SimulateExecution", m_simulateExecution = false );
  declareProperty("Optimizer", m_optimizationMode = "", "The following modes are currently available: PCE, COD, DRE, E" );
  declareProperty("DumpIntraEventDynamics", m_dumpIntraEventDynamics = false, "Dump intra-event concurrency dynamics to csv file" );
}

//---------------------------------------------------------------------------
ForwardSchedulerSvc::~ForwardSchedulerSvc(){}
//---------------------------------------------------------------------------

StatusCode ForwardSchedulerSvc::initialize(){

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

  // Get the algo resource pool
  m_algResourcePool = serviceLocator()->service("AlgResourcePool");
  if (!m_algResourcePool.isValid())
    error() << "Error retrieving AlgoResourcePool" << endmsg;

  // Get Whiteboard
  m_whiteboard = serviceLocator()->service(m_whiteboardSvcName);
  if (!m_whiteboard.isValid())
    fatal() << "Error retrieving EventDataSvc interface IHiveWhiteBoard." << endmsg;

  // Check the MaxEventsInFlight parameters and react
  // Deprecated for the moment
  size_t numberOfWBSlots = m_whiteboard->getNumberOfStores();
  if (m_maxEventsInFlight!=0){
    warning() << "Property MaxEventsInFlight was set. This works but it's deprecated. "
              << "Please migrate your code options files." << endmsg;

    if (m_maxEventsInFlight != (int)numberOfWBSlots){
      warning() << "In addition, the number of events in flight ("
                << m_maxEventsInFlight << ") differs from the slots in the whiteboard ("
                << numberOfWBSlots << "). Setting the number of events in flight to "
                << numberOfWBSlots << endmsg;
    }
  }

  // Align the two quantities
  m_maxEventsInFlight = numberOfWBSlots;

  // Set the number of free slots
  m_freeSlots=m_maxEventsInFlight;

  // Get the list of algorithms
  const std::list<IAlgorithm*>& algos = m_algResourcePool->getFlatAlgList();
  const unsigned int algsNumber = algos.size();
  info() << "Found " <<  algsNumber << " algorithms" << endmsg;

  const unsigned int algosDependenciesSize=m_algosDependencies.size();
  info() << "Algodependecies size is " << algosDependenciesSize << endmsg;

  /* Dependencies
   0) Read deps from config file
   1) Look for handles in algo, if none
   2) Assume none are required
  */
  if (algosDependenciesSize == 0) {
    for (IAlgorithm* ialgoPtr : algos) {
      Algorithm* algoPtr = dynamic_cast<Algorithm*>(ialgoPtr);
      if (nullptr == algoPtr)
        fatal() << "Could not convert IAlgorithm into Algorithm: this will result in a crash." << endmsg;

#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
      const std::vector<MinimalDataObjectHandle*>& algoHandles(algoPtr->handles());
#pragma GCC diagnostic pop
      std::vector<std::string> algoDependencies;
      if (!algoHandles.empty()) {
        info() << "Algorithm " << algoPtr->name() << " data dependencies:" << endmsg;
        for (MinimalDataObjectHandle* handlePtr : algoHandles) {
          if (handlePtr->isValid()) {
            if (handlePtr->accessType() == MinimalDataObjectHandle::AccessType::READ) {
              const std::string& productName = handlePtr->dataProductName();
              info() << "  o READ Handle found for product " << productName << endmsg;
              algoDependencies.emplace_back(productName);

              //just for info output alternative locations
              if (handlePtr->alternativeDataProductNames().size() != 0) {
                info() << "\t\t alternative locations";
                for (auto s : handlePtr->alternativeDataProductNames())
                  info() << " " << s;
                info() << endmsg;
              }
            } else {
              //output WRITE handles just for info
              info() << "  o WRITE Handle found for product " << handlePtr->dataProductName() << endmsg;
            }
          }
        }
      } else {
        info() << "Algorithm " << algoPtr->name() << " has no data dependencies." << endmsg;
      }
      m_algosDependencies.emplace_back(algoDependencies);
    }
  } else {
    if (algsNumber != algosDependenciesSize){
      error() << "number of Algorithms is different from size of Data Dependency list!" << endmsg;
      return StatusCode::FAILURE;
    }
  }

  // Fill the containers to convert algo names to index
  m_algname_vect.reserve(algsNumber);
  unsigned int index=0;
  for (IAlgorithm* algo : algos){
    const std::string& name = algo->name();
    m_algname_index_map[name]=index;
    m_algname_vect.emplace_back(name);
    index++;
  }

  // prepare the control flow part
  if (m_CFNext) m_DFNext = true; //force usage of new data flow machinery when new control flow is used
  if (!m_CFNext && !m_optimizationMode.empty()) {
    fatal() << "Execution optimization is only available with the graph-based execution flow management" << endmsg;
    return StatusCode::FAILURE;
  }
  const AlgResourcePool* algPool = dynamic_cast<const AlgResourcePool*>(m_algResourcePool.get());
  sc = m_efManager.initialize(algPool->getExecutionFlowGraph(), m_algname_index_map, m_eventSlots, m_optimizationMode);
  unsigned int controlFlowNodeNumber = m_efManager.getExecutionFlowGraph()->getControlFlowNodeCounter();
  // Shortcut for the message service
  SmartIF<IMessageSvc> messageSvc (serviceLocator());
  if (!messageSvc.isValid())
    error() << "Error retrieving MessageSvc interface IMessageSvc." << endmsg;

  m_eventSlots.assign(m_maxEventsInFlight,EventSlot(m_algosDependencies,algsNumber,controlFlowNodeNumber,messageSvc));
  std::for_each(m_eventSlots.begin(),m_eventSlots.end(),[](EventSlot& slot){slot.complete=true;});

  // Clearly inform about the level of concurrency
  info() << "Concurrency level information:" << endmsg;
  info() << " o Number of events in flight: " << m_maxEventsInFlight << endmsg;
  info() << " o Number of algorithms in flight: " << m_maxAlgosInFlight << endmsg;
  info() << " o TBB thread pool size: " << m_threadPoolSize << endmsg;

  // Simulating execution flow by only analyzing the graph topology and logic
  if (m_simulateExecution) {
    auto vis = concurrency::RunSimulator(0);
    m_efManager.simulateExecutionFlow(vis);
  }

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

  return sc;

}
//---------------------------------------------------------------------------

StatusCode ForwardSchedulerSvc::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 Scheduler thread" << endmsg;
  m_thread.join();

  //m_efManager.getExecutionFlowGraph()->dumpExecutionPlan();

  return sc;

  }
//---------------------------------------------------------------------------
void ForwardSchedulerSvc::activate(){

  // Now it's running
  m_isActive=true;

  tbb::task_scheduler_init* TBBSchedInit = nullptr;

  // -100 prevents the creation of the pool and the scheduler directly executes
  // the tasks.
  if (-100 != m_threadPoolSize) {
    debug() << "Initialising a TBB thread pool of requested size " << m_threadPoolSize << endmsg;
    // Leave -1 in case selected, increment otherwise
    int thePoolSize = m_threadPoolSize;
    if (thePoolSize == -1)
      debug() << "...default TBB thread pool size amounts to " << tbb::task_scheduler_init::default_num_threads()<< endmsg;
    if (thePoolSize != -1)
      thePoolSize += 1;
    TBBSchedInit = new tbb::task_scheduler_init (thePoolSize);
  } else {
    debug() << "Thread pool size is one. Pool not initialised." << endmsg;
  }
  // 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 actionsQueue" << endmsg;
  while(m_isActive or m_actionsQueue.size()!=0){
    m_actionsQueue.pop(thisAction);
    sc = thisAction();
    if (sc!=StatusCode::SUCCESS)
      verbose() << "Action did not succeed (which is not bad per se)." << endmsg;
    else
      verbose() << "Action succeeded." << endmsg;
  }

  if (TBBSchedInit)
    delete TBBSchedInit;
}

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

StatusCode ForwardSchedulerSvc::deactivate(){

  if (m_isActive){
    // Drain the scheduler
    m_actionsQueue.push(std::bind(&ForwardSchedulerSvc::m_drain,
                                  this));
    // This would be the last action
    m_actionsQueue.push([this]() -> StatusCode {m_isActive=false;return StatusCode::SUCCESS;});
  }

  return StatusCode::SUCCESS;
}

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

//===========================================================================
// Utils and shortcuts

inline const std::string& ForwardSchedulerSvc::index2algname (unsigned int index) {
  return m_algname_vect[index];
}

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

inline unsigned int ForwardSchedulerSvc::algname2index(const std::string& algoname) {
  unsigned int index = m_algname_index_map[algoname];
  return index;
}

//===========================================================================
// EventSlot management
StatusCode ForwardSchedulerSvc::pushNewEvent(EventContext* eventContext){

  if (!eventContext){
    fatal() << "Event context is nullptr" << endmsg;
    return StatusCode::FAILURE;
  }

  if (m_freeSlots.load() == 0) {
    if (msgLevel(MSG::DEBUG))
      debug() << "A free processing slot could not be found." << endmsg;
    return StatusCode::FAILURE;
  }

  //no problem as push new event is only called from one thread (event loop manager)
  m_freeSlots--;

  auto action = [this,eventContext] () -> StatusCode {
    // Event processing slot forced to be the same as the wb slot
    const unsigned int thisSlotNum = eventContext->slot();
    EventSlot& thisSlot = m_eventSlots[thisSlotNum];
    if (!thisSlot.complete)
      fatal() << "The slot " << thisSlotNum << " is supposed to be a finished event but it's not" << endmsg;
    info() << "A free processing slot was found." << endmsg;
    thisSlot.reset(eventContext);
    // XXX: CF tests
    if (m_CFNext) {
      auto vis = concurrency::Trigger(thisSlotNum);
      m_efManager.touchReadyAlgorithms(vis);
    }

    return this->updateStates(thisSlotNum);
  }; // end of lambda

  // Kick off the scheduling!
  if (msgLevel(MSG::VERBOSE)) {
    verbose() << "Pushing the action to update the scheduler for slot " <<  eventContext->slot() << endmsg;
    verbose() << "Free slots available " <<  m_freeSlots.load() << endmsg;
  }
  m_actionsQueue.push(action);

  return StatusCode::SUCCESS;
}

//---------------------------------------------------------------------------
StatusCode ForwardSchedulerSvc::pushNewEvents(std::vector<EventContext*>& eventContexts){
  StatusCode sc;
  for (auto context : eventContexts){
    sc = pushNewEvent(context);
    if (sc != StatusCode::SUCCESS) return sc;
  }
  return sc;
}

//---------------------------------------------------------------------------
unsigned int ForwardSchedulerSvc::freeSlots(){
  return std::max(m_freeSlots.load(),0);
}

//---------------------------------------------------------------------------
StatusCode ForwardSchedulerSvc::m_drain(){

  unsigned int slotNum=0;
  for (auto& thisSlot : m_eventSlots){
    if (not thisSlot.algsStates.allAlgsExecuted() and not thisSlot.complete){
      updateStates(slotNum);
    }
    slotNum++;
  }
  return StatusCode::SUCCESS;
}

//---------------------------------------------------------------------------
StatusCode ForwardSchedulerSvc::popFinishedEvent(EventContext*& eventContext){
  if (m_freeSlots.load() == m_maxEventsInFlight or
      !m_isActive) {
    return StatusCode::FAILURE;
  } else {
    m_finishedEvents.pop(eventContext);
    m_freeSlots++;
    debug() << "Popped slot " << eventContext->slot() << "(event "
            << eventContext->evt() << ")" << endmsg;
    return StatusCode::SUCCESS;
  }
}

//---------------------------------------------------------------------------
StatusCode ForwardSchedulerSvc::tryPopFinishedEvent(EventContext*& eventContext){
  if (m_finishedEvents.try_pop(eventContext)) {
    if (msgLevel(MSG::DEBUG))
      debug() << "Try Pop successful slot " << eventContext->slot()
              << "(event " << eventContext->evt() << ")" << endmsg;
    m_freeSlots++;
    return StatusCode::SUCCESS;
  }
  return StatusCode::FAILURE;
}

//---------------------------------------------------------------------------
StatusCode ForwardSchedulerSvc::eventFailed(EventContext* eventContext){

  // Set the number of slots available to an error code
  m_freeSlots.store(0);

  fatal() << "*** Event " << eventContext->evt() << " on slot "
          << eventContext->slot() << " failed! ***" << endmsg;

  //dumpSchedulerState(-1);

  // Empty queue and deactivate the service
  action thisAction;
  while(m_actionsQueue.try_pop(thisAction)){};
  deactivate();

  // Push into the finished events queue the failed context
  EventContext* thisEvtContext;
  while(m_finishedEvents.try_pop(thisEvtContext)) { m_finishedEvents.push(thisEvtContext); };
  m_finishedEvents.push(eventContext);

  return StatusCode::FAILURE;

}

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

//===========================================================================
// States Management

StatusCode ForwardSchedulerSvc::updateStates(int si, const std::string& algo_name){

  m_updateNeeded=true;

  // Fill a map of initial state / action using closures.
  // done to update the states w/o several if/elses
  // Posterchild for constexpr with gcc4.7 onwards!
  /*const std::map<AlgsExecutionStates::State, std::function<StatusCode(unsigned int iAlgo, int si)>>
   statesTransitions = {
  {AlgsExecutionStates::CONTROLREADY, std::bind(&ForwardSchedulerSvc::promoteToDataReady,
                                      this,
                                      std::placeholders::_1,
                                      std::placeholders::_2)},
  {AlgsExecutionStates::DATAREADY, std::bind(&ForwardSchedulerSvc::promoteToScheduled,
                                   this,
                                   std::placeholders::_1,
                                   std::placeholders::_2)}
  };*/

  StatusCode global_sc(StatusCode::FAILURE);
  StatusCode partial_sc;

   // Sort from the oldest to the newest event
   // Prepare a vector of pointers to the slots to avoid copies
   std::vector<EventSlot*> eventSlotsPtrs;

   // Consider all slots if si <0 or just one otherwise
   if (si<0) {
     const int eventsSlotsSize(m_eventSlots.size());
     eventSlotsPtrs.reserve(eventsSlotsSize);
     for (auto slotIt=m_eventSlots.begin();slotIt!=m_eventSlots.end();slotIt++) {
       if (!slotIt->complete)
         eventSlotsPtrs.push_back(&(*slotIt));
     }
     std::sort(eventSlotsPtrs.begin(),
               eventSlotsPtrs.end(),
               [](EventSlot* a, EventSlot* b) {return a->eventContext->evt() < b->eventContext->evt();});
   } else {
     eventSlotsPtrs.push_back(&m_eventSlots[si]);
   }

  for (EventSlot* thisSlotPtr : eventSlotsPtrs) {
    int iSlot = thisSlotPtr->eventContext->slot();

    // Cache the states of the algos to improve readability and performance
    auto& thisSlot = m_eventSlots[iSlot];
    AlgsExecutionStates& thisAlgsStates = thisSlot.algsStates;

    // Take care of the control ready update
    // XXX: CF tests
    if (!m_CFNext) {
      m_efManager.updateEventState(thisAlgsStates,thisSlot.controlFlowState);
    } else {
      if (!algo_name.empty())
        m_efManager.updateDecision(algo_name,iSlot,thisAlgsStates,thisSlot.controlFlowState);
    }


    //DF note: all this this is a loop over all algs and applies CR->DR and DR->SCHD transistions
    /*for (unsigned int iAlgo=0;iAlgo<m_algname_vect.size();++iAlgo){
        const AlgsExecutionStates::State& algState = thisAlgsStates[iAlgo];
        if (algState==AlgsExecutionStates::ERROR)
            error() << " Algo " << index2algname(iAlgo) << " is in ERROR state." << endmsg;
        // Loop on state transitions from the one suited to algo state up to the one for SCHEDULED.
        partial_sc=StatusCode::SUCCESS;
        for (auto state_transition = statesTransitions.find(algState);
                state_transition!=statesTransitions.end() && partial_sc.isSuccess();
                state_transition++){
            partial_sc = state_transition->second(iAlgo,iSlot);
            if (partial_sc.isFailure()){
                debug() << "Could not apply transition from "
                        << AlgsExecutionStates::stateNames[thisAlgsStates[iAlgo]]
                                                           << " for algorithm " << index2algname(iAlgo)
                                                           << " on processing slot " << iSlot << endmsg;
            }
            else{global_sc=partial_sc;}
        } // end loop on transitions
    }*/ // end loop on algos


    StatusCode partial_sc;
    //first update CONTROLREADY to DATAREADY
    if (!m_CFNext) {
      for(auto it = thisAlgsStates.begin(AlgsExecutionStates::State::CONTROLREADY);
          it != thisAlgsStates.end(AlgsExecutionStates::State::CONTROLREADY); ++it) {

        uint algIndex = *it;
        partial_sc = promoteToDataReady(algIndex, iSlot);
        if (partial_sc.isFailure())
          if (msgLevel(MSG::DEBUG))
            debug() << "Could not apply transition from "
                    << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::CONTROLREADY]
                    << " for algorithm " << index2algname(algIndex) << " on processing slot " << iSlot << endmsg;
      }
    }

    //now update DATAREADY to SCHEDULED
    if (!m_optimizationMode.empty()) {
      auto comp_nodes = [this] (const uint& i,const uint& j) {
          return (m_efManager.getExecutionFlowGraph()->getAlgorithmNode(index2algname(i))->getRank() <
          m_efManager.getExecutionFlowGraph()->getAlgorithmNode(index2algname(j))->getRank());
      };
      std::priority_queue<uint,std::vector<uint>,std::function<bool(const uint&,const uint&)>> buffer(comp_nodes,std::vector<uint>());
      for(auto it = thisAlgsStates.begin(AlgsExecutionStates::State::DATAREADY);
          it != thisAlgsStates.end(AlgsExecutionStates::State::DATAREADY); ++it)
        buffer.push(*it);
      /*std::stringstream s;
      auto buffer2 = buffer;
      while (!buffer2.empty()) {
        s << m_efManager.getExecutionFlowGraph()->getAlgorithmNode(index2algname(buffer2.top()))->getRank() << ", ";
        buffer2.pop();
      }
      info() << "DRBuffer is: [ " << s.str() << " ]  <--" << algo_name << " executed" << endmsg;*/

      while (!buffer.empty()) {
        partial_sc = promoteToScheduled(buffer.top(), iSlot);
        if (partial_sc.isFailure())
          if (msgLevel(MSG::DEBUG))
            debug() << "Could not apply transition from "
                    << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
                    << " for algorithm " << index2algname(buffer.top()) << " on processing slot " << iSlot << endmsg;
        buffer.pop();
      }

    } else {
      for(auto it = thisAlgsStates.begin(AlgsExecutionStates::State::DATAREADY);
          it != thisAlgsStates.end(AlgsExecutionStates::State::DATAREADY); ++it) {
        uint algIndex = *it;
        partial_sc = promoteToScheduled(algIndex, iSlot);
        if (partial_sc.isFailure())
          if (msgLevel(MSG::DEBUG))
            debug() << "Could not apply transition from "
                    << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
                    << " for algorithm " << index2algname(algIndex) << " on processing slot " << iSlot << endmsg;

      }
    }

    if (m_dumpIntraEventDynamics) {
      std::stringstream s;
      s << algo_name << ", " << thisAlgsStates.sizeOfSubset(State::CONTROLREADY)
                     << ", " << thisAlgsStates.sizeOfSubset(State::DATAREADY)
                     << ", " << thisAlgsStates.sizeOfSubset(State::SCHEDULED) << "\n";
      auto threads = (m_threadPoolSize != -1) ? std::to_string(m_threadPoolSize)
                                              : std::to_string(tbb::task_scheduler_init::default_num_threads());
      std::ofstream myfile;
      myfile.open("IntraEventConcurrencyDynamics_" + threads + "T.csv", std::ios::app);
      myfile << s.str();
      myfile.close();
    }


    // Not complete because this would mean that the slot is already free!
    if (!thisSlot.complete &&
        m_efManager.rootDecisionResolved(thisSlot.controlFlowState) &&
        !thisSlot.algsStates.algsPresent(AlgsExecutionStates::CONTROLREADY) &&
        !thisSlot.algsStates.algsPresent(AlgsExecutionStates::DATAREADY) &&
        !thisSlot.algsStates.algsPresent(AlgsExecutionStates::SCHEDULED)) {

      thisSlot.complete=true;
      // if the event did not fail, add it to the finished events
      // otherwise it is taken care of in the error handling already
      if (!thisSlot.eventContext->evtFail()) {
        m_finishedEvents.push(thisSlot.eventContext);
        if (msgLevel(MSG::DEBUG))
          debug() << "Event " << thisSlot.eventContext->evt() << " finished (slot "
                  << thisSlot.eventContext->slot() << ")." << endmsg;
      }
      // now let's return the fully evaluated result of the control flow
      if (msgLevel(MSG::DEBUG)) {
        std::stringstream ss;
        m_efManager.printEventState(ss, thisSlot.algsStates, thisSlot.controlFlowState,0);
        debug() << ss.str() << endmsg;
      }

      thisSlot.eventContext= nullptr;
    } else {
      StatusCode eventStalledSC = isStalled(iSlot);
      if (! eventStalledSC.isSuccess())
        eventFailed(thisSlot.eventContext);
    }
  } // end loop on slots

  verbose() << "States Updated." << endmsg;

  return global_sc;
}

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

StatusCode ForwardSchedulerSvc::isStalled(int iSlot) {
  // Get the slot
  EventSlot& thisSlot = m_eventSlots[iSlot];

  if (m_actionsQueue.empty() &&
      m_algosInFlight == 0 &&
      (!thisSlot.algsStates.algsPresent(AlgsExecutionStates::DATAREADY))) {

    info() << "About to declare a stall" << endmsg;
    fatal() << "*** Stall detected! ***\n" << endmsg;
    dumpSchedulerState(iSlot);
    //throw GaudiException ("Stall detected",name(),StatusCode::FAILURE);

    return StatusCode::FAILURE;
  }
  return StatusCode::SUCCESS;
}

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

void ForwardSchedulerSvc::dumpSchedulerState(int iSlot) {

  // To have just one big message
  std::stringstream outputMessageStream;

  int slotCount = -1;
  for (auto thisSlot : m_eventSlots){
    slotCount++;
    outputMessageStream.str(std::string());
    if ( thisSlot.complete )
      continue;

    outputMessageStream << "Dump of Scheduler State for slot " << thisSlot.eventContext->evt() << std::endl;

    if ( 0 > iSlot or iSlot == slotCount) {
        outputMessageStream << "Algorithms states for event " << thisSlot.eventContext->evt() << std::endl;

      const std::vector<std::string>& wbSlotContent ( thisSlot.dataFlowMgr.content() );
      for (unsigned int algoIdx=0; algoIdx < thisSlot.algsStates.size(); ++algoIdx ) {
        outputMessageStream << " o " << index2algname(algoIdx)
                            << " was in state " << AlgsExecutionStates::stateNames[thisSlot.algsStates[algoIdx]]
                            << ". Its data dependencies are ";
        std::vector<std::string> deps (thisSlot.dataFlowMgr.dataDependencies(algoIdx));
        const int depsSize=deps.size();
        if (depsSize==0)
          outputMessageStream << " none.";

        for (int i=0;i<depsSize;++i)
          outputMessageStream << deps[i] << (i==(depsSize-1) ? "" :", ");

        // Now list what dependencies were available
        // With std::algorithms, move the ones which are missing at the beginning of the vector
        std::vector<std::string>::iterator missinngDepsEndIt =
                        std::remove_if(deps.begin(), // from the beginning of the deps
                                       deps.end(),   // to their end
                                       [&wbSlotContent] (std::string dep) { // remove if this lambda returns true
          return std::count(wbSlotContent.begin(),wbSlotContent.end(),dep)!=0; //look for dep in wb content
        });

        if (deps.begin() != missinngDepsEndIt) {
          outputMessageStream << ". The following are missing: ";
          for (std::vector<std::string>::iterator missingDep=deps.begin();missingDep!=missinngDepsEndIt;++missingDep)
            outputMessageStream << *missingDep << (missingDep==(missinngDepsEndIt-1)?"":", ");
        }

        outputMessageStream << std::endl;
      }

      fatal() << outputMessageStream.str() << endmsg;
      outputMessageStream.str(std::string());

      // Snapshot of the WhiteBoard
      outputMessageStream << "The content of the whiteboard for this event was:\n";
      for (auto& product : wbSlotContent )
        outputMessageStream << " o " << product << std::endl;

      fatal() << outputMessageStream.str()<< endmsg;
      outputMessageStream.str(std::string());

      // Snapshot of the ControlFlow
      outputMessageStream << "The status of the control flow for this event was:\n";
      std::stringstream cFlowStateStringStream;
      m_efManager.printEventState(cFlowStateStringStream, thisSlot.algsStates, thisSlot.controlFlowState,0);

      outputMessageStream << cFlowStateStringStream.str();

      fatal() <<  outputMessageStream.str() << endmsg;
    }
  }

}

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

StatusCode ForwardSchedulerSvc::promoteToControlReady(unsigned int iAlgo, int si) {

  // Do the control flow
  StatusCode sc = m_eventSlots[si].algsStates.updateState(iAlgo,AlgsExecutionStates::CONTROLREADY);
  if (sc.isSuccess())
    if (msgLevel(MSG::DEBUG))
      debug() << "Promoting " << index2algname(iAlgo) << " to CONTROLREADY" << endmsg;

  return sc;

}

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

StatusCode ForwardSchedulerSvc::promoteToDataReady(unsigned int iAlgo, int si) {

  StatusCode sc;
  if (!m_DFNext) {
    sc = m_eventSlots[si].dataFlowMgr.canAlgorithmRun(iAlgo);
  } else {
    sc = m_efManager.algoDataDependenciesSatisfied(index2algname(iAlgo),si);
  }

  StatusCode updateSc(StatusCode::FAILURE);
  if (sc == StatusCode::SUCCESS)
    updateSc = m_eventSlots[si].algsStates.updateState(iAlgo,AlgsExecutionStates::DATAREADY);

  if (updateSc.isSuccess())
    if (msgLevel(MSG::DEBUG))
      debug() << "Promoting " << index2algname(iAlgo) << " to DATAREADY" << endmsg;

  return updateSc;

}

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

StatusCode ForwardSchedulerSvc::promoteToScheduled(unsigned int iAlgo, int si) {

  if (m_algosInFlight == m_maxAlgosInFlight)
    return StatusCode::FAILURE;

  const std::string& algName(index2algname(iAlgo));

  IAlgorithm* ialgoPtr=nullptr;
  StatusCode sc ( m_algResourcePool->acquireAlgorithm(algName,ialgoPtr) );

  if (sc.isSuccess()) {
    Algorithm* algoPtr = dynamic_cast<Algorithm*> (ialgoPtr); // DP: expose the setter of the context?
    EventContext* eventContext ( m_eventSlots[si].eventContext );
    if (!eventContext)
      fatal() << "Event context for algorithm " << algName << " is a nullptr (slot " << si<< ")" << endmsg;

    algoPtr->setContext(m_eventSlots[si].eventContext);
    ++m_algosInFlight;
    // Avoid to use tbb if the pool size is 1 and run in this thread
    if (-100 != m_threadPoolSize) {
      tbb::task* t = new( tbb::task::allocate_root() ) AlgoExecutionTask(ialgoPtr, iAlgo, serviceLocator(), this);
      tbb::task::enqueue( *t);
    } else {
      AlgoExecutionTask theTask(ialgoPtr, iAlgo, serviceLocator(), this);
      theTask.execute();
    }

    if (msgLevel(MSG::DEBUG))
      debug() << "Algorithm " << algName << " was submitted on event " << eventContext->evt()
              << ". Algorithms scheduled are " << m_algosInFlight << endmsg;

    StatusCode updateSc ( m_eventSlots[si].algsStates.updateState(iAlgo,AlgsExecutionStates::SCHEDULED) );

    if (updateSc.isSuccess())
      if (msgLevel(MSG::DEBUG))
        debug() << "Promoting " << index2algname(iAlgo) << " to SCHEDULED" << endmsg;
    return updateSc;
  } else {
    if (msgLevel(MSG::DEBUG))
      debug() << "Could not acquire instance for algorithm " << index2algname(iAlgo) << " on slot " << si << endmsg;
    return sc;
  }

}

//---------------------------------------------------------------------------
StatusCode ForwardSchedulerSvc::promoteToExecuted(unsigned int iAlgo, int si, IAlgorithm* algo) {

  // Put back the instance
  Algorithm* castedAlgo = dynamic_cast<Algorithm*>(algo); // DP: expose context getter in IAlgo?
  if (!castedAlgo)
    fatal() << "The casting did not succeed!" << endmsg;
  EventContext* eventContext = castedAlgo->getContext();

  // Check if the execution failed
  if (eventContext->evtFail())
    eventFailed(eventContext);

  StatusCode sc = m_algResourcePool->releaseAlgorithm(algo->name(),algo);

  if (!sc.isSuccess()) {
    error() << "[Event " << eventContext->evt() << ", Slot " << eventContext->slot()  << "] "
            << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
    return StatusCode::FAILURE;
    }

  m_algosInFlight--;

  EventSlot& thisSlot = m_eventSlots[si];
  // XXX: CF tests
  if (!m_DFNext) {
    // Update the catalog: some new products may be there
    m_whiteboard->selectStore(eventContext->slot()).ignore();

    // update prods in the dataflow
    // DP: Handles could be used. Just update what the algo wrote
    std::vector<std::string> new_products;
    m_whiteboard->getNewDataObjects(new_products).ignore();
    for (const auto& new_product : new_products)
      if (msgLevel(MSG::DEBUG))
        debug() << "Found in WB: " << new_product << endmsg;
    thisSlot.dataFlowMgr.updateDataObjectsCatalog(new_products);
  }

  if (msgLevel(MSG::DEBUG))
    debug() << "Algorithm " << algo->name() << " executed. Algorithms scheduled are " << m_algosInFlight << endmsg;

  // Limit number of updates
  if (m_CFNext) m_updateNeeded = true; // XXX: CF tests: with the new CF traversal the if clause below has to be removed
  if (m_updateNeeded) {
    // Schedule an update of the status of the algorithms
    auto updateAction = std::bind(&ForwardSchedulerSvc::updateStates, this, -1, algo->name());
    m_actionsQueue.push(updateAction);
    m_updateNeeded = false;
  }

  if (msgLevel(MSG::DEBUG))
    debug() << "Trying to handle execution result of " << index2algname(iAlgo) << "." << endmsg;
  State state;
  if (algo->filterPassed()) {
    state = State::EVTACCEPTED;
  } else {
    state = State::EVTREJECTED;
  }

  sc = thisSlot.algsStates.updateState(iAlgo,state);

  if (sc.isSuccess())
    if (msgLevel(MSG::DEBUG))
      debug() << "Promoting " << index2algname(iAlgo) << " on slot " << si << " to "
              << AlgsExecutionStates::stateNames[state] << endmsg;

  return sc;
}

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