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
typedef std::vector< std::pair< std::string, std::string > >	TypeKeyPairs

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

Protected Types
typedef ServiceHandle< IClassIDSvc >	IClassIDSvc_t
typedef std::map< std::string, unsigned int >	CounterMapType
	map to record number of writes per object More...
typedef std::recursive_mutex	mutex_t
	mutex for this Stream write() and handle() methods More...

Protected Member Functions
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...

Protected Attributes
ServiceHandle< StoreGateSvc >	m_dataStore
	handle to the StoreGateSvc store where the data we want to write out resides More...
ServiceHandle< StoreGateSvc >	m_metadataStore
ServiceHandle< StoreGateSvc > *	m_currentStore
ServiceHandle< IItemListSvc >	m_itemSvc
ServiceHandle< MetaDataSvc >	m_metaDataSvc
ServiceHandle< IDictLoaderSvc >	m_dictLoader
ServiceHandle< ITPCnvSvc >	m_tpCnvSvc
ServiceHandle< IIncidentSvc >	m_incidentSvc
std::string	m_persName
	Name of the persistency service capable to write data from the store. More...
StringProperty	m_writingTool
	Name of the OutputStreamTool used for writing. More...
std::string	m_outputName
	Name of the output file. More...
std::string	m_outputAttributes
IClassIDSvc_t	m_pCLIDSvc
ServiceHandle< OutputStreamSequencerSvc >	m_outSeqSvc
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...
StringArrayProperty	m_excludeList {this,"ExcludeList",{},"List of metadata items to write","OrderedSet<std::string>"}
	Vector of item names. More...
StringArrayProperty	m_compressionListHigh
	Vector of item names. More...
StringArrayProperty	m_compressionListLow
	Vector of item names. More...
unsigned int	m_compressionBitsHigh
	Number of mantissa bits in the float compression. More...
unsigned int	m_compressionBitsLow
	Number of mantissa bits in the float compression. More...
StringArrayProperty	m_transientItems
	List of items that are known to be present in the transient store (and hence we can make input dependencies on them). More...
ToolHandle< SG::IFolder >	m_p2BWritten
	the top-level folder with items to be written More...
ToolHandle< SG::IFolder >	m_decoder
	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...
std::atomic< int >	m_events
	Number of events written to this output stream. More...
bool	m_forceRead
	set to true to force read of data objects in item list More...
bool	m_extendProvenanceRecord
	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...
bool	m_writeOnExecute
	set to true to trigger streaming of data on execute() More...
bool	m_writeOnFinalize
	set to true to trigger streaming of data on finalize() More...
bool	m_itemListFromTool
	set to write out everything in input DataHeader More...
bool	m_checkNumberOfWrites
	set to true to check for number of times each object is written More...
CounterMapType	m_objectWriteCounter
ToolHandle< IAthenaOutputStreamTool >	m_streamer
	pointer to AthenaOutputStreamTool More...
ToolHandleArray< IAthenaOutputTool >	m_helperTools
	vector of AlgTools that that are executed by this stream More...
bool	m_writeMetadataAndDisconnect = false
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

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

implement IAlgorithm
virtual StatusCode	initialize () override
virtual StatusCode	finalize () override
virtual StatusCode	execute () override
virtual StatusCode	stop () 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
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
void	tokenizeAtSep (std::vector< std::string > &, const std::string &, const std::string &) const
	tokenize a string based on a substring More...
bool	matchKey (const std::vector< std::string > &key, const std::string &proxyName) const
	Try to match a DataProxy name to a vector of strings. 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...

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

CounterMapType

typedef std::map<std::string, unsigned int> AthenaOutputStream::CounterMapType

protected

map to record number of writes per object

Definition at line 142 of file AthenaOutputStream.h.

IClassIDSvc_t

typedef ServiceHandle<IClassIDSvc> AthenaOutputStream::IClassIDSvc_t

protected

Definition at line 83 of file AthenaOutputStream.h.

Items

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

Definition at line 58 of file AthenaOutputStream.h.

mutex_t

typedef std::recursive_mutex AthenaOutputStream::mutex_t

protected

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

Definition at line 163 of file AthenaOutputStream.h.

TypeKeyPairs

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

Definition at line 188 of file AthenaOutputStream.h.

Constructor & Destructor Documentation

AthenaOutputStream()

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

Standard algorithm Constructor.

Definition at line 156 of file AthenaOutputStream.cxx.

      : base_class(name, pSvcLocator),
        m_dataStore("StoreGateSvc", name),
        m_metadataStore("MetaDataStore", name),
        m_currentStore(&m_dataStore),
        m_itemSvc("ItemListSvc", name),
        m_metaDataSvc("MetaDataSvc", name),
        m_dictLoader("AthDictLoaderSvc", name),
        m_tpCnvSvc("AthTPCnvSvc", name),
        m_incidentSvc("IncidentSvc", name),
        m_outputAttributes(),
        m_pCLIDSvc("ClassIDSvc", name),
        m_outSeqSvc("OutputStreamSequencerSvc", name),
        m_p2BWritten(string("SG::Folder/") + name + string("_TopFolder"), this),
        m_decoder(string("SG::Folder/") + name + string("_excluded"), this),
        m_compressionDecoderHigh(string("SG::Folder/") + name + string("_compressed_high"), this),
        m_compressionDecoderLow(string("SG::Folder/") + name + string("_compressed_low"), this),
        m_transient(string("SG::Folder/") + name + string("_transient"), this),
        m_events(0),
        m_streamer(string("AthenaOutputStreamTool/") + name + string("Tool"), this),
        m_helperTools(this)
{
   assert(pSvcLocator);
   declareProperty("TransientItems",         m_transientItems);
   declareProperty("OutputFile",             m_outputName="DidNotNameOutput.root");
   declareProperty("EvtConversionSvc",       m_persName="EventPersistencySvc");
   declareProperty("WritingTool",            m_streamer);
   declareProperty("Store",                  m_dataStore);
   declareProperty("MetadataStore",          m_metadataStore);
   declareProperty("ForceRead",              m_forceRead=true);
   declareProperty("ExtendProvenanceRecord", m_extendProvenanceRecord=true);
   declareProperty("WriteOnExecute",         m_writeOnExecute=true);
   declareProperty("WriteOnFinalize",        m_writeOnFinalize=false);
   declareProperty("TakeItemsFromInput",     m_itemListFromTool=false);
   declareProperty("CheckNumberOfWrites",    m_checkNumberOfWrites=true);
   declareProperty("HelperTools",            m_helperTools);
   declareProperty("CompressionListHigh",    m_compressionListHigh);
   declareProperty("CompressionListLow",     m_compressionListLow);
   declareProperty("CompressionBitsHigh",    m_compressionBitsHigh = 7);
   declareProperty("CompressionBitsLow",     m_compressionBitsLow = 15);
   
   // Associate action handlers with the AcceptAlgs,
   // RequireAlgs & VetoAlgs properties
   m_itemList.declareUpdateHandler(&AthenaOutputStream::itemListHandler, this);
   m_excludeList.declareUpdateHandler(&AthenaOutputStream::excludeListHandler, this);
   m_compressionListHigh.declareUpdateHandler(&AthenaOutputStream::compressionListHandlerHigh, this);
   m_compressionListLow.declareUpdateHandler(&AthenaOutputStream::compressionListHandlerLow, this);
}

~AthenaOutputStream()

AthenaOutputStream::~AthenaOutputStream ( )

virtual

Standard Destructor.

Definition at line 206 of file AthenaOutputStream.cxx.

                                        {
   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 769 of file AthenaOutputStream.cxx.

{
   // anything after a dot is a list of dynamic Aux attributes, separated by dots
   size_t dotpos = item.key().find('.');
   string item_key, aux_attr;
   if( dotpos != 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("addItemObjects(" << item_id << ",\"" << item_key << "\") called");
   ATH_MSG_DEBUG("           Key:" << item_key );
   if( aux_attr.size() ) {
   ATH_MSG_DEBUG("      Aux Attr:" << aux_attr );
   }
   static const std::string wildCard = "*";
   std::set<std::string> clidKeys;
   for (SG::IFolder::const_iterator iter = m_decoder->begin(), iterEnd = m_decoder->end();
           iter != iterEnd; ++iter) {
      if (iter->id() == item_id) {
         clidKeys.insert(iter->key());
      }
   }
 
   // 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("     Comp Attr " << it.second.size() << " with " << it.first << " mantissa bits.");
     if ( it.second.size() > 0 ) {
       for( const auto& attr : it.second ) {
          ATH_MSG_DEBUG("       >> " << 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;
      // For item list entry
      // Check for wildcard within string, i.e. 'xxx*yyy', and save the matching parts
      std::vector<std::string> keyTokens;
      keyTokens.reserve(2);
      std::vector<std::string> xkeyTokens;
      xkeyTokens.reserve(2);
      ATH_MSG_VERBOSE("Calling tokenizeAtStep( " << keyTokens << ", " << item_key << ", " << wildCard << ")" );
      this->tokenizeAtSep( keyTokens, item_key, wildCard );
      ATH_MSG_VERBOSE("Done calling tokenizeAtStep( " << keyTokens << ", " << item_key << ", " << wildCard << ")" );
      // Now loop over any found proxies
      for (; iter != end; ++iter) {
         SG::DataProxy* itemProxy(iter->second);
         string proxyName = itemProxy->name();
         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) {
            ATH_MSG_VERBOSE("Calling matchKey( " << keyTokens << ", " << proxyName << ")" );
            keyMatch = matchKey(keyTokens, proxyName);
            ATH_MSG_VERBOSE("Done calling matchKey( " << keyTokens << ", " << proxyName << ") with result: " << keyMatch );
        }
 
         // Now undo the flag based on a similar analysis of excluded wildcard keys
         bool xkeyMatch = false;
         for (std::set<std::string>::const_iterator c2k_it = clidKeys.begin(), c2k_itEnd = clidKeys.end();
                 keyMatch && c2k_it != c2k_itEnd; ++c2k_it) {
            if (*c2k_it == wildCard) {
               xkeyMatch = true;  // wildcard first
            } else {
               // Look for wildcard in key
               std::string::size_type xsep = c2k_it->find(wildCard);
               // If wildcard not found, then check whether the second is an excluded key
               if (xsep == std::string::npos) {
                  if (*c2k_it == proxyName) {
                     xkeyMatch = true;
                  }
               } else { // Otherwise take before and after wildcard for later use
                  this->tokenizeAtSep( xkeyTokens, *c2k_it, wildCard );
                  ATH_MSG_DEBUG("x Proxy name=" << proxyName );
                  xkeyMatch = matchKey(xkeyTokens, proxyName);
               }
            }
         }
         if( !stream.empty() and stream != m_outputName ) {
            // reject keys that are marked for a different output stream
            ATH_MSG_DEBUG("Rejecting key: " << itemProxy->name() << " in output: " << m_outputName);
            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(" 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("Failed to retrieve object from MetaCont with key=" << item_key << ", for EventRangeID=" << 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(" 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(" 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( "Error in casting object with CLID "
                                   << itemProxy->clID() << " to SG::IConstAuxStore*" );
                    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( "Container " << key << " has " << it.second.size() <<
                                       " variables that'll be lossy float compressed"
                                       " with " << it.first << " mantissa bits" );
                      }
                    } // End of loop over variables to be lossy float compressed
                  } // End of lossy float compression logic
 
               }
 
               added = true;
               if (m_checkNumberOfWrites) {
                  CounterMapType::iterator cit = m_objectWriteCounter.find(tn);
                  if (cit == m_objectWriteCounter.end()) {
                    // First time through
                    m_objectWriteCounter.insert(CounterMapType::value_type(tn, 1));
                  } else {
                     // set to next iteration (to avoid double counting)
                     // StreamTools will eliminate duplicates.
                     (*cit).second = m_events + 1;
                  }
               }
               if (m_itemSvc->addStreamItem(this->name(),tns.str()).isFailure()) {
                  ATH_MSG_WARNING("Unable to record item " << tns.str() << " in Svc");
               }
            }
         } else if (keyMatch && xkeyMatch) {
            removed = true;
         }
      } // proxy loop
      if (!added && !removed) {
         ATH_MSG_DEBUG(" No object matching " << item_id << ",\"" << item_key  << "\" found");
      } else if (removed) {
         ATH_MSG_DEBUG(" Object being excluded based on property setting "
                 << item_id << ",\"" << item_key  << "\". Skipping");
      }
   } else {
      ATH_MSG_DEBUG(" Failed to receive proxy iterators from StoreGate for "
              << item_id << ",\"" << item_key  << "\". Skipping");
   }
   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 1034 of file AthenaOutputStream.cxx.

{
  // Create an empty result
  std::set<std::string> result;
 
  // Check the item is indeed Aux.
  if(item_key.find("Aux.") == string::npos) {
    return result;
  }
 
  // First the high compression list
  for (SG::IFolder::const_iterator iter = handle->begin(), iterEnd = handle->end();
         iter != iterEnd; ++iter) {
    // 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('.');
    string comp_item_key{""}, comp_str{""};
    if(seppos != 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 701 of file AthenaOutputStream.cxx.

                                            {
   m_objects.erase(m_objects.begin(), m_objects.end());
   m_ownedObjects.clear();
   m_altObjects.clear();
}

collectAllObjects()

StatusCode AthenaOutputStream::collectAllObjects

(

)

Collect data objects for output streamer list.

Definition at line 707 of file AthenaOutputStream.cxx.

                                                 {
   if (m_itemListFromTool) {
      if (!m_streamer->getInputItemList(&*m_p2BWritten).isSuccess()) {
         ATH_MSG_WARNING("collectAllObjects() could not get ItemList from Tool.");
      }
   }
 
   auto vetoes = std::make_unique<SG::SelectionVetoes>();
   auto compInfo = std::make_unique<SG::CompressionInfo>();
 
   m_p2BWritten->updateItemList(true);
   std::vector<CLID> folderclids;
   // Collect all objects that need to be persistified:
   //FIXME refactor: move this in folder. Treat as composite
   for (SG::IFolder::const_iterator i = m_p2BWritten->begin(), iEnd = m_p2BWritten->end(); i != iEnd; ++i) {
      ATH_CHECK( addItemObjects(*i, *vetoes, *compInfo) );
      folderclids.push_back(i->id());
   }
 
   // FIXME This is a bruteforce hack to remove items erroneously 
   // added somewhere in the morass of the addItemObjects logic
   IDataSelector prunedList;
   for (auto it = m_objects.begin(); it != m_objects.end(); ++it) {
      if (std::find(folderclids.begin(),folderclids.end(),(*it)->clID())!=folderclids.end()) {
         if (SG::isTransientKey ((*it)->name())) {
           ATH_MSG_ERROR("Request to write transient object key " <<
                         (*it)->name() << " ignored");
         }
         else {
           prunedList.push_back(*it);  // build new list that is correct
         }
      }
      else {
         ATH_MSG_DEBUG("Object " << (*it)->clID() <<","<< (*it)->name() << " found that was not in itemlist");
      }
   }
   m_objects.clear();  // clear previous list
   for (auto it = prunedList.begin(); it != prunedList.end(); ++it) {
      if ((*it)->name().length() > 4 && (*it)->name().substr((*it)->name().length() - 4) == "Aux.") {
         m_objects.push_back(*it);  // first copy aux store new into previous
      }
   }
   for (auto it = prunedList.begin(); it != prunedList.end(); ++it) {
      if ((*it)->name().length() <= 4 || (*it)->name().substr((*it)->name().length() - 4) != "Aux.") {
         m_objects.push_back(*it);  // then copy others new into previous
      }
   }
 
   // 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 1148 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 1157 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.

Definition at line 1139 of file AthenaOutputStream.cxx.

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

execute()

StatusCode AthenaOutputStream::execute ( )

overridevirtual

Reimplemented in AthenaConditionStream.

Definition at line 559 of file AthenaOutputStream.cxx.

                                       {
   bool failed = false;
   for (ToolHandle<IAthenaOutputTool>& tool : m_helperTools) {
      if (!tool->preExecute().isSuccess()) {
         failed = true;
      }
   }
   // Test whether this event should be output
   if (m_writeOnExecute && isEventAccepted()) {
      if (write().isFailure()) {
         failed = true;
      }
   }
   for (ToolHandle<IAthenaOutputTool>& tool : m_helperTools) {
      if(!tool->postExecute().isSuccess()) {
         failed = true;
      }
   }
   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 533 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");
   if (!m_helperTools.release().isSuccess()) {
      failed = true;
   }
   if (!m_streamer.release().isSuccess()) {
      failed = true;
   }
   if (failed) {
      return(StatusCode::FAILURE);
   }
   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 444 of file AthenaOutputStream.cxx.

{
   ATH_MSG_DEBUG("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("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 368 of file AthenaOutputStream.cxx.

{
   ATH_MSG_DEBUG("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("Slot range change: '" << rangeFN << "' -> '" << newRangeFN << "'");
            ATH_MSG_DEBUG("There are " << m_slotRangeMap.size() << " slots in use");
            for(auto range : m_slotRangeMap ) {
               ATH_MSG_DEBUG("Slot: " << range.first << "  FN=" << range.second);
            }
            if( count_events_in_range(rangeFN) == 1 ) {
               finalizeRange( rangeFN );
            }
         }
         ATH_MSG_INFO("slot " << slot << " processing event in range: " << 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("There are " << m_slotRangeMap.size() << " slots in use");
         for(auto range : m_slotRangeMap ) {
            ATH_MSG_DEBUG("Slot: " << range.first << "  FN=" << 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("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

Definition at line 1077 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("Unable to record item " << temp.str() << " in Svc");
      }
    }
  }
 
  // 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 211 of file AthenaOutputStream.cxx.

                                          {
   ATH_CHECK( this->FilteredAlgorithm::initialize() );
   ATH_MSG_DEBUG("In initialize");
   // Reset the number of events written
   m_events = 0;
 
   // set up the SG service:
   ATH_CHECK( m_dataStore.retrieve() );
   ATH_MSG_DEBUG("Found " << m_dataStore.typeAndName() << " store.");
   assert(static_cast<bool>(m_dataStore));
   if (!m_metadataItemList.value().empty()) {
      ATH_CHECK( m_metadataStore.retrieve() );
      ATH_MSG_DEBUG("Found " << m_metadataStore.typeAndName() << " store.");
      assert(static_cast<bool>(m_metadataStore));
   }
 
   // set up the CLID service:
   ATH_CHECK( m_pCLIDSvc.retrieve() );
 
   ATH_CHECK( m_dictLoader.retrieve() );
   ATH_CHECK( m_tpCnvSvc.retrieve() );
 
   // set up the ItemListSvc service:
   assert(static_cast<bool>(m_pCLIDSvc));
   ATH_CHECK( m_itemSvc.retrieve() );
   assert(static_cast<bool>(m_itemSvc));
 
   // set up the OutputStreamSequencer service:
   ATH_CHECK( m_outSeqSvc.retrieve() );
   assert(static_cast<bool>(m_outSeqSvc));
 
   // 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 << endmsg << "Data output: " << m_outputName);
 
   for (ToolHandle<IAthenaOutputTool>& 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(), 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("Float compression mantissa bits for high compression " <<
                  "(" << m_compressionBitsHigh << ") is outside the allowed range of [5, 23].");
     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("Float compression mantissa bits for low compression " <<
                  "(" << m_compressionBitsLow << ") is outside the allowed range of [5, 23].");
     ATH_MSG_INFO("Setting it to the appropriate limit.");
     m_compressionBitsLow = m_compressionBitsLow < 5 ? 5 : 23;
   }
   if(m_compressionBitsLow <= m_compressionBitsHigh) {
     ATH_MSG_ERROR("Float compression mantissa bits for low compression " <<
                   "(" << m_compressionBitsLow << ") is lower than or equal to high compression " <<
                   "(" << m_compressionBitsHigh << ")! Please check the configuration! ");
     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("High compression will use " << m_compressionBitsHigh << " mantissa bits, and " <<
                  "low compression will use " << m_compressionBitsLow << " mantissa bits.");
   }
 
   // Setup stream name
   if (m_streamName.empty()) {
      m_streamName.setValue(this->name());
   }
 
   // Set SG key for selected variable information.
   m_selVetoesKey = "SelectionVetoes_" + m_streamName;
   ATH_CHECK( m_selVetoesKey.initialize() );
 
   m_compInfoKey = "CompressionInfo_" + m_streamName;
   ATH_CHECK( m_compInfoKey.initialize() );
 
   ATH_MSG_DEBUG("End initialize");
   return StatusCode::SUCCESS;
}

io_finalize()

StatusCode AthenaOutputStream::io_finalize ( )

overridevirtual

Definition at line 1248 of file AthenaOutputStream.cxx.

                                           {
   ATH_MSG_INFO("I/O finalization...");
   for (ToolHandle<IAthenaOutputTool>& 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 1231 of file AthenaOutputStream.cxx.

                                         {
   ATH_MSG_INFO("I/O reinitialization...");
   // For 'write on finalize', we set up listener for 'MetaDataStop'
   // and perform write at this point. This happens at 'stop' of the
   // event selector. RDS 04/2010
   // Set to be listener for end of event
   m_incidentSvc->removeListener(this, "MetaDataStop"); // Remove any existing listener to avoid handling the incident multiple times
   m_incidentSvc->addListener(this, "MetaDataStop", 50);
   for (ToolHandle<IAthenaOutputTool>& 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 1129 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 1270 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());
  }
}

matchKey()

bool AthenaOutputStream::matchKey ( const std::vector< std::string > &  key, const std::string &  proxyName  ) const

private

Try to match a DataProxy name to a vector of strings.

Definition at line 1188 of file AthenaOutputStream.cxx.

                                                                 {
  bool keyMatch = true; // default return
 
  // Get an iterator to the first (not zeroth!) string in the vector
  std::vector<std::string>::const_iterator itrEnd = key.cend();
  std::vector<std::string>::const_iterator itr = key.cbegin();
 
  // Walk through the whole proxyName string and try to match to all sub-keys
  // We are using that: std::string::npos!=string.find("") is always true
  std::string::size_type proxyNamePos=0;
  while ( itr != itrEnd &&
          std::string::npos != ( proxyNamePos = proxyName.find(*itr, proxyNamePos) )
          ) {
    // If we are at the begin iterator and the first element is Not an empty string
    ATH_MSG_VERBOSE("If we are at the begin iterator and the first element is Not an empty string");
    if ( !(key.front().empty()) && itr == key.cbegin() && proxyNamePos != 0 ) {
      // We had to match a precise name at the beginning, but didn't find it at the beginning
      ATH_MSG_VERBOSE("We had to match a precise name at the beginning, but didn't find it at the beginning");
      break;
    }
    // If we are at the end iterator and the last element is Not an empty string
    if ( !(key.back().empty()) && itr == --(key.cend()) && (proxyNamePos+itr->size()!=proxyName.size()) ) {
      // We had to match a precise name at the end, but didn't find it at the end
      ATH_MSG_VERBOSE("We had to match a precise name at the end, but didn't find it at the end");
      break;
    }
    ATH_MSG_VERBOSE("Found a match of subkey: " << *itr << " in string: " << proxyName
                    << " at position: " << proxyNamePos );
    // If we have a good match, increment the iterator and the search position
    proxyNamePos += itr->size();
    ++itr;
  }
  // Didn't find everything
  if ( itr != itrEnd ) {
    keyMatch = false;
    ATH_MSG_VERBOSE("Couldn't match every sub-string... return: " << keyMatch);
  }
  else { ATH_MSG_VERBOSE("Did match every sub-string... return: " << keyMatch); }
 
  return(keyMatch);
}

selectedObjects()

IDataSelector* AthenaOutputStream::selectedObjects ( )

inline

Return the list of selected objects.

Definition at line 209 of file AthenaOutputStream.h.

                                    {
      return &m_objects;
   }

stop()

StatusCode AthenaOutputStream::stop ( )

overridevirtual

Definition at line 361 of file AthenaOutputStream.cxx.

{
   ATH_MSG_DEBUG("AthenaOutputStream " << this->name() << " ::stop()");
   return StatusCode::SUCCESS;
}

tokenizeAtSep()

void AthenaOutputStream::tokenizeAtSep ( std::vector< std::string > &  subStrings, const std::string &  portia, const std::string &  sepstr  ) const

private

tokenize a string based on a substring

Definition at line 1166 of file AthenaOutputStream.cxx.

                                                                        {
  subStrings.clear(); // clear from previous iteration step
  // If the portia starts with a wildcard, add an empty string
  if (portia.starts_with( sepstr )) {
    subStrings.push_back("");
  }
  boost::char_separator<char> csep(sepstr.c_str());
  boost::tokenizer<char_separator<char>> tokens(portia, csep);
  for (const std::string& t : tokens) {
    //ATH_MSG_VERBOSE("Now on token: " << t);
    subStrings.push_back(t);
  }
  // If the portia ends with a wildcard, add an empty string
  if ( portia.size() >= sepstr.size() &&
       portia.compare( portia.size() - sepstr.size(), sepstr.size(), sepstr) == 0 ) {
    subStrings.push_back("");
  }
  return;
}

write()

StatusCode AthenaOutputStream::write ( )

virtual

Stream the data.

Definition at line 592 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( "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("Unable to initialize OutputStreamTool for " << outputFN );
            return StatusCode::FAILURE;
         }
         m_streamerMap[ outputFN ].reset( streamer );
      }
   }
   
   // Clear any previously existing item list
   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 );
   
   // print out info about objects collected
   if (m_checkNumberOfWrites) {
      bool checkCountError = false;
      ATH_MSG_DEBUG(" Collected objects:");
      bool first = true;
      unsigned int lastCount = 0;
      for (CounterMapType::iterator cit = m_objectWriteCounter.begin(),
              clast = m_objectWriteCounter.end(); cit != clast; ++cit) {
         bool isError = false;
         if (first) {
            lastCount = (*cit).second;
            first = false;
         } else if (lastCount != (*cit).second) {
            isError = true;
            //Complain, but don't abort
            checkCountError = true;
         }
         if (isError) {
            ATH_MSG_ERROR(" INCORRECT Object/count: "
                          << (*cit).first << ", " << (*cit).second << " should be: " << lastCount);
         } else {
            ATH_MSG_DEBUG(" Object/count: " << (*cit).first << ", " << (*cit).second);
         }
      }
      if (checkCountError) {
         ATH_MSG_FATAL("Check number of writes failed. See messages above "
                       "to identify which container is not always written");
         return(StatusCode::FAILURE);
      }
   }
   // prepare before releasing lock because m_outputAttributes change in metadataStop
   const std::string connectStr = outputFN + m_outputAttributes;
 
   for (ToolHandle<IAthenaOutputTool>& 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("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 465 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 (ToolHandle<IAthenaOutputTool>& 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 );
 
   // Always force a final commit in stop - mainly applies to AthenaPool
   if (m_writeOnFinalize) {
      if (write().isFailure()) {  // true mean write AND commit
         throw GaudiException("Cannot write on finalize", name(), StatusCode::FAILURE);
      }
      ATH_MSG_INFO("Records written: " << m_events);
   }
   // 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_checkNumberOfWrites = false;
      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_INFO("Metadata records written: " << m_events);
   }
}

Member Data Documentation

m_altObjects

IDataSelector AthenaOutputStream::m_altObjects

protected

Objects overridden by ‘exact’ handling.

Definition at line 121 of file AthenaOutputStream.h.

m_checkNumberOfWrites

bool AthenaOutputStream::m_checkNumberOfWrites

protected

set to true to check for number of times each object is written

Definition at line 140 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 117 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 174 of file AthenaOutputStream.h.

m_compressionBitsHigh

unsigned int AthenaOutputStream::m_compressionBitsHigh

protected

Number of mantissa bits in the float compression.

Definition at line 100 of file AthenaOutputStream.h.

m_compressionBitsLow

unsigned int AthenaOutputStream::m_compressionBitsLow

protected

Number of mantissa bits in the float compression.

Definition at line 102 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 111 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 113 of file AthenaOutputStream.h.

m_compressionListHigh

StringArrayProperty AthenaOutputStream::m_compressionListHigh

protected

Vector of item names.

Definition at line 96 of file AthenaOutputStream.h.

m_compressionListLow

StringArrayProperty AthenaOutputStream::m_compressionListLow

protected

Vector of item names.

Definition at line 98 of file AthenaOutputStream.h.

m_currentStore

ServiceHandle<StoreGateSvc>* AthenaOutputStream::m_currentStore

protected

Definition at line 65 of file AthenaOutputStream.h.

m_dataStore

ServiceHandle<StoreGateSvc> AthenaOutputStream::m_dataStore

protected

handle to the StoreGateSvc store where the data we want to write out resides

Definition at line 63 of file AthenaOutputStream.h.

m_decoder

ToolHandle<SG::IFolder> AthenaOutputStream::m_decoder

protected

the top-level folder with items to be written

Definition at line 109 of file AthenaOutputStream.h.

m_dictLoader

ServiceHandle<IDictLoaderSvc> AthenaOutputStream::m_dictLoader

protected

Definition at line 70 of file AthenaOutputStream.h.

m_events

std::atomic<int> AthenaOutputStream::m_events

protected

Number of events written to this output stream.

Definition at line 127 of file AthenaOutputStream.h.

m_excludeList

StringArrayProperty AthenaOutputStream::m_excludeList {this,"ExcludeList",{},"List of metadata items to write","OrderedSet<std::string>"}

protected

Vector of item names.

Definition at line 94 of file AthenaOutputStream.h.

m_extendProvenanceRecord

bool AthenaOutputStream::m_extendProvenanceRecord

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 132 of file AthenaOutputStream.h.

m_forceRead

bool AthenaOutputStream::m_forceRead

protected

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

Definition at line 129 of file AthenaOutputStream.h.

m_helperTools

ToolHandleArray<IAthenaOutputTool> AthenaOutputStream::m_helperTools

protected

vector of AlgTools that that are executed by this stream

Definition at line 148 of file AthenaOutputStream.h.

m_incidentSvc

ServiceHandle<IIncidentSvc> AthenaOutputStream::m_incidentSvc

protected

Definition at line 73 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 90 of file AthenaOutputStream.h.

m_itemListFromTool

bool AthenaOutputStream::m_itemListFromTool

protected

set to write out everything in input DataHeader

Definition at line 138 of file AthenaOutputStream.h.

m_itemSvc

ServiceHandle<IItemListSvc> AthenaOutputStream::m_itemSvc

protected

Definition at line 67 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 92 of file AthenaOutputStream.h.

m_metadataStore

ServiceHandle<StoreGateSvc> AthenaOutputStream::m_metadataStore

protected

Definition at line 64 of file AthenaOutputStream.h.

m_metaDataSvc

ServiceHandle<MetaDataSvc> AthenaOutputStream::m_metaDataSvc

protected

Definition at line 68 of file AthenaOutputStream.h.

m_mutex

mutex_t AthenaOutputStream::m_mutex

protected

Definition at line 164 of file AthenaOutputStream.h.

m_objects

IDataSelector AthenaOutputStream::m_objects

protected

Collection of objects being selected.

Definition at line 119 of file AthenaOutputStream.h.

m_objectWriteCounter

CounterMapType AthenaOutputStream::m_objectWriteCounter

protected

Definition at line 143 of file AthenaOutputStream.h.

m_outputAttributes

std::string AthenaOutputStream::m_outputAttributes

protected

Definition at line 81 of file AthenaOutputStream.h.

m_outputName

std::string AthenaOutputStream::m_outputName

protected

Name of the output file.

Definition at line 80 of file AthenaOutputStream.h.

m_outSeqSvc

ServiceHandle<OutputStreamSequencerSvc> AthenaOutputStream::m_outSeqSvc

protected

Definition at line 85 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 125 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 107 of file AthenaOutputStream.h.

m_pCLIDSvc

IClassIDSvc_t AthenaOutputStream::m_pCLIDSvc

protected

Definition at line 84 of file AthenaOutputStream.h.

m_persName

std::string AthenaOutputStream::m_persName

protected

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

Definition at line 76 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 158 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 169 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 155 of file AthenaOutputStream.h.

m_streamer

ToolHandle<IAthenaOutputStreamTool> AthenaOutputStream::m_streamer

protected

pointer to AthenaOutputStreamTool

Definition at line 146 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 161 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 88 of file AthenaOutputStream.h.

m_tpCnvSvc

ServiceHandle<ITPCnvSvc> AthenaOutputStream::m_tpCnvSvc

protected

Definition at line 71 of file AthenaOutputStream.h.

m_transient

ToolHandle<SG::IFolder> AthenaOutputStream::m_transient

protected

Decoded list of transient ids.

Definition at line 115 of file AthenaOutputStream.h.

m_transientItems

StringArrayProperty AthenaOutputStream::m_transientItems

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 105 of file AthenaOutputStream.h.

m_writeMetadataAndDisconnect

bool AthenaOutputStream::m_writeMetadataAndDisconnect = false

protected

Definition at line 151 of file AthenaOutputStream.h.

m_writeOnExecute

bool AthenaOutputStream::m_writeOnExecute

protected

set to true to trigger streaming of data on execute()

Definition at line 134 of file AthenaOutputStream.h.

m_writeOnFinalize

bool AthenaOutputStream::m_writeOnFinalize

protected

set to true to trigger streaming of data on finalize()

Definition at line 136 of file AthenaOutputStream.h.

m_writingTool

StringProperty AthenaOutputStream::m_writingTool

protected

Name of the OutputStreamTool used for writing.

Definition at line 78 of file AthenaOutputStream.h.

---

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

• AthenaOutputStream.h
• AthenaOutputStream.cxx