HiveEventLoopMgr Class Reference

#include <GaudiHive/HiveEventLoopMgr.h>

Inheritance diagram for HiveEventLoopMgr:

Collaboration diagram for HiveEventLoopMgr:

Public Member Functions
	HiveEventLoopMgr (const std::string &nam, ISvcLocator *svcLoc)
	Standard Constructor. More...
virtual	~HiveEventLoopMgr ()
	Standard Destructor. More...
StatusCode	getEventRoot (IOpaqueAddress *&refpAddr)
	Create event address using event selector. More...
virtual StatusCode	initialize ()
	implementation of IService::initialize More...
virtual StatusCode	reinitialize ()
	implementation of IService::reinitialize More...
virtual StatusCode	stop ()
	implementation of IService::stop More...
virtual StatusCode	finalize ()
	implementation of IService::finalize More...
virtual StatusCode	nextEvent (int maxevt)
	implementation of IService::nextEvent More...
virtual StatusCode	executeEvent (void *par)
	implementation of IEventProcessor::executeEvent(void* par) More...
virtual StatusCode	executeRun (int maxevt)
	implementation of IEventProcessor::executeRun() More...
void	taskFinished (IAlgorithm *&algo)
	Decrement the number of algos in flight and put algo back in manager - maybe private. More...
Public Member Functions inherited from MinimalEventLoopMgr
	MinimalEventLoopMgr (const std::string &nam, ISvcLocator *svcLoc)
	Standard Constructor. More...
	~MinimalEventLoopMgr () override=default
	Standard Destructor. More...
StatusCode	initialize () override
	implementation of IService::initialize More...
StatusCode	start () override
	implementation of IService::start More...
StatusCode	stop () override
	implementation of IService::stop More...
StatusCode	finalize () override
	implementation of IService::finalize More...
StatusCode	reinitialize () override
	implementation of IService::reinitialize More...
StatusCode	restart () override
	implementation of IService::restart More...
StatusCode	nextEvent (int maxevt) override
	implementation of IEventProcessor::nextEvent More...
StatusCode	executeEvent (void *par) override
	implementation of IEventProcessor::executeEvent(void* par) More...
StatusCode	executeRun (int maxevt) override
	implementation of IEventProcessor::executeRun( ) More...
StatusCode	stopRun () override
	implementation of IEventProcessor::stopRun( ) More...
void	topAlgHandler (Property &p)
	Top algorithm List handler. More...
StatusCode	decodeTopAlgs ()
	decodeTopAlgNameList & topAlgNameListHandler More...
void	outStreamHandler (Property &p)
	Output stream List handler. More...
StatusCode	decodeOutStreams ()
	decodeOutStreamNameList & outStreamNameListHandler More...
	MinimalEventLoopMgr (const std::string &nam, ISvcLocator *svcLoc)
	Standard Constructor. More...
	~MinimalEventLoopMgr () override=default
	Standard Destructor. More...
StatusCode	initialize () override
	implementation of IService::initialize More...
StatusCode	start () override
	implementation of IService::start More...
StatusCode	stop () override
	implementation of IService::stop More...
StatusCode	finalize () override
	implementation of IService::finalize More...
StatusCode	reinitialize () override
	implementation of IService::reinitialize More...
StatusCode	restart () override
	implementation of IService::restart More...
StatusCode	nextEvent (int maxevt) override
	implementation of IEventProcessor::nextEvent More...
StatusCode	executeEvent (void *par) override
	implementation of IEventProcessor::executeEvent(void* par) More...
StatusCode	executeRun (int maxevt) override
	implementation of IEventProcessor::executeRun( ) More...
StatusCode	stopRun () override
	implementation of IEventProcessor::stopRun( ) More...
void	topAlgHandler (Property &p)
	Top algorithm List handler. More...
StatusCode	decodeTopAlgs ()
	decodeTopAlgNameList & topAlgNameListHandler More...
void	outStreamHandler (Property &p)
	Output stream List handler. More...
StatusCode	decodeOutStreams ()
	decodeOutStreamNameList & outStreamNameListHandler More...
Public Member Functions inherited from extends< BASE, Interfaces >
void *	i_cast (const InterfaceID &tid) const override
	Implementation of IInterface::i_cast. More...
StatusCode	queryInterface (const InterfaceID &ti, void **pp) override
	Implementation of IInterface::queryInterface. More...
std::vector< std::string >	getInterfaceNames () const override
	Implementation of IInterface::getInterfaceNames. More...
	~extends () override=default
	Virtual destructor. More...
void *	i_cast (const InterfaceID &tid) const override
	Implementation of IInterface::i_cast. More...
StatusCode	queryInterface (const InterfaceID &ti, void **pp) override
	Implementation of IInterface::queryInterface. More...
std::vector< std::string >	getInterfaceNames () const override
	Implementation of IInterface::getInterfaceNames. More...
	~extends () override=default
	Virtual destructor. More...
Public Member Functions inherited from extend_interfaces< Interfaces...>
	~extend_interfaces () override=default
	Virtual destructor. More...
	~extend_interfaces () override=default
	Virtual destructor. More...

Protected Types
typedef std::vector< std::vector< std::string > >	algosDependenciesCollection
	Algorithms Inputs. More...
Protected Types inherited from MinimalEventLoopMgr
enum	State {   OFFLINE, CONFIGURED, FINALIZED, INITIALIZED,   OFFLINE, CONFIGURED, FINALIZED, INITIALIZED }
enum	State {   OFFLINE, CONFIGURED, FINALIZED, INITIALIZED,   OFFLINE, CONFIGURED, FINALIZED, INITIALIZED }

