Gaudi::Examples::QueueingApplication Class Reference

Inheritance diagram for Gaudi::Examples::QueueingApplication:

Collaboration diagram for Gaudi::Examples::QueueingApplication:

Private Member Functions
int	run () override
	Implement the application main logic: More...
	Application (Options opts)
	Construct and configure the application from the provided options. More...

Additional Inherited Members
Public Types inherited from Gaudi.Application
using	Options = std::map< std::string, std::string >
using	Factory = Gaudi::PluginService::Factory< Application(Options)>
Public Member Functions inherited from Gaudi.Application
def	__init__ (self, opts, appType="Gaudi::Application")
def	create (cls, appType, opts)
def	run (self)
def	__del__ (self)
	Application (Options opts)
	Construct and configure the application from the provided options. More...
virtual	~Application ()
int	run (std::function< int(SmartIF< IStateful > &)> action)
	Run a user provided implementation of the application main logic. More...
Static Public Member Functions inherited from Gaudi.Application
static Factory::ReturnType	create (std::string_view type, Options opts)
	Factory function to instantiate a derived class via GaudiPluginService. More...
Protected Attributes inherited from Gaudi.Application
SmartIF< IStateful >	app
	Handle to the ApplicationMgr instance. More...

Detailed Description

Definition at line 48 of file QueueingApplication.cpp.

Member Function Documentation

Application()

Gaudi.Application::Application

private

Construct and configure the application from the provided options.

Definition at line 33 of file Application.cpp.

                                                               {
  // # Prepare the application
  // - instantiate
  app = Gaudi::createApplicationMgr();
  GAUDI_ASSERT_THROW_NAME( app, "failure instantiating ApplicationMgr", "Gaudi::Application" );
 
  // - main configuration
  consumeOptions( app.as<IProperty>(), options );
  // - start minimal services
  GAUDI_ASSERT_THROW_NAME( app->configure(), "failure creating basic services", "Gaudi::Application" );
 
  consumeOptions( SmartIF<IMessageSvc>{ app }.as<IProperty>(), options );
 
  // - prepare job configuration
  {
    auto  sloc = app.as<ISvcLocator>();
    auto& jos  = sloc->getOptsSvc();
    for ( const auto& [name, value] : options ) jos.set( name, value );
  }
}

run()

int Gaudi::Examples::QueueingApplication::run ( )

inlineoverrideprivatevirtual

Implement the application main logic:

• prepare for processing (initialize + start)
• loop over events
• terminate (stop + finalize)

Reimplemented from Gaudi.Application.

Definition at line 53 of file QueueingApplication.cpp.

                       {
      // - parameters for the job
      const std::size_t n_of_batches  = 2;
      const std::size_t evts_in_batch = 5;
 
      // - get ready to process events
      app->initialize().ignore();
 
      // - this MsgStream is useful to have uniform printout
      MsgStream log( app.as<IMessageSvc>(), "<main>" );
 
      // we print the parameters here so that they appear close to the Capacity in the log
      log << MSG::INFO << " n_of_batches: " << n_of_batches << endmsg;
      log << MSG::INFO << " evts_in_batch: " << evts_in_batch << endmsg;
 
      app->start().ignore(); // this starts the QueueingEventLoopMgr processing thread
 
      // - main processing loop
      {
        // - get the IQueueingEventProcessor interface of the application
        SmartIF<Gaudi::Interfaces::IQueueingEventProcessor> qep{ app };
 
        // - processing state informations
        //   - events ready to be processed
        std::queue<EventContext> ready;
        //   - count of events enqueued
        std::size_t evt_count = 0;
 
        // - loop over input batches
        for ( std::size_t batch = 1; batch <= n_of_batches; ++batch ) {
          // - prepare the batch
          log << MSG::INFO << "prepare batch of events n. " << batch << endmsg;
          if ( batch == 2 ) {
            log << MSG::INFO << "  (pretend we need time so that the processing thread drains the input queue)"
                << endmsg;
            sleep_for( 4s );
            log << MSG::INFO << "  (all events in the queue should have been processed by now)" << endmsg;
          }
          for ( std::size_t i = 0; i < evts_in_batch; ++i ) {
            // - create a new EventContext for each event in the batch
            auto ctx = qep->createEventContext();
            // ... here you can do something with the context ... like setting I/O related stuff
            // - declare the event as ready to be enqueued
            ready.push( std::move( ctx ) );
          }
 
          // - once the batch is ready we can push all events, relying on the push to block if needed
          log << MSG::INFO << "looping over the batch" << endmsg;
          while ( !ready.empty() ) {
            ++evt_count;
            log << MSG::INFO << "- pushing event " << evt_count << " (" << ready.front() << ")..." << endmsg;
            qep->push( std::move( ready.front() ) ); // this blocks if the system is full
            ready.pop();
            log << MSG::INFO << "  ... event " << evt_count << " pushed" << endmsg;
 
            // - for each push we try a pop, to avoid that the output queue gets too big
            log << MSG::INFO << "- checking for completed events" << endmsg;
            if ( auto result = qep->pop() ) { // this never blocks, but evaulates to false if there was nothing to pop
              // - if an event completed, we can do something (e.g. I/O)
              auto&& [sc, ctx] = std::move( *result );
              log << MSG::INFO << "  " << ctx << " -> " << sc << endmsg;
            }
          }
        }
        log << MSG::INFO << "no more inputs: let's drain the output queue" << endmsg;
        while ( !qep->empty() ) {
          if ( auto result = qep->pop() ) {
            auto&& [sc, ctx] = std::move( *result );
            log << MSG::INFO << "  " << ctx << " -> " << sc << endmsg;
          } else {
            sleep_for( 10ms );
          }
        }
 
        // - nothing else to do on the events
        log << MSG::INFO << "all done" << endmsg;
      }
 
      // - terminate the application
      app->stop().ignore(); // this stops the QueueingEventLoopMgr processing thread
      app->finalize().ignore();
 
      // - get and propagate the return code the ApplicationMgr whishes to expose
      return getAppReturnCode( app.as<IProperty>() );
    }

---

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

• GaudiExamples/src/QueueingEventProcessor/QueueingApplication.cpp