OutputStreamSequencerSvc Class Reference

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 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...

Static Public Member Functions
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...
BooleanProperty	m_replaceRangeMode
	Flag to put in ReplaceRangeMode (i.e. 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 29 of file OutputStreamSequencerSvc.h.

Member Typedef Documentation

RangeReport_ptr

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

Definition at line 35 of file OutputStreamSequencerSvc.h.

RangeReport_t

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

Definition at line 34 of file OutputStreamSequencerSvc.h.

Constructor & Destructor Documentation

OutputStreamSequencerSvc()

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

Standard Service Constructor.

Definition at line 24 of file OutputStreamSequencerSvc.cxx.

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

~OutputStreamSequencerSvc()

OutputStreamSequencerSvc::~OutputStreamSequencerSvc ( )

virtual

Destructor.

Definition at line 32 of file OutputStreamSequencerSvc.cxx.

                                                    {
}

Member Function Documentation

buildSequenceFileName()

std::string OutputStreamSequencerSvc::buildSequenceFileName

(

const std::string &

orgFileName

)

Returns sequenced file name for output stream.

Definition at line 176 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 );
   if (!m_replaceRangeMode) {
     // 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();
   } else {
     std::string_view origFileNameView = orgFileName;
     std::size_t open = origFileNameView.find('[');
     std::size_t close = origFileNameView.find(']');
     // If we don't find a [  ] enclosed section, just append the rangeID to the
     // end
     if (open == std::string_view::npos || close == std::string_view::npos) {
       m_lastFileName = std::format("{}.{}", origFileNameView, rangeID);
     } else {
       // build list of elems to substitute from
       ATH_MSG_DEBUG("Building element list");
       std::vector<std::string_view> elems{};
       std::size_t pos = open + 1;
       for (std::size_t comma = origFileNameView.find(',', pos);
            comma < close;
            comma = origFileNameView.find(',', pos)) {
     std::string_view item = origFileNameView.substr(pos, comma - pos);
     ATH_MSG_DEBUG("(start) pos = " << pos << ", (end) comma = " << comma << ", item = " << item);
         elems.push_back(item);
         pos = comma + 1;
       }
       std::string_view last_item = origFileNameView.substr(pos, close - pos);
       ATH_MSG_DEBUG("(start) pos = " << pos << ", (end) close = " << close << ", item = " << last_item);
       elems.push_back(last_item);
       // substitute
       std::size_t rangeIdx{};
       auto rangeIdxParseRes = std::from_chars(
           rangeID.data(), rangeID.data() + rangeID.size(), rangeIdx);
       if (rangeIdxParseRes.ec != std::errc()) {
         ATH_MSG_ERROR(
             "Error parsing rangeID to integer. Replacing [] list with "
             "rangeID.");
         m_lastFileName =
             std::format("{}{}{}", origFileNameView.substr(0, open), rangeID,
                         origFileNameView.substr(close + 1));
       } else if (rangeIdx >= elems.size()) {
         ATH_MSG_WARNING(
             "Number of elements in [] list <= rangeID. Replacing [] list with "
             "rangeID.");
         m_lastFileName =
             std::format("{}{}{}", origFileNameView.substr(0, open), rangeID,
                         origFileNameView.substr(close + 1));
       } else {
         m_lastFileName = std::format(
             "{}{}{}", origFileNameView.substr(0, open), elems.at(rangeIdx),
             origFileNameView.substr(close + 1));
         ATH_MSG_DEBUG("Output file: " << m_lastFileName);
       }
     }
   }
 
   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 254 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 70 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 285 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 90 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();
        // Handle BeginInputFile
        if (inc.type() == IncidentType::BeginInputFile) {
          rangeID = "INFILE";
        }
        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 };
            std::lock_guard lockg( m_mutex );
            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);
         }
         else if (rangeID == "INFILE") {
             rangeID = std::to_string(m_fileSequenceNumber);
         }
         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 ] = std::move(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
      std::lock_guard lockg( m_mutex );
      ATH_MSG_DEBUG("Assigning rangeID = " << m_currentRangeID << " to slot " << slot);
      // 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 59 of file OutputStreamSequencerSvc.h.

{ return m_incidentName.value(); }

inConcurrentEventsMode()

bool OutputStreamSequencerSvc::inConcurrentEventsMode ( )

static

Are there concurrent events? (threads>1)

Definition at line 79 of file OutputStreamSequencerSvc.cxx.

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

initialize()

StatusCode OutputStreamSequencerSvc::initialize ( )

finaloverridevirtual

Required of all Gaudi services:

Definition at line 35 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( );
   std::lock_guard lockg( m_mutex );
   m_finishedRange = m_fnToRangeId.end();
 
   return(StatusCode::SUCCESS);
}

inUse()

bool OutputStreamSequencerSvc::inUse

(

)

const

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

Definition at line 84 of file OutputStreamSequencerSvc.cxx.

                                              {
   std::lock_guard lockg( m_mutex );
   return m_fileSequenceNumber >= 0;
}

lastIncident()

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

inline

Last incident type that was handled.

Definition at line 74 of file OutputStreamSequencerSvc.h.

{ return m_lastIncident; }

publishRangeReport()

void OutputStreamSequencerSvc::publishRangeReport

(

const std::string &

outputFile

)

Definition at line 279 of file OutputStreamSequencerSvc.cxx.

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

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 265 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 83 of file OutputStreamSequencerSvc.h.

m_fileSequenceNumber

int OutputStreamSequencerSvc::m_fileSequenceNumber {}

private

The event sequence number.

Definition at line 80 of file OutputStreamSequencerSvc.h.

m_finishedRange

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

private

Definition at line 111 of file OutputStreamSequencerSvc.h.

m_fnToRangeId

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

private

Definition at line 110 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 96 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 86 of file OutputStreamSequencerSvc.h.

m_lastIncident

std::string OutputStreamSequencerSvc::m_lastIncident

private

Last incident type that was handled.

Definition at line 89 of file OutputStreamSequencerSvc.h.

m_metaDataSvc

ServiceHandle<MetaDataSvc> OutputStreamSequencerSvc::m_metaDataSvc

private

Definition at line 77 of file OutputStreamSequencerSvc.h.

m_mutex

std::mutex OutputStreamSequencerSvc::m_mutex

mutableprivate

Definition at line 113 of file OutputStreamSequencerSvc.h.

m_rangeIDinSlot

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

private

EventRange ID for all slots.

Definition at line 92 of file OutputStreamSequencerSvc.h.

m_replaceRangeMode

BooleanProperty OutputStreamSequencerSvc::m_replaceRangeMode

private

Initial value:

{
       this, "ReplaceRangeMode", false,
       "If True, everything between [ and ] in the output filename is treated "
       "as a comma-separated list, and the range_idth element of the list is "
       "selected"}

Flag to put in ReplaceRangeMode (i.e.

everything between [ and ] (inclusive) is replaced with the range_idth element of that list)

Definition at line 104 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 99 of file OutputStreamSequencerSvc.h.

---

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

• OutputStreamSequencerSvc.h
• OutputStreamSequencerSvc.cxx