Protected Member Functions
void	find_dependencies ()
	Get the input and output collections. More...

Protected Attributes
SmartIF< IDataManagerSvc >	m_evtDataMgrSvc
	Reference to the Event Data Service's IDataManagerSvc interface. More...
SmartIF< IDataProviderSvc >	m_evtDataSvc
	Reference to the Event Data Service's IDataProviderSvc interface. More...
SmartIF< IEvtSelector >	m_evtSelector
	Reference to the Event Selector. More...
IEvtSelector::Context *	m_evtContext
	Event Iterator. More...
std::string	m_evtsel
	Event selector. More...
SmartIF< IDataManagerSvc >	m_histoDataMgrSvc
	Reference to the Histogram Data Service. More...
SmartIF< IConversionSvc >	m_histoPersSvc
	Reference to the Histogram Persistency Service. More...
SmartIF< IHiveWhiteBoard >	m_whiteboard
	Reference to the Histogram Persistency Service. More...
SmartIF< IAlgResourcePool >	m_algResourcePool
	Reference to the Algorithm resource pool. More...
std::string	m_histPersName
	Name of the Hist Pers type. More...
SmartIF< IProperty >	m_appMgrProperty
	Property interface of ApplicationMgr. More...
bool	m_endEventFired
	Flag to avoid to fire the EnvEvent incident twice in a row (and also not before the first event) More...
bool	m_warnings
	Flag to disable warning messages when using external input. More...
unsigned int	m_max_parallel
	Maximum number of parallel running algorithms. More...
tbb::task_scheduler_init *	m_tbb_scheduler_init
	Pointer to tbb task scheduler. More...
state_type	m_termination_requirement
	The termination requirement. More...
std::vector< state_type >	m_all_requirements
	All requirements. More...
std::map< std::string, unsigned int >	m_product_indices
	Register of input products. More...
std::atomic_uint	m_total_algos_in_flight
	Total number of algos in flight across all events. More...
unsigned int	m_numberOfAlgos
	Total number of algos. More...
bool	m_DumpQueues
	Dump the algorithm queues. More...
unsigned int	m_evts_parallel
	Number of events in parallel. More...
unsigned int	m_num_threads
	Total numbers of threads. More...
bool	m_CloneAlgorithms
	Clone algorithms to run them simultaneously. More...
algosDependenciesCollection	m_AlgosDependencies
unsigned int	m_nProducts
Protected Attributes inherited from MinimalEventLoopMgr
SmartIF< IAppMgrUI >	m_appMgrUI
	Reference to the IAppMgrUI interface of the application manager. More...
SmartIF< IIncidentSvc >	m_incidentSvc
	Reference to the incident service. More...
ListAlg	m_topAlgList
	List of top level algorithms. More...
ListAlg	m_outStreamList
	List of output streams. More...
std::string	m_outStreamType
	Out Stream type. More...
StringArrayProperty	m_topAlgNames
	List of top level algorithms names. More...
StringArrayProperty	m_outStreamNames
	List of output stream names. More...
State	m_state = OFFLINE
	State of the object. More...
bool	m_scheduledStop = false
	Scheduled stop of event processing. More...
SmartIF< IIncidentListener >	m_abortEventListener
	Instance of the incident listener waiting for AbortEvent. More...
bool	m_abortEvent = false
	Flag signalling that the event being processedhas to be aborted (skip all following top algs). More...
std::string	m_abortEventSource
	Source of the AbortEvent incident. More...

Additional Inherited Members
Public Types inherited from MinimalEventLoopMgr
typedef std::vector< SmartIF< IAlgorithm > >	ListAlg
typedef std::vector< SmartIF< IAlgorithm > >	ListAlg
Public Types inherited from extends< BASE, Interfaces >
using	base_class = extends
	Typedef to this class. More...
using	extend_interfaces_base = extend_interfaces< Interfaces...>
	Typedef to the base of this class. More...
using	base_class = extends
	Typedef to this class. More...
using	extend_interfaces_base = extend_interfaces< Interfaces...>
	Typedef to the base of this class. More...
Public Types inherited from extend_interfaces< Interfaces...>
using	ext_iids = typename Gaudi::interface_list_cat< typename Interfaces::ext_iids...>::type
	take union of the ext_iids of all Interfaces... More...
using	ext_iids = typename Gaudi::interface_list_cat< typename Interfaces::ext_iids...>::type
	take union of the ext_iids of all Interfaces... More...

Detailed Description

Definition at line 32 of file HiveEventLoopMgr.h.

Member Typedef Documentation

typedef std::vector<std::vector<std::string> > HiveEventLoopMgr::algosDependenciesCollection

protected

Algorithms Inputs.

Definition at line 91 of file HiveEventLoopMgr.h.

Constructor & Destructor Documentation

HiveEventLoopMgr::HiveEventLoopMgr	(	const std::string &	nam,
		ISvcLocator *	svcLoc
	)

Standard Constructor.

Definition at line 79 of file HiveEventLoopMgr.cpp.

