Gaudi::TestSuite::QueueingEventLoopMgr Class Reference

Inheritance diagram for Gaudi::TestSuite::QueueingEventLoopMgr:

Collaboration diagram for Gaudi::TestSuite::QueueingEventLoopMgr:

Public Member Functions
StatusCode	start () override
StatusCode	stop () override
StatusCode	executeEvent (EventContext &&ctx) override
void	push (EventContext &&ctx) override
bool	empty () const override
	Tell if the processor has events in the queues.
std::optional< ResultType >	pop () override
	Get the next available result, if any.
Public Member Functions inherited from extends< MinimalEventLoopMgr, Gaudi::Interfaces::IQueueingEventProcessor >
void const *	i_cast (const InterfaceID &tid) const override
	Implementation of IInterface::i_cast.
StatusCode	queryInterface (const InterfaceID &ti, void **pp) override
	Implementation of IInterface::queryInterface.
std::vector< std::string >	getInterfaceNames () const override
	Implementation of IInterface::getInterfaceNames.
Public Member Functions inherited from MinimalEventLoopMgr
	MinimalEventLoopMgr (const std::string &nam, ISvcLocator *svcLoc)
	Standard Constructor.
	MinimalEventLoopMgr (const MinimalEventLoopMgr &)=delete
	No copy allowed.
MinimalEventLoopMgr &	operator= (const MinimalEventLoopMgr &)=delete
	No copy allowed.
StatusCode	initialize () override
	implementation of IService::initialize
StatusCode	start () override
	implementation of IService::start
StatusCode	stop () override
	implementation of IService::stop
StatusCode	finalize () override
	implementation of IService::finalize
StatusCode	reinitialize () override
	implementation of IService::reinitialize
StatusCode	restart () override
	implementation of IService::restart
EventContext	createEventContext () override
	implementation of IEventProcessor::createEventContext()
StatusCode	nextEvent (int maxevt) override
	implementation of IEventProcessor::nextEvent
StatusCode	executeEvent (EventContext &&ctx) override
	implementation of IEventProcessor::executeEvent(EventContext&&)
StatusCode	executeRun (int maxevt) override
	implementation of IEventProcessor::executeRun( )
StatusCode	stopRun () override
	implementation of IEventProcessor::stopRun( )
void	topAlgHandler (Gaudi::Details::PropertyBase &p)
	Top algorithm List handler.
StatusCode	decodeTopAlgs ()
	decodeTopAlgNameList & topAlgNameListHandler
void	outStreamHandler (Gaudi::Details::PropertyBase &p)
	Output stream List handler.
StatusCode	decodeOutStreams ()
	decodeOutStreamNameList & outStreamNameListHandler
Public Member Functions inherited from extends< Service, IEventProcessor >
void const *	i_cast (const InterfaceID &tid) const override
	Implementation of IInterface::i_cast.
StatusCode	queryInterface (const InterfaceID &ti, void **pp) override
	Implementation of IInterface::queryInterface.
std::vector< std::string >	getInterfaceNames () const override
	Implementation of IInterface::getInterfaceNames.
Public Member Functions inherited from Service
const std::string &	name () const override
	Retrieve name of the service.
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.
StatusCode	sysStart () override
	Initialize Service.
StatusCode	sysStop () override
	Initialize Service.
StatusCode	sysFinalize () override
	Finalize Service.
StatusCode	sysReinitialize () override
	Re-initialize the Service.
StatusCode	sysRestart () override
	Re-initialize the Service.
	Service (std::string name, ISvcLocator *svcloc)
	Standard Constructor.
SmartIF< ISvcLocator > &	serviceLocator () const override
	Retrieve pointer to service locator.
template<typename IFace = IService>
SmartIF< IFace >	service (const std::string &name, bool createIf=true) const
template<class T>
Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, ToolHandle< T > &hndl, const std::string &doc="none")
template<class T>
StatusCode	declareTool (ToolHandle< T > &handle, bool createIf=true)
template<class T>
StatusCode	declareTool (ToolHandle< T > &handle, const std::string &toolTypeAndName, bool createIf=true)
	Declare used tool.
template<class T>
Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, ToolHandleArray< T > &hndlArr, const std::string &doc="none")
template<class T>
void	addToolsArray (ToolHandleArray< T > &hndlArr)
const std::vector< IAlgTool * > &	tools () const
SmartIF< IAuditorSvc > &	auditorSvc () const
	The standard auditor service.May not be invoked before sysInitialize() has been invoked.
Public Member Functions inherited from PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
StatusCode	setProperty (const Gaudi::Details::PropertyBase &p)
	Set the property from a property.
StatusCode	setProperty (const std::string &name, const char *v)
	Special case for string literals.
StatusCode	setProperty (const std::string &name, const std::string &v)
	Special case for std::string.
StatusCode	setProperty (const std::string &name, const TYPE &value)
	set the property form the value
	PropertyHolder ()=default
Gaudi::Details::PropertyBase &	declareProperty (Gaudi::Details::PropertyBase &prop)
	Declare a property.
Gaudi::Details::PropertyBase *	declareRemoteProperty (const std::string &name, IProperty *rsvc, const std::string &rname="")
	Declare a remote property.
StatusCode	setPropertyRepr (const std::string &n, const std::string &r) override
	set the property from name and value string representation
StatusCode	getProperty (Gaudi::Details::PropertyBase *p) const override
	get the property
const std::vector< Gaudi::Details::PropertyBase * > &	getProperties () const override
	get all properties
bool	hasProperty (std::string_view name) const override
	Return true if we have a property with the given name.
Gaudi::Details::PropertyBase *	property (std::string_view name) const
	\fixme property and bindPropertiesTo should be protected
void	bindPropertiesTo (Gaudi::Interfaces::IOptionsSvc &optsSvc)
PropertyHolder &	operator= (const PropertyHolder &)=delete
Public Member Functions inherited from CommonMessagingBase
virtual	~CommonMessagingBase ()=default
	Virtual destructor.
const SmartIF< IMessageSvc > &	msgSvc () const
	The standard message service.
MsgStream &	msgStream () const
	Return an uninitialized MsgStream.
MsgStream &	msgStream (const MSG::Level level) const
	Predefined configurable message stream for the efficient printouts.
MsgStream &	always () const
	shortcut for the method msgStream(MSG::ALWAYS)
MsgStream &	fatal () const
	shortcut for the method msgStream(MSG::FATAL)
MsgStream &	err () const
	shortcut for the method msgStream(MSG::ERROR)
MsgStream &	error () const
	shortcut for the method msgStream(MSG::ERROR)
MsgStream &	warning () const
	shortcut for the method msgStream(MSG::WARNING)
MsgStream &	info () const
	shortcut for the method msgStream(MSG::INFO)
MsgStream &	debug () const
	shortcut for the method msgStream(MSG::DEBUG)
MsgStream &	verbose () const
	shortcut for the method msgStream(MSG::VERBOSE)
MsgStream &	msg () const
	shortcut for the method msgStream(MSG::INFO)
MSG::Level	msgLevel () const
	get the cached level (originally extracted from the embedded MsgStream)
bool	msgLevel (MSG::Level lvl) const
	get the output level from the embedded MsgStream

Private Member Functions
std::tuple< StatusCode, EventContext >	processEvent (EventContext &&context)

Private Attributes
tbb::concurrent_bounded_queue< EventContext >	m_incoming
tbb::concurrent_bounded_queue< ResultType >	m_done
std::atomic< std::size_t >	m_inFlight { 0 }
Gaudi::Property< std::size_t >	m_queueCapacity
std::thread	m_evtLoopThread

Additional Inherited Members
Public Types inherited from extends< MinimalEventLoopMgr, Gaudi::Interfaces::IQueueingEventProcessor >
using	base_class
	Typedef to this class.
using	extend_interfaces_base
	Typedef to the base of this class.
Public Types inherited from MinimalEventLoopMgr
typedef std::vector< SmartIF< IAlgorithm > >	ListAlg
Public Types inherited from extends< Service, IEventProcessor >
using	base_class
	Typedef to this class.
using	extend_interfaces_base
	Typedef to the base of this class.
Public Types inherited from Service
using	Factory = Gaudi::PluginService::Factory<IService*( const std::string&, ISvcLocator* )>
Public Types inherited from PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
using	PropertyHolderImpl
	Typedef used to refer to this class from derived classes, as in.
Public Types inherited from CommonMessaging< implements< IService, IProperty, IStateful > >
using	base_class
Public Types inherited from extend_interfaces< Interfaces... >
using	ext_iids
	take union of the ext_iids of all Interfaces...
Protected Types inherited from MinimalEventLoopMgr
enum	State { OFFLINE , CONFIGURED , FINALIZED , INITIALIZED }
Protected Member Functions inherited from Service
std::vector< IAlgTool * > &	tools ()
	~Service () override
int	outputLevel () const
	get the Service's output level
Protected Member Functions inherited from CommonMessaging< implements< IService, IProperty, IStateful > >
MSG::Level	setUpMessaging () const override final
	Set up local caches.
MSG::Level	resetMessaging ()
	Reinitialize internal states.
void	updateMsgStreamOutputLevel (int level)
	Update the output level of the cached MsgStream.
Protected Attributes inherited from MinimalEventLoopMgr
Gaudi::Property< std::vector< std::string > >	m_topAlgNames
Gaudi::Property< std::vector< std::string > >	m_outStreamNames
Gaudi::Property< bool >	m_printCFExp
SmartIF< IAppMgrUI >	m_appMgrUI
	Reference to the IAppMgrUI interface of the application manager.
SmartIF< IIncidentSvc >	m_incidentSvc
	Reference to the incident service.
SmartIF< IAlgExecStateSvc >	m_aess
	List of top level algorithms.
ListAlg	m_topAlgList
ListAlg	m_outStreamList
	List of output streams.
State	m_state = OFFLINE
	State of the object.
bool	m_scheduledStop = false
	Scheduled stop of event processing.
AbortEventListener	m_abortEventListener
	Instance of the incident listener waiting for AbortEvent.
SmartIF< IHiveWhiteBoard >	m_WB
	< Event data service (whiteboard)
size_t	m_nevt { 0 }
Protected Attributes inherited from Service
Gaudi::StateMachine::State	m_state = Gaudi::StateMachine::OFFLINE
	Service state.
Gaudi::StateMachine::State	m_targetState = Gaudi::StateMachine::OFFLINE
	Service state.
Gaudi::Property< int >	m_outputLevel { this, "OutputLevel", MSG::NIL, "output level" }
	flag indicating whether ToolHandle tools have been added to m_tools
Gaudi::Property< bool >	m_auditorInitialize { this, "AuditInitialize", false, "trigger auditor on initialize()" }
Gaudi::Property< bool >	m_auditorStart { this, "AuditStart", false, "trigger auditor on start()" }
Gaudi::Property< bool >	m_auditorStop { this, "AuditStop", false, "trigger auditor on stop()" }
Gaudi::Property< bool >	m_auditorFinalize { this, "AuditFinalize", false, "trigger auditor on finalize()" }
Gaudi::Property< bool >	m_auditorReinitialize { this, "AuditReinitialize", false, "trigger auditor on reinitialize()" }
Gaudi::Property< bool >	m_auditorRestart { this, "AuditRestart", false, "trigger auditor on restart()" }
Gaudi::Property< bool >	m_autoRetrieveTools
Gaudi::Property< bool >	m_checkToolDeps
SmartIF< IAuditorSvc >	m_pAuditorSvc
	Auditor Service.

Detailed Description

Definition at line 25 of file QueueingEventLoopMgr.cpp.

Member Function Documentation

empty()

bool Gaudi::TestSuite::QueueingEventLoopMgr::empty ( ) const

inlineoverride

Tell if the processor has events in the queues.

Definition at line 45 of file QueueingEventLoopMgr.cpp.

                                {
      // because of the way we count "in flight" (+1 while waiting that we get something from the queue)
      // and the way tbb:concurrent_bounded_queue reports the size while waiting for pop to return (-1)
      // this is a correct definition of "empty" (nothing pending, nothing being processed and no results
      // to be popped)
      return !( m_inFlight + m_done.size() + m_incoming.size() );
    }

executeEvent()

StatusCode Gaudi::TestSuite::QueueingEventLoopMgr::executeEvent ( EventContext && ctx )

inlineoverride

Definition at line 34 of file QueueingEventLoopMgr.cpp.

                                                           {
      using namespace std::chrono_literals;
      push( std::move( ctx ) );
      std::optional<ResultType> result;
      while ( !( result = pop() ) ) std::this_thread::sleep_for( 1ms );
      return std::get<0>( std::move( *result ) );
    }

pop()

std::optional< ResultType > Gaudi::TestSuite::QueueingEventLoopMgr::pop ( )

inlineoverride

Get the next available result, if any.

Definition at line 54 of file QueueingEventLoopMgr.cpp.

                                           {
      ResultType out;
      if ( m_done.try_pop( out ) )
        return out;
      else
        return std::nullopt;
    }

processEvent()

std::tuple< StatusCode, EventContext > QueueingEventLoopMgr::processEvent ( EventContext && context )

private

Definition at line 128 of file QueueingEventLoopMgr.cpp.

                                                                                              {
  DEBMSG << "processing " << context << endmsg;
 
  bool eventfailed = false;
 
  m_aess->reset( context );
  Gaudi::Hive::setCurrentContext( context );
 
  // select the appropriate store
  if ( m_WB.isValid() ) m_WB->selectStore( context.slot() ).ignore();
 
  // Get the IProperty interface of the ApplicationMgr to pass it to RetCodeGuard
  const auto appmgr = serviceLocator()->as<IProperty>();
  // Call the execute() method of all top algorithms
  for ( auto& ita : m_topAlgList ) {
    StatusCode sc( StatusCode::FAILURE );
    try {
      if ( m_abortEventListener.abortEvent ) {
        DEBMSG << "AbortEvent incident fired by " << m_abortEventListener.abortEventSource << endmsg;
        m_abortEventListener.abortEvent = false;
        sc.ignore();
        break;
      }
      RetCodeGuard rcg( appmgr, Gaudi::ReturnCode::UnhandledException );
      sc = ita->sysExecute( context );
      rcg.ignore(); // disarm the guard
    } catch ( const GaudiException& Exception ) {
      fatal() << ".executeEvent(): Exception with tag=" << Exception.tag() << " thrown by " << ita->name() << endmsg;
      error() << Exception << endmsg;
    } catch ( const std::exception& Exception ) {
      fatal() << ".executeEvent(): Standard std::exception thrown by " << ita->name() << endmsg;
      error() << Exception.what() << endmsg;
    } catch ( ... ) { fatal() << ".executeEvent(): UNKNOWN Exception thrown by " << ita->name() << endmsg; }
 
    if ( !sc.isSuccess() ) {
      warning() << "Execution of algorithm " << ita->name() << " failed" << endmsg;
      eventfailed = true;
    }
  }
 
  m_aess->updateEventStatus( eventfailed, context );
 
  // ensure that the abortEvent flag is cleared before the next event
  if ( m_abortEventListener.abortEvent ) {
    DEBMSG << "AbortEvent incident fired by " << m_abortEventListener.abortEventSource << endmsg;
    m_abortEventListener.abortEvent = false;
  }
 
  // Call the execute() method of all output streams
  for ( auto& ito : m_outStreamList ) {
    AlgExecStateRef state = m_aess->algExecState( ito, context );
    state.setFilterPassed( true );
    StatusCode sc = ito->sysExecute( context );
    if ( !sc.isSuccess() ) {
      warning() << "Execution of output stream " << ito->name() << " failed" << endmsg;
      eventfailed = true;
    }
  }
 
  StatusCode outcome = StatusCode::SUCCESS;
  // Check if there was an error processing current event
  if ( eventfailed ) {
    error() << "Error processing event loop." << endmsg;
    std::ostringstream ost;
    m_aess->dump( ost, context );
    DEBMSG << "Dumping AlgExecStateSvc status:\n" << ost.str() << endmsg;
    outcome = StatusCode::FAILURE;
  }
 
  return { std::move( outcome ), std::move( context ) };
}

push()

void Gaudi::TestSuite::QueueingEventLoopMgr::push ( EventContext && ctx )

inlineoverride

Definition at line 42 of file QueueingEventLoopMgr.cpp.

{ m_incoming.push( std::move( ctx ) ); }

start()

StatusCode QueueingEventLoopMgr::start ( )

override

Definition at line 100 of file QueueingEventLoopMgr.cpp.

                                       {
  auto ok = base_class::start();
  if ( !ok ) return ok;
 
  info() << m_queueCapacity << endmsg;
 
  m_evtLoopThread = std::thread( [this]() {
    while ( true ) {
      EventContext ctx;
      ++m_inFlight; // yes, this is not accurate.
      m_incoming.pop( ctx );
      if ( ctx.valid() ) {
        // the sleep is not strictly needed, but it should make the output more stable for the test
        std::this_thread::sleep_for( std::chrono::milliseconds( 100 ) );
        m_done.push( processEvent( std::move( ctx ) ) );
        --m_inFlight; // yes, this is not accurate.
      } else {
        DEBMSG << "exiting event loop thread" << endmsg;
        m_done.emplace( StatusCode::SUCCESS, std::move( ctx ) );
        --m_inFlight; // yes, this is not accurate.
        break;
      }
    }
  } );
 
  return ok;
}

stop()

StatusCode QueueingEventLoopMgr::stop ( )

override

Definition at line 200 of file QueueingEventLoopMgr.cpp.

                                      {
  // Send an invalid context to stop the processing thread
  push( EventContext{} );
  m_evtLoopThread.join();
 
  return base_class::stop();
}

Member Data Documentation

m_done

tbb::concurrent_bounded_queue<ResultType> Gaudi::TestSuite::QueueingEventLoopMgr::m_done

private

Definition at line 64 of file QueueingEventLoopMgr.cpp.

m_evtLoopThread

std::thread Gaudi::TestSuite::QueueingEventLoopMgr::m_evtLoopThread

private

Definition at line 74 of file QueueingEventLoopMgr.cpp.

m_incoming

tbb::concurrent_bounded_queue<EventContext> Gaudi::TestSuite::QueueingEventLoopMgr::m_incoming

private

Definition at line 63 of file QueueingEventLoopMgr.cpp.

m_inFlight

std::atomic<std::size_t> Gaudi::TestSuite::QueueingEventLoopMgr::m_inFlight { 0 }

private

Definition at line 65 of file QueueingEventLoopMgr.cpp.

{ 0 };

m_queueCapacity

Gaudi::Property<std::size_t> Gaudi::TestSuite::QueueingEventLoopMgr::m_queueCapacity

private

Initial value:

{
        this, "Capacity", m_incoming.capacity() + 1,
        [this]( Gaudi::Details::PropertyBase& ) { m_incoming.set_capacity( m_queueCapacity - 1 ); } }

Definition at line 68 of file QueueingEventLoopMgr.cpp.

                                            {
        this, "Capacity", m_incoming.capacity() + 1,
        [this]( Gaudi::Details::PropertyBase& ) { m_incoming.set_capacity( m_queueCapacity - 1 ); } };

---

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

• GaudiTestSuite/src/QueueingEventProcessor/QueueingEventLoopMgr.cpp