AthenaPoolCnvSvc.h

Go to the documentation of this file.

/*
  Copyright (C) 2002-2024 CERN for the benefit of the ATLAS collaboration
*/
 
#ifndef ATHENAPOOLCNVSVC_ATHENAPOOLCNVSVC_H
#define ATHENAPOOLCNVSVC_ATHENAPOOLCNVSVC_H
 
#include "AthenaPoolCnvSvc/IAthenaPoolCnvSvc.h"
 
#include "GaudiKernel/IClassIDSvc.h"
#include "GaudiKernel/IIncidentListener.h"
#include "GaudiKernel/IIoComponent.h"
#include "GaudiKernel/ServiceHandle.h"
#include "GaudiKernel/ToolHandle.h"
 
#include "StorageSvc/DbType.h"
#include "AthenaBaseComps/AthCnvSvc.h"
#include "AthenaKernel/IAthenaIPCTool.h"
#include "PmbCxxUtils/BasicStopWatch.h"
#include "PoolSvc/IPoolSvc.h"
 
#include <vector>
#include <map>
#include <mutex>
#include <limits>
 
// Forward declarations
class IAthenaSerializeSvc;
class Guid;
 
template <class TYPE> class SvcFactory;
 
class AthenaPoolCnvSvc : public extends<AthCnvSvc,
                                        IAthenaPoolCnvSvc,
                                        IIncidentListener,
                                        IIoComponent> {
   // Allow the factory class access to the constructor
   friend class SvcFactory<AthenaPoolCnvSvc>;
 
public:
 
   virtual StatusCode initialize() override;
   virtual StatusCode io_reinit() override;
   virtual StatusCode finalize() override;
   virtual StatusCode io_finalize() override;
   virtual StatusCode stop() override final;
 
   virtual StatusCode createObj(IOpaqueAddress* pAddress, DataObject*& refpObject) override;
 
   virtual StatusCode createRep(DataObject* pObject, IOpaqueAddress*& refpAddress) override;
 
   virtual StatusCode fillRepRefs(IOpaqueAddress* pAddress, DataObject* pObject) override;
 
   virtual StatusCode connectOutput(const std::string& outputConnectionSpec,
           const std::string& openMode) override;
 
   virtual StatusCode connectOutput(const std::string& outputConnectionSpec) override;
 
   virtual StatusCode commitOutput(const std::string& outputConnectionSpec, bool doCommit) override;
 
   virtual StatusCode disconnectOutput(const std::string& outputConnectionSpec) override;
 
   virtual IPoolSvc* getPoolSvc() override;
 
   virtual Token* registerForWrite(Placement* placement, const void* obj, const RootType& classDesc) override;
 
   virtual void setObjPtr(void*& obj, const Token* token) override;
 
   virtual bool useDetailChronoStat() const override;
 
   StatusCode createAddress(long svcType,
           const CLID& clid,
           const std::string* par,
           const unsigned long* ip,
           IOpaqueAddress*& refpAddress) override;
 
   virtual StatusCode createAddress(long svcType,
           const CLID& clid,
           const std::string& refAddress,
           IOpaqueAddress*& refpAddress) override;
 
   virtual StatusCode convertAddress(const IOpaqueAddress* pAddress, std::string& refAddress) override;
 
   virtual StatusCode decodeOutputSpec(std::string& connectionSpec, int& outputTech) const override;
 
   virtual StatusCode registerCleanUp(IAthenaPoolCleanUp* cnv) override;
 
   virtual StatusCode cleanUp(const std::string& connection) override;
 
   virtual StatusCode setInputAttributes(const std::string& fileName) override;
 
   virtual StatusCode makeServer(int num) override;
 
   virtual StatusCode makeClient(int num) override;
 
   virtual StatusCode readData() override;
 
   virtual StatusCode commitCatalog() override;
 
   StatusCode abortSharedWrClients(int client_n);
 
   virtual void handle(const Incident& incident) override;
 
   void flushDataHeaderForms(const std::string& streamName = "*");
 
   AthenaPoolCnvSvc(const std::string& name, ISvcLocator* pSvcLocator);
   virtual ~AthenaPoolCnvSvc() = default;
 
private: // member functions
   void extractPoolAttributes(const StringArrayProperty& property,
       std::vector<std::vector<std::string> >* contAttr,
       std::vector<std::vector<std::string> >* dbAttr,
       std::vector<std::vector<std::string> >* domAttr = 0) const;
 
   StatusCode processPoolAttributes(std::vector<std::vector<std::string> >& attr,
       const std::string& fileName,
       unsigned long contextId,
       bool doGet = true,
       bool doSet = true,
       bool doClear = true) const;
 
private: // data
   pool::DbType                  m_dbType;
   std::string                   m_lastInputFileName;
   ServiceHandle<IPoolSvc>       m_poolSvc{this,"PoolSvc","PoolSvc"};
   ServiceHandle<IClassIDSvc>    m_clidSvc{this,"ClassIDSvc","ClassIDSvc"};
   ServiceHandle<IAthenaSerializeSvc> m_serializeSvc{this,"AthenaRootSerializeSvc","AthenaRootSerializeSvc"};
   ToolHandle<IAthenaIPCTool>    m_inputStreamingTool{this,"InputStreamingTool",{}};
   ToolHandle<IAthenaIPCTool>    m_outputStreamingTool{this,"OutputStreamingTool",{}};
   bool m_streamServerActive=false;
   int m_metadataClient=0;
 
   PMonUtils::BasicStopWatchResultMap_t m_chronoMap{};
 
private: // properties
   BooleanProperty m_useDetailChronoStat{this,"UseDetailChronoStat",false};
 
   StringProperty  m_storageTechProp{this,"StorageTechnology", "ROOTTREEINDEX"};
   StringProperty  m_containerPrefixProp{this,"PoolContainerPrefix","Default"};
   StringProperty  m_containerNameHintProp{this,"TopLevelContainerName",""};
   StringProperty  m_branchNameHintProp{this,"SubLevelBranchName", "<type>/<key>"};
 
   StringArrayProperty m_poolAttr{this,"PoolAttributes",{},"Pool Attributes","OrderedSet<std::string>"};
   std::vector<std::vector<std::string> > m_domainAttr;
   std::vector<std::vector<std::string> > m_databaseAttr;
   std::vector<std::vector<std::string> > m_containerAttr;
   std::vector<unsigned int> m_contextAttr;
   std::map<std::string, int> m_fileCommitCounter;
   std::map<std::string, int> m_fileFlushSetting;
   StringArrayProperty m_inputPoolAttr{this,"InputPoolAttributes",{}};
   std::vector<std::vector<std::string> > m_inputAttr;
   StringArrayProperty m_inputPoolAttrPerEvent{this,"PrintInputAttrPerEvt",{}};
   std::vector<std::vector<std::string> > m_inputAttrPerEvent;
 
   StringArrayProperty m_maxFileSizes{this,"MaxFileSizes",{}};
   long long m_domainMaxFileSize=std::numeric_limits<long long>::max();
   std::map<std::string, long long> m_databaseMaxFileSize;
 
   BooleanProperty m_persSvcPerOutput{this,"PersSvcPerOutput",true};
   unsigned outputContextId(const std::string& outputConnection);
 
   StringProperty m_persSvcPerInputType{this,"PersSvcPerInputType",""};
   std::mutex  m_mutex;
 
   StringProperty  m_metadataContainerProp{this,"OutputMetadataContainer","MetaData"};
   StringArrayProperty m_metadataContainersAug{this, "OutputMetadataContainers", {}, "Metadata containers used for augmentations"};
 
   IntegerProperty m_makeStreamingToolClient{this,"MakeStreamingToolClient",0};
   IntegerProperty m_streamingTechnology{this,"StreamingTechnology",-1};
   BooleanProperty m_parallelCompression{this,"ParallelCompression",true};
   StringProperty  m_streamPortString{this,"StreamPortString","?pmerge=localhost:0"};
   IntegerProperty m_numberEventsPerWrite{this,"NumberEventsPerWrite",-1};
 
   BooleanProperty m_oneDataHeaderForm { this, "OneDataHeaderForm", false };
   IntegerProperty m_DHFormCacheSize { this, "maxDHFormCacheSize", 100 };
   BooleanProperty m_DHFilterAliases { this, "doFilterDHAliases", true };
 
};
 
#endif