: MinimalEventLoopMgr(nam, svcLoc)
{
    m_histoDataMgrSvc   = 0;
    m_histoPersSvc      = 0;
    m_evtDataMgrSvc     = 0;
    m_evtDataSvc        = 0;
    m_evtSelector       = 0;
    m_evtContext        = 0;
        m_algResourcePool   = 0;
    m_endEventFired     = true;
    m_total_algos_in_flight=0;
    m_max_parallel = 1;
    m_evts_parallel = 1;
    m_num_threads = 1;
    m_DumpQueues = true;
        m_nProducts = 0;

    // Declare properties
    declareProperty("HistogramPersistency", m_histPersName = "");
    declareProperty("EvtSel", m_evtsel );
    declareProperty("Warnings",m_warnings=true,
            "Set this property to false to suppress warning messages");
    declareProperty("MaxAlgosParallel", m_max_parallel );
    declareProperty("MaxEventsParallel", m_evts_parallel);
    declareProperty("NumThreads", m_num_threads);
    declareProperty("DumpQueues", m_DumpQueues= false);
    declareProperty("CloneAlgorithms", m_CloneAlgorithms= false);
    declareProperty("AlgosDependencies", m_AlgosDependencies);
}

HiveEventLoopMgr::~HiveEventLoopMgr ( )

virtual

Standard Destructor.

Definition at line 113 of file HiveEventLoopMgr.cpp.

                                      {
    if( m_histoDataMgrSvc ) m_histoDataMgrSvc->release();
    if( m_histoPersSvc ) m_histoPersSvc->release();
    if( m_evtDataMgrSvc ) m_evtDataMgrSvc->release();
    if( m_evtDataSvc ) m_evtDataSvc->release();
    if( m_evtSelector ) m_evtSelector->release();
    if( m_evtContext ) delete m_evtContext;
}

Member Function Documentation

StatusCode HiveEventLoopMgr::executeEvent ( void *  par )

virtual

implementation of IEventProcessor::executeEvent(void* par)

Definition at line 370 of file HiveEventLoopMgr.cpp.

                                                    {

    // Fire BeginEvent "Incident"
    m_incidentSvc->fireIncident(Incident(name(),IncidentType::BeginEvent));
    // An incident may schedule a stop, in which case is better to exit before the actual execution.
    if ( m_scheduledStop ) {
        always() << "Terminating event processing loop due to a stop scheduled by an incident listener" << endmsg;
        return StatusCode::SUCCESS;
    }

    // Execute Algorithms
    m_incidentSvc->fireIncident(Incident(name(), IncidentType::BeginProcessing));

    // Prepare the event context for concurrency


    // Call the resetExecuted() method of ALL "known" algorithms
    // (before we were reseting only the topalgs)
    SmartIF<IAlgManager> algMan(serviceLocator());
    if (LIKELY(algMan.isValid())) {
        for(auto ialg: algMan->getAlgorithms()) {
            if (LIKELY(0 != ialg)) ialg->resetExecuted();
        }
    }

    bool eventfailed = false;//run_parallel();

    // ensure that the abortEvent flag is cleared before the next event
    if (UNLIKELY(m_abortEvent)) {
        DEBMSG << "AbortEvent incident fired by " << m_abortEventSource << endmsg;
        m_abortEvent = false;
    }

    // Call the execute() method of all output streams
    for (ListAlg::iterator ito = m_outStreamList.begin(); ito != m_outStreamList.end(); ito++ ) {
        (*ito)->resetExecuted();
        StatusCode sc;
        sc = (*ito)->sysExecute();
        if (UNLIKELY(!sc.isSuccess())) {
            warning() << "Execution of output stream " << (*ito)->name() << " failed" << endmsg;
            eventfailed = true;
        }
    }

    m_incidentSvc->fireIncident(Incident(name(), IncidentType::EndProcessing));

    // Check if there was an error processing current event
    if (UNLIKELY(eventfailed)){
        error() << "Error processing event loop." << endmsg;
        return StatusCode(StatusCode::FAILURE,true);
    }
    return StatusCode(StatusCode::SUCCESS,true);

}

StatusCode HiveEventLoopMgr::executeRun ( int  maxevt )

virtual

implementation of IEventProcessor::executeRun()

Definition at line 428 of file HiveEventLoopMgr.cpp.

                                                       {
    StatusCode  sc;
    // initialize the base class
    sc = MinimalEventLoopMgr::executeRun(maxevt);
    return sc;
}

StatusCode HiveEventLoopMgr::finalize ( )

virtual

implementation of IService::finalize

Definition at line 299 of file HiveEventLoopMgr.cpp.

                                         {
    StatusCode sc;

    // Finalize base class
    sc = MinimalEventLoopMgr::finalize();
    if (! sc.isSuccess()) {
        error() << "Error finalizing base class" << endmsg;
        return sc;
    }

    // Save Histograms Now
    if ( m_histoPersSvc != 0 )    {
        HistogramAgent agent;
        sc = m_histoDataMgrSvc->traverseTree( &agent );
        if( sc.isSuccess() )   {
            IDataSelector* objects = agent.selectedObjects();
            // skip /stat entry!
            if ( objects->size() > 0 )    {
                IDataSelector::iterator i;
                for ( i = objects->begin(); i != objects->end(); i++ )    {
                    IOpaqueAddress* pAddr = 0;
                    StatusCode iret = m_histoPersSvc->createRep(*i, pAddr);
                    if ( iret.isSuccess() )     {
                        (*i)->registry()->setAddress(pAddr);
                    }
                    else  {
                        sc = iret;
                    }
                }
                for ( i = objects->begin(); i != objects->end(); i++ )    {
                    IRegistry* reg = (*i)->registry();
                    StatusCode iret = m_histoPersSvc->fillRepRefs(reg->address(), *i);
                    if ( !iret.isSuccess() )    {
                        sc = iret;
                    }
                }
            }
            if ( sc.isSuccess() )    {
                info() << "Histograms converted successfully according to request." << endmsg;
            }
            else  {
                error() << "Error while saving Histograms." << endmsg;
            }
        }
        else {
            error() << "Error while traversing Histogram data store" << endmsg;
        }
    }

    // Release event selector context
    if ( m_evtSelector && m_evtContext )   {
        m_evtSelector->releaseContext(m_evtContext).ignore();
        m_evtContext = 0;
    }

    // Release all interfaces...
    m_histoDataMgrSvc = 0;
    m_histoPersSvc    = 0;

    m_evtSelector     = 0;
    m_evtDataSvc      = 0;
    m_evtDataMgrSvc   = 0;

    delete m_tbb_scheduler_init;

    return StatusCode::SUCCESS;
}

void HiveEventLoopMgr::find_dependencies ( )

protected

Get the input and output collections.

Compute dependencies between the algorithms.

Definition at line 743 of file HiveEventLoopMgr.cpp.

                                    {

        // Count how many products are actually requested
        for (auto& thisAlgoDependencies : m_AlgosDependencies){
            m_nProducts += thisAlgoDependencies.size();
        }
    const unsigned int n_algos = m_topAlgList.size();
    std::vector<state_type> all_requirements(n_algos,state_type(m_nProducts));

    unsigned int algo_counter=0;
    unsigned int input_counter=0;

        MsgStream log(msgSvc(), name());
    // loop on the dependencies
    for (const auto& algoDependencies : m_AlgosDependencies){ // loop on algo dependencies lists
        state_type requirements(m_nProducts);
        log << MSG::DEBUG << "Algorithm " << algo_counter << " dependencies: " << endmsg;
        for (const auto& dependency : algoDependencies){ // loop on dependencies
            log << MSG::DEBUG << " - " << dependency << endmsg;
            auto ret = m_product_indices.insert(std::pair<std::string, unsigned int>("/Event/"+dependency,input_counter));
            // insert successful means == wasn't known before. So increment counter
            if (ret.second==true) ++input_counter;
            // in any case the return value holds the proper product index
            requirements[ret.first->second] = true;
            log << MSG::DEBUG << "  - Requirements now: " << requirements[ret.first->second] << endmsg;
        }// end loop on single dependencies

        all_requirements[algo_counter] = requirements;
        ++algo_counter;
    } // end loop on algo dependencies lists

    // Loop on the product indices
    log << MSG::DEBUG << "Product indices:" << endmsg;
    for (auto& prod_index: m_product_indices)
        log << MSG::DEBUG << " - " << prod_index.first << " " << prod_index.second << endmsg;

    m_numberOfAlgos = algo_counter;
    m_all_requirements = all_requirements;

}

StatusCode HiveEventLoopMgr::getEventRoot

(

IOpaqueAddress *&

refpAddr

)

Create event address using event selector.

Definition at line 716 of file HiveEventLoopMgr.cpp.

                                                                    {
    refpAddr = 0;
    StatusCode sc = m_evtSelector->next(*m_evtContext);
    if ( !sc.isSuccess() )  {
        return sc;
    }
    // Create root address and assign address to data service
    sc = m_evtSelector->createAddress(*m_evtContext,refpAddr);
    if( !sc.isSuccess() )  {
        sc = m_evtSelector->next(*m_evtContext);
        if ( sc.isSuccess() )  {
            sc = m_evtSelector->createAddress(*m_evtContext,refpAddr);
            if ( !sc.isSuccess() )  {
                warning() << "Error creating IOpaqueAddress." << endmsg;
            }
        }
    }
    return sc;
}

StatusCode HiveEventLoopMgr::initialize ( )

virtual

implementation of IService::initialize

Definition at line 125 of file HiveEventLoopMgr.cpp.

                                           {
    // Initialize the base class
    StatusCode sc = MinimalEventLoopMgr::initialize();
    if( !sc.isSuccess() ) {
        DEBMSG << "Error Initializing base class MinimalEventLoopMgr." << endmsg;
        return sc;
    }

    find_dependencies();

    // Setup access to event data services
    m_evtDataMgrSvc = serviceLocator()->service("EventDataSvc");
    if( !m_evtDataMgrSvc.isValid() )  {
        fatal() << "Error retrieving EventDataSvc interface IDataManagerSvc." << endmsg;
        return StatusCode::FAILURE;
    }
    m_evtDataSvc = serviceLocator()->service("EventDataSvc");
    if( !m_evtDataSvc.isValid() )  {
        fatal() << "Error retrieving EventDataSvc interface IDataProviderSvc." << endmsg;
        return StatusCode::FAILURE;
    }
    m_whiteboard = serviceLocator()->service("EventDataSvc");
    if( !m_whiteboard.isValid() )  {
        fatal() << "Error retrieving EventDataSvc interface IHiveWhiteBoard." << endmsg;
        return StatusCode::FAILURE;
    }
    m_whiteboard->setNumberOfStores(m_evts_parallel).ignore();

    // Obtain the IProperty of the ApplicationMgr
    m_appMgrProperty = serviceLocator();
    if ( ! m_appMgrProperty.isValid() )   {
        fatal() << "IProperty interface not found in ApplicationMgr." << endmsg;
        return StatusCode::FAILURE;
    }

    // We do not expect a Event Selector necessarily being declared
    setProperty(m_appMgrProperty->getProperty("EvtSel")).ignore();

    if( m_evtsel != "NONE" || m_evtsel.length() == 0) {
        m_evtSelector = serviceLocator()->service("EventSelector");
        if( m_evtSelector.isValid() ) {
            // Setup Event Selector
            sc=m_evtSelector->createContext(m_evtContext);
            if( !sc.isSuccess() )   {
                fatal() << "Can not create the event selector Context." << endmsg;
                return sc;
            }
        }
        else {
            fatal() << "EventSelector not found." << endmsg;
            return sc;
        }
    }
    else {
        m_evtSelector = 0;
        m_evtContext = 0;
        if ( m_warnings ) {
            warning() << "Unable to locate service \"EventSelector\" " << endmsg;
            warning() << "No events will be processed from external input." << endmsg;
        }
    }

    // Setup access to histogramming services
    m_histoDataMgrSvc = serviceLocator()->service("HistogramDataSvc");
    if( !m_histoDataMgrSvc.isValid() )  {
        fatal() << "Error retrieving HistogramDataSvc." << endmsg;
        return sc;
    }
    // Setup histogram persistency
    m_histoPersSvc = serviceLocator()->service("HistogramPersistencySvc");
    if( !m_histoPersSvc.isValid() ) {
        warning() << "Histograms cannot not be saved - though required." << endmsg;
        return sc;
    }

        // Setup algorithm resource pool
        m_algResourcePool = serviceLocator()->service("AlgResourcePool");
        if( !m_algResourcePool.isValid() ) {
      fatal() << "Error retrieving AlgResourcePool" << endmsg;
      return StatusCode::FAILURE;
        }

    // Setup tbb task scheduler
    // TODO: shouldn't be in this case
    // One more for the current thread
    m_tbb_scheduler_init = new tbb::task_scheduler_init(m_num_threads+1);

    return StatusCode::SUCCESS;
}

StatusCode HiveEventLoopMgr::nextEvent ( int  maxevt )

virtual

implementation of IService::nextEvent

Definition at line 442 of file HiveEventLoopMgr.cpp.

                                                   {
    // Collapse executeEvent and run_parallel in the same method
    // TODO _very_ sporty on conditions and checks!!

    auto start_time = tbb::tick_count::now();
    auto secsFromStart = [&start_time]()->double{
        return (tbb::tick_count::now()-start_time).seconds();
    };

    typedef std::tuple<EventContext*,EventSchedulingState*> contextSchedState_tuple;

    MsgStream log(msgSvc(), name());


    // Reset the application return code.
    Gaudi::setAppReturnCode(m_appMgrProperty, Gaudi::ReturnCode::Success, true).ignore();

    // Lambda to check if an event has finished
    auto has_finished = [] // acquire nothing
         (contextSchedState_tuple evtContext_evtstate) // argument is a tuple
         { return std::get<1>(evtContext_evtstate)->hasFinished();}; // true if finished

    // Useful for the Logs
    always() << "Running with "
             << m_evts_parallel << " parallel events, "
             << m_max_parallel << " max concurrent algorithms, "
             << m_num_threads << " threads."
             << endmsg;

    int n_processed_events = 0;
    bool eof = false;
    StatusCode sc;

    // Events in flight
    std::list<contextSchedState_tuple> events_in_flight;

    // Loop until no more evts are there

    while( maxevt == -1 ? !eof : n_processed_events < maxevt ){// TODO Fix the condition in case of -1

        const unsigned int n_events_in_flight = events_in_flight.size();
        const unsigned int n_evts_to_process = maxevt - n_processed_events - n_events_in_flight;

        unsigned int n_acquirable_events = m_evts_parallel - n_events_in_flight ;
        if (n_acquirable_events > n_evts_to_process)
            n_acquirable_events = n_evts_to_process;

        log << MSG::INFO << "Evts in flight: " <<  n_events_in_flight << endmsg;
        log << MSG::INFO << "Evts processed: " <<  n_processed_events<< endmsg;
        log << MSG::INFO << "Evts parallel: " << m_evts_parallel << endmsg;
        log << MSG::INFO << "Acquirable Events are " << n_acquirable_events << endmsg;

        // Initialisation section ------------------------------------------------

        // Loop on events to be initialised
        for (unsigned int offset=0; offset< n_acquirable_events; ++offset){

            EventContext* evtContext(new EventContext);
            const int evt_num =  n_processed_events + offset + n_events_in_flight;
        evtContext->set(evt_num,  m_whiteboard->allocateStore(evt_num) );
            m_whiteboard->selectStore(evtContext->slot()).ignore();

            if( m_evtContext ) {
                //---This is the "event iterator" context from EventSelector
                IOpaqueAddress* pAddr = 0;
                sc = getEventRoot(pAddr);
                if( !sc.isSuccess() )  {
                    info() << "No more events in event selection " << endmsg;
                    eof = true;
                    maxevt = evt_num;  // Set the maxevt to the determined maximum
                    break;
                }
                sc = m_evtDataMgrSvc->setRoot ("/Event", pAddr);
                if( !sc.isSuccess() )  {
                    warning() << "Error declaring event root address." << endmsg;
                }
            }
            else {
                //---In case of no event selector----------------
                sc = m_evtDataMgrSvc->setRoot ("/Event", new DataObject());
                if( !sc.isSuccess() )  {
                    warning() << "Error declaring event root DataObject" << endmsg;
                }
            }

            EventSchedulingState* event_state = new EventSchedulingState(m_topAlgList.size(),m_nProducts);
            events_in_flight.push_back(std::make_tuple(evtContext,event_state));
            info()  << "Started event " << evt_num << " at " << secsFromStart() << endmsg;

        }// End initialisation loop on acquired events

        // End initialisation section --------------------------------------------

        // Scheduling section ----------------------------------------------------
        auto in_flight_end = events_in_flight.end();
        auto in_flight_begin = events_in_flight.begin();
        // loop until at least one evt finished
        while (in_flight_end == find_if(in_flight_begin, in_flight_end ,has_finished)){
            bool no_algo_can_run = true;
            for (auto& evtContext_evtstate : events_in_flight){ // loop on evts

                EventContext* event_Context = std::get<0>(evtContext_evtstate);
                EventSchedulingState* event_state = std::get<1>(evtContext_evtstate);

                for (unsigned int algo_counter=0; algo_counter<m_topAlgList.size(); algo_counter++) { // loop on algos
                    // check whether all requirements/dependencies for the algorithm are fulfilled...
                    const state_type& algo_requirements = m_all_requirements[algo_counter];
                    // Very verbose!
                    //      log << MSG::VERBOSE << "Checking dependencies for algo " << algo_counter << ":\n"
                    //          << "  o Requirements: " <<  algo_requirements << std::endl
                    //          << "  o State: " << event_state->state() << endmsg;

                    // ...and whether the algorithm was already started and if it can be started
                    bool algo_not_started_and_dependencies_there = (algo_requirements.is_subset_of(event_state->state()) &&
                                    (event_state->hasStarted(algo_counter) ) == false);

                    // It could run, just the maximum number of algos in flight has been reached
                    if (algo_not_started_and_dependencies_there)
                        no_algo_can_run = false;
                    if (algo_not_started_and_dependencies_there &&
                       (m_total_algos_in_flight < m_max_parallel )) {
                       // Pick the algorithm if available and if not and requested create one
                       IAlgorithm* ialgo=NULL;
                       // To be transferred to the algomanager, this is inefficient
                       ListAlg::iterator algoIt = m_topAlgList.begin();
                       std::advance(algoIt, algo_counter);
                       if(m_algResourcePool->acquireAlgorithm(algoIt->get()->name(),ialgo)){
                           log << MSG::INFO << "Launching algo " << algo_counter<<  " on event " << event_Context->evt() << endmsg;
                            // Attach context to the algo
                            Algorithm* algo = dynamic_cast<Algorithm*> (ialgo);
                            algo->setContext(event_Context);

                            tbb::task* t = new( tbb::task::allocate_root() ) HiveAlgoTask(ialgo, event_state, this);
                            tbb::task::enqueue( *t);

                            event_state->algoStarts(algo_counter);
                            ++m_total_algos_in_flight;

                           log << MSG::INFO << "Algos in flight: " <<  m_total_algos_in_flight << endmsg;
                       }
            } // End scheduling if block

               }// end loop on algo indices

               // update the event state with what has been put into the DataSvc
               std::vector<std::string> new_products;
               m_whiteboard->selectStore(event_Context->slot()).ignore();
               sc = m_whiteboard->getNewDataObjects(new_products);
               if( !sc.isSuccess() ){
                   warning() << "Error getting recent new products (since last time called)" << endmsg;
               }
               for (const auto& newProduct : new_products) {
                   log << MSG::DEBUG << "New Product: " << newProduct << " in the store." << endmsg;
                   if (m_product_indices.count( newProduct ) == 1) { // only products with dependencies upon need to be announced to other algos
                       log << MSG::DEBUG << "  - Used as input by some algorithm. Updating the event state." << endmsg;
                       event_state->update_state(m_product_indices[newProduct]);
                   }
               }


               /* Check if we stall on the current event
                * One should check if:
                * - Nothing can run
                * - Nothing is running
                * - No new product is available
                * - The event is not finished
                * At this point one could claim a stall.
                * The implementation poses a challenge though, which resides in the
                * asyncronous termination of algorithm and potential writing in the
                * store. Therefore one checks the 4 aforementioned conditions.
                * Then, the store is again checked (without removing the new
                * elements). If something new is there the stall is not sure
                * anymore.
                * Another possibility could be to check if any algo terminated
                * during the checks made to the wb probably.
                */
                if (no_algo_can_run && // nothing could run
                    m_total_algos_in_flight==0 && // nothing is running
                    new_products.size() == 0 && // no new product available
                    ! event_state->hasFinished() ){ // the event is not finished

                    // Check if something arrived on the wb meanwhile
                    if(!m_whiteboard->newDataObjectsPresent()){

                        std::string errorMessage("No algorithm can run, "
                                                 "no algorithm in flight, "
                                                 "no new products in the store, "
                                                 "event not complete: this is a stall.");
                        fatal() << errorMessage << std::endl
                                << "Algorithms that ran for event " << event_Context->evt() << std::endl;
                        unsigned int algo_counter=0;
                        for (auto& algo : m_topAlgList){
                            bool has_started = event_state->hasStarted(algo_counter);
                            if (has_started)
                                fatal() << " o " << algo->name() << " could run" << std::endl;
                            else
                                fatal() << " o " << algo->name() << " could NOT run" << std::endl;
                            algo_counter++;
                        } // End ofloop on algos
                        fatal() << endmsg;
                        throw GaudiException (errorMessage,"HiveEventLoopMgr",StatusCode::FAILURE);
                    }
        }
            }// end loop on evts in flight
        }// end loop until at least one evt in flight finished

        // Remove from the in flight events the finished ones
        std::list<contextSchedState_tuple>::iterator it=events_in_flight.begin();

        while (it!=events_in_flight.end()){
        // Now proceed to deletion
        if (std::get<1>(*it)->hasFinished()){
            const unsigned int evt_num = std::get<0>(*it)->evt();
            const unsigned int evt_slot = std::get<0>(*it)->slot();
                log << MSG::INFO << "Event "<< evt_num << " finished. Events in flight are "
                        << events_in_flight.size() << ". Processed events are "
                        <<  n_processed_events << endmsg;
                info() << "Event "<< evt_num << " finished. now is " <<  secsFromStart() << endmsg;

                // Calculate min and max event num
                unsigned int min_event_num=0xFFFFFFFF;
                unsigned int max_event_num=0;

                for (auto& evtContext_evtstate : events_in_flight){
                    const unsigned int evt_num = std::get<0>(evtContext_evtstate)->evt();
                    // Update min and max for backlog calculation
                    if (evt_num > max_event_num) max_event_num=evt_num;
                    if (evt_num < min_event_num) min_event_num=evt_num;
                }
                unsigned int evt_backlog=max_event_num-min_event_num;
                info() << "Event backlog (max= " << max_event_num << ", min= "
                        << min_event_num<<" ) = " << evt_backlog << endmsg;


         // Output
         // Call the execute() method of all output streams
  for (ListAlg::iterator ito = m_outStreamList.begin(); ito != m_outStreamList.end(); ito++ ) {
    (*ito)->resetExecuted();
    StatusCode sc;
    sc = (*ito)->sysExecute();
    if (UNLIKELY(!sc.isSuccess())) {
      warning() << "Execution of output stream " << (*ito)->name() << " failed" << endmsg;
    }
  }

                sc = m_whiteboard->clearStore(evt_slot);
                if( !sc.isSuccess() )  {
                    warning() << "Clear of Event data store failed" << endmsg;
                }
                else {
                    info() << "Cleared store " << evt_slot << endmsg;
                }
                m_whiteboard->freeStore(evt_slot).ignore();

                delete std::get<0>(*it);
                delete std::get<1>(*it);
                it=events_in_flight.erase(it) ;

                n_processed_events++;

            } else{
                ++it;
            }
        }
        // End scheduling session ------------------------------------------------

    } // End while loop on events

    always() << "---> Loop Finished (seconds): " << secsFromStart() <<endmsg;

    return StatusCode::SUCCESS;
}

StatusCode HiveEventLoopMgr::reinitialize ( )

virtual

implementation of IService::reinitialize

Definition at line 217 of file HiveEventLoopMgr.cpp.

                                          {

    // Initialize the base class
    StatusCode sc = MinimalEventLoopMgr::reinitialize();
    if( !sc.isSuccess() ) {
        DEBMSG << "Error Initializing base class MinimalEventLoopMgr." << endmsg;
        return sc;
    }

    // Check to see whether a new Event Selector has been specified
    setProperty(m_appMgrProperty->getProperty("EvtSel"));
    if( m_evtsel != "NONE" || m_evtsel.length() == 0) {
        SmartIF<IService> theSvc(serviceLocator()->service("EventSelector"));
        SmartIF<IEvtSelector> theEvtSel(theSvc);
        if( theEvtSel.isValid() && ( theEvtSel.get() != m_evtSelector.get() ) ) {
            // Setup Event Selector
            if ( m_evtSelector.get() && m_evtContext ) {
                // Need to release context before switching to new event selector
                m_evtSelector->releaseContext(m_evtContext);
                m_evtContext = 0;
            }
            m_evtSelector = theEvtSel;
            if (theSvc->FSMState() == Gaudi::StateMachine::INITIALIZED) {
                sc = theSvc->reinitialize();
                if( !sc.isSuccess() ) {
                    error() << "Failure Reinitializing EventSelector "
                            << theSvc->name( ) << endmsg;
                    return sc;
                }
            }
            else {
                sc = theSvc->sysInitialize();
                if( !sc.isSuccess() ) {
                    error() << "Failure Initializing EventSelector "
                            << theSvc->name( ) << endmsg;
                    return sc;
                }
            }
            sc = m_evtSelector->createContext(m_evtContext);
            if( !sc.isSuccess() ) {
                error() << "Can not create Context " << theSvc->name( ) << endmsg;
                return sc;
            }
            info() << "EventSelector service changed to "
                    << theSvc->name( ) << endmsg;
        }
        else if ( m_evtSelector.isValid() ) {
            if ( m_evtContext ) {
                m_evtSelector->releaseContext(m_evtContext);
                m_evtContext = 0;
            }
            sc = m_evtSelector->createContext(m_evtContext);
            if( !sc.isSuccess() ) {
                error() << "Can not create Context " << theSvc->name( ) << endmsg;
                return sc;
            }
        }
    }
    else if ( m_evtSelector.isValid() && m_evtContext ) {
        m_evtSelector->releaseContext(m_evtContext);
        m_evtSelector = 0;
        m_evtContext = 0;
    }
    return StatusCode::SUCCESS;
}

StatusCode HiveEventLoopMgr::stop ( )

virtual

implementation of IService::stop

Definition at line 287 of file HiveEventLoopMgr.cpp.

                                     {
    if ( ! m_endEventFired ) {
        // Fire pending EndEvent incident
        m_incidentSvc->fireIncident(Incident(name(),IncidentType::EndEvent));
        m_endEventFired = true;
    }
    return MinimalEventLoopMgr::stop();
}

void HiveEventLoopMgr::taskFinished

(

IAlgorithm *&

algo

)

Decrement the number of algos in flight and put algo back in manager - maybe private.

Definition at line 786 of file HiveEventLoopMgr.cpp.

                                                    {
    m_algResourcePool->releaseAlgorithm(algo->name(),algo);
    --m_total_algos_in_flight;
    MsgStream log(msgSvc(), name());
    log << MSG::DEBUG << "[taskFinished] Algos in flight: " <<  m_total_algos_in_flight << endmsg;
}

Member Data Documentation

algosDependenciesCollection HiveEventLoopMgr::m_AlgosDependencies

protected

Definition at line 93 of file HiveEventLoopMgr.h.

SmartIF<IAlgResourcePool> HiveEventLoopMgr::m_algResourcePool

protected

Reference to the Algorithm resource pool.

Definition at line 53 of file HiveEventLoopMgr.h.

std::vector<state_type> HiveEventLoopMgr::m_all_requirements

protected

All requirements.

Definition at line 74 of file HiveEventLoopMgr.h.

SmartIF<IProperty> HiveEventLoopMgr::m_appMgrProperty

protected

Property interface of ApplicationMgr.

Definition at line 57 of file HiveEventLoopMgr.h.

bool HiveEventLoopMgr::m_CloneAlgorithms

protected

Clone algorithms to run them simultaneously.

Definition at line 88 of file HiveEventLoopMgr.h.

bool HiveEventLoopMgr::m_DumpQueues

protected

Dump the algorithm queues.

Definition at line 82 of file HiveEventLoopMgr.h.

bool HiveEventLoopMgr::m_endEventFired

protected

Flag to avoid to fire the EnvEvent incident twice in a row (and also not before the first event)

Definition at line 60 of file HiveEventLoopMgr.h.

IEvtSelector::Context* HiveEventLoopMgr::m_evtContext

protected

Event Iterator.

Definition at line 43 of file HiveEventLoopMgr.h.

SmartIF<IDataManagerSvc> HiveEventLoopMgr::m_evtDataMgrSvc

protected

Reference to the Event Data Service's IDataManagerSvc interface.

Definition at line 37 of file HiveEventLoopMgr.h.

SmartIF<IDataProviderSvc> HiveEventLoopMgr::m_evtDataSvc

protected

Reference to the Event Data Service's IDataProviderSvc interface.

Definition at line 39 of file HiveEventLoopMgr.h.

unsigned int HiveEventLoopMgr::m_evts_parallel

protected

Number of events in parallel.

Definition at line 84 of file HiveEventLoopMgr.h.

std::string HiveEventLoopMgr::m_evtsel

protected

Event selector.

Definition at line 45 of file HiveEventLoopMgr.h.

SmartIF<IEvtSelector> HiveEventLoopMgr::m_evtSelector

protected

Reference to the Event Selector.

Definition at line 41 of file HiveEventLoopMgr.h.

SmartIF<IDataManagerSvc> HiveEventLoopMgr::m_histoDataMgrSvc

protected

Reference to the Histogram Data Service.

Definition at line 47 of file HiveEventLoopMgr.h.

SmartIF<IConversionSvc> HiveEventLoopMgr::m_histoPersSvc

protected

Reference to the Histogram Persistency Service.

Definition at line 49 of file HiveEventLoopMgr.h.

std::string HiveEventLoopMgr::m_histPersName

protected

Name of the Hist Pers type.

Definition at line 55 of file HiveEventLoopMgr.h.

unsigned int HiveEventLoopMgr::m_max_parallel

protected

Maximum number of parallel running algorithms.

Definition at line 66 of file HiveEventLoopMgr.h.

unsigned int HiveEventLoopMgr::m_nProducts

protected

Definition at line 95 of file HiveEventLoopMgr.h.

unsigned int HiveEventLoopMgr::m_num_threads

protected

Total numbers of threads.

Definition at line 86 of file HiveEventLoopMgr.h.

unsigned int HiveEventLoopMgr::m_numberOfAlgos

protected

Total number of algos.

Definition at line 80 of file HiveEventLoopMgr.h.

std::map<std::string,unsigned int> HiveEventLoopMgr::m_product_indices

protected

Register of input products.

Definition at line 76 of file HiveEventLoopMgr.h.

tbb::task_scheduler_init* HiveEventLoopMgr::m_tbb_scheduler_init

protected

Pointer to tbb task scheduler.

Definition at line 68 of file HiveEventLoopMgr.h.

state_type HiveEventLoopMgr::m_termination_requirement

protected

The termination requirement.

Definition at line 72 of file HiveEventLoopMgr.h.

std::atomic_uint HiveEventLoopMgr::m_total_algos_in_flight

protected

Total number of algos in flight across all events.

Definition at line 78 of file HiveEventLoopMgr.h.

bool HiveEventLoopMgr::m_warnings

protected

Flag to disable warning messages when using external input.

Definition at line 62 of file HiveEventLoopMgr.h.

SmartIF<IHiveWhiteBoard> HiveEventLoopMgr::m_whiteboard

protected

Reference to the Histogram Persistency Service.

Definition at line 51 of file HiveEventLoopMgr.h.

---

The documentation for this class was generated from the following files:

• GaudiHive/GaudiHive/HiveEventLoopMgr.h
• GaudiHive/src/HiveEventLoopMgr.cpp