de/d26/AthenaOutputStream_8h_source.html

// Dear emacs, this is -*- C++ -*-


/*

  Copyright (C) 2002-2024 CERN for the benefit of the ATLAS collaboration

*/


#ifndef ATHENASERVICES_ATHENAOUTPUTSTREAM_H

#define ATHENASERVICES_ATHENAOUTPUTSTREAM_H


// STL include files

#include <memory>

#include <map>

#include <set>

#include <vector>

#include <string>

#include <mutex>


// Required for inheritance

#include "GaudiKernel/IDataSelector.h"

#include "Gaudi/Property.h"

#include "GaudiKernel/ServiceHandle.h"

#include "GaudiKernel/ToolHandle.h"


#include "GaudiKernel/IIncidentListener.h"

#include "AthenaBaseComps/FilteredAlgorithm.h"

#include "StoreGate/WriteHandleKey.h"

#include "GaudiKernel/IIoComponent.h"


#include "SelectionVetoes.h"

#include "CompressionInfo.h"


// forward declarations

class IClassIDSvc;

class OutputStreamSequencerSvc;

class StoreGateSvc;

class IAthenaOutputStreamTool;

class IAthenaOutputTool;

class IItemListSvc;

class IDictLoaderSvc;

class ITPCnvSvc;

class MetaDataSvc;


namespace SG {

   class DataProxy;

   class IFolder;

   class IAuxStoreIO;

   class FolderItem;

}


class AthenaOutputStream : public extends<FilteredAlgorithm,

                                          IIncidentListener, IIoComponent> {


public:

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


protected:

   ServiceHandle<StoreGateSvc> m_dataStore;

   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;

   StringProperty           m_writingTool;

   std::string              m_outputName;

   std::string              m_outputAttributes;


   typedef ServiceHandle<IClassIDSvc> IClassIDSvc_t;

   IClassIDSvc_t m_pCLIDSvc;

   ServiceHandle<OutputStreamSequencerSvc>  m_outSeqSvc;


   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"};

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

   StringArrayProperty      m_compressionListHigh;

   StringArrayProperty      m_compressionListLow;

   unsigned int             m_compressionBitsHigh;

   unsigned int             m_compressionBitsLow;

   StringArrayProperty      m_transientItems;

   ToolHandle<SG::IFolder>  m_p2BWritten;

   ToolHandle<SG::IFolder>  m_decoder;

   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;

   std::atomic<int> m_events;

   bool m_forceRead;

   bool m_extendProvenanceRecord;

   bool m_writeOnExecute;

   bool m_writeOnFinalize;

   bool m_itemListFromTool;

   bool m_checkNumberOfWrites;

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

   CounterMapType  m_objectWriteCounter;


   ToolHandle<IAthenaOutputStreamTool> m_streamer;

   ToolHandleArray<IAthenaOutputTool> m_helperTools;


   // flag set by MetaDataStop if OutputSequencer is used with EndEvent

   bool m_writeMetadataAndDisconnect =  false;


   // ------- 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;

   typedef std::recursive_mutex mutex_t;

   mutex_t  m_mutex;


private:

   SG::WriteHandleKey<SG::SelectionVetoes> m_selVetoesKey

   { this, "SelVetoesKey", "" };


   SG::WriteHandleKey<SG::CompressionInfo> m_compInfoKey

   { this, "CompInfoKey", "" };


protected:

   void itemListHandler(Gaudi::Details::PropertyBase& /* theProp */);

   void excludeListHandler(Gaudi::Details::PropertyBase& /* theProp */);

   void compressionListHandlerHigh(Gaudi::Details::PropertyBase& /* theProp */);

   void compressionListHandlerLow(Gaudi::Details::PropertyBase& /* theProp */);


public:

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


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

   virtual ~AthenaOutputStream();


   virtual StatusCode initialize() override;

   virtual StatusCode finalize() override;

   virtual StatusCode execute() override;

   virtual StatusCode stop() 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;


private:

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


   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;


   bool matchKey(const std::vector<std::string>& key, const std::string& proxyName) 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);

};


#endif // ATHENASERVICES_OUTPUTSTREAM_H