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...
Public Member Functions inherited from extends< Service, IEventProcessor >
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 Service
const std::string &	name () const override
	Retrieve name of the service. More...
StatusCode	configure () override
StatusCode	initialize () override
StatusCode	start () override
StatusCode	stop () override
StatusCode	finalize () override
StatusCode	terminate () override
Gaudi::StateMachine::State	FSMState () const override
Gaudi::StateMachine::State	targetFSMState () const override
StatusCode	reinitialize () override
StatusCode	restart () override
StatusCode	sysInitialize () override
	Initialize Service. More...
StatusCode	sysStart () override
	Initialize Service. More...
StatusCode	sysStop () override
	Initialize Service. More...
StatusCode	sysFinalize () override
	Finalize Service. More...
StatusCode	sysReinitialize () override
	Re-initialize the Service. More...
StatusCode	sysRestart () override
	Re-initialize the Service. More...
StatusCode	setProperty (const Property &p) override
StatusCode	setProperty (const std::string &s) override
StatusCode	setProperty (const std::string &n, const std::string &v) override
StatusCode	getProperty (Property *p) const override
const Property &	getProperty (const std::string &name) const override
StatusCode	getProperty (const std::string &n, std::string &v) const override
const std::vector< Property * > &	getProperties () const override
bool	hasProperty (const std::string &name) const override
template<class TYPE >
StatusCode	setProperty (const std::string &name, const TYPE &value)
	set the property form the value More...
	Service (std::string name, ISvcLocator *svcloc)
	Standard Constructor. More...
SmartIF< ISvcLocator > &	serviceLocator () const override
	Retrieve pointer to service locator. More...
StatusCode	setProperties ()
	Method for setting declared properties to the values specified for the job. More...
template<class T >
StatusCode	service (const std::string &name, const T *&psvc, bool createIf=true) const
	Access a service by name, creating it if it doesn't already exist. More...
template<class T >
StatusCode	service (const std::string &name, T *&psvc, bool createIf=true) const
template<typename IFace = IService>
SmartIF< IFace >	service (const std::string &name, bool createIf=true) const
template<class T >
StatusCode	service (const std::string &svcType, const std::string &svcName, T *&psvc) const
	Access a service by name and type, creating it if it doesn't already exist. More...
template<class T >
Property *	declareProperty (const std::string &name, T &property, const std::string &doc="none") const
	Declare the named property. More...
Property *	declareRemoteProperty (const std::string &name, IProperty *rsvc, const std::string &rname="") const
	Declare remote named properties. More...
template<class T >
StatusCode	declarePrivateTool (ToolHandle< T > &handle, std::string toolTypeAndName="", bool createIf=true)
	Declare used Private tool. More...
template<class T >
StatusCode	declarePublicTool (ToolHandle< T > &handle, std::string toolTypeAndName="", bool createIf=true)
	Declare used Public tool. More...
SmartIF< IAuditorSvc > &	auditorSvc () const
	The standard auditor service.May not be invoked before sysInitialize() has been invoked. More...
Public Member Functions inherited from CommonMessagingBase
virtual	~CommonMessagingBase ()=default
	Virtual destructor. More...
SmartIF< IMessageSvc > &	msgSvc () const
	The standard message service. More...
MsgStream &	msgStream () const
	Return an uninitialized MsgStream. More...
MsgStream &	msgStream (const MSG::Level level) const
	Predefined configurable message stream for the efficient printouts. More...
MsgStream &	always () const
	shortcut for the method msgStream(MSG::ALWAYS) More...
MsgStream &	fatal () const
	shortcut for the method msgStream(MSG::FATAL) More...
MsgStream &	err () const
	shortcut for the method msgStream(MSG::ERROR) More...
MsgStream &	error () const
	shortcut for the method msgStream(MSG::ERROR) More...
MsgStream &	warning () const
	shortcut for the method msgStream(MSG::WARNING) More...
MsgStream &	info () const
	shortcut for the method msgStream(MSG::INFO) More...
MsgStream &	debug () const
	shortcut for the method msgStream(MSG::DEBUG) More...
MsgStream &	verbose () const
	shortcut for the method msgStream(MSG::VERBOSE) More...
MsgStream &	msg () const
	shortcut for the method msgStream(MSG::INFO) More...
MSG::Level	msgLevel () const
	get the output level from the embedded MsgStream More...
MSG::Level	outputLevel () const __attribute__((deprecated))
	Backward compatibility function for getting the output level. More...
bool	msgLevel (MSG::Level lvl) const
	get the output level from the embedded MsgStream More...
Public Member Functions inherited from extend_interfaces< Interfaces...>
	~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 }

Protected Member Functions
void	find_dependencies ()
	Get the input and output collections. More...
Protected Member Functions inherited from Service
	~Service () override
	Standard Destructor. More...
int	outputLevel () const
	get the Service's output level More...
Protected Member Functions inherited from CommonMessaging< implements< IService, IProperty, IStateful > >
void	updateMsgStreamOutputLevel (int level)
	Update the output level of the cached MsgStream. 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< DataObjID, 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...
Protected Attributes inherited from Service
IntegerProperty	m_outputLevel = MSG::NIL
	Service output level. More...
Gaudi::StateMachine::State	m_state = Gaudi::StateMachine::OFFLINE
	Service state. More...
Gaudi::StateMachine::State	m_targetState = Gaudi::StateMachine::OFFLINE
	Service state. More...

Additional Inherited Members
Public Types inherited from MinimalEventLoopMgr
typedef std::vector< SmartIF< IAlgorithm > >	ListAlg
Public Types inherited from extends< Service, IEventProcessor >
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 Service
typedef Gaudi::PluginService::Factory< IService *, const std::string &, ISvcLocator * >	Factory
Public Types inherited from CommonMessaging< implements< IService, IProperty, IStateful > >
using	base_class = CommonMessaging
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...

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 744 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 717 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;
    DataObjIDColl 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 787 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<DataObjID,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