This class provides configuration properties to enable OutputStream file sequences. More...

#include <OutputStreamSequencerSvc.h>

Inheritance diagram for OutputStreamSequencerSvc:

Collaboration diagram for OutputStreamSequencerSvc:

Public Types
typedef std::pair< std::string, std::string >	RangeReport_t

typedef std::unique_ptr< RangeReport_t >	RangeReport_ptr

Public Member Functions
	OutputStreamSequencerSvc (const std::string &name, ISvcLocator *pSvcLocator)
	Standard Service Constructor. More...

virtual	~OutputStreamSequencerSvc ()
	Destructor. More...

virtual StatusCode	initialize () override final
	Required of all Gaudi services: More...

virtual StatusCode	finalize () override final
	Required of all Gaudi services: More...

virtual StatusCode	queryInterface (const InterfaceID &riid, void **ppvInterface) override final
	Required of all Gaudi services: see Gaudi documentation for details. More...

virtual void	handle (const Incident &) override final
	Incident service handle. More...

std::string	buildSequenceFileName (const std::string &)
	Returns sequenced file name for output stream. More...

void	publishRangeReport (const std::string &outputFile)

RangeReport_ptr	getRangeReport ()

std::string	incidentName () const
	The name of the incident that starts a new event sequence. More...

std::string	currentRangeID () const
	The current Event Range ID (only one range is returned) More...

std::string	setRangeID (const std::string &rangeID)
	set the RangeID (possibly temporarily) so the right Range Filename may be generated More...

bool	inUse () const
	Is the service in active use? (true after the first range incident is handled) More...

const std::string &	lastIncident ()
	Last incident type that was handled. More...

MsgStream &	msg () const

MsgStream &	msg (const MSG::Level lvl) const

bool	msgLvl (const MSG::Level lvl) const

Static Public Member Functions
static const InterfaceID &	interfaceID ()
	Retrieve interface ID. More...

static bool	inConcurrentEventsMode ()
	Are there concurrent events? (threads>1) More...

Private Attributes
ServiceHandle< MetaDataSvc >	m_metaDataSvc

int	m_fileSequenceNumber
	The event sequence number. More...

std::string	m_currentRangeID
	Current EventRange ID constructed on the last NextRange incident. More...

std::string	m_lastFileName
	Recently constructed full file name (useful in single threaded processing) More...

std::string	m_lastIncident
	Last incident type that was handled. More...

std::vector< std::string >	m_rangeIDinSlot
	EventRange ID for all slots. More...

StringProperty	m_incidentName
	SequenceIncidentName, incident name for triggering file sequencing. More...

BooleanProperty	m_reportingOn
	Flag to switch on storage of reporting info in fnToRangeId. More...

std::map< std::string, std::string >	m_fnToRangeId

std::map< std::string, std::string >::iterator	m_finishedRange

std::mutex	m_mutex

Detailed Description

This class provides configuration properties to enable OutputStream file sequences.

Definition at line 32 of file OutputStreamSequencerSvc.h.

Member Typedef Documentation

◆ RangeReport_ptr

typedef std::unique_ptr<RangeReport_t> OutputStreamSequencerSvc::RangeReport_ptr

Definition at line 37 of file OutputStreamSequencerSvc.h.

◆ RangeReport_t

typedef std::pair<std::string,std::string> OutputStreamSequencerSvc::RangeReport_t

Definition at line 36 of file OutputStreamSequencerSvc.h.

Constructor & Destructor Documentation

◆ OutputStreamSequencerSvc()

OutputStreamSequencerSvc::OutputStreamSequencerSvc	(	const std::string &	name,
		ISvcLocator *	pSvcLocator
	)

Standard Service Constructor.

Definition at line 20 of file OutputStreamSequencerSvc.cxx.

                                                                                                   : ::AthService(name, pSvcLocator),
     m_metaDataSvc("MetaDataSvc", name),
     m_fileSequenceNumber(-1)
 {
 }

◆ ~OutputStreamSequencerSvc()

OutputStreamSequencerSvc::~OutputStreamSequencerSvc ( )

virtual

Destructor.

Definition at line 27 of file OutputStreamSequencerSvc.cxx.

27 {

28 }

Member Function Documentation

◆ buildSequenceFileName()

std::string OutputStreamSequencerSvc::buildSequenceFileName ( const std::string & orgFileName )

Returns sequenced file name for output stream.

Definition at line 170 of file OutputStreamSequencerSvc.cxx.

 {
    if( !inUse() ) {
       // Event sequences not in use, just return the original filename
       return orgFileName;
    }
    std::string rangeID = currentRangeID();
    std::lock_guard lockg( m_mutex );
    // build the full output file name for this event range
    std::string fileNameCore = orgFileName, fileNameExt;
    std::size_t sepPos = orgFileName.find('[');
    if (sepPos != std::string::npos) {
       fileNameCore = orgFileName.substr(0, sepPos);
       fileNameExt = orgFileName.substr(sepPos);
    }
    std::ostringstream n;
    n << fileNameCore << "." << rangeID << fileNameExt;
    m_lastFileName = n.str();
  
    if( m_reportingOn.value() ) {
       m_fnToRangeId.insert( std::pair(m_lastFileName, rangeID) );
    }
  
    return m_lastFileName;
 }

◆ currentRangeID()

std::string OutputStreamSequencerSvc::currentRangeID ( ) const

The current Event Range ID (only one range is returned)

Definition at line 197 of file OutputStreamSequencerSvc.cxx.

 {
    if( !inUse() )  return "";
    auto slot = Gaudi::Hive::currentContext().slot();
    if( slot == EventContext::INVALID_CONTEXT_ID )  slot = 0; 
    std::lock_guard lockg( m_mutex );
    if( slot >= m_rangeIDinSlot.size() ) return "";
    return m_rangeIDinSlot[ slot ];
 }

◆ finalize()

StatusCode OutputStreamSequencerSvc::finalize ( )

finaloverridevirtual

Required of all Gaudi services:

Definition at line 64 of file OutputStreamSequencerSvc.cxx.

                                               {
    // Release MetaDataSvc 
    if (!m_metaDataSvc.release().isSuccess()) {
       ATH_MSG_WARNING("Cannot release MetaDataSvc.");
    }
    return(StatusCode::SUCCESS);
 }

◆ getRangeReport()

OutputStreamSequencerSvc::RangeReport_ptr OutputStreamSequencerSvc::getRangeReport ( )

Definition at line 228 of file OutputStreamSequencerSvc.cxx.

 {
   RangeReport_ptr report;
   if( !m_reportingOn.value() ) {
      ATH_MSG_WARNING("Reporting not turned on - set " << m_reportingOn.name() << " to True");
   } else {
      std::lock_guard lockg( m_mutex );
      if(m_finishedRange!=m_fnToRangeId.end()) {
         report = std::make_unique<RangeReport_t>(m_finishedRange->second,m_finishedRange->first);
         m_fnToRangeId.erase(m_finishedRange);
         m_finishedRange=m_fnToRangeId.end();
      }
   }
   return report;
 }

◆ handle()

void OutputStreamSequencerSvc::handle ( const Incident & inc )

finaloverridevirtual

Incident service handle.

Definition at line 94 of file OutputStreamSequencerSvc.cxx.

 {
    auto slot = Gaudi::Hive::currentContext().slot();
    bool has_context = ( slot != EventContext::INVALID_CONTEXT_ID );
    // in AthenaSP there is no context so go with the first slot
    if( !has_context )  slot = 0;
    m_lastIncident = inc.type();
    ATH_MSG_INFO("Handling incident of type " << m_lastIncident << " for slot=" << slot
                 << (!has_context? " NO event context":"") );
  
    if( inc.type() == incidentName() ) {  // NextEventRange 
       std::string rangeID;
       const FileIncident* fileInc  = dynamic_cast<const FileIncident*>(&inc);
       if (fileInc != nullptr) {
          rangeID = fileInc->fileName();
          ATH_MSG_DEBUG("Requested (through incident) Next Event Range filename extension: " << rangeID);
       }
  
       if( rangeID == "dummy" ) {
          if( not inConcurrentEventsMode() ) {
             // finish the previous Range here only in SEQUENTIAL (threads<2) event processing
             // Write metadata on the incident finishing a Range (filename=="dummy") in ES MP
             ATH_MSG_DEBUG("MetaData transition");
             // immediate write and disconnect for ES, otherwise do it after Event write is done
             bool disconnect { true };
             if( !m_metaDataSvc->transitionMetaDataFile( m_lastFileName, disconnect ).isSuccess() ) {
                throw GaudiException("Cannot transition MetaData", name(), StatusCode::FAILURE);
             }
          }
          // exit now, wait for the next (real) incident that will start the next range
          return;
       }
       {
          // start a new range
          std::lock_guard lockg( m_mutex );
          m_fileSequenceNumber++;
          if( rangeID.empty() ) {
             std::ostringstream n;
             n << "_" << std::setw(4) << std::setfill('0') << m_fileSequenceNumber;
             rangeID = n.str();
             ATH_MSG_DEBUG("Default next event range filename extension: " << rangeID);
          } 
          if( slot >= m_rangeIDinSlot.size() ) {
             // MN - late resize, is there a better place for it?
             m_rangeIDinSlot.resize( std::max(slot+1, Gaudi::Concurrency::ConcurrencyFlags::numConcurrentEvents()) );
          }
          // from now on new events will use the new rangeID
          m_currentRangeID = rangeID;
          // for ESMT these incidents are asynchronous, so wait for BeginProcessing to update the range map
          if( not inConcurrentEventsMode() or has_context ) {
             m_rangeIDinSlot[ slot ] = rangeID;
          }
       }
       if( not inConcurrentEventsMode() and not fileInc ) {
          // non-file incident case (filename=="") in regular SP LoopMgr
          ATH_MSG_DEBUG("MetaData transition");
          bool disconnect { false };
          // MN: may not know the full filename yet, but that is only needed for disconnect==true
          if( !m_metaDataSvc->transitionMetaDataFile( "" /*m_lastFileName*/, disconnect ).isSuccess() ) {
             throw GaudiException("Cannot transition MetaData", name(), StatusCode::FAILURE);
          }
       }
    }
    else if( inc.type() == IncidentType::BeginProcessing ) {
       // new event start - assing current rangeId to its slot
       ATH_MSG_DEBUG("Assigning rangeID = " << m_currentRangeID << " to slot " << slot);
       std::lock_guard lockg( m_mutex );
       // If this service is enabled but not getting NextRange incidents, need to resize here
       if( slot >= m_rangeIDinSlot.size() ) {
          m_rangeIDinSlot.resize( std::max(slot+1, Gaudi::Concurrency::ConcurrencyFlags::numConcurrentEvents()) );
       }
       m_rangeIDinSlot[ slot ] = m_currentRangeID;
    }
 }

◆ incidentName()

std::string OutputStreamSequencerSvc::incidentName ( ) const

inline

The name of the incident that starts a new event sequence.

Definition at line 65 of file OutputStreamSequencerSvc.h.

65 { return m_incidentName.value(); }

◆ inConcurrentEventsMode()

bool OutputStreamSequencerSvc::inConcurrentEventsMode ( )

static

Are there concurrent events? (threads>1)

Definition at line 84 of file OutputStreamSequencerSvc.cxx.

                                                          {
    return Gaudi::Concurrency::ConcurrencyFlags::numConcurrentEvents() > 1;
 }

◆ initialize()

StatusCode OutputStreamSequencerSvc::initialize ( )

finaloverridevirtual

Required of all Gaudi services:

Definition at line 30 of file OutputStreamSequencerSvc.cxx.

                                                 {
    ATH_MSG_DEBUG("Initializing " << name());
  
    // Set to be listener for end of event
    ServiceHandle<IIncidentSvc> incsvc("IncidentSvc", this->name());
    if (!incsvc.retrieve().isSuccess()) {
       ATH_MSG_FATAL("Cannot get IncidentSvc.");
       return(StatusCode::FAILURE);
    }
    if( !incidentName().empty() ) {
       incsvc->addListener(this, incidentName(), 100);
       incsvc->addListener(this, IncidentType::BeginProcessing, 100);
       ATH_MSG_DEBUG("Listening to " << incidentName() << " incidents" );
       ATH_MSG_DEBUG("Reporting is " << (m_reportingOn.value()? "ON" : "OFF") );
       // Retrieve MetaDataSvc
       if( !m_metaDataSvc.isValid() and !m_metaDataSvc.retrieve().isSuccess() ) {
          ATH_MSG_ERROR("Cannot get MetaDataSvc");
          return StatusCode::FAILURE;
       }
    }
  
    if( inConcurrentEventsMode() ) {
       ATH_MSG_DEBUG("Concurrent events mode");
    } else {
       ATH_MSG_VERBOSE("Sequential events mode");
    }
  
    // Gaudi::Concurrency::ConcurrencyFlags::numConcurrentEvents() not set yet
    // m_rangeIDinSlot.resize( );
    m_finishedRange = m_fnToRangeId.end();
  
    return(StatusCode::SUCCESS);
 }

◆ interfaceID()

static const InterfaceID& OutputStreamSequencerSvc::interfaceID ( )

inlinestatic

Retrieve interface ID.

Definition at line 46 of file OutputStreamSequencerSvc.h.

46 { return IID_OutputStreamSequencerSvc; }

◆ inUse()

bool OutputStreamSequencerSvc::inUse ( ) const

Is the service in active use? (true after the first range incident is handled)

Definition at line 89 of file OutputStreamSequencerSvc.cxx.

                                               {
    return m_fileSequenceNumber >= 0;
 }

◆ lastIncident()

const std::string& OutputStreamSequencerSvc::lastIncident ( )

inline

Last incident type that was handled.

Definition at line 80 of file OutputStreamSequencerSvc.h.

80 { return m_lastIncident; }

◆ msg() [1/2]

MsgStream& AthCommonMsg< Service >::msg ( ) const

inlineinherited

Definition at line 24 of file AthCommonMsg.h.

                                 {
     return this->msgStream();
   }

◆ msg() [2/2]

MsgStream& AthCommonMsg< Service >::msg ( const MSG::Level lvl ) const

inlineinherited

Definition at line 27 of file AthCommonMsg.h.

                                                   {
     return this->msgStream(lvl);
   }

◆ msgLvl()

bool AthCommonMsg< Service >::msgLvl ( const MSG::Level lvl ) const

inlineinherited

Definition at line 30 of file AthCommonMsg.h.

                                                {
     return this->msgLevel(lvl);
   }

◆ publishRangeReport()

void OutputStreamSequencerSvc::publishRangeReport ( const std::string & outputFile )

Definition at line 222 of file OutputStreamSequencerSvc.cxx.

 {
    std::lock_guard lockg( m_mutex );
    m_finishedRange = m_fnToRangeId.find(outputFile);
 }

◆ queryInterface()

StatusCode OutputStreamSequencerSvc::queryInterface	(	const InterfaceID &	riid,
		void **	ppvInterface
	)

finaloverridevirtual

Required of all Gaudi services: see Gaudi documentation for details.

Definition at line 72 of file OutputStreamSequencerSvc.cxx.

                                                                                                 {
    if (riid == this->interfaceID()) {
       *ppvInterface = this;
    } else {
       // Interface is not directly available: try out a base class
       return(::AthService::queryInterface(riid, ppvInterface));
    }
    addRef();
    return(StatusCode::SUCCESS);
 }

◆ setRangeID()

std::string OutputStreamSequencerSvc::setRangeID ( const std::string & rangeID )

set the RangeID (possibly temporarily) so the right Range Filename may be generated

Definition at line 208 of file OutputStreamSequencerSvc.cxx.

 {
    auto slot = Gaudi::Hive::currentContext().slot();
    if( slot == EventContext::INVALID_CONTEXT_ID )  slot = 0;
    std::lock_guard lockg( m_mutex );
    if( slot >= m_rangeIDinSlot.size() ) {
       throw std::runtime_error("OutputStreamSequencer::setRangeID(): slot out of range");
    }
    std::string oldrange =  m_rangeIDinSlot[ slot ];
    m_rangeIDinSlot[ slot ] = rangeID;
    return oldrange;
 }

Member Data Documentation

◆ m_currentRangeID

std::string OutputStreamSequencerSvc::m_currentRangeID

private

Current EventRange ID constructed on the last NextRange incident.

Definition at line 89 of file OutputStreamSequencerSvc.h.

◆ m_fileSequenceNumber

int OutputStreamSequencerSvc::m_fileSequenceNumber

private

The event sequence number.

Definition at line 86 of file OutputStreamSequencerSvc.h.

◆ m_finishedRange

std::map<std::string,std::string>::iterator OutputStreamSequencerSvc::m_finishedRange

private

Definition at line 109 of file OutputStreamSequencerSvc.h.

◆ m_fnToRangeId

std::map<std::string,std::string> OutputStreamSequencerSvc::m_fnToRangeId

private

Definition at line 108 of file OutputStreamSequencerSvc.h.

◆ m_incidentName

StringProperty OutputStreamSequencerSvc::m_incidentName

private

Initial value:

{this, "SequenceIncidentName", "",

"Name of the incident that signals the next Event Range start" }

SequenceIncidentName, incident name for triggering file sequencing.

Definition at line 102 of file OutputStreamSequencerSvc.h.

◆ m_lastFileName

std::string OutputStreamSequencerSvc::m_lastFileName

private

Recently constructed full file name (useful in single threaded processing)

Definition at line 92 of file OutputStreamSequencerSvc.h.

◆ m_lastIncident

std::string OutputStreamSequencerSvc::m_lastIncident

private

Last incident type that was handled.

Definition at line 95 of file OutputStreamSequencerSvc.h.

◆ m_metaDataSvc

ServiceHandle<MetaDataSvc> OutputStreamSequencerSvc::m_metaDataSvc

private

Definition at line 83 of file OutputStreamSequencerSvc.h.

◆ m_mutex

std::mutex OutputStreamSequencerSvc::m_mutex

mutableprivate

Definition at line 111 of file OutputStreamSequencerSvc.h.

◆ m_rangeIDinSlot

std::vector<std::string> OutputStreamSequencerSvc::m_rangeIDinSlot

private

EventRange ID for all slots.

Definition at line 98 of file OutputStreamSequencerSvc.h.

◆ m_reportingOn

BooleanProperty OutputStreamSequencerSvc::m_reportingOn

private

Initial value:

{this, "ReportingOn", false,

"If True, keep info about Ranges for getRangeReport() calls"}

Flag to switch on storage of reporting info in fnToRangeId.

Definition at line 105 of file OutputStreamSequencerSvc.h.

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

Public Types

Public Member Functions

Static Public Member Functions

Private Attributes

Detailed Description

Member Typedef Documentation

◆ RangeReport_ptr

◆ RangeReport_t

Constructor & Destructor Documentation

◆ OutputStreamSequencerSvc()

◆ ~OutputStreamSequencerSvc()

Member Function Documentation

◆ buildSequenceFileName()

◆ currentRangeID()

◆ finalize()

◆ getRangeReport()

◆ handle()

◆ incidentName()

◆ inConcurrentEventsMode()

◆ initialize()

◆ interfaceID()

◆ inUse()

◆ lastIncident()

◆ msg() [1/2]

◆ msg() [2/2]

◆ msgLvl()

◆ publishRangeReport()

◆ queryInterface()

◆ setRangeID()

Member Data Documentation

◆ m_currentRangeID

◆ m_fileSequenceNumber

◆ m_finishedRange

◆ m_fnToRangeId

◆ m_incidentName

◆ m_lastFileName

◆ m_lastIncident

◆ m_metaDataSvc

◆ m_mutex

◆ m_rangeIDinSlot

◆ m_reportingOn