AthenaOutputStream Class Reference

algorithm that marks for write data objects in SG More...

#include <AthenaOutputStream.h>

Inheritance diagram for AthenaOutputStream:

Collaboration diagram for AthenaOutputStream:

Public Types
typedef std::vector< SG::DataProxy * >	Items
	typedefs More...
typedef std::vector< std::pair< std::string, std::string > >	TypeKeyPairs
typedef std::recursive_mutex	mutex_t

Public Member Functions
	AthenaOutputStream (const std::string &name, ISvcLocator *pSvcLocator)
	Standard algorithm Constructor. More...
virtual	~AthenaOutputStream ()
	Standard Destructor. More...

implement IAlgorithm
ServiceHandle< StoreGateSvc >	m_dataStore {this, "Store", "StoreGateSvc/StoreGateSvc", "Handle to event store"}
	Handle to the StoreGateSvc store where the data we want to write out resides. More...
ServiceHandle< StoreGateSvc >	m_metadataStore {this, "MetadataStore", "StoreGateSvc/MetaDataStore", "Handle to metadata store"}
ServiceHandle< StoreGateSvc > *	m_currentStore
ServiceHandle< IItemListSvc >	m_itemSvc {this, "ItemListSvc", "ItemListSvc"}
	Handles to all the necessary services. More...
ServiceHandle< MetaDataSvc >	m_metaDataSvc {this, "MetaDataSvc", "MetaDataSvc"}
ServiceHandle< IDictLoaderSvc >	m_dictLoader {this, "AthDictLoaderSvc", "AthDictLoaderSvc"}
ServiceHandle< ITPCnvSvc >	m_tpCnvSvc {this, "AthTPCnvSvc", "AthTPCnvSvc"}
ServiceHandle< IIncidentSvc >	m_incidentSvc {this, "IncidentSvc", "IncidentSvc"}
ServiceHandle< IClassIDSvc >	m_pCLIDSvc {this, "ClassIDSvc", "ClassIDSvc"}
ServiceHandle< OutputStreamSequencerSvc >	m_outSeqSvc {this, "OutputStreamSequencerSvc", "OutputStreamSequencerSvc"}
StringProperty	m_streamName {this, "StreamName", "", "Name of the output stream"}
	Stream name (defaults to algorithm name) More...
StringArrayProperty	m_itemList {this, "ItemList", {}, "List of items to write", "OutputStreamItemList"}
	Vector of item names. More...
StringArrayProperty	m_metadataItemList {this, "MetadataItemList", {}, "List of metadata items to write","OutputStreamItemList"}
	Vector of item names. More...
StringProperty	m_keepProvenances
	Provenance record selection. More...
StringArrayProperty	m_compressionListHigh {this, "CompressionListHigh", {}, "Lossy float compression list (high)"}
	Vector of item names. More...
StringArrayProperty	m_compressionListLow {this, "CompressionListLow", {}, "Lossy float compression list (low)"}
	Vector of item names. More...
UnsignedIntegerProperty	m_compressionBitsHigh {this, "CompressionBitsHigh", 7, "Lossy float compression bits (high)"}
	Number of mantissa bits in the float compression. More...
UnsignedIntegerProperty	m_compressionBitsLow {this, "CompressionBitsLow", 15, "Lossy float compression bits (low)"}
	Number of mantissa bits in the float compression. More...
StringArrayProperty	m_transientItems {this, "TransientItems", {}, "Transient item list"}
	List of items that are known to be present in the transient store (and hence we can make input dependencies on them). More...
StringProperty	m_outputName {this, "OutputFile", "DidNotNameOutput.root", "Name of the output file"}
	Name of the output file. More...
StringProperty	m_persName {this, "EvtConversionSvc", "EventPersistencySvc", "Name of the persistency service writing data"}
	Name of the persistency service capable to write data from the store. More...
BooleanProperty	m_forceRead {this, "ForceRead", true, "Force read data objects in ItemList"}
	set to true to force read of data objects in item list More...
BooleanProperty	m_extendProvenanceRecord {this, "ExtendProvenanceRecord", true, "Extend provenance record"}
	Set to false to omit adding the current DataHeader into the DataHeader history This will cause the input file to be neglected for back navigation (replace mode). More...
BooleanProperty	m_itemListFromTool {this, "TakeItemsFromInput", false, "Write everything in input DataHeader to output"}
	Set to write out everything in input DataHeader. More...
ToolHandle< SG::IFolder >	m_p2BWritten
	The top-level folder with items to be written. More...
ToolHandle< SG::IFolder >	m_compressionDecoderHigh
	The top-level folder with items to be compressed high. More...
ToolHandle< SG::IFolder >	m_compressionDecoderLow
	The top-level folder with items to be compressed low. More...
ToolHandle< SG::IFolder >	m_transient
	Decoded list of transient ids. More...
std::multimap< CLID, std::string >	m_CLIDKeyPairs
	Map of (clid,key) pairs to be excluded (comes from m_excludeList) More...
IDataSelector	m_objects
	Collection of objects being selected. More...
IDataSelector	m_altObjects
	Objects overridden by ‘exact’ handling. More...
std::vector< std::unique_ptr< DataObject > >	m_ownedObjects
	Collection of DataObject instances owned by this service. More...
ToolHandle< IAthenaOutputStreamTool >	m_streamer
	pointer to AthenaOutputStreamTool More...
ToolHandleArray< IAthenaOutputTool >	m_helperTools {this, "HelperTools", {}, "List of AlgTools used by this stream"}
	vector of AlgTools that that are executed by this stream More...
bool	m_writeMetadataAndDisconnect {false}
std::atomic< int >	m_events {0}
	Number of events written to this output stream. More...
std::map< unsigned, std::string >	m_slotRangeMap
	map of filenames assigned to active slots More...
std::map< std::string, std::string >	m_rangeIDforRangeFN
	map of RangeIDs (as used by the Sequencer) for each Range filename generated More...
std::map< std::string, std::unique_ptr< IAthenaOutputStreamTool > >	m_streamerMap
	map of streamerTools handling event ranges in MT More...
mutex_t	m_mutex
	mutex for this Stream write() and handle() methods More...
SG::WriteHandleKey< SG::SelectionVetoes >	m_selVetoesKey { this, "SelVetoesKey", "" }
	Key used for recording selected dynamic variable information to the event store. More...
SG::WriteHandleKey< SG::CompressionInfo >	m_compInfoKey { this, "CompInfoKey", "" }
	Key used for recording lossy float compressed variable information to the event store. More...
std::string	m_outputAttributes
	Output attributes. More...
virtual StatusCode	initialize () override
virtual StatusCode	finalize () override
virtual StatusCode	execute () override
virtual StatusCode	write ()
	Stream the data. More...
void	clearSelection ()
	Clear list of selected objects. More...
StatusCode	collectAllObjects ()
	Collect data objects for output streamer list. More...
IDataSelector *	selectedObjects ()
	Return the list of selected objects. More...
virtual void	handle (const Incident &incident) override
	Incident service handle listening for MetaDataStop. More...
virtual StatusCode	io_reinit () override
	Callback method to reinitialize the internal state of the component for I/O purposes (e.g. upon fork(2)) More...
virtual StatusCode	io_finalize () override
void	itemListHandler (Gaudi::Details::PropertyBase &)
	Handler for ItemNames Property. More...
void	excludeListHandler (Gaudi::Details::PropertyBase &)
	Handler for ItemNames Property. More...
void	compressionListHandlerHigh (Gaudi::Details::PropertyBase &)
	Handler for ItemNames Property. More...
void	compressionListHandlerLow (Gaudi::Details::PropertyBase &)
	Handler for ItemNames Property. More...
StatusCode	addItemObjects (const SG::FolderItem &, SG::SelectionVetoes &vetoes, SG::CompressionInfo &compInfo)
	Add item data objects to output streamer list. More...
void	handleVariableSelection (const SG::IConstAuxStore &auxstore, SG::DataProxy &itemProxy, const std::string &tns, const std::string &aux_attr, SG::SelectionVetoes &vetoes) const
	Here we build the vetoed AuxIDs. More...
void	writeMetaData (const std::string &outputFN="")
	Write MetaData for this stream (by default) or for a substream outputFN (in ES mode) More...
std::set< std::string >	buildCompressionSet (const ToolHandle< SG::IFolder > &handle, const CLID &item_id, const std::string &item_key) const
	Helper function for building the compression lists. More...
void	finalizeRange (const std::string &rangeFN)
void	loadDict (CLID clid)
	Helper function to load dictionaries (both transient and persistent) for a given type. More...
bool	simpleMatch (const std::string &pattern, const std::string &text)
	Glob-style matcher, where the only meta-character is '*'. More...

Detailed Description

algorithm that marks for write data objects in SG

Author

srini.nosp@m.r@bn.nosp@m.l.gov

Definition at line 54 of file AthenaOutputStream.h.

Member Typedef Documentation

Items

typedef std::vector<SG::DataProxy*> AthenaOutputStream::Items

typedefs

Definition at line 59 of file AthenaOutputStream.h.

mutex_t

typedef std::recursive_mutex AthenaOutputStream::mutex_t

Definition at line 61 of file AthenaOutputStream.h.

TypeKeyPairs

typedef std::vector<std::pair<std::string, std::string> > AthenaOutputStream::TypeKeyPairs

Definition at line 60 of file AthenaOutputStream.h.

Constructor & Destructor Documentation

AthenaOutputStream()

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

Standard algorithm Constructor.

Definition at line 43 of file AthenaOutputStream.cxx.

      : base_class(name, pSvcLocator),
        m_currentStore(&m_dataStore),
        m_p2BWritten(std::format("SG::Folder/{}_TopFolder", name), this),
        m_compressionDecoderHigh(std::format("SG::Folder/{}_compressed_high", name), this),
        m_compressionDecoderLow(std::format("SG::Folder/{}_compressed_low", name), this),
        m_transient(std::format("SG::Folder/{}_transient", name), this),
        m_streamer(std::format("AthenaOutputStreamTool/{}Tool", name), this)
{
   // Ensure the service locater is good
   assert(pSvcLocator);
 
   // This property depends on the name that's known at construction time
   // Therefore, do it the old fashioned way
   declareProperty("WritingTool", m_streamer);
 
   // Associate action handlers with the AcceptAlgs,
   // RequireAlgs & VetoAlgs properties
   m_itemList.declareUpdateHandler(&AthenaOutputStream::itemListHandler, this);
   m_compressionListHigh.declareUpdateHandler(&AthenaOutputStream::compressionListHandlerHigh, this);
   m_compressionListLow.declareUpdateHandler(&AthenaOutputStream::compressionListHandlerLow, this);
}

~AthenaOutputStream()

AthenaOutputStream::~AthenaOutputStream ( )

virtual

Standard Destructor.

Definition at line 67 of file AthenaOutputStream.cxx.

                                        {
   // Clear the internal caches
   m_streamerMap.clear();
}

Member Function Documentation

addItemObjects()

StatusCode AthenaOutputStream::addItemObjects ( const SG::FolderItem &  item, SG::SelectionVetoes &  vetoes, SG::CompressionInfo &  compInfo  )

private

Add item data objects to output streamer list.

Handle variable selections. Both variable selection and lossy float compression are limited to event data for the time being

Definition at line 564 of file AthenaOutputStream.cxx.

{
   // anything after a dot is a list of dynamic Aux attributes, separated by dots
   size_t dotpos = item.key().find('.');
   std::string item_key, aux_attr;
   if( dotpos != std::string::npos ) {
      item_key = item.key().substr(0, dotpos+1);
      aux_attr = item.key().substr(dotpos+1);
   } else {
      item_key = item.key();
   }
   CLID item_id = item.id();
   ATH_MSG_DEBUG(std::format("addItemObjects({},\"{}\") called", item_id, item_key));
   ATH_MSG_DEBUG(std::format("           Key:{}", item_key));
   if( aux_attr.size() ) {
      ATH_MSG_DEBUG(std::format("      Aux Attr:{}", aux_attr));
   }
 
   // Here we build the list of attributes for the lossy float compression
   // Note that we do not allow m_compressionBitsHigh >= m_compressionBitsLow
   // Otherwise is, in any case, a logical error and they'd potentially overwrite each other
   std::map< unsigned int, std::set< std::string > > comp_attr_map;
   comp_attr_map[ m_compressionBitsHigh ] = buildCompressionSet( m_compressionDecoderHigh, item_id, item_key );
   comp_attr_map[ m_compressionBitsLow  ] = buildCompressionSet( m_compressionDecoderLow, item_id, item_key );
 
   // Print some debugging information regarding the lossy float compression configuration
   for( const auto& it : comp_attr_map ) {
     ATH_MSG_DEBUG(std::format("     Comp Attr {} with {} mantissa bits.", it.second.size(), it.first));
     if ( it.second.size() > 0 ) {
       for( const auto& attr : it.second ) {
          ATH_MSG_DEBUG(std::format("       >> {}", attr));
       }
     }
   }
 
   // For MetaData objects of type T that are kept in MetaContainers get the MetaCont<T> ID
   const CLID remapped_item_id = m_metaDataSvc->remapMetaContCLID( item_id );
   SG::ConstProxyIterator iter, end;
   SG::ProxyMap map;
   bool gotProxies = false;
   // Look for the clid in storegate
   SG::DataProxy* match = (*m_currentStore)->proxy(remapped_item_id, item_key, true);
   if (match != nullptr) {
      map.insert({item_key, match});
      iter = map.begin();
      end = map.end();
      gotProxies = true;
   }
   // Look for the clid in storegate
   if (!gotProxies && ((*m_currentStore)->proxyRange(remapped_item_id, iter, end)).isSuccess()) {
      gotProxies = true;
   }
   if (gotProxies) {
      bool added = false, removed = false;
      // Now loop over any found proxies
      for (; iter != end; ++iter) {
         SG::DataProxy* itemProxy(iter->second);
         std::string proxyName = itemProxy->name();
         std::string stream;
         if( m_currentStore == &m_metadataStore ) {
            // only check metadata keys
            stream = m_metaDataSvc->removeStreamFromKey(proxyName);  // can modify proxyName
         }
         // Does this key match the proxy key name - allow for wildcarding and aliases
         bool keyMatch = ( item_key == "*" ||
                           item_key == proxyName ||
                           itemProxy->hasAlias(item_key) );
         if (!keyMatch) {
            // For item list we currently allow wildcards ('*'), which has limited use, e.g.:
            // xAOD::CutBookkeeperAuxContainer#IncompleteCutBookkeepers*Aux.
            // Here we look for those few cases...
            keyMatch = simpleMatch(item_key, proxyName);
            ATH_MSG_DEBUG(std::format("Result of checking {} against {} to see if it matches is {}",
                                      proxyName, item_key, keyMatch));
         }
 
         // Now check if this item is marked for another output stream, if so we reject it
         // We also reject keys that are marked transient at this point
         bool xkeyMatch = false;
         if( (!stream.empty() and stream != m_outputName) || SG::isTransientKey(proxyName) ) {
            // reject keys that are marked for a different output stream
            ATH_MSG_DEBUG(std::format("Rejecting key: {} in output: {}", itemProxy->name(), m_outputName.toString()));
            xkeyMatch = true;
         }
 
         // All right, it passes key match find in itemList, but not in excludeList
         if (keyMatch && !xkeyMatch) {
            if (m_forceRead && itemProxy->isValid()) {
               if (nullptr == itemProxy->accessData()) {
                  ATH_MSG_ERROR(std::format(" Could not get data object for id {},\"{}\"", remapped_item_id, proxyName));
               }
            }
            if (nullptr != itemProxy->object()) {
               if( std::find(m_objects.begin(), m_objects.end(), itemProxy->object()) == m_objects.end() &&
                   std::find(m_altObjects.begin(), m_altObjects.end(), itemProxy->object()) == m_altObjects.end() )
               {
                  if( item_id != remapped_item_id ) {
                     // For MetaCont<T>: -
                     // create a temporary DataObject for an entry in the  container to pass to CnvSvc
                     DataBucketBase* dbb = static_cast<DataBucketBase*>( itemProxy->object() );
                     const MetaContBase* metaCont = static_cast<MetaContBase*>( dbb->cast( ClassID_traits<MetaContBase>::ID() ) );
                     void* obj = metaCont? metaCont->getAsVoid( m_outSeqSvc->currentRangeID() ) : nullptr;
                     if( obj ) {
                        auto altbucket = std::make_unique<AltDataBucket>(
                           obj, item_id, *CLIDRegistry::CLIDToTypeinfo(item_id), proxyName );
                        m_objects.push_back( altbucket.get() );
                        m_ownedObjects.push_back( std::move(altbucket) );
                        m_altObjects.push_back( itemProxy->object() ); // only for duplicate prevention
                     } else {
                        ATH_MSG_ERROR(std::format("Failed to retrieve object from MetaCont with key={}, for EventRangeID={}",
                                      item_key, m_outSeqSvc->currentRangeID()));
                        return StatusCode::FAILURE;
                     }
                  } else if (item.exact()) {
                   // If the exact flag is set, make a new DataObject
                   // holding the object as the requested type.
                   DataBucketBase* dbb = dynamic_cast<DataBucketBase*> (itemProxy->object());
                   if (!dbb) std::abort();
                   void* ptr = dbb->cast (item_id);
                   if (!ptr) {
                     // Hard cast
                     ptr = dbb->object();
                   }
                   auto altbucket =
                     std::make_unique<AltDataBucket>
                       (ptr, item_id,
                        *CLIDRegistry::CLIDToTypeinfo (item_id),
                        *itemProxy);
                   m_objects.push_back(altbucket.get());
                   m_ownedObjects.push_back (std::move(altbucket));
                   m_altObjects.push_back (itemProxy->object());
                 }
                 else
                   m_objects.push_back(itemProxy->object());
                 ATH_MSG_DEBUG(std::format(" Added object {},\"{}\"", item_id, proxyName));
               }
 
               // Build ItemListSvc string
               std::string tn;
               std::stringstream tns;
               if (!m_pCLIDSvc->getTypeNameOfID(item_id, tn).isSuccess()) {
                  ATH_MSG_ERROR(std::format(" Could not get type name for id {},\"{}\"", item_id, proxyName));
                  tns << item_id << '_' << proxyName;
               } else {
                  tn += '_' + proxyName;
                  tns << tn;
               }
 
               if ((*m_currentStore)->storeID() == StoreID::EVENT_STORE &&
                   item_key.find( RootAuxDynIO::AUX_POSTFIX ) == ( item_key.size() - 4 )) {
 
                  const SG::IConstAuxStore* auxstore( nullptr );
                  try {
                    SG::fromStorable( itemProxy->object(), auxstore, true );
                  } catch( const std::exception& ) {
                    ATH_MSG_DEBUG(std::format("Error in casting object with CLID {} to SG::IConstAuxStore*", itemProxy->clID()));
                    auxstore = nullptr;
                  }
 
                  if (auxstore) {
                    handleVariableSelection (*auxstore, *itemProxy,
                                             tns.str(), aux_attr,
                                             vetoes);
 
                    // Here comes the compression logic using ThinningInfo
                    // Get a hold of all AuxIDs for this store (static, dynamic etc.)
                    const SG::auxid_set_t allVars = auxstore->getAuxIDs();
 
                    // Get a handle on the compression information for this store
                    std::string key = item_key;
                    key.erase (key.size()-4, 4);
 
                    // Build the compression list, retrieve the relevant AuxIDs and
                    // store it in the relevant map that is going to be inserted into
                    // the ThinningCache later on by the ThinningCacheTool
                    xAOD::AuxCompression compression;
                    compression.setCompressedAuxIDs( comp_attr_map );
                    for( const auto& it : compression.getCompressedAuxIDs( allVars ) ) {
                      if( it.second.size() > 0 ) { // insert only if the set is non-empty
                        compInfo[ key ][ it.first ] = it.second;
                        ATH_MSG_DEBUG(std::format("Container {} has {} variables that'll be "
                                      "lossy float compressed with {} mantissa bits",
                                      key, it.second.size(), it.first));
                      }
                    } // End of loop over variables to be lossy float compressed
                  } // End of lossy float compression logic
 
               }
 
               added = true;
               if (m_itemSvc->addStreamItem(this->name(),tns.str()).isFailure()) {
                  ATH_MSG_WARNING(std::format("Unable to record item {} in Svc", tns.str()));
               }
            }
         } else if (keyMatch && xkeyMatch) {
            removed = true;
         }
      } // proxy loop
      if (!added && !removed) {
         ATH_MSG_DEBUG(std::format(" No object matching {},\"{}\" found", item_id, item_key));
      } else if (removed) {
         ATH_MSG_DEBUG(std::format(" Object being excluded based on property setting {},\"{}\". Skipping",
                                   item_id, item_key));
      }
   } else {
      ATH_MSG_DEBUG(std::format(" Failed to receive proxy iterators from StoreGate for {},\"{}\". Skipping",
                                item_id, item_key));
   }
   return StatusCode::SUCCESS;
}

buildCompressionSet()

std::set< std::string > AthenaOutputStream::buildCompressionSet ( const ToolHandle< SG::IFolder > &  handle, const CLID &  item_id, const std::string &  item_key  ) const

private

Helper function for building the compression lists.

Here we build the list of attributes for the float compression CompressionList follows the same logic as the ItemList We find the matching keys, read the string after "Aux.", tokenize by "." and build an std::set of these to be communicated to ThinningInfo elsewhere in the code.

Definition at line 786 of file AthenaOutputStream.cxx.

{
  // Create an empty result
  std::set<std::string> result;
 
  // Check the item is indeed Aux.
  if(item_key.find("Aux.") == std::string::npos) {
    return result;
  }
 
  // First the high compression list
  for (const auto& iter : *handle) {
    // First match the IDs for early rejection.
    if (iter.id() != item_id) {
      continue;
    }
    // Then find the compression item key and the compression list string
    size_t seppos = iter.key().find('.');
    std::string comp_item_key{""}, comp_str{""};
    if(seppos != std::string::npos) {
      comp_item_key = iter.key().substr(0, seppos+1);
      comp_str = iter.key().substr(seppos+1);
    } else {
      comp_item_key = iter.key();
    }
    // Proceed only if the keys match and the
    // compression list string is not empty
    if (!comp_str.empty() && comp_item_key == item_key) {
      std::stringstream ss(comp_str);
      std::string attr;
      while( std::getline(ss, attr, '.') ) {
         result.insert(attr);
      }
    }
  }
 
  // All done, return the result
  return result;
}

clearSelection()

void AthenaOutputStream::clearSelection

(

)

Clear list of selected objects.

Definition at line 523 of file AthenaOutputStream.cxx.

                                            {
   m_objects.clear();
   m_ownedObjects.clear();
   m_altObjects.clear();
}

collectAllObjects()

StatusCode AthenaOutputStream::collectAllObjects

(

)

Collect data objects for output streamer list.

Definition at line 530 of file AthenaOutputStream.cxx.

                                                 {
   if (m_itemListFromTool) {
      if (!m_streamer->getInputItemList(&*m_p2BWritten).isSuccess()) {
         ATH_MSG_WARNING("collectAllObjects() could not get ItemList from Tool.");
      }
   }
 
   // This holds the vetoes for the AuxID selection
   auto vetoes = std::make_unique<SG::SelectionVetoes>();
   // This holds the lossy float compression information
   auto compInfo = std::make_unique<SG::CompressionInfo>();
 
   m_p2BWritten->updateItemList(true);
   // Collect all objects that need to be persistified:
   for (const auto& i : *m_p2BWritten) {
      ATH_CHECK( addItemObjects(i, *vetoes, *compInfo) );
   }
 
   // If there were any variable selections, record the information in SG.
   if (!vetoes->empty()) {
     ATH_CHECK( SG::makeHandle (m_selVetoesKey).record (std::move (vetoes)) );
   }
 
   // Store the lossy float compression information in the SG.
   if (!compInfo->empty()) {
     ATH_CHECK( SG::makeHandle (m_compInfoKey).record (std::move (compInfo)) );
   }
 
   return StatusCode::SUCCESS;
}

compressionListHandlerHigh()

void AthenaOutputStream::compressionListHandlerHigh ( Gaudi::Details::PropertyBase &  )

protected

Handler for ItemNames Property.

Definition at line 891 of file AthenaOutputStream.cxx.

                                                                               {
   IProperty *pAsIProp(nullptr);
   if ((m_compressionDecoderHigh.retrieve()).isFailure() ||
           nullptr == (pAsIProp = dynamic_cast<IProperty*>(&*m_compressionDecoderHigh)) ||
           (pAsIProp->setProperty("ItemList", m_compressionListHigh.toString())).isFailure()) {
      throw GaudiException("Folder property [ItemList] not found", name(), StatusCode::FAILURE);
   }
}

compressionListHandlerLow()

void AthenaOutputStream::compressionListHandlerLow ( Gaudi::Details::PropertyBase &  )

protected

Handler for ItemNames Property.

Definition at line 900 of file AthenaOutputStream.cxx.

                                                                              {
   IProperty *pAsIProp(nullptr);
   if ((m_compressionDecoderLow.retrieve()).isFailure() ||
           nullptr == (pAsIProp = dynamic_cast<IProperty*>(&*m_compressionDecoderLow)) ||
           (pAsIProp->setProperty("ItemList", m_compressionListLow.toString())).isFailure()) {
      throw GaudiException("Folder property [ItemList] not found", name(), StatusCode::FAILURE);
   }
}

excludeListHandler()

void AthenaOutputStream::excludeListHandler ( Gaudi::Details::PropertyBase &  )

protected

Handler for ItemNames Property.

execute()

StatusCode AthenaOutputStream::execute ( )

overridevirtual

Reimplemented in AthenaConditionStream.

Definition at line 407 of file AthenaOutputStream.cxx.

                                       {
   bool failed = false;
   // Call tool preExecute prior to writing
   for (auto& tool : m_helperTools) {
      if (!tool->preExecute().isSuccess()) {
         failed = true;
      }
   }
   // Write the event if the event is accepted
   if (isEventAccepted()) {
      if (write().isFailure()) {
         failed = true;
      }
   }
   // Call tool postExecute after writing
   for (auto& tool : m_helperTools) {
      if(!tool->postExecute().isSuccess()) {
         failed = true;
      }
   }
   // See if we should write metadata and do if so
   if( m_writeMetadataAndDisconnect ) {
      writeMetaData();
      m_writeMetadataAndDisconnect = false;
      // finalize will disconnect output
      if( !finalize().isSuccess() ) {
         failed = true;
      }
   }
   if (failed) {
      return(StatusCode::FAILURE);
   }
   return(StatusCode::SUCCESS);
}

finalize()

StatusCode AthenaOutputStream::finalize ( )

overridevirtual

Reimplemented in AthenaConditionStream.

Definition at line 379 of file AthenaOutputStream.cxx.

{
   bool failed = false;
   ATH_MSG_DEBUG("finalize: Optimize output");
   // Connect the output file to the service
   if (!m_streamer->finalizeOutput().isSuccess()) {
      failed = true;
   }
   ATH_MSG_DEBUG("finalize: end optimize output");
   // Release the tools
   if (!m_helperTools.release().isSuccess()) {
      failed = true;
   }
   if (!m_streamer.release().isSuccess()) {
      failed = true;
   }
   if (failed) {
      return(StatusCode::FAILURE);
   }
   // Clear the internal caches
   m_objects.clear();
   m_objects.shrink_to_fit();
   m_ownedObjects.clear();
   m_altObjects.clear();
   return(StatusCode::SUCCESS);
}

finalizeRange()

void AthenaOutputStream::finalizeRange ( const std::string &  rangeFN )

private

Definition at line 299 of file AthenaOutputStream.cxx.

{
   ATH_MSG_DEBUG(std::format("Writing MetaData to {}", rangeFN));
   // MN: not calling StopMetaData Incident here but directly writeMetaData() - OK for Sim, check others
   // metadata tools like CutFlowSvc are not able to handle this yet
   const std::string rememberID = m_outSeqSvc->setRangeID( m_rangeIDforRangeFN[ rangeFN ] );
   writeMetaData( rangeFN );
   m_outSeqSvc->setRangeID( rememberID );
 
   ATH_MSG_INFO(std::format("Finished writing Event Sequence to {}", rangeFN));
   auto strm_iter = m_streamerMap.find( rangeFN );
   strm_iter->second->finalizeOutput().ignore();
   strm_iter->second->finalize().ignore();
   m_streamerMap.erase( strm_iter );
   m_outSeqSvc->publishRangeReport( rangeFN );
}

handle()

void AthenaOutputStream::handle ( const Incident &  incident )

overridevirtual

Incident service handle listening for MetaDataStop.

Definition at line 224 of file AthenaOutputStream.cxx.

{
   ATH_MSG_DEBUG(std::format("handle() incident type: {}", inc.type()));
   // mutex shared with write() which is called from writeMetaData
   std::unique_lock<mutex_t>  lock(m_mutex);
 
   if( inc.type() == "MetaDataStop" )  {
      if( m_outSeqSvc->inUse() ) {
         if( m_outSeqSvc->inConcurrentEventsMode() ) {
            // EventService MT - write metadata and close all remaining substreams
            while( m_streamerMap.size() > 0 ) {
               finalizeRange( m_streamerMap.begin()->first );
            }
            return;
         }
         if( m_outSeqSvc->lastIncident() == "EndEvent" ) {
            // in r22 EndEvent comes before output writing
            // - queue metadata writing and disconnect for after Event write
            m_writeMetadataAndDisconnect = true;
            return;
         }
      }
      // not in Event Service
      writeMetaData();
   }
   else if( m_outSeqSvc->inUse() ) {
      // Handle Event Ranges for Event Service
      EventContext::ContextID_t slot = inc.context().slot();
      if( slot == EventContext::INVALID_CONTEXT_ID ) {
         throw GaudiException("Received Incident with invalid slot in ES mode", name(), StatusCode::FAILURE);
      }
      auto count_events_in_range = [&](const std::string& range) {
         return std::count_if(m_slotRangeMap.cbegin(), m_slotRangeMap.cend(),
                              [&](auto& el){return el.second == range;} );
      };
      if( inc.type() == IncidentType::BeginProcessing ) {
         // get the current/old range filename for this slot
         const std::string rangeFN = m_slotRangeMap[ slot ];
         // build the new range filename for this slot
         const std::string newRangeFN = m_outSeqSvc->buildSequenceFileName( m_outputName );
         if( !rangeFN.empty() and rangeFN != newRangeFN ) {
            ATH_MSG_INFO(std::format("Slot range change: '{}' -> '{}'", rangeFN, newRangeFN));
            ATH_MSG_DEBUG(std::format("There are {} slots in use",m_slotRangeMap.size()));
            for(const auto & range : m_slotRangeMap ) {
               ATH_MSG_DEBUG(std::format("Slot: {}  FN={}", range.first, range.second));
            }
            if( count_events_in_range(rangeFN) == 1 ) {
               finalizeRange( rangeFN );
            }
         }
         ATH_MSG_INFO(std::format("slot {} processing event in range: {}", slot, newRangeFN));
         m_slotRangeMap[ slot ] = newRangeFN;
         // remember the RangeID for this slot so we can write metadata *after* a range change
         m_rangeIDforRangeFN[ newRangeFN ] = m_outSeqSvc->currentRangeID();
      }
      else if( inc.type() == IncidentType::EndProcessing ) {
         ATH_MSG_DEBUG(std::format("There are {} slots in use", m_slotRangeMap.size()));
         for( const auto& range : m_slotRangeMap ) {
            ATH_MSG_DEBUG(std::format("Slot: {}  FN={}", range.first, range.second));
         }
         if( m_slotRangeMap.size() > 1 ) {
            // if there are multiple slots, we can detect if the range ended with this event
            // - except the last range, because there is no next range to clear the slot map
            const std::string rangeFN = m_slotRangeMap[ slot ];
            if( count_events_in_range(rangeFN) == 1 ) {
               finalizeRange( rangeFN );
               m_slotRangeMap[ slot ].clear();
            }
         }
      }
   }
   ATH_MSG_DEBUG(std::format("Leaving incident handler for {}", inc.type()));
}

handleVariableSelection()

void AthenaOutputStream::handleVariableSelection ( const SG::IConstAuxStore &  auxstore, SG::DataProxy &  itemProxy, const std::string &  tns, const std::string &  aux_attr, SG::SelectionVetoes &  vetoes  ) const

private

Here we build the vetoed AuxIDs.

Definition at line 829 of file AthenaOutputStream.cxx.

{
  // Collect dynamic Aux selection (parse the line, attributes separated by dot)
  std::set<std::string> attributes;
  if( aux_attr.size() ) {
    std::stringstream ss(aux_attr);
    std::string attr;
    while( std::getline(ss, attr, '.') ) {
      attributes.insert(attr);
      std::stringstream temp;
      temp << tns << attr;
      if (m_itemSvc->addStreamItem(this->name(),temp.str()).isFailure()) {
        ATH_MSG_WARNING(std::format("Unable to record item {} in Svc", temp.str()));
      }
    }
  }
 
  // Return early if there's no selection.
  if (attributes.empty()) {
    return;
  }
 
  std::string key = itemProxy.name();
  if (key.size() >= 4 && key.compare (key.size()-4, 4, "Aux.")==0)
  {
    key.erase (key.size()-4, 4);
  }
 
  // Find the entry for the selection.
  SG::auxid_set_t& vset = vetoes[key];
 
  // Form the veto mask for this object.
  xAOD::AuxSelection sel;
  sel.selectAux (attributes);
 
  // Get all the AuxIDs that we know of and the selected ones
  SG::auxid_set_t all = auxstore.getAuxIDs();
  SG::auxid_set_t selected = sel.getSelectedAuxIDs( all );
 
  // Loop over all and build a list of vetoed AuxIDs from non selected ones
  for( const SG::auxid_t auxid : all ) {
    if ( !selected.test( auxid ) ) {
      vset.insert( auxid );
    }
  }
}

initialize()

StatusCode AthenaOutputStream::initialize ( )

overridevirtual

Definition at line 73 of file AthenaOutputStream.cxx.

                                          {
   ATH_MSG_DEBUG("In initialize");
 
   // Initialize the FilteredAlgorithm base
   ATH_CHECK( this->FilteredAlgorithm::initialize() );
 
   // Reset the number of events written
   m_events = 0;
 
   // Set up the SG services
   ATH_CHECK( m_dataStore.retrieve() );
   ATH_MSG_DEBUG(std::format("Found {} store.", m_dataStore.typeAndName()));
   if (!m_metadataItemList.value().empty()) {
      ATH_CHECK( m_metadataStore.retrieve() );
      ATH_MSG_DEBUG(std::format("Found {} store.", m_metadataStore.typeAndName()));
   }
 
   // Set up various services
   ATH_CHECK( m_pCLIDSvc.retrieve() );
   ATH_CHECK( m_dictLoader.retrieve() );
   ATH_CHECK( m_tpCnvSvc.retrieve() );
   ATH_CHECK( m_itemSvc.retrieve() );
   ATH_CHECK( m_outSeqSvc.retrieve() );
 
   // Get Output Stream tool for writing
   ATH_CHECK( m_streamer.retrieve() );
   ATH_CHECK( m_streamer->connectServices(m_dataStore.typeAndName(), m_persName, m_extendProvenanceRecord) );
 
   ATH_CHECK( m_helperTools.retrieve() );
   ATH_MSG_INFO("Found " << m_helperTools);
   ATH_MSG_INFO(std::format("Data output: {}", m_outputName.toString()));
 
   for (auto& tool : m_helperTools) {
     ATH_CHECK( tool->postInitialize() );
   }
 
   // Register this algorithm for 'I/O' events
   ServiceHandle<IIoComponentMgr> iomgr("IoComponentMgr", name());
   ATH_CHECK( iomgr.retrieve() );
   ATH_CHECK( iomgr->io_register(this) );
   ATH_CHECK( iomgr->io_register(this, IIoComponentMgr::IoMode::WRITE, m_outputName) );
   ATH_CHECK( this->io_reinit() );
 
   // Add an explicit input dependency for everything in our item list
   // that we know from the configuration is in the transient store.
   // We don't want to add everything on the list, because configurations
   // often initialize this with a maximal static list of everything
   // that could possibly be written.
   {
     ATH_CHECK( m_transient.retrieve() );
     IProperty *pAsIProp = dynamic_cast<IProperty*> (&*m_transient);
     if (!pAsIProp) {
       ATH_MSG_FATAL ("Bad folder interface");
       return StatusCode::FAILURE;
     }
     ATH_CHECK (pAsIProp->setProperty("ItemList", m_transientItems.toString()));
 
     for (const SG::FolderItem& item : *m_p2BWritten) {
       // Load ROOT dictionaries now.
       loadDict (item.id());
 
       const std::string& k = item.key();
       if (k.find('*') != std::string::npos) continue;
       if (k.find('.') != std::string::npos) continue;
       for (const SG::FolderItem& titem : *m_transient) {
         if (titem.id() == item.id() && titem.key() == k) {
           DataObjID id (item.id(), std::format("{}+{}", m_dataStore.name(), k));
           this->addDependency (id, Gaudi::DataHandle::Reader);
           break;
         }
       }
     }
     m_transient->clear();
   }
 
   // Also load dictionaries for metadata classes.
   if (!m_metadataItemList.value().empty()) {
     IProperty *pAsIProp = dynamic_cast<IProperty*> (&*m_transient);
     if (!pAsIProp) {
       ATH_MSG_FATAL ("Bad folder interface");
       return StatusCode::FAILURE;
     }
     ATH_CHECK (pAsIProp->setProperty("ItemList", m_metadataItemList.toString()));
     for (const SG::FolderItem& item : *m_transient) {
       loadDict (item.id());
     }
     m_transient->clear();
   }
 
   // Also make sure we have the dictionary for Token.
   m_dictLoader->load_type ("Token");
 
   // Listen to event range incidents if incident name is configured
   ATH_CHECK( m_incidentSvc.retrieve() );
   if( !m_outSeqSvc->incidentName().empty() ) {
      // use priority 95 to make sure the Output Sequencer goes first (it has priority 100)
      m_incidentSvc->addListener(this, IncidentType::BeginProcessing, 95);
      m_incidentSvc->addListener(this, IncidentType::EndProcessing, 95);
   }
 
   // Check compression settings and print some information about the configuration
   // Both should be between [5, 23] and high compression should be < low compression
   if(m_compressionBitsHigh < 5 || m_compressionBitsHigh > 23) {
     ATH_MSG_INFO(std::format("Float compression mantissa bits for high compression "
                              "({}) is outside the allowed range of [5, 23].",
                              m_compressionBitsHigh.toString()));
     ATH_MSG_INFO("Setting it to the appropriate limit.");
     m_compressionBitsHigh = m_compressionBitsHigh < 5 ? 5 : 23;
   }
   if(m_compressionBitsLow < 5 || m_compressionBitsLow > 23) {
     ATH_MSG_INFO(std::format("Float compression mantissa bits for low compression "
                              "({}) is outside the allowed range of [5, 23].",
                              m_compressionBitsLow.toString()));
     ATH_MSG_INFO("Setting it to the appropriate limit.");
     m_compressionBitsLow = m_compressionBitsLow < 5 ? 5 : 23;
   }
   if(m_compressionBitsLow <= m_compressionBitsHigh) {
     ATH_MSG_ERROR(std::format("Float compression mantissa bits for low compression "
                               "({}) is lower than or equal to high compression "
                               "({})! Please check the configuration! ",
                               m_compressionBitsLow.toString(),
                               m_compressionBitsHigh.toString()));
     return StatusCode::FAILURE;
   }
   if(m_compressionListHigh.value().empty() && m_compressionListLow.value().empty()) {
     ATH_MSG_VERBOSE("Both high and low float compression lists are empty. Float compression will NOT be applied.");
   } else {
     ATH_MSG_INFO("Either high or low (or both) float compression lists are defined. Float compression will be applied.");
     ATH_MSG_INFO(std::format("High compression will use {} mantissa bits, and "
                              "low compression will use {} mantissa bits.",
                              m_compressionBitsHigh.toString(),
                              m_compressionBitsLow.toString()));
   }
 
   // Setup stream name
   if (m_streamName.empty()) {
      m_streamName.setValue(this->name());
   }
 
   // Set SG key for selected variable information.
   m_selVetoesKey = std::format("SelectionVetoes_{}", m_streamName.toString());
   ATH_CHECK( m_selVetoesKey.initialize() );
 
   m_compInfoKey = std::format("CompressionInfo_{}", m_streamName.toString());
   ATH_CHECK( m_compInfoKey.initialize() );
 
   ATH_MSG_DEBUG("End initialize");
   return StatusCode::SUCCESS;
}

io_finalize()

StatusCode AthenaOutputStream::io_finalize ( )

overridevirtual

Definition at line 925 of file AthenaOutputStream.cxx.

                                           {
   ATH_MSG_INFO("I/O finalization...");
   for (auto& tool : m_helperTools) {
      if (!tool->preFinalize().isSuccess()) {
          ATH_MSG_ERROR("Cannot finalize helper tool");
      }
   }
   const Incident metaDataStopIncident(name(), "MetaDataStop");
   this->handle(metaDataStopIncident);
   m_incidentSvc->removeListener(this, "MetaDataStop");
   if (m_dataStore->clearStore().isFailure()) {
      ATH_MSG_WARNING("Cannot clear the DataStore");
   }
   return StatusCode::SUCCESS;
}

io_reinit()

StatusCode AthenaOutputStream::io_reinit ( )

overridevirtual

Callback method to reinitialize the internal state of the component for I/O purposes (e.g. upon fork(2))

Definition at line 911 of file AthenaOutputStream.cxx.

                                         {
   ATH_MSG_INFO("I/O reinitialization...");
   m_incidentSvc->removeListener(this, "MetaDataStop"); // Remove any existing listener to avoid handling the incident multiple times
   m_incidentSvc->addListener(this, "MetaDataStop", 50);
   for (auto& tool : m_helperTools) {
      if (!tool->postInitialize().isSuccess()) {
          ATH_MSG_ERROR("Cannot initialize helper tool");
      }
   }
   return StatusCode::SUCCESS;
}

itemListHandler()

void AthenaOutputStream::itemListHandler ( Gaudi::Details::PropertyBase &  )

protected

Handler for ItemNames Property.

Definition at line 881 of file AthenaOutputStream.cxx.

                                                                    {
   // Assuming concrete SG::Folder also has an itemList property
   IProperty *pAsIProp(nullptr);
   if ((m_p2BWritten.retrieve()).isFailure() ||
           nullptr == (pAsIProp = dynamic_cast<IProperty*>(&*m_p2BWritten)) ||
           (pAsIProp->setProperty(m_itemList)).isFailure()) {
      throw GaudiException("Folder property [itemList] not found", name(), StatusCode::FAILURE);
   }
}

loadDict()

void AthenaOutputStream::loadDict ( CLID  clid )

private

Helper function to load dictionaries (both transient and persistent) for a given type.

We want to to this explicitly during initialization to avoid sporadic failures seen loading dictionaries while multiple threads are running. See ATEAM-697 and ATEAM-749.

Definition at line 946 of file AthenaOutputStream.cxx.

{
  m_dictLoader->load_type (clid);
 
  // Also load the persistent class dictionary, if applicable.
  std::unique_ptr<ITPCnvBase> tpcnv = m_tpCnvSvc->t2p_cnv_unique (clid);
  if (tpcnv) {
    m_dictLoader->load_type (tpcnv->persistentTInfo());
  }
}

selectedObjects()

IDataSelector* AthenaOutputStream::selectedObjects ( )

inline

Return the list of selected objects.

Definition at line 86 of file AthenaOutputStream.h.

                                    {
      return &m_objects;
   }

simpleMatch()

bool AthenaOutputStream::simpleMatch ( const std::string &  pattern, const std::string &  text  )

private

Glob-style matcher, where the only meta-character is '*'.

Definition at line 958 of file AthenaOutputStream.cxx.

                                                            {
   size_t pi = 0, ti = 0, star = std::string::npos, match = 0;
   while (ti < text.size()) {
      if (pi < pattern.size() && (pattern[pi] == text[ti] || pattern[pi] == '*')) {
         if (pattern[pi] == '*') { star = pi++; match = ti; }
         else { ++pi; ++ti; }
      } else if (star != std::string::npos) {
         pi = star + 1;
         ti = ++match;
      } else return false;
   }
   while (pi < pattern.size() && pattern[pi] == '*') ++pi;
   return pi == pattern.size();
}

write()

StatusCode AthenaOutputStream::write ( )

virtual

Stream the data.

Definition at line 443 of file AthenaOutputStream.cxx.

                                     {
   bool failed = false;
   IAthenaOutputStreamTool* streamer = &*m_streamer;
   std::string outputFN;
 
   std::unique_lock<mutex_t>  lock(m_mutex);
   outputFN = m_outSeqSvc->buildSequenceFileName( m_outputName );
 
   // Handle Event Ranges
   if( m_outSeqSvc->inUse() and m_outSeqSvc->inConcurrentEventsMode() ) {
      ATH_MSG_DEBUG(std::format("Writing event sequence to {}", outputFN));
      streamer = m_streamerMap[ outputFN ].get();
      if( !streamer ) {
         // new range, needs a new streamer tool
         IAlgTool* st = AlgTool::Factory::create( m_streamer->type(), m_streamer->type(), m_streamer->name(), this ).release();
         st->addRef();
         streamer = dynamic_cast<IAthenaOutputStreamTool*>( st );
         IProperty *mstreamer_props = dynamic_cast<IProperty*> (&*m_streamer);
         IProperty *streamer_props = dynamic_cast<IProperty*> (&*streamer);
         for ( const auto& prop : mstreamer_props->getProperties() ) {
            ATH_CHECK( streamer_props->setProperty( *prop ) );
         }
         if( !streamer or streamer->initialize().isFailure()
             or streamer->connectServices(m_dataStore.typeAndName(), m_persName, m_extendProvenanceRecord).isFailure() ) {
            ATH_MSG_FATAL(std::format("Unable to initialize OutputStreamTool for {}", outputFN));
            return StatusCode::FAILURE;
         }
         m_streamerMap[ outputFN ].reset( streamer );
      }
   }
 
   // Clear any previously existing item list
   // and collect all objects that are asked to be written out
   clearSelection();
   ATH_CHECK( collectAllObjects() );
 
   // keep a local copy of the object lists so they are not overwritten when we release the lock
   IDataSelector objects = std::move( m_objects );
   IDataSelector altObjects = std::move( m_altObjects );
   std::vector<std::unique_ptr<DataObject> > ownedObjects = std::move( m_ownedObjects );
 
   // prepare before releasing lock because m_outputAttributes change in metadataStop
   const std::string connectStr = outputFN + m_outputAttributes;
 
   for (auto& tool : m_helperTools) {
     ATH_CHECK( tool->preStream() );
   }
 
   // MN: would be nice to release the Stream lock here
   // lock.unlock();
 
   // Connect the output file to the service
   if (!streamer->connectOutput(connectStr).isSuccess()) {
      ATH_MSG_FATAL("Could not connectOutput");
      return StatusCode::FAILURE;
   }
   ATH_MSG_DEBUG(std::format("connectOutput done for {}", outputFN));
   StatusCode currentStatus = streamer->streamObjects(objects, connectStr);
   // Do final check of streaming
   if (!currentStatus.isSuccess()) {
      if (!currentStatus.isRecoverable()) {
         ATH_MSG_FATAL("streamObjects failed.");
         failed = true;
      } else {
         ATH_MSG_DEBUG("streamObjects failed.");
      }
   }
   bool doCommit = false;
   if (!streamer->commitOutput(doCommit).isSuccess()) {
      ATH_MSG_FATAL("commitOutput failed.");
      failed = true;
   }
   if (failed) {
      return(StatusCode::FAILURE);
   }
   m_events++;
   return(StatusCode::SUCCESS);
}

writeMetaData()

void AthenaOutputStream::writeMetaData ( const std::string &  outputFN = "" )

private

Write MetaData for this stream (by default) or for a substream outputFN (in ES mode)

Definition at line 319 of file AthenaOutputStream.cxx.

{
   // use main stream tool by default, or per outputFile in ES mode
   IAthenaOutputStreamTool* streamer = outputFN.empty()? &*m_streamer : m_streamerMap[outputFN].get();
 
   for (auto& tool : m_helperTools) {
      if (!tool->preFinalize().isSuccess()) {
         throw GaudiException("Cannot finalize helper tool", name(), StatusCode::FAILURE);
      }
   }
   if( m_metaDataSvc->prepareOutput(outputFN).isFailure() ) {
      throw GaudiException("Failed on MetaDataSvc prepareOutput", name(), StatusCode::FAILURE);
   }
   // lock all metadata to prevent updates during writing
   MetaDataSvc::ToolLockGuard   tool_guard( *m_metaDataSvc );
 
   // Prepare the WriteDataHeaderForms incident
   std::string DHFWriteIncidentfileName = m_outSeqSvc->buildSequenceFileName(m_outputName);
   // remove technology from the name
   size_t pos = DHFWriteIncidentfileName.find(':');
   if( pos != std::string::npos ) DHFWriteIncidentfileName = DHFWriteIncidentfileName.substr(pos+1);
   FileIncident incident(name(), "WriteDataHeaderForms", DHFWriteIncidentfileName);
   m_incidentSvc->fireIncident(incident);
 
   ATH_MSG_DEBUG("metadataItemList: " << m_metadataItemList.value() );
   if (!m_metadataItemList.value().empty()) {
      m_currentStore = &m_metadataStore;
      StatusCode status = streamer->connectServices(m_metadataStore.typeAndName(), m_persName, false);
      if (status.isFailure()) {
         throw GaudiException("Unable to connect metadata services", name(), StatusCode::FAILURE);
      }
      m_outputAttributes = "[OutputCollection=MetaDataHdr][PoolContainerPrefix=MetaData][AttributeListKey=]";
      m_p2BWritten->clear();
      IProperty *pAsIProp(nullptr);
      if ((m_p2BWritten.retrieve()).isFailure() ||
          nullptr == (pAsIProp = dynamic_cast<IProperty*>(&*m_p2BWritten)) ||
          (pAsIProp->setProperty("ItemList", m_metadataItemList.toString())).isFailure()) {
         throw GaudiException("Folder property [metadataItemList] not found", name(), StatusCode::FAILURE);
      }
      if (write().isFailure()) {
         throw GaudiException("Cannot write metadata", name(), StatusCode::FAILURE);
      }
      FileIncident incident(name(), "WriteDataHeaderForms", DHFWriteIncidentfileName + m_outputAttributes);
      m_incidentSvc->fireIncident(incident);
 
      m_outputAttributes.clear();
      m_currentStore = &m_dataStore;
      status = streamer->connectServices(m_dataStore.typeAndName(), m_persName, m_extendProvenanceRecord);
      if (status.isFailure()) {
         throw GaudiException("Unable to re-connect services", name(), StatusCode::FAILURE);
      }
      m_p2BWritten->clear();
      if ((pAsIProp->setProperty(m_itemList)).isFailure()) {
         throw GaudiException("Folder property [itemList] not found", name(), StatusCode::FAILURE);
      }
      ATH_MSG_DEBUG(std::format("Metadata items written: {}", m_metadataItemList.value().size()));
   }
}

Member Data Documentation

m_altObjects

IDataSelector AthenaOutputStream::m_altObjects

protected

Objects overridden by ‘exact’ handling.

Definition at line 177 of file AthenaOutputStream.h.

m_CLIDKeyPairs

std::multimap<CLID,std::string> AthenaOutputStream::m_CLIDKeyPairs

protected

Map of (clid,key) pairs to be excluded (comes from m_excludeList)

Definition at line 171 of file AthenaOutputStream.h.

m_compInfoKey

SG::WriteHandleKey<SG::CompressionInfo> AthenaOutputStream::m_compInfoKey { this, "CompInfoKey", "" }

private

Key used for recording lossy float compressed variable information to the event store.

Definition at line 229 of file AthenaOutputStream.h.

m_compressionBitsHigh

UnsignedIntegerProperty AthenaOutputStream::m_compressionBitsHigh {this, "CompressionBitsHigh", 7, "Lossy float compression bits (high)"}

protected

Number of mantissa bits in the float compression.

Definition at line 133 of file AthenaOutputStream.h.

m_compressionBitsLow

UnsignedIntegerProperty AthenaOutputStream::m_compressionBitsLow {this, "CompressionBitsLow", 15, "Lossy float compression bits (low)"}

protected

Number of mantissa bits in the float compression.

Definition at line 136 of file AthenaOutputStream.h.

m_compressionDecoderHigh

ToolHandle<SG::IFolder> AthenaOutputStream::m_compressionDecoderHigh

protected

The top-level folder with items to be compressed high.

Definition at line 162 of file AthenaOutputStream.h.

m_compressionDecoderLow

ToolHandle<SG::IFolder> AthenaOutputStream::m_compressionDecoderLow

protected

The top-level folder with items to be compressed low.

Definition at line 165 of file AthenaOutputStream.h.

m_compressionListHigh

StringArrayProperty AthenaOutputStream::m_compressionListHigh {this, "CompressionListHigh", {}, "Lossy float compression list (high)"}

protected

Vector of item names.

Definition at line 127 of file AthenaOutputStream.h.

m_compressionListLow

StringArrayProperty AthenaOutputStream::m_compressionListLow {this, "CompressionListLow", {}, "Lossy float compression list (low)"}

protected

Vector of item names.

Definition at line 130 of file AthenaOutputStream.h.

m_currentStore

ServiceHandle<StoreGateSvc>* AthenaOutputStream::m_currentStore

protected

Definition at line 102 of file AthenaOutputStream.h.

m_dataStore

ServiceHandle<StoreGateSvc> AthenaOutputStream::m_dataStore {this, "Store", "StoreGateSvc/StoreGateSvc", "Handle to event store"}

protected

Handle to the StoreGateSvc store where the data we want to write out resides.

Definition at line 100 of file AthenaOutputStream.h.

m_dictLoader

ServiceHandle<IDictLoaderSvc> AthenaOutputStream::m_dictLoader {this, "AthDictLoaderSvc", "AthDictLoaderSvc"}

protected

Definition at line 107 of file AthenaOutputStream.h.

m_events

std::atomic<int> AthenaOutputStream::m_events {0}

protected

Number of events written to this output stream.

Definition at line 194 of file AthenaOutputStream.h.

m_extendProvenanceRecord

BooleanProperty AthenaOutputStream::m_extendProvenanceRecord {this, "ExtendProvenanceRecord", true, "Extend provenance record"}

protected

Set to false to omit adding the current DataHeader into the DataHeader history This will cause the input file to be neglected for back navigation (replace mode).

Definition at line 153 of file AthenaOutputStream.h.

m_forceRead

BooleanProperty AthenaOutputStream::m_forceRead {this, "ForceRead", true, "Force read data objects in ItemList"}

protected

set to true to force read of data objects in item list

Definition at line 149 of file AthenaOutputStream.h.

m_helperTools

ToolHandleArray<IAthenaOutputTool> AthenaOutputStream::m_helperTools {this, "HelperTools", {}, "List of AlgTools used by this stream"}

protected

vector of AlgTools that that are executed by this stream

Definition at line 188 of file AthenaOutputStream.h.

m_incidentSvc

ServiceHandle<IIncidentSvc> AthenaOutputStream::m_incidentSvc {this, "IncidentSvc", "IncidentSvc"}

protected

Definition at line 109 of file AthenaOutputStream.h.

m_itemList

StringArrayProperty AthenaOutputStream::m_itemList {this, "ItemList", {}, "List of items to write", "OutputStreamItemList"}

protected

Vector of item names.

Definition at line 117 of file AthenaOutputStream.h.

m_itemListFromTool

BooleanProperty AthenaOutputStream::m_itemListFromTool {this, "TakeItemsFromInput", false, "Write everything in input DataHeader to output"}

protected

Set to write out everything in input DataHeader.

Definition at line 156 of file AthenaOutputStream.h.

m_itemSvc

ServiceHandle<IItemListSvc> AthenaOutputStream::m_itemSvc {this, "ItemListSvc", "ItemListSvc"}

protected

Handles to all the necessary services.

Definition at line 105 of file AthenaOutputStream.h.

m_keepProvenances

StringProperty AthenaOutputStream::m_keepProvenances

protected

Initial value:

{this, "KeepProvenanceTagsRegEx", {".*"},
                              "RegEx pattern to select processing tags for which DataHeader should retain provenances"}

Provenance record selection.

Definition at line 123 of file AthenaOutputStream.h.

m_metadataItemList

StringArrayProperty AthenaOutputStream::m_metadataItemList {this, "MetadataItemList", {}, "List of metadata items to write","OutputStreamItemList"}

protected

Vector of item names.

Definition at line 120 of file AthenaOutputStream.h.

m_metadataStore

ServiceHandle<StoreGateSvc> AthenaOutputStream::m_metadataStore {this, "MetadataStore", "StoreGateSvc/MetaDataStore", "Handle to metadata store"}

protected

Definition at line 101 of file AthenaOutputStream.h.

m_metaDataSvc

ServiceHandle<MetaDataSvc> AthenaOutputStream::m_metaDataSvc {this, "MetaDataSvc", "MetaDataSvc"}

protected

Definition at line 106 of file AthenaOutputStream.h.

m_mutex

mutex_t AthenaOutputStream::m_mutex

protected

mutex for this Stream write() and handle() methods

Definition at line 207 of file AthenaOutputStream.h.

m_objects

IDataSelector AthenaOutputStream::m_objects

protected

Collection of objects being selected.

Definition at line 174 of file AthenaOutputStream.h.

m_outputAttributes

std::string AthenaOutputStream::m_outputAttributes

private

Output attributes.

Definition at line 233 of file AthenaOutputStream.h.

m_outputName

StringProperty AthenaOutputStream::m_outputName {this, "OutputFile", "DidNotNameOutput.root", "Name of the output file"}

protected

Name of the output file.

Definition at line 143 of file AthenaOutputStream.h.

m_outSeqSvc

ServiceHandle<OutputStreamSequencerSvc> AthenaOutputStream::m_outSeqSvc {this, "OutputStreamSequencerSvc", "OutputStreamSequencerSvc"}

protected

Definition at line 111 of file AthenaOutputStream.h.

m_ownedObjects

std::vector<std::unique_ptr<DataObject> > AthenaOutputStream::m_ownedObjects

protected

Collection of DataObject instances owned by this service.

FIXME: it would be simpler to just have m_objects be a vector of DataObjectSharedPtr<DataObject>, but that implies interface changes.

Definition at line 182 of file AthenaOutputStream.h.

m_p2BWritten

ToolHandle<SG::IFolder> AthenaOutputStream::m_p2BWritten

protected

The top-level folder with items to be written.

Definition at line 159 of file AthenaOutputStream.h.

m_pCLIDSvc

ServiceHandle<IClassIDSvc> AthenaOutputStream::m_pCLIDSvc {this, "ClassIDSvc", "ClassIDSvc"}

protected

Definition at line 110 of file AthenaOutputStream.h.

m_persName

StringProperty AthenaOutputStream::m_persName {this, "EvtConversionSvc", "EventPersistencySvc", "Name of the persistency service writing data"}

protected

Name of the persistency service capable to write data from the store.

Definition at line 146 of file AthenaOutputStream.h.

m_rangeIDforRangeFN

std::map< std::string, std::string > AthenaOutputStream::m_rangeIDforRangeFN

protected

map of RangeIDs (as used by the Sequencer) for each Range filename generated

Definition at line 201 of file AthenaOutputStream.h.

m_selVetoesKey

SG::WriteHandleKey<SG::SelectionVetoes> AthenaOutputStream::m_selVetoesKey { this, "SelVetoesKey", "" }

private

Key used for recording selected dynamic variable information to the event store.

Definition at line 224 of file AthenaOutputStream.h.

m_slotRangeMap

std::map< unsigned, std::string > AthenaOutputStream::m_slotRangeMap

protected

map of filenames assigned to active slots

Definition at line 198 of file AthenaOutputStream.h.

m_streamer

ToolHandle<IAthenaOutputStreamTool> AthenaOutputStream::m_streamer

protected

pointer to AthenaOutputStreamTool

Definition at line 185 of file AthenaOutputStream.h.

m_streamerMap

std::map< std::string, std::unique_ptr<IAthenaOutputStreamTool> > AthenaOutputStream::m_streamerMap

protected

map of streamerTools handling event ranges in MT

Definition at line 204 of file AthenaOutputStream.h.

m_streamName

StringProperty AthenaOutputStream::m_streamName {this, "StreamName", "", "Name of the output stream"}

protected

Stream name (defaults to algorithm name)

Definition at line 114 of file AthenaOutputStream.h.

m_tpCnvSvc

ServiceHandle<ITPCnvSvc> AthenaOutputStream::m_tpCnvSvc {this, "AthTPCnvSvc", "AthTPCnvSvc"}

protected

Definition at line 108 of file AthenaOutputStream.h.

m_transient

ToolHandle<SG::IFolder> AthenaOutputStream::m_transient

protected

Decoded list of transient ids.

Definition at line 168 of file AthenaOutputStream.h.

m_transientItems

StringArrayProperty AthenaOutputStream::m_transientItems {this, "TransientItems", {}, "Transient item list"}

protected

List of items that are known to be present in the transient store (and hence we can make input dependencies on them).

Definition at line 140 of file AthenaOutputStream.h.

m_writeMetadataAndDisconnect

bool AthenaOutputStream::m_writeMetadataAndDisconnect {false}

protected

Definition at line 191 of file AthenaOutputStream.h.

---

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

• AthenaOutputStream.h
• AthenaOutputStream.cxx