EventSelectorAthenaPool.h

Go to the documentation of this file.

/*
  Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
*/
 
#ifndef EVENTSELECTORATHENAPOOL_H
#define EVENTSELECTORATHENAPOOL_H

 
#include "GaudiKernel/IEvtSelector.h"
#include "GaudiKernel/IIoComponent.h"
#include "GaudiKernel/IIncidentListener.h"
#include "GaudiKernel/ServiceHandle.h"
#include "GaudiKernel/ToolHandle.h"
#include "PersistentDataModel/Guid.h"
 
#include "AthenaKernel/IAthenaIPCTool.h"
#include "AthenaKernel/IAthenaSelectorTool.h"
#include "AthenaKernel/IEvtSelectorSeek.h"
#include "AthenaKernel/IEventShare.h"
#include "AthenaKernel/ISecondaryEventSelector.h"
#include "AthenaKernel/SlotSpecificObj.h"
#include "AthenaPoolCnvSvc/IAthenaPoolCnvSvc.h"
#include "AthenaBaseComps/AthService.h"
 
#include <map>
#include <atomic>
#include <cstdint>
#include <memory>
 
// Forward declarations
class IIncidentSvc;
class IOpaqueAddress;
class ISvcLocator;
class EventContextAthenaPool;
class PoolCollectionConverter;
class StoreGateSvc;
namespace pool {
   class ICollectionCursor;
}

class EventSelectorAthenaPool :
  public extends<::AthService, IEvtSelector, IEvtSelectorSeek, IEventShare, IIoComponent, IIncidentListener, ISecondaryEventSelector>
{
 
public: // Constructor and Destructor
   EventSelectorAthenaPool(const std::string& name, ISvcLocator* pSvcLocator);
   virtual ~EventSelectorAthenaPool();

   virtual StatusCode initialize() override;
   virtual StatusCode start() override;
   virtual StatusCode stop() override;
   virtual StatusCode finalize() override;
 
   //-------------------------------------------------
   // IEventSelector
   virtual StatusCode createContext(IEvtSelector::Context*& ctxt) const override;

   virtual StatusCode next(IEvtSelector::Context& ctxt) const override;
   virtual StatusCode next(IEvtSelector::Context& ctxt, int jump) const override;

   virtual StatusCode previous(IEvtSelector::Context& ctxt) const override;
   virtual StatusCode previous(IEvtSelector::Context& ctxt, int jump) const override;

   virtual StatusCode last(IEvtSelector::Context& ctxt) const override;

   virtual StatusCode rewind(IEvtSelector::Context& ctxt) const override;

   virtual StatusCode createAddress(const IEvtSelector::Context& ctxt, IOpaqueAddress*& iop) const override;

   virtual StatusCode releaseContext(IEvtSelector::Context*& ctxt) const override;

   virtual StatusCode resetCriteria(const std::string& criteria, IEvtSelector::Context& ctxt) const override;
 
   //-------------------------------------------------
   // IEventSelectorSeek
   virtual StatusCode seek(Context& ctxt, int evtnum) const override;

   virtual int curEvent (const Context& ctxt) const override;

   virtual int size (Context& ctxt) const override;
 
   //-------------------------------------------------
   // IEventShare
   virtual StatusCode makeServer(int num) override;

   virtual StatusCode makeClient(int num) override;

   virtual StatusCode share(int evtnum) override;

   virtual StatusCode readEvent(int maxevt) override;
 
   //-------------------------------------------------
   // IIoComponent
   virtual StatusCode io_reinit() override;
   virtual StatusCode io_finalize() override;
 
   //-------------------------------------------------
   // IIncidentListener
   virtual void handle(const Incident& incident) override;
 
protected:
   //-------------------------------------------------
   // ISecondaryEventSelector
   virtual StatusCode nextHandleFileTransition(IEvtSelector::Context& ctxt) const override;
   virtual StatusCode nextWithSkip(IEvtSelector::Context& ctxt) const override;
   virtual StatusCode recordAttributeList() const override;
   virtual StatusCode fillAttributeList(coral::AttributeList *attrList, const std::string &suffix, bool copySource) const override;
   // Disconnect DB if all events from the source FID were processed and the Selector moved to another file
   virtual bool disconnectIfFinished(const SG::SourceID &fid) const override;
 
private: // internal member functions
   StoreGateSvc* eventStore() const;
   StatusCode reinit() const;
   std::unique_ptr<PoolCollectionConverter> getCollectionCnv(bool throwIncidents = false) const;
   int findEvent(int evtNum) const;
   void fireEndFileIncidents(bool isLastFile) const;
 
private: // data
   EventContextAthenaPool*         m_endIter{};
 
   mutable std::unique_ptr<PoolCollectionConverter> m_poolCollectionConverter ATLAS_THREAD_SAFE {};
   mutable pool::ICollectionCursor* m_headerIterator ATLAS_THREAD_SAFE {};
   mutable Guid m_guid ATLAS_THREAD_SAFE {};
   mutable std::map<SG::SourceID, int> m_activeEventsPerSource ATLAS_THREAD_SAFE;
 
   ServiceHandle<IAthenaPoolCnvSvc> m_athenaPoolCnvSvc{this, "ConversionService", "AthenaPoolCnvSvc", ""};
   ServiceHandle<IIncidentSvc> m_incidentSvc{this, "IncidentSvc", "IncidentSvc", ""};
 
private: // properties
   Gaudi::Property<bool> m_isSecondary{this, "IsSecondary", false, ""};
   Gaudi::Property<bool> m_processMetadata{this, "ProcessMetadata", true, ""};
   Gaudi::Property<std::string> m_collectionType{this, "CollectionType", "ImplicitCollection", ""};

   std::string m_attrListKey{"Input"};

   Gaudi::Property<std::vector<std::string>> m_inputCollectionsProp{this, "InputCollections", {}, ""};
   mutable std::vector<std::string>::const_iterator m_inputCollectionsIterator ATLAS_THREAD_SAFE;
   void inputCollectionsHandler(Gaudi::Details::PropertyBase&);
   mutable bool m_inputCollectionsChanged ATLAS_THREAD_SAFE;

   Gaudi::Property<bool> m_keepInputFilesOpen{this, "KeepInputFilesOpen", false, ""};

   mutable ToolHandleArray<IAthenaSelectorTool> m_helperTools ATLAS_THREAD_SAFE {this, "HelperTools", {}, ""};
   ToolHandle<IAthenaSelectorTool> m_counterTool{this, "CounterTool", "", ""};
   ToolHandle<IAthenaIPCTool> m_eventStreamingTool{this, "SharedMemoryTool", "", ""};
   Gaudi::Property<int> m_makeStreamingToolClient{this, "MakeStreamingToolClient",0};

   Gaudi::CheckedProperty<uint32_t> m_runNo{this, "RunNumber", 0, ""};
   Gaudi::CheckedProperty<uint32_t> m_oldRunNo{this, "OldRunNumber", 0, ""};
   Gaudi::Property<bool> m_overrideRunNumber{this, "OverrideRunNumber", false, ""};
   Gaudi::Property<bool> m_overrideRunNumberFromInput{this, "OverrideRunNumberFromInput", false, ""};
   // TODO: check if not really used
   Gaudi::CheckedProperty<uint64_t> m_firstEventNo{this, "FirstEvent", 1, ""};
   // TODO: check if not really used
   Gaudi::CheckedProperty<uint64_t> m_eventsPerRun{this, "EventsPerRun", 1000000, ""};
   Gaudi::CheckedProperty<uint32_t> m_firstLBNo{this, "FirstLB", 0, ""};
   Gaudi::CheckedProperty<uint32_t> m_eventsPerLB{this, "EventsPerLB", 1000, ""};
   Gaudi::CheckedProperty<uint32_t> m_initTimeStamp{this, "InitialTimeStamp", 0, ""};
   Gaudi::Property<uint32_t> m_timeStampInterval{this, "TimeStampInterval", 0, ""};
 
   mutable std::atomic_long m_curCollection{};
   mutable std::vector<int> m_numEvt ATLAS_THREAD_SAFE;
   mutable std::vector<int> m_firstEvt ATLAS_THREAD_SAFE;

   Gaudi::Property<int> m_skipEvents{this, "SkipEvents", 0, ""};
   Gaudi::Property<std::vector<long>> m_skipEventSequenceProp{this, "SkipEventSequence", {}, ""};
   Gaudi::Property<std::string> m_skipEventRangesProp{this, "SkipEventRanges", {}, ""};
   mutable std::vector<std::pair<long,long>> m_skipEventRanges ATLAS_THREAD_SAFE;
 
   mutable std::atomic_int m_evtCount{}; // internal count of events
   mutable std::atomic_bool m_firedIncident{};
 
   typedef std::mutex CallMutex;
   mutable CallMutex m_callLock;
 
   SG::SlotSpecificObj<SG::SourceID> m_sourceID;

   friend class DoubleEventSelectorAthenaPool;
};
 
#endif