ATLAS Offline Software
Loading...
Searching...
No Matches
AthenaOutputStream Class Reference

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

#include <AthenaOutputStream.h>

Inheritance diagram for AthenaOutputStream:

Public Types

typedef std::recursive_mutex mutex_t

Public Member Functions

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

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.
ServiceHandle< StoreGateSvc > m_metadataStore {this, "MetadataStore", "StoreGateSvc/MetaDataStore", "Handle to metadata store"}
ServiceHandle< MetaDataSvc > m_metaDataSvc {this, "MetaDataSvc", "MetaDataSvc"}
 Handles to all the necessary services.
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).
StringArrayProperty m_itemList {this, "ItemList", {}, "List of items to write", "OutputStreamItemList"}
 Vector of item names.
StringArrayProperty m_metadataItemList {this, "MetadataItemList", {}, "List of metadata items to write","OutputStreamItemList"}
 Vector of item names.
StringProperty m_keepProvenances
 Provenance record selection.
StringArrayProperty m_compressionListHigh {this, "CompressionListHigh", {}, "Lossy float compression list (high)"}
 Vector of item names.
StringArrayProperty m_compressionListLow {this, "CompressionListLow", {}, "Lossy float compression list (low)"}
 Vector of item names.
UnsignedIntegerProperty m_compressionBitsHigh {this, "CompressionBitsHigh", 7, "Lossy float compression bits (high)"}
 Number of mantissa bits in the float compression.
UnsignedIntegerProperty m_compressionBitsLow {this, "CompressionBitsLow", 15, "Lossy float compression bits (low)"}
 Number of mantissa bits in the float compression.
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).
StringProperty m_outputName {this, "OutputFile", "DidNotNameOutput.root", "Name of the output file"}
 Name of the output file.
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.
BooleanProperty m_forceRead {this, "ForceRead", true, "Force read data objects in ItemList"}
 set to true to force read of data objects in item list
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).
BooleanProperty m_itemListFromTool {this, "TakeItemsFromInput", false, "Write everything in input DataHeader to output"}
 Set to write out everything in input DataHeader.
ToolHandle< SG::IFolder > m_p2BWritten
 The top-level folder with items to be written.
ToolHandle< SG::IFolder > m_compressionDecoderHigh
 The top-level folder with items to be compressed high.
ToolHandle< SG::IFolder > m_compressionDecoderLow
 The top-level folder with items to be compressed low.
ToolHandle< SG::IFolder > m_transient
 Decoded list of transient ids.
IDataSelector m_objects
 Collection of objects being selected.
IDataSelector m_altObjects
 Objects overridden by `exact' handling.
std::vector< std::unique_ptr< DataObject > > m_ownedObjects
 Collection of DataObject instances owned by this service.
ToolHandle< IAthenaOutputStreamTool > m_streamer
 pointer to AthenaOutputStreamTool
ToolHandleArray< IAthenaOutputTool > m_helperTools {this, "HelperTools", {}, "List of AlgTools used by this stream"}
 vector of AlgTools that that are executed by this stream
bool m_writeMetadataAndDisconnect {false}
std::atomic< int > m_events {0}
 Number of events written to this output stream.
std::map< unsigned, std::string > m_slotRangeMap
 map of filenames assigned to active slots
std::map< std::string, std::string > m_rangeIDforRangeFN
 map of RangeIDs (as used by the Sequencer) for each Range filename generated
std::map< std::string, std::unique_ptr< IAthenaOutputStreamTool > > m_streamerMap
 map of streamerTools handling event ranges in MT
mutex_t m_mutex
 mutex for this Stream write() and handle() methods
SG::WriteHandleKey< SG::SelectionVetoes > m_selVetoesKey { this, "SelVetoesKey", "" }
 Key used for recording selected dynamic variable information to the event store.
SG::WriteHandleKey< SG::CompressionInfo > m_compInfoKey { this, "CompInfoKey", "" }
 Key used for recording lossy float compressed variable information to the event store.
std::string m_outputAttributes
 Output attributes.
virtual StatusCode initialize () override
virtual StatusCode finalize () override
virtual StatusCode execute (const EventContext &ctx) override
virtual StatusCode write (const EventContext &ctx)
 Stream the data from the event store.
StatusCode write (const EventContext &ctx, StoreGateSvc &store)
 Stream the data from the given store.
void clearSelection ()
 Clear list of selected objects.
StatusCode collectAllObjects (const EventContext &ctx, StoreGateSvc &store)
 Collect data objects for output streamer list.
IDataSelector * selectedObjects ()
 Return the list of selected objects.
virtual void handle (const Incident &incident) override
 Incident service handle listening for MetaDataStop.
virtual StatusCode io_reinit () override
 Callback method to reinitialize the internal state of the component for I/O purposes (e.g. upon fork(2)).
virtual StatusCode io_finalize () override
void itemListHandler (Gaudi::Details::PropertyBase &)
 Handler for ItemNames Property.
void compressionListHandlerHigh (Gaudi::Details::PropertyBase &)
 Handler for ItemNames Property.
void compressionListHandlerLow (Gaudi::Details::PropertyBase &)
 Handler for ItemNames Property.
StatusCode addItemObjects (const EventContext &, StoreGateSvc &store, const SG::FolderItem &, SG::SelectionVetoes &vetoes, SG::CompressionInfo &compInfo)
 Add item data objects to output streamer list.
void handleVariableSelection (const SG::IConstAuxStore &auxstore, SG::DataProxy &itemProxy, const std::string &aux_attr, SG::SelectionVetoes &vetoes) const
 Here we build the vetoed AuxIDs.
void writeMetaData (const EventContext &ctx, const std::string &outputFN="")
 Write MetaData for this stream (by default) or for a substream outputFN (in ES mode).
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.
void finalizeRange (const EventContext &ctx, const std::string &rangeFN)
void loadDict (CLID clid)
 Helper function to load dictionaries (both transient and persistent) for a given type.
bool simpleMatch (const std::string &pattern, const std::string &text)
 Glob-style matcher, where the only meta-character is '*'.

Detailed Description

algorithm that marks for write data objects in SG

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

Definition at line 53 of file AthenaOutputStream.h.

Member Typedef Documentation

◆ mutex_t

typedef std::recursive_mutex AthenaOutputStream::mutex_t

Definition at line 57 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.

44 : base_class(name, pSvcLocator),
45 m_p2BWritten(std::format("SG::Folder/{}_TopFolder", name), this),
46 m_compressionDecoderHigh(std::format("SG::Folder/{}_compressed_high", name), this),
47 m_compressionDecoderLow(std::format("SG::Folder/{}_compressed_low", name), this),
48 m_transient(std::format("SG::Folder/{}_transient", name), this),
49 m_streamer(std::format("AthenaOutputStreamTool/{}Tool", name), this)
50{
51 // Ensure the service locater is good
52 assert(pSvcLocator);
53
54 // This property depends on the name that's known at construction time
55 // Therefore, do it the old fashioned way
56 declareProperty("WritingTool", m_streamer);
57
58 // Associate action handlers with the AcceptAlgs,
59 // RequireAlgs & VetoAlgs properties
60 m_itemList.declareUpdateHandler(&AthenaOutputStream::itemListHandler, this);
63}
ToolHandle< SG::IFolder > m_compressionDecoderLow
The top-level folder with items to be compressed low.
ToolHandle< SG::IFolder > m_p2BWritten
The top-level folder with items to be written.
StringArrayProperty m_compressionListLow
Vector of item names.
ToolHandle< IAthenaOutputStreamTool > m_streamer
pointer to AthenaOutputStreamTool
ToolHandle< SG::IFolder > m_compressionDecoderHigh
The top-level folder with items to be compressed high.
StringArrayProperty m_itemList
Vector of item names.
void compressionListHandlerLow(Gaudi::Details::PropertyBase &)
Handler for ItemNames Property.
StringArrayProperty m_compressionListHigh
Vector of item names.
ToolHandle< SG::IFolder > m_transient
Decoded list of transient ids.
void compressionListHandlerHigh(Gaudi::Details::PropertyBase &)
Handler for ItemNames Property.
void itemListHandler(Gaudi::Details::PropertyBase &)
Handler for ItemNames Property.

◆ ~AthenaOutputStream()

AthenaOutputStream::~AthenaOutputStream ( )
virtual

Standard Destructor.

Definition at line 66 of file AthenaOutputStream.cxx.

66 {
67 // Clear the internal caches
68 m_streamerMap.clear();
69}
std::map< std::string, std::unique_ptr< IAthenaOutputStreamTool > > m_streamerMap
map of streamerTools handling event ranges in MT

Member Function Documentation

◆ addItemObjects()

StatusCode AthenaOutputStream::addItemObjects ( const EventContext & ctx,
StoreGateSvc & store,
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 569 of file AthenaOutputStream.cxx.

574{
575 // anything after a dot is a list of dynamic Aux attributes, separated by dots
576 size_t dotpos = item.key().find('.');
577 std::string item_key, aux_attr;
578 if( dotpos != std::string::npos ) {
579 item_key = item.key().substr(0, dotpos+1);
580 aux_attr = item.key().substr(dotpos+1);
581 } else {
582 item_key = item.key();
583 }
584 CLID item_id = item.id();
585 ATH_MSG_DEBUG("addItemObjects({},\"{}\") called", item_id, item_key);
586 ATH_MSG_DEBUG(" Key:{}", item_key);
587 if( aux_attr.size() ) {
588 ATH_MSG_DEBUG(" Aux Attr:{}", aux_attr);
589 }
590
591 // Here we build the list of attributes for the lossy float compression
592 // Note that we do not allow m_compressionBitsHigh >= m_compressionBitsLow
593 // Otherwise is, in any case, a logical error and they'd potentially overwrite each other
594 std::map< unsigned int, std::set< std::string > > comp_attr_map;
595 comp_attr_map[ m_compressionBitsHigh ] = buildCompressionSet( m_compressionDecoderHigh, item_id, item_key );
596 comp_attr_map[ m_compressionBitsLow ] = buildCompressionSet( m_compressionDecoderLow, item_id, item_key );
597
598 // Print some debugging information regarding the lossy float compression configuration
599 for( const auto& it : comp_attr_map ) {
600 ATH_MSG_DEBUG(" Comp Attr {} with {} mantissa bits.", it.second.size(), it.first);
601 if ( it.second.size() > 0 ) {
602 for( const auto& attr : it.second ) {
603 ATH_MSG_DEBUG(" >> {}", attr);
604 }
605 }
606 }
607
608 // For MetaData objects of type T that are kept in MetaContainers get the MetaCont<T> ID
609 // (only metadata is kept in MetaContainers)
610 const bool isMetaData = store.storeID() == StoreID::METADATA_STORE;
611 const CLID remapped_item_id = isMetaData ? m_metaDataSvc->remapMetaContCLID( item_id ) : item_id;
612 SG::TypelessIterator iter, end;
613 bool gotProxies = false;
614 // Look for the clid in storegate
615 SG::DataProxy* match = store.proxy(remapped_item_id, item_key, true);
616 if (match != nullptr) {
617 iter.setState (&match, &match+1).ignore();
618 end.setState (&match+1, &match+1).ignore();
619 gotProxies = true;
620 }
621 // Look for the clid in storegate
622 if (!gotProxies && (store.retrieve(remapped_item_id, iter, end)).isSuccess()) {
623 gotProxies = true;
624 }
625 if (gotProxies) {
626 bool added = false, removed = false;
627 // Now loop over any found proxies
628 for (; iter != end; ++iter) {
629 SG::DataProxy* itemProxy = *iter;
630 std::string proxyName = itemProxy->name();
631 std::string stream;
632 if( isMetaData ) {
633 // only check metadata keys
634 stream = m_metaDataSvc->removeStreamFromKey(proxyName); // can modify proxyName
635 }
636 // Does this key match the proxy key name - allow for wildcarding and aliases
637 bool keyMatch = ( item_key == "*" ||
638 item_key == proxyName ||
639 itemProxy->hasAlias(item_key) );
640 if (!keyMatch) {
641 // For item list we currently allow wildcards ('*'), which has limited use, e.g.:
642 // xAOD::CutBookkeeperAuxContainer#IncompleteCutBookkeepers*Aux.
643 // Here we look for those few cases...
644 keyMatch = simpleMatch(item_key, proxyName);
645 ATH_MSG_DEBUG("Result of checking {} against {} to see if it matches is {}",
646 proxyName, item_key, keyMatch);
647 }
648
649 // Now check if this item is marked for another output stream, if so we reject it
650 // We also reject keys that are marked transient at this point
651 bool xkeyMatch = false;
652 if( (!stream.empty() and stream != m_outputName) || SG::isTransientKey(proxyName) ) {
653 // reject keys that are marked for a different output stream
654 ATH_MSG_DEBUG("Rejecting key: {} in output: {}", itemProxy->name(), m_outputName.toString());
655 xkeyMatch = true;
656 }
657
658 // All right, it passes key match find in itemList, but not in excludeList
659 if (keyMatch && !xkeyMatch) {
660 if (m_forceRead && itemProxy->isValid()) {
661 if (nullptr == itemProxy->accessData()) {
662 ATH_MSG_ERROR(" Could not get data object for id {},\"{}\"", remapped_item_id, proxyName);
663 }
664 }
665 if (nullptr != itemProxy->object()) {
666 if( std::find(m_objects.begin(), m_objects.end(), itemProxy->object()) == m_objects.end() &&
667 std::find(m_altObjects.begin(), m_altObjects.end(), itemProxy->object()) == m_altObjects.end() )
668 {
669 if( item_id != remapped_item_id ) {
670 // For MetaCont<T>: -
671 // create a temporary DataObject for an entry in the container to pass to CnvSvc
672 DataBucketBase* dbb = static_cast<DataBucketBase*>( itemProxy->object() );
673 const MetaContBase* metaCont = static_cast<MetaContBase*>( dbb->cast( ClassID_traits<MetaContBase>::ID() ) );
674 void* obj = metaCont? metaCont->getAsVoid( m_outSeqSvc->currentRangeID(ctx) ) : nullptr;
675 if( obj ) {
676 auto altbucket = std::make_unique<AltDataBucket>(
677 obj, item_id, *CLIDRegistry::CLIDToTypeinfo(item_id), proxyName );
678 m_objects.push_back( altbucket.get() );
679 m_ownedObjects.push_back( std::move(altbucket) );
680 m_altObjects.push_back( itemProxy->object() ); // only for duplicate prevention
681 } else {
682 ATH_MSG_ERROR("Failed to retrieve object from MetaCont with key={}, for EventRangeID={}",
683 item_key, m_outSeqSvc->currentRangeID(ctx));
684 return StatusCode::FAILURE;
685 }
686 } else if (item.exact()) {
687 // If the exact flag is set, make a new DataObject
688 // holding the object as the requested type.
689 DataBucketBase* dbb = dynamic_cast<DataBucketBase*> (itemProxy->object());
690 if (!dbb) std::abort();
691 void* ptr = dbb->cast (item_id);
692 if (!ptr) {
693 // Hard cast
694 ptr = dbb->object();
695 }
696 auto altbucket =
697 std::make_unique<AltDataBucket>
698 (ptr, item_id,
700 *itemProxy);
701 m_objects.push_back(altbucket.get());
702 m_ownedObjects.push_back (std::move(altbucket));
703 m_altObjects.push_back (itemProxy->object());
704 }
705 else
706 m_objects.push_back(itemProxy->object());
707 ATH_MSG_DEBUG(" Added object {},\"{}\"", item_id, proxyName);
708 }
709
714 item_key.find( RootAuxDynIO::AUX_POSTFIX ) == ( item_key.size() - 4 )) {
715
716 const SG::IConstAuxStore* auxstore( nullptr );
717 try {
718 SG::fromStorable( itemProxy->object(), auxstore, true );
719 } catch( const std::exception& ) {
720 ATH_MSG_DEBUG("Error in casting object with CLID {} to SG::IConstAuxStore*", itemProxy->clID());
721 auxstore = nullptr;
722 }
723
724 if (auxstore) {
725 handleVariableSelection (*auxstore, *itemProxy,
726 aux_attr, vetoes);
727
728 // Here comes the compression logic using ThinningInfo
729 // Get a hold of all AuxIDs for this store (static, dynamic etc.)
730 const SG::auxid_set_t allVars = auxstore->getAuxIDs();
731
732 // Get a handle on the compression information for this store
733 std::string key = item_key;
734 key.erase (key.size()-4, 4);
735
736 // Build the compression list, retrieve the relevant AuxIDs and
737 // store it in the relevant map that is going to be inserted into
738 // the ThinningCache later on by the ThinningCacheTool
739 xAOD::AuxCompression compression;
740 compression.setCompressedAuxIDs( comp_attr_map );
741 for( const auto& it : compression.getCompressedAuxIDs( allVars ) ) {
742 if( it.second.size() > 0 ) { // insert only if the set is non-empty
743 compInfo[ key ][ it.first ] = it.second;
744 ATH_MSG_DEBUG("Container {} has {} variables that'll be "
745 "lossy float compressed with {} mantissa bits",
746 key, it.second.size(), it.first);
747 }
748 } // End of loop over variables to be lossy float compressed
749 } // End of lossy float compression logic
750
751 }
752
753 added = true;
754 }
755 } else if (keyMatch && xkeyMatch) {
756 removed = true;
757 }
758 } // proxy loop
759 if (!added && !removed) {
760 ATH_MSG_DEBUG(" No object matching {},\"{}\" found", item_id, item_key);
761 } else if (removed) {
762 ATH_MSG_DEBUG(" Object being excluded based on property setting {},\"{}\". Skipping",
763 item_id, item_key);
764 }
765 } else {
766 ATH_MSG_DEBUG(" Failed to receive proxy iterators from StoreGate for {},\"{}\". Skipping",
767 item_id, item_key);
768 }
769 return StatusCode::SUCCESS;
770}
#define ATH_MSG_DEBUG(x,...)
#define ATH_MSG_ERROR(x,...)
uint32_t CLID
The Class ID type.
IDataSelector m_objects
Collection of objects being selected.
UnsignedIntegerProperty m_compressionBitsLow
Number of mantissa bits in the float compression.
UnsignedIntegerProperty m_compressionBitsHigh
Number of mantissa bits in the float compression.
ServiceHandle< MetaDataSvc > m_metaDataSvc
Handles to all the necessary services.
StringProperty m_outputName
Name of the output file.
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.
void handleVariableSelection(const SG::IConstAuxStore &auxstore, SG::DataProxy &itemProxy, const std::string &aux_attr, SG::SelectionVetoes &vetoes) const
Here we build the vetoed AuxIDs.
IDataSelector m_altObjects
Objects overridden by `exact' handling.
BooleanProperty m_forceRead
set to true to force read of data objects in item list
std::vector< std::unique_ptr< DataObject > > m_ownedObjects
Collection of DataObject instances owned by this service.
ServiceHandle< OutputStreamSequencerSvc > m_outSeqSvc
bool simpleMatch(const std::string &pattern, const std::string &text)
Glob-style matcher, where the only meta-character is '*'.
static const std::type_info * CLIDToTypeinfo(CLID clid)
Translate between CLID and type_info.
virtual void * object()=0
T * cast(SG::IRegisterTransient *irt=0, bool isConst=true)
Return the contents of the DataBucket, converted to type T.
virtual void * getAsVoid(const SourceID &sid) const =0
virtual StoreID::type storeID() const override
Definition TestStore.cxx:32
virtual SG::DataProxy * proxy(const void *const pTransient) const override
CLID clID() const
Retrieve clid.
virtual const name_type & name() const override final
Retrieve data object key == string.
bool isValid() const
called by destructor
DataObject * accessData()
Access DataObject on-demand using conversion service.
bool hasAlias(const std::string &key) const
Test to see if a given string is in the alias set.
CLID id() const
const std::string & key() const
bool exact() const
@ EVENT_STORE
Definition StoreID.h:26
@ METADATA_STORE
Definition StoreID.h:29
virtual std::map< unsigned int, SG::auxid_set_t > getCompressedAuxIDs(const SG::auxid_set_t &fullset) const
Return those variables that are selected to be compressed per compression setting.
virtual void setCompressedAuxIDs(const std::map< unsigned int, std::set< std::string > > &attributes)
Set which variables should be compressed per compression setting.
bool match(std::string s1, std::string s2)
match the individual directories of two strings
Definition hcg.cxx:359
constexpr char AUX_POSTFIX[]
Common post-fix for the names of auxiliary containers in StoreGate.
TestStore store
Definition TestStore.cxx:23
bool fromStorable(DataObject *pDObj, T *&pTrans, bool quiet=false, IRegisterTransient *irt=0, bool isConst=true)
void * ptr(T *p)
Definition SGImplSvc.cxx:75

◆ 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 778 of file AthenaOutputStream.cxx.

781{
782 // Create an empty result
783 std::set<std::string> result;
784
785 // Check the item is indeed Aux.
786 if(item_key.find("Aux.") == std::string::npos) {
787 return result;
788 }
789
790 // First the high compression list
791 for (const auto& iter : *handle) {
792 // First match the IDs for early rejection.
793 if (iter.id() != item_id) {
794 continue;
795 }
796 // Then find the compression item key and the compression list string
797 size_t seppos = iter.key().find('.');
798 std::string comp_item_key{""}, comp_str{""};
799 if(seppos != std::string::npos) {
800 comp_item_key = iter.key().substr(0, seppos+1);
801 comp_str = iter.key().substr(seppos+1);
802 } else {
803 comp_item_key = iter.key();
804 }
805 // Proceed only if the keys match and the
806 // compression list string is not empty
807 if (!comp_str.empty() && comp_item_key == item_key) {
808 std::stringstream ss(comp_str);
809 std::string attr;
810 while( std::getline(ss, attr, '.') ) {
811 result.insert(attr);
812 }
813 }
814 }
815
816 // All done, return the result
817 return result;
818}
static Double_t ss
virtual void handle(const Incident &incident) override
Incident service handle listening for MetaDataStop.

◆ clearSelection()

void AthenaOutputStream::clearSelection ( )

Clear list of selected objects.

Definition at line 528 of file AthenaOutputStream.cxx.

528 {
529 m_objects.clear();
530 m_ownedObjects.clear();
531 m_altObjects.clear();
532}

◆ collectAllObjects()

StatusCode AthenaOutputStream::collectAllObjects ( const EventContext & ctx,
StoreGateSvc & store )

Collect data objects for output streamer list.

Definition at line 535 of file AthenaOutputStream.cxx.

535 {
536 if (m_itemListFromTool) {
537 if (!m_streamer->getInputItemList(&*m_p2BWritten).isSuccess()) {
538 ATH_MSG_WARNING("collectAllObjects() could not get ItemList from Tool.");
539 }
540 }
541
542 // This holds the vetoes for the AuxID selection
543 auto vetoes = std::make_unique<SG::SelectionVetoes>();
544 // This holds the lossy float compression information
545 auto compInfo = std::make_unique<SG::CompressionInfo>();
546
547 m_p2BWritten->updateItemList(true);
548 // Collect all objects that need to be persistified:
549 for (const auto& i : *m_p2BWritten) {
550 ATH_CHECK( addItemObjects(ctx, store, i, *vetoes, *compInfo) );
551 }
552
553 // If there were any variable selections, record the information in SG.
554 if (!vetoes->empty()) {
555 ATH_CHECK( SG::makeHandle (m_selVetoesKey, ctx).record (std::move (vetoes)) );
556 }
557
558 // Store the lossy float compression information in the SG.
559 if (!compInfo->empty()) {
560 ATH_CHECK( SG::makeHandle (m_compInfoKey, ctx).record (std::move (compInfo)) );
561 }
562
563 return StatusCode::SUCCESS;
564}
#define ATH_CHECK
Evaluate an expression and check for errors.
#define ATH_MSG_WARNING(x,...)
BooleanProperty m_itemListFromTool
Set to write out everything in input DataHeader.
SG::WriteHandleKey< SG::CompressionInfo > m_compInfoKey
Key used for recording lossy float compressed variable information to the event store.
SG::WriteHandleKey< SG::SelectionVetoes > m_selVetoesKey
Key used for recording selected dynamic variable information to the event store.
StatusCode addItemObjects(const EventContext &, StoreGateSvc &store, const SG::FolderItem &, SG::SelectionVetoes &vetoes, SG::CompressionInfo &compInfo)
Add item data objects to output streamer list.
SG::ReadCondHandle< T > makeHandle(const SG::ReadCondHandleKey< T > &key, const EventContext &ctx=Gaudi::Hive::currentContext())

◆ compressionListHandlerHigh()

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

Handler for ItemNames Property.

Definition at line 877 of file AthenaOutputStream.cxx.

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

◆ compressionListHandlerLow()

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

Handler for ItemNames Property.

Definition at line 886 of file AthenaOutputStream.cxx.

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

◆ execute()

StatusCode AthenaOutputStream::execute ( const EventContext & ctx)
overridevirtual

Definition at line 404 of file AthenaOutputStream.cxx.

404 {
405 bool failed = false;
406 // Call tool preExecute prior to writing
407 for (auto& tool : m_helperTools) {
408 if (!tool->preExecute().isSuccess()) {
409 failed = true;
410 }
411 }
412 // Write the event if the event is accepted
413 if (isEventAccepted(ctx)) {
414 if (write(ctx).isFailure()) {
415 failed = true;
416 }
417 }
418 // Call tool postExecute after writing
419 for (auto& tool : m_helperTools) {
420 if(!tool->postExecute().isSuccess()) {
421 failed = true;
422 }
423 }
424 // See if we should write metadata and do if so
426 writeMetaData(ctx);
428 // finalize will disconnect output
429 if( !finalize().isSuccess() ) {
430 failed = true;
431 }
432 }
433 if (failed) {
434 return(StatusCode::FAILURE);
435 }
436 return(StatusCode::SUCCESS);
437}
virtual StatusCode finalize() override
ToolHandleArray< IAthenaOutputTool > m_helperTools
vector of AlgTools that that are executed by this stream
virtual StatusCode write(const EventContext &ctx)
Stream the data from the event store.
void writeMetaData(const EventContext &ctx, const std::string &outputFN="")
Write MetaData for this stream (by default) or for a substream outputFN (in ES mode).

◆ finalize()

StatusCode AthenaOutputStream::finalize ( )
overridevirtual

Definition at line 376 of file AthenaOutputStream.cxx.

377{
378 bool failed = false;
379 ATH_MSG_DEBUG("finalize: Optimize output");
380 // Connect the output file to the service
381 if (!m_streamer->finalizeOutput().isSuccess()) {
382 failed = true;
383 }
384 ATH_MSG_DEBUG("finalize: end optimize output");
385 // Release the tools
386 if (!m_helperTools.release().isSuccess()) {
387 failed = true;
388 }
389 if (!m_streamer.release().isSuccess()) {
390 failed = true;
391 }
392 if (failed) {
393 return(StatusCode::FAILURE);
394 }
395 // Clear the internal caches
396 m_objects.clear();
397 m_objects.shrink_to_fit();
398 m_ownedObjects.clear();
399 m_altObjects.clear();
400 return(StatusCode::SUCCESS);
401}

◆ finalizeRange()

void AthenaOutputStream::finalizeRange ( const EventContext & ctx,
const std::string & rangeFN )
private

Definition at line 298 of file AthenaOutputStream.cxx.

299{
300 ATH_MSG_DEBUG("Writing MetaData to {}", rangeFN);
301 // MN: not calling StopMetaData Incident here but directly writeMetaData() - OK for Sim, check others
302 // metadata tools like CutFlowSvc are not able to handle this yet
303 const std::string rememberID = m_outSeqSvc->setRangeID( ctx, m_rangeIDforRangeFN[ rangeFN ] );
304 writeMetaData( ctx, rangeFN );
305 m_outSeqSvc->setRangeID( ctx, rememberID );
306
307 ATH_MSG_INFO("Finished writing Event Sequence to {}", rangeFN);
308 auto strm_iter = m_streamerMap.find( rangeFN );
309 strm_iter->second->finalizeOutput().ignore();
310 strm_iter->second->finalize().ignore();
311 m_streamerMap.erase( strm_iter );
312 m_outSeqSvc->publishRangeReport( rangeFN );
313}
#define ATH_MSG_INFO(x,...)
std::map< std::string, std::string > m_rangeIDforRangeFN
map of RangeIDs (as used by the Sequencer) for each Range filename generated

◆ handle()

void AthenaOutputStream::handle ( const Incident & incident)
overridevirtual

Incident service handle listening for MetaDataStop.

Definition at line 223 of file AthenaOutputStream.cxx.

224{
225 ATH_MSG_DEBUG("handle() incident type: {}", inc.type());
226 // mutex shared with write() which is called from writeMetaData
227 std::unique_lock<mutex_t> lock(m_mutex);
228
229 if( inc.type() == "MetaDataStop" ) {
230 if( m_outSeqSvc->inUse() ) {
231 if( m_outSeqSvc->inConcurrentEventsMode() ) {
232 // EventService MT - write metadata and close all remaining substreams
233 while( m_streamerMap.size() > 0 ) {
234 finalizeRange( inc.context(), m_streamerMap.begin()->first );
235 }
236 return;
237 }
238 if( m_outSeqSvc->lastIncident() == "EndEvent" ) {
239 // in r22 EndEvent comes before output writing
240 // - queue metadata writing and disconnect for after Event write
242 return;
243 }
244 }
245 // not in Event Service
246 writeMetaData(inc.context());
247 }
248 else if( m_outSeqSvc->inUse() ) {
249 // Handle Event Ranges for Event Service
250 EventContext::ContextID_t slot = inc.context().slot();
251 if( slot == EventContext::INVALID_CONTEXT_ID ) {
252 throw GaudiException("Received Incident with invalid slot in ES mode", name(), StatusCode::FAILURE);
253 }
254 auto count_events_in_range = [&](const std::string& range) {
255 return std::count_if(m_slotRangeMap.cbegin(), m_slotRangeMap.cend(),
256 [&](auto& el){return el.second == range;} );
257 };
258 if( inc.type() == IncidentType::BeginProcessing ) {
259 // get the current/old range filename for this slot
260 const std::string rangeFN = m_slotRangeMap[ slot ];
261 // build the new range filename for this slot
262 const std::string newRangeFN = m_outSeqSvc->buildSequenceFileName(inc.context(), m_outputName );
263 if( !rangeFN.empty() and rangeFN != newRangeFN ) {
264 ATH_MSG_INFO("Slot range change: '{}' -> '{}'", rangeFN, newRangeFN);
265 ATH_MSG_DEBUG("There are {} slots in use",m_slotRangeMap.size());
266 for(const auto & range : m_slotRangeMap ) {
267 ATH_MSG_DEBUG("Slot: {} FN={}", range.first, range.second);
268 }
269 if( count_events_in_range(rangeFN) == 1 ) {
270 finalizeRange( inc.context(), rangeFN );
271 }
272 }
273 ATH_MSG_INFO("slot {} processing event in range: {}", slot, newRangeFN);
274 m_slotRangeMap[ slot ] = newRangeFN;
275 // remember the RangeID for this slot so we can write metadata *after* a range change
276 m_rangeIDforRangeFN[ newRangeFN ] = m_outSeqSvc->currentRangeID(inc.context());
277 }
278 else if( inc.type() == IncidentType::EndProcessing ) {
279 ATH_MSG_DEBUG("There are {} slots in use", m_slotRangeMap.size());
280 for( const auto& range : m_slotRangeMap ) {
281 ATH_MSG_DEBUG("Slot: {} FN={}", range.first, range.second);
282 }
283 if( m_slotRangeMap.size() > 1 ) {
284 // if there are multiple slots, we can detect if the range ended with this event
285 // - except the last range, because there is no next range to clear the slot map
286 const std::string rangeFN = m_slotRangeMap[ slot ];
287 if( count_events_in_range(rangeFN) == 1 ) {
288 finalizeRange( inc.context(), rangeFN );
289 m_slotRangeMap[ slot ].clear();
290 }
291 }
292 }
293 }
294 ATH_MSG_DEBUG("Leaving incident handler for {}", inc.type());
295}
virtual void lock()=0
Interface to allow an object to lock itself when made const in SG.
std::map< unsigned, std::string > m_slotRangeMap
map of filenames assigned to active slots
void finalizeRange(const EventContext &ctx, const std::string &rangeFN)
mutex_t m_mutex
mutex for this Stream write() and handle() methods

◆ handleVariableSelection()

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

Here we build the vetoed AuxIDs.

Definition at line 821 of file AthenaOutputStream.cxx.

825{
826 // Collect dynamic Aux selection (parse the line, attributes separated by dot)
827 std::set<std::string> attributes;
828 if( aux_attr.size() ) {
829 std::stringstream ss(aux_attr);
830 std::string attr;
831 while( std::getline(ss, attr, '.') ) {
832 attributes.insert(attr);
833 }
834 }
835
836 // Return early if there's no selection.
837 if (attributes.empty()) {
838 return;
839 }
840
841 std::string key = itemProxy.name();
842 if (key.size() >= 4 && key.compare (key.size()-4, 4, "Aux.")==0)
843 {
844 key.erase (key.size()-4, 4);
845 }
846
847 // Find the entry for the selection.
848 SG::auxid_set_t& vset = vetoes[key];
849
850 // Form the veto mask for this object.
851 xAOD::AuxSelection sel;
852 sel.selectAux (attributes);
853
854 // Get all the AuxIDs that we know of and the selected ones
855 SG::auxid_set_t all = auxstore.getAuxIDs();
856 SG::auxid_set_t selected = sel.getSelectedAuxIDs( all );
857
858 // Loop over all and build a list of vetoed AuxIDs from non selected ones
859 for( const SG::auxid_t auxid : all ) {
860 if ( !selected.test( auxid ) ) {
861 vset.insert( auxid );
862 }
863 }
864}
size_t auxid_t
Identifier for a particular aux data item.
Definition AuxTypes.h:27

◆ initialize()

StatusCode AthenaOutputStream::initialize ( )
overridevirtual

Definition at line 72 of file AthenaOutputStream.cxx.

72 {
73 ATH_MSG_DEBUG("In initialize");
74
75 // Initialize the FilteredAlgorithm base
77
78 // Reset the number of events written
79 m_events = 0;
80
81 // Set up the SG services
82 ATH_CHECK( m_dataStore.retrieve() );
83 ATH_MSG_DEBUG("Found {} store.", m_dataStore.typeAndName());
84 if (!m_metadataItemList.value().empty()) {
85 ATH_CHECK( m_metadataStore.retrieve() );
86 ATH_MSG_DEBUG("Found {} store.", m_metadataStore.typeAndName());
87 }
88
89 // Set up various services
90 ATH_CHECK( m_pCLIDSvc.retrieve() );
91 ATH_CHECK( m_dictLoader.retrieve() );
92 ATH_CHECK( m_tpCnvSvc.retrieve() );
93 ATH_CHECK( m_outSeqSvc.retrieve() );
94 ATH_CHECK( m_metaDataSvc.retrieve() );
95
96 // Get Output Stream tool for writing
97 ATH_CHECK( m_streamer.retrieve() );
98 ATH_CHECK( m_streamer->connectServices(m_dataStore.typeAndName(), m_persName, m_extendProvenanceRecord) );
99
100 ATH_CHECK( m_helperTools.retrieve() );
101 ATH_MSG_INFO("Found " << m_helperTools);
102 ATH_MSG_INFO("Data output: {}", m_outputName.toString());
103
104 for (auto& tool : m_helperTools) {
105 ATH_CHECK( tool->postInitialize() );
106 }
107
108 // Register this algorithm for 'I/O' events
109 ServiceHandle<IIoComponentMgr> iomgr("IoComponentMgr", name());
110 ATH_CHECK( iomgr.retrieve() );
111 ATH_CHECK( iomgr->io_register(this) );
112 ATH_CHECK( iomgr->io_register(this, IIoComponentMgr::IoMode::WRITE, m_outputName) );
113 ATH_CHECK( this->io_reinit() );
114
115 // Add an explicit input dependency for everything in our item list
116 // that we know from the configuration is in the transient store.
117 // We don't want to add everything on the list, because configurations
118 // often initialize this with a maximal static list of everything
119 // that could possibly be written.
120 {
121 ATH_CHECK( m_transient.retrieve() );
122 IProperty *pAsIProp = dynamic_cast<IProperty*> (&*m_transient);
123 if (!pAsIProp) {
124 ATH_MSG_FATAL ("Bad folder interface");
125 return StatusCode::FAILURE;
126 }
127 ATH_CHECK (pAsIProp->setProperty("ItemList", m_transientItems.toString()));
128
129 for (const SG::FolderItem& item : *m_p2BWritten) {
130 // Load ROOT dictionaries now.
131 loadDict (item.id());
132
133 const std::string& k = item.key();
134 if (k.find('*') != std::string::npos) continue;
135 if (k.find('.') != std::string::npos) continue;
136 for (const SG::FolderItem& titem : *m_transient) {
137 if (titem.id() == item.id() && titem.key() == k) {
138 DataObjID id (item.id(), std::format("{}+{}", m_dataStore.name(), k));
139 this->addDependency (id, Gaudi::DataHandle::Reader);
140 break;
141 }
142 }
143 }
144 m_transient->clear();
145 }
146
147 // Also load dictionaries for metadata classes.
148 if (!m_metadataItemList.value().empty()) {
149 IProperty *pAsIProp = dynamic_cast<IProperty*> (&*m_transient);
150 if (!pAsIProp) {
151 ATH_MSG_FATAL ("Bad folder interface");
152 return StatusCode::FAILURE;
153 }
154 ATH_CHECK (pAsIProp->setProperty("ItemList", m_metadataItemList.toString()));
155 for (const SG::FolderItem& item : *m_transient) {
156 loadDict (item.id());
157 }
158 m_transient->clear();
159 }
160
161 // Also make sure we have the dictionary for Token.
162 m_dictLoader->load_type ("Token");
163
164 // Listen to event range incidents if incident name is configured
165 ATH_CHECK( m_incidentSvc.retrieve() );
166 if( !m_outSeqSvc->incidentName().empty() ) {
167 // use priority 95 to make sure the Output Sequencer goes first (it has priority 100)
168 m_incidentSvc->addListener(this, IncidentType::BeginProcessing, 95);
169 m_incidentSvc->addListener(this, IncidentType::EndProcessing, 95);
170 }
171
172 // Check compression settings and print some information about the configuration
173 // Both should be between [5, 23] and high compression should be < low compression
175 ATH_MSG_INFO("Float compression mantissa bits for high compression "
176 "({}) is outside the allowed range of [5, 23].",
177 m_compressionBitsHigh.toString());
178 ATH_MSG_INFO("Setting it to the appropriate limit.");
180 }
182 ATH_MSG_INFO("Float compression mantissa bits for low compression "
183 "({}) is outside the allowed range of [5, 23].",
184 m_compressionBitsLow.toString());
185 ATH_MSG_INFO("Setting it to the appropriate limit.");
187 }
189 ATH_MSG_ERROR("Float compression mantissa bits for low compression "
190 "({}) is lower than or equal to high compression "
191 "({})! Please check the configuration! ",
192 m_compressionBitsLow.toString(),
193 m_compressionBitsHigh.toString());
194 return StatusCode::FAILURE;
195 }
196 if(m_compressionListHigh.value().empty() && m_compressionListLow.value().empty()) {
197 ATH_MSG_VERBOSE("Both high and low float compression lists are empty. Float compression will NOT be applied.");
198 } else {
199 ATH_MSG_INFO("Either high or low (or both) float compression lists are defined. Float compression will be applied.");
200 ATH_MSG_INFO("High compression will use {} mantissa bits, and "
201 "low compression will use {} mantissa bits.",
202 m_compressionBitsHigh.toString(),
203 m_compressionBitsLow.toString());
204 }
205
206 // Setup stream name
207 if (m_streamName.empty()) {
208 m_streamName.setValue(this->name());
209 }
210
211 // Set SG key for selected variable information.
212 m_selVetoesKey = std::format("SelectionVetoes_{}", m_streamName.toString());
213 ATH_CHECK( m_selVetoesKey.initialize() );
214
215 m_compInfoKey = std::format("CompressionInfo_{}", m_streamName.toString());
216 ATH_CHECK( m_compInfoKey.initialize() );
217
218 ATH_MSG_DEBUG("End initialize");
219 return StatusCode::SUCCESS;
220}
#define ATH_MSG_VERBOSE(x,...)
#define ATH_MSG_FATAL(x,...)
ServiceHandle< StoreGateSvc > m_metadataStore
StringProperty m_streamName
Stream name (defaults to algorithm name).
StringArrayProperty m_transientItems
List of items that are known to be present in the transient store (and hence we can make input depend...
ServiceHandle< IClassIDSvc > m_pCLIDSvc
ServiceHandle< IIncidentSvc > m_incidentSvc
ServiceHandle< IDictLoaderSvc > m_dictLoader
StringProperty m_persName
Name of the persistency service capable to write data from the store.
void loadDict(CLID clid)
Helper function to load dictionaries (both transient and persistent) for a given type.
ServiceHandle< ITPCnvSvc > m_tpCnvSvc
StringArrayProperty m_metadataItemList
Vector of item names.
ServiceHandle< StoreGateSvc > m_dataStore
Handle to the StoreGateSvc store where the data we want to write out resides.
virtual StatusCode io_reinit() override
Callback method to reinitialize the internal state of the component for I/O purposes (e....
BooleanProperty m_extendProvenanceRecord
Set to false to omit adding the current DataHeader into the DataHeader history This will cause the in...
std::atomic< int > m_events
Number of events written to this output stream.
virtual StatusCode initialize() override

◆ io_finalize()

StatusCode AthenaOutputStream::io_finalize ( )
overridevirtual

Definition at line 911 of file AthenaOutputStream.cxx.

911 {
912 ATH_MSG_INFO("I/O finalization...");
913 for (auto& tool : m_helperTools) {
914 if (!tool->preFinalize().isSuccess()) {
915 ATH_MSG_ERROR("Cannot finalize helper tool");
916 }
917 }
918 const Incident metaDataStopIncident(name(), "MetaDataStop");
919 this->handle(metaDataStopIncident);
920 m_incidentSvc->removeListener(this, "MetaDataStop");
921 if (m_dataStore->clearStore().isFailure()) {
922 ATH_MSG_WARNING("Cannot clear the DataStore");
923 }
924 return StatusCode::SUCCESS;
925}

◆ 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 897 of file AthenaOutputStream.cxx.

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

◆ itemListHandler()

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

Handler for ItemNames Property.

Definition at line 867 of file AthenaOutputStream.cxx.

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

◆ 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 932 of file AthenaOutputStream.cxx.

933{
934 m_dictLoader->load_type (clid);
935
936 // Also load the persistent class dictionary, if applicable.
937 std::unique_ptr<ITPCnvBase> tpcnv = m_tpCnvSvc->t2p_cnv_unique (clid);
938 if (tpcnv) {
939 m_dictLoader->load_type (tpcnv->persistentTInfo());
940 }
941}

◆ selectedObjects()

IDataSelector * AthenaOutputStream::selectedObjects ( )
inline

Return the list of selected objects.

Definition at line 84 of file AthenaOutputStream.h.

84 {
85 return &m_objects;
86 }

◆ 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 944 of file AthenaOutputStream.cxx.

945 {
946 size_t pi = 0, ti = 0, star = std::string::npos, match = 0;
947 while (ti < text.size()) {
948 if (pi < pattern.size() && (pattern[pi] == text[ti] || pattern[pi] == '*')) {
949 if (pattern[pi] == '*') { star = pi++; match = ti; }
950 else { ++pi; ++ti; }
951 } else if (star != std::string::npos) {
952 pi = star + 1;
953 ti = ++match;
954 } else return false;
955 }
956 while (pi < pattern.size() && pattern[pi] == '*') ++pi;
957 return pi == pattern.size();
958}
#define pi

◆ write() [1/2]

StatusCode AthenaOutputStream::write ( const EventContext & ctx)
virtual

Stream the data from the event store.

Definition at line 440 of file AthenaOutputStream.cxx.

440 {
441 return write(ctx, *m_dataStore);
442}

◆ write() [2/2]

StatusCode AthenaOutputStream::write ( const EventContext & ctx,
StoreGateSvc & store )

Stream the data from the given store.

Definition at line 444 of file AthenaOutputStream.cxx.

444 {
445 bool failed = false;
446 IAthenaOutputStreamTool* streamer = &*m_streamer;
447 std::string outputFN;
448
449 std::unique_lock<mutex_t> lock(m_mutex);
450 outputFN = m_outSeqSvc->buildSequenceFileName( ctx, m_outputName );
451
452 // Handle Event Ranges
453 if( m_outSeqSvc->inUse() and m_outSeqSvc->inConcurrentEventsMode() ) {
454 ATH_MSG_DEBUG("Writing event sequence to {}", outputFN);
455 streamer = m_streamerMap[ outputFN ].get();
456 if( !streamer ) {
457 // new range, needs a new streamer tool
458 IAlgTool* st = AlgTool::Factory::create( m_streamer->type(), m_streamer->type(), m_streamer->name(), this ).release();
459 st->addRef();
460 streamer = dynamic_cast<IAthenaOutputStreamTool*>( st );
461 IProperty *mstreamer_props = dynamic_cast<IProperty*> (&*m_streamer);
462 IProperty *streamer_props = dynamic_cast<IProperty*> (&*streamer);
463 if (!mstreamer_props || !streamer_props) {
464 ATH_MSG_FATAL("Cannot cast streamer to IProperty");
465 return StatusCode::FAILURE;
466 }
467 for ( const auto& prop : mstreamer_props->getProperties() ) {
468 ATH_CHECK( streamer_props->setProperty( *prop ) );
469 }
470 if( !streamer or streamer->initialize().isFailure()
471 or streamer->connectServices(m_dataStore.typeAndName(), m_persName, m_extendProvenanceRecord).isFailure() ) {
472 ATH_MSG_FATAL("Unable to initialize OutputStreamTool for {}", outputFN);
473 return StatusCode::FAILURE;
474 }
475 m_streamerMap[ outputFN ].reset( streamer );
476 }
477 }
478
479 // Clear any previously existing item list
480 // and collect all objects that are asked to be written out
482 ATH_CHECK( collectAllObjects(ctx, store) );
483
484 // keep a local copy of the object lists so they are not overwritten when we release the lock
485 IDataSelector objects = std::move( m_objects );
486 IDataSelector altObjects = std::move( m_altObjects );
487 std::vector<std::unique_ptr<DataObject> > ownedObjects = std::move( m_ownedObjects );
488
489 // prepare before releasing lock because m_outputAttributes change in metadataStop
490 const std::string connectStr = outputFN + m_outputAttributes;
491
492 for (auto& tool : m_helperTools) {
493 ATH_CHECK( tool->preStream() );
494 }
495
496 // MN: would be nice to release the Stream lock here
497 // lock.unlock();
498
499 // Connect the output file to the service
500 if (!streamer->connectOutput(connectStr).isSuccess()) {
501 ATH_MSG_FATAL("Could not connectOutput");
502 return StatusCode::FAILURE;
503 }
504 ATH_MSG_DEBUG("connectOutput done for {}", outputFN);
505 StatusCode currentStatus = streamer->streamObjects(objects, connectStr);
506 // Do final check of streaming
507 if (!currentStatus.isSuccess()) {
508 if (!currentStatus.isRecoverable()) {
509 ATH_MSG_FATAL("streamObjects failed.");
510 failed = true;
511 } else {
512 ATH_MSG_DEBUG("streamObjects failed.");
513 }
514 }
515 bool doCommit = false;
516 if (!streamer->commitOutput(doCommit).isSuccess()) {
517 ATH_MSG_FATAL("commitOutput failed.");
518 failed = true;
519 }
520 if (failed) {
521 return(StatusCode::FAILURE);
522 }
523 m_events++;
524 return(StatusCode::SUCCESS);
525}
void clearSelection()
Clear list of selected objects.
std::string m_outputAttributes
Output attributes.
StatusCode collectAllObjects(const EventContext &ctx, StoreGateSvc &store)
Collect data objects for output streamer list.
virtual StatusCode connectServices(const std::string &dataStore, const std::string &cnvSvc, bool extendProvenenceRecord=false)=0
Specify which data store and conversion service to use and whether to extend provenence Only use if o...
virtual StatusCode streamObjects(const TypeKeyPairs &typeKeys, const std::string &outputName="")=0
virtual StatusCode connectOutput(const std::string &outputName="")=0
Connect to the output stream Must connectOutput BEFORE streaming Only specify "outputName" if one wan...
virtual StatusCode commitOutput(bool doCommit=false)=0
Commit the output stream after having streamed out objects Must commitOutput AFTER streaming.
::StatusCode StatusCode
StatusCode definition for legacy code.

◆ writeMetaData()

void AthenaOutputStream::writeMetaData ( const EventContext & ctx,
const std::string & outputFN = "" )
private

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

Definition at line 318 of file AthenaOutputStream.cxx.

319{
320 // use main stream tool by default, or per outputFile in ES mode
321 IAthenaOutputStreamTool* streamer = outputFN.empty()? &*m_streamer : m_streamerMap[outputFN].get();
322
323 for (auto& tool : m_helperTools) {
324 if (!tool->preFinalize().isSuccess()) {
325 throw GaudiException("Cannot finalize helper tool", name(), StatusCode::FAILURE);
326 }
327 }
328 if( m_metaDataSvc->prepareOutput(outputFN).isFailure() ) {
329 throw GaudiException("Failed on MetaDataSvc prepareOutput", name(), StatusCode::FAILURE);
330 }
331 // lock all metadata to prevent updates during writing
332 MetaDataSvc::ToolLockGuard tool_guard( *m_metaDataSvc );
333
334 // Prepare the WriteDataHeaderForms incident
335 std::string DHFWriteIncidentfileName = m_outSeqSvc->buildSequenceFileName(ctx, m_outputName);
336 // remove technology from the name
337 size_t pos = DHFWriteIncidentfileName.find(':');
338 if( pos != std::string::npos ) DHFWriteIncidentfileName = DHFWriteIncidentfileName.substr(pos+1);
339 FileIncident incident(name(), "WriteDataHeaderForms", DHFWriteIncidentfileName);
340 m_incidentSvc->fireIncident(incident);
341
342 ATH_MSG_DEBUG("metadataItemList: {}", m_metadataItemList.value() );
343 if (!m_metadataItemList.value().empty()) {
344 StatusCode status = streamer->connectServices(m_metadataStore.typeAndName(), m_persName, false);
345 if (status.isFailure()) {
346 throw GaudiException("Unable to connect metadata services", name(), StatusCode::FAILURE);
347 }
348 m_outputAttributes = "[OutputCollection=MetaDataHdr][PoolContainerPrefix=MetaData][AttributeListKey=]";
349 m_p2BWritten->clear();
350 IProperty *pAsIProp(nullptr);
351 if ((m_p2BWritten.retrieve()).isFailure() ||
352 nullptr == (pAsIProp = dynamic_cast<IProperty*>(&*m_p2BWritten)) ||
353 (pAsIProp->setProperty("ItemList", m_metadataItemList.toString())).isFailure()) {
354 throw GaudiException("Folder property [metadataItemList] not found", name(), StatusCode::FAILURE);
355 }
356 if (write(ctx, *m_metadataStore).isFailure()) {
357 throw GaudiException("Cannot write metadata", name(), StatusCode::FAILURE);
358 }
359 FileIncident incident(name(), "WriteDataHeaderForms", DHFWriteIncidentfileName + m_outputAttributes);
360 m_incidentSvc->fireIncident(incident);
361
362 m_outputAttributes.clear();
364 if (status.isFailure()) {
365 throw GaudiException("Unable to re-connect services", name(), StatusCode::FAILURE);
366 }
367 m_p2BWritten->clear();
368 if ((pAsIProp->setProperty(m_itemList)).isFailure()) {
369 throw GaudiException("Folder property [itemList] not found", name(), StatusCode::FAILURE);
370 }
371 ATH_MSG_DEBUG("Metadata items written: {}", m_metadataItemList.value().size());
372 }
373}
status
Definition merge.py:16

Member Data Documentation

◆ m_altObjects

IDataSelector AthenaOutputStream::m_altObjects
protected

Objects overridden by `exact' handling.

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

220{ this, "CompInfoKey", "" };

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

129{this, "CompressionBitsHigh", 7, "Lossy float compression bits (high)"};

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

132{this, "CompressionBitsLow", 15, "Lossy float compression bits (low)"};

◆ m_compressionDecoderHigh

ToolHandle<SG::IFolder> AthenaOutputStream::m_compressionDecoderHigh
protected

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

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

123{this, "CompressionListHigh", {}, "Lossy float compression list (high)"};

◆ m_compressionListLow

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

Vector of item names.

Definition at line 126 of file AthenaOutputStream.h.

126{this, "CompressionListLow", {}, "Lossy float compression list (low)"};

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

98{this, "Store", "StoreGateSvc/StoreGateSvc", "Handle to event store"};

◆ m_dictLoader

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

Definition at line 103 of file AthenaOutputStream.h.

103{this, "AthDictLoaderSvc", "AthDictLoaderSvc"};

◆ m_events

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

Number of events written to this output stream.

Definition at line 187 of file AthenaOutputStream.h.

187{0};

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

149{this, "ExtendProvenanceRecord", true, "Extend provenance record"};

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

145{this, "ForceRead", true, "Force read data objects in ItemList"};

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

181{this, "HelperTools", {}, "List of AlgTools used by this stream"};

◆ m_incidentSvc

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

Definition at line 105 of file AthenaOutputStream.h.

105{this, "IncidentSvc", "IncidentSvc"};

◆ m_itemList

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

Vector of item names.

Definition at line 113 of file AthenaOutputStream.h.

113{this, "ItemList", {}, "List of items to write", "OutputStreamItemList"};

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

152{this, "TakeItemsFromInput", false, "Write everything in input DataHeader to output"};

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

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

◆ m_metadataItemList

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

Vector of item names.

Definition at line 116 of file AthenaOutputStream.h.

116{this, "MetadataItemList", {}, "List of metadata items to write","OutputStreamItemList"};

◆ m_metadataStore

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

Definition at line 99 of file AthenaOutputStream.h.

99{this, "MetadataStore", "StoreGateSvc/MetaDataStore", "Handle to metadata store"};

◆ m_metaDataSvc

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

Handles to all the necessary services.

Definition at line 102 of file AthenaOutputStream.h.

102{this, "MetaDataSvc", "MetaDataSvc"};

◆ m_mutex

mutex_t AthenaOutputStream::m_mutex
protected

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

Definition at line 200 of file AthenaOutputStream.h.

◆ m_objects

IDataSelector AthenaOutputStream::m_objects
protected

Collection of objects being selected.

Definition at line 167 of file AthenaOutputStream.h.

◆ m_outputAttributes

std::string AthenaOutputStream::m_outputAttributes
private

Output attributes.

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

139{this, "OutputFile", "DidNotNameOutput.root", "Name of the output file"};

◆ m_outSeqSvc

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

Definition at line 107 of file AthenaOutputStream.h.

107{this, "OutputStreamSequencerSvc", "OutputStreamSequencerSvc"};

◆ 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 175 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 155 of file AthenaOutputStream.h.

◆ m_pCLIDSvc

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

Definition at line 106 of file AthenaOutputStream.h.

106{this, "ClassIDSvc", "ClassIDSvc"};

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

142{this, "EvtConversionSvc", "EventPersistencySvc", "Name of the persistency service writing data"};

◆ 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 194 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 214 of file AthenaOutputStream.h.

215{ this, "SelVetoesKey", "" };

◆ m_slotRangeMap

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

map of filenames assigned to active slots

Definition at line 191 of file AthenaOutputStream.h.

◆ m_streamer

ToolHandle<IAthenaOutputStreamTool> AthenaOutputStream::m_streamer
protected

pointer to AthenaOutputStreamTool

Definition at line 178 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 197 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 110 of file AthenaOutputStream.h.

110{this, "StreamName", "", "Name of the output stream"};

◆ m_tpCnvSvc

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

Definition at line 104 of file AthenaOutputStream.h.

104{this, "AthTPCnvSvc", "AthTPCnvSvc"};

◆ m_transient

ToolHandle<SG::IFolder> AthenaOutputStream::m_transient
protected

Decoded list of transient ids.

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

136{this, "TransientItems", {}, "Transient item list"};

◆ m_writeMetadataAndDisconnect

bool AthenaOutputStream::m_writeMetadataAndDisconnect {false}
protected

Definition at line 184 of file AthenaOutputStream.h.

184{false};

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