AthenaOutputStream.h

Go to the documentation of this file.

// Dear emacs, this is -*- C++ -*-
 
/*
  Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
*/
 
#ifndef ATHENASERVICES_ATHENAOUTPUTSTREAM_H
#define ATHENASERVICES_ATHENAOUTPUTSTREAM_H
 
// STL include files
#include <map>
#include <memory>
#include <mutex>
#include <set>
#include <string>
#include <vector>
 
// Required for inheritance
#include "Gaudi/Property.h"
#include "GaudiKernel/IDataSelector.h"
#include "GaudiKernel/ServiceHandle.h"
#include "GaudiKernel/ToolHandle.h"
 
// Framework include files
#include "AthenaBaseComps/FilteredAlgorithm.h"
#include "AthenaKernel/IAthenaOutputStreamTool.h"
#include "AthenaKernel/IAthenaOutputTool.h"
#include "AthenaKernel/IDictLoaderSvc.h"
#include "AthenaKernel/ITPCnvSvc.h"
#include "GaudiKernel/IClassIDSvc.h"
#include "GaudiKernel/IIncidentListener.h"
#include "GaudiKernel/IIoComponent.h"
#include "StoreGate/WriteHandleKey.h"
 
// Local include files
#include "CompressionInfo.h"
#include "MetaDataSvc.h"
#include "OutputStreamSequencerSvc.h"
#include "SelectionVetoes.h"
 
// forward declarations
namespace SG {
   class DataProxy;
   class IFolder;
   class IAuxStoreIO;
   class FolderItem;
}
 
class AthenaOutputStream : public extends<FilteredAlgorithm,
                                          IIncidentListener, IIoComponent> {
 
public:
   typedef std::vector<SG::DataProxy*>     Items;
   typedef std::vector<std::pair<std::string, std::string> > TypeKeyPairs;
   typedef std::recursive_mutex mutex_t;
 
   AthenaOutputStream(const std::string& name, ISvcLocator* pSvcLocator);
 
   virtual ~AthenaOutputStream();
 
 
   virtual StatusCode initialize() override;
   virtual StatusCode finalize() override;
   virtual StatusCode execute() override;
 
   virtual StatusCode write();
 
   void clearSelection();
 
   StatusCode collectAllObjects();
 
   IDataSelector* selectedObjects() {
      return &m_objects;
   }
 
   virtual void handle(const Incident& incident) override;
 
   virtual StatusCode io_reinit() override;
   virtual StatusCode io_finalize() override;
 
protected:
   ServiceHandle<StoreGateSvc>  m_dataStore{this, "Store", "StoreGateSvc/StoreGateSvc", "Handle to event store"};
   ServiceHandle<StoreGateSvc>  m_metadataStore{this, "MetadataStore", "StoreGateSvc/MetaDataStore", "Handle to metadata store"};
   ServiceHandle<StoreGateSvc>* m_currentStore;
 
   ServiceHandle<MetaDataSvc>              m_metaDataSvc{this, "MetaDataSvc", "MetaDataSvc"};
   ServiceHandle<IDictLoaderSvc>           m_dictLoader{this, "AthDictLoaderSvc", "AthDictLoaderSvc"};
   ServiceHandle<ITPCnvSvc>                m_tpCnvSvc{this, "AthTPCnvSvc", "AthTPCnvSvc"};
   ServiceHandle<IIncidentSvc>             m_incidentSvc{this, "IncidentSvc", "IncidentSvc"};
   ServiceHandle<IClassIDSvc>              m_pCLIDSvc{this, "ClassIDSvc", "ClassIDSvc"};
   ServiceHandle<OutputStreamSequencerSvc> m_outSeqSvc{this, "OutputStreamSequencerSvc", "OutputStreamSequencerSvc"};
 
   StringProperty           m_streamName{this, "StreamName", "", "Name of the output stream"};
 
   StringArrayProperty      m_itemList{this, "ItemList", {}, "List of items to write", "OutputStreamItemList"};
 
   StringArrayProperty      m_metadataItemList{this, "MetadataItemList", {}, "List of metadata items to write","OutputStreamItemList"};
 
   StringProperty           m_keepProvenances {this, "KeepProvenanceTagsRegEx", {".*"},
                              "RegEx pattern to select processing tags for which DataHeader should retain provenances"};
 
   StringArrayProperty      m_compressionListHigh{this, "CompressionListHigh", {}, "Lossy float compression list (high)"};
 
   StringArrayProperty      m_compressionListLow{this, "CompressionListLow", {}, "Lossy float compression list (low)"};
 
   UnsignedIntegerProperty  m_compressionBitsHigh{this, "CompressionBitsHigh", 7, "Lossy float compression bits (high)"};
 
   UnsignedIntegerProperty  m_compressionBitsLow{this, "CompressionBitsLow", 15, "Lossy float compression bits (low)"};
 
   StringArrayProperty      m_transientItems{this, "TransientItems", {}, "Transient item list"};
 
   StringProperty           m_outputName{this, "OutputFile", "DidNotNameOutput.root", "Name of the output file"};
 
   StringProperty           m_persName{this, "EvtConversionSvc", "EventPersistencySvc", "Name of the persistency service writing data"};
 
   BooleanProperty m_forceRead{this, "ForceRead", true, "Force read data objects in ItemList"};
 
   BooleanProperty m_extendProvenanceRecord{this, "ExtendProvenanceRecord", true, "Extend provenance record"};
 
   BooleanProperty m_itemListFromTool{this, "TakeItemsFromInput", false, "Write everything in input DataHeader to output"};
 
   ToolHandle<SG::IFolder>  m_p2BWritten;
 
   ToolHandle<SG::IFolder>  m_compressionDecoderHigh;
 
   ToolHandle<SG::IFolder>  m_compressionDecoderLow;
 
   ToolHandle<SG::IFolder>  m_transient;
 
   std::multimap<CLID,std::string> m_CLIDKeyPairs;
 
   IDataSelector m_objects;
 
   IDataSelector m_altObjects;
 
   std::vector<std::unique_ptr<DataObject> > m_ownedObjects;
 
   ToolHandle<IAthenaOutputStreamTool> m_streamer;
 
   ToolHandleArray<IAthenaOutputTool> m_helperTools{this, "HelperTools", {}, "List of AlgTools used by this stream"};
 
   // flag set by MetaDataStop if OutputSequencer is used with EndEvent
   bool m_writeMetadataAndDisconnect{false};
 
   std::atomic<int> m_events{0};
 
   // ------- Event Ranges handling in MT -------
   std::map< unsigned, std::string >  m_slotRangeMap;
 
   std::map< std::string, std::string > m_rangeIDforRangeFN;
 
   std::map< std::string, std::unique_ptr<IAthenaOutputStreamTool> > m_streamerMap;
 
   mutex_t  m_mutex;
 
   void itemListHandler(Gaudi::Details::PropertyBase& /* theProp */);
 
   void excludeListHandler(Gaudi::Details::PropertyBase& /* theProp */);
 
   void compressionListHandlerHigh(Gaudi::Details::PropertyBase& /* theProp */);
 
   void compressionListHandlerLow(Gaudi::Details::PropertyBase& /* theProp */);
 
private:
   SG::WriteHandleKey<SG::SelectionVetoes> m_selVetoesKey
   { this, "SelVetoesKey", "" };
 
   SG::WriteHandleKey<SG::CompressionInfo> m_compInfoKey
   { this, "CompInfoKey", "" };
 
   std::string m_outputAttributes;
 
   StatusCode addItemObjects(const SG::FolderItem&, SG::SelectionVetoes& vetoes, SG::CompressionInfo& compInfo);
 
   void handleVariableSelection (const SG::IConstAuxStore& auxstore,
                                 SG::DataProxy& itemProxy,
                                 const std::string& aux_attr,
                                 SG::SelectionVetoes& vetoes) const;
 
   void writeMetaData( const std::string& outputFN="" );
 
   std::set<std::string> buildCompressionSet (const ToolHandle<SG::IFolder>& handle,
                                              const CLID& item_id,
                                              const std::string& item_key) const;
 
   // close an EventService substream that was writing to 'rangeFN' output
   void finalizeRange( const std::string & rangeFN );
 
  void loadDict (CLID clid);
 
  bool simpleMatch(const std::string& pattern, const std::string& text);
};
 
#endif // ATHENASERVICES_OUTPUTSTREAM_H