EventSelectorByteStream.h

Go to the documentation of this file.

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

 
#include "GaudiKernel/IEvtSelector.h"
#include "GaudiKernel/ServiceHandle.h"
#include "GaudiKernel/ToolHandle.h"
#include "GaudiKernel/IIoComponent.h"
 
#include "AthenaKernel/IAthenaIPCTool.h"
#include "AthenaKernel/IAthenaSelectorTool.h"
#include "AthenaKernel/IEvtSelectorSeek.h"
#include "AthenaKernel/IEventShare.h"
#include "AthenaKernel/ISecondaryEventSelector.h"
#include "AthenaKernel/InputFileIncidentGuard.h"
#include "AthenaBaseComps/AthService.h"
 
#include "ByteStreamCnvSvcBase/IByteStreamInputSvc.h"
#include "ByteStreamData/RawEvent.h"
#include "CxxUtils/checker_macros.h"
#include <mutex>
#include <optional>
#include <utility>
 
// Forward declarations.
class ISvcLocator;
class EventContextByteStream;
class IROBDataProviderSvc;
 
// Class EventSelectorByteStream.
class EventSelectorByteStream :
   public extends<::AthService, IEvtSelector, IEvtSelectorSeek, IEventShare, IIoComponent, ISecondaryEventSelector>
{
public:
   EventSelectorByteStream(const std::string& name, ISvcLocator* svcloc);
   virtual ~EventSelectorByteStream();

   virtual StatusCode initialize() override;
   virtual StatusCode start() override;
   virtual StatusCode stop() override;
   virtual StatusCode finalize() override;
 
   //-------------------------------------------------
   // IEventSelector
   virtual StatusCode createContext(Context*& it) const override;
   virtual StatusCode next(Context& it) const override;
   virtual StatusCode next(Context& it, int jump) const override;
   virtual StatusCode previous(Context& it) const override;
   virtual StatusCode previous(Context& it, int jump) const override;

   virtual StatusCode last(Context& it) const override;
   virtual StatusCode rewind(Context& it) const override;

   virtual StatusCode createAddress(const Context& it,IOpaqueAddress*& iop) const override;

   virtual StatusCode releaseContext(Context*& it) const override;

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

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

   virtual int size (Context& it) 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;
 
protected:
   using mutex_t = std::recursive_mutex;
   using lock_t = std::lock_guard<mutex_t>;
 
   //-------------------------------------------------
   // 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
   StatusCode nextImpl(Context& it, lock_t& lock) const;
   StatusCode nextImpl(Context& it, int jump, lock_t& lock) const;
   StatusCode previousImpl(Context& it, lock_t& lock) const;
   StatusCode previousImpl(Context& it, int jump, lock_t& lock) const;
   StatusCode nextHandleFileTransitionImpl(IEvtSelector::Context& ctxt,
                                           lock_t& lock) const;
   StatusCode nextWithSkipImpl(IEvtSelector::Context& ctxt,
                               lock_t& lock) const;
   StatusCode recordAttributeListImpl(lock_t& lock) const;
   StatusCode fillAttributeListImpl(coral::AttributeList *attrList, const std::string &suffix, bool copySource,
                                    lock_t& lock) const;

   StatusCode reinit(lock_t& lock);
   StatusCode openNewRun(lock_t& lock) const;
   long openBlock(const std::string& fileName, lock_t& lock) const;
   void fileTransition(lock_t& lock) const;
   void nextFile(lock_t& lock) const;
   int findEvent(int evtNum, lock_t& lock) const;
   StoreGateSvc* eventStore() const;
 
private: // properties
   // FIXME: A Gaudi EventSelector is not meant to have mutable state.
   // Any needed state is meant to be kept in the Context object.
   // So the mutable members here should instead be kept in
   // EventContextByteStream.  However, making that work would
   // require redesigning other Athena interfaces.  So for now,
   // just add a mutex to protect access to them.
   mutable mutex_t m_mutex;
   mutable int           m_fileCount ATLAS_THREAD_SAFE = 0;  
   mutable std::vector<int>     m_numEvt ATLAS_THREAD_SAFE;
   mutable std::vector<int>     m_firstEvt ATLAS_THREAD_SAFE;
   mutable std::vector<std::string>::const_iterator m_inputCollectionsIterator ATLAS_THREAD_SAFE;
   mutable std::vector<long> m_skipEventSequence ATLAS_THREAD_SAFE;
   mutable long m_NumEvents ATLAS_THREAD_SAFE = 0; // Number of Events read so far.
   mutable ToolHandle<IAthenaIPCTool> m_eventStreamingTool ATLAS_THREAD_SAFE {this, "SharedMemoryTool", "", ""};

   Gaudi::Property<bool> m_isSecondary{this, "IsSecondary", false, ""};
 
   Gaudi::Property<std::string> m_eventSourceName{this, "ByteStreamInputSvc", "", ""};
   Gaudi::Property<bool> m_procBadEvent{this, "ProcessBadEvent", false, ""}; 
   Gaudi::Property<int>  m_maxBadEvts{this, "MaxBadEvents", -1, ""};         
 
   EventContextByteStream*  m_beginIter{};
   EventContextByteStream*  m_endIter{};
   SmartIF<IByteStreamInputSvc> m_eventSource;
   Gaudi::Property<std::vector<std::string>> m_inputCollectionsProp{this, "Input", {}, ""};
   void inputCollectionsHandler(Gaudi::Details::PropertyBase&);
   ServiceHandle<IIncidentSvc> m_incidentSvc{this, "IncidentSvc", "IncidentSvc", ""};
 
   Gaudi::Property<long> m_skipEvents{this, "SkipEvents", 0, ""}; // Number of events to skip at the beginning
   Gaudi::Property<std::vector<long>> m_skipEventSequenceProp{this, "SkipEventSequence", {}, ""};
 
   bool m_firstFileFired{};
   mutable std::optional<InputFileIncidentGuard> m_inputFileGuard ATLAS_THREAD_SAFE;
   mutable std::optional<std::pair<std::string, std::string>> m_pendingFile ATLAS_THREAD_SAFE;

   ToolHandleArray<IAthenaSelectorTool> m_helperTools{this, "HelperTools", {}, "Helper tools executed by EventSelector"};
   ToolHandle<IAthenaSelectorTool>      m_counterTool{this, "CounterTool", "", ""};

   Gaudi::Property<bool> m_overrideRunNumber{this, "OverrideRunNumber", false, ""};
   Gaudi::Property<bool> m_filebased{this, "FileBased", true, ""};
 
   Gaudi::CheckedProperty<uint32_t> m_runNo{this, "RunNumber", 0, ""};
   Gaudi::CheckedProperty<uint64_t> m_firstEventNo{this, "FirstEvent", 1, ""};
   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, ""};
};
 
#endif // BYTESTREAMCNVSVC_EVENTSELECTORBYTESTREAM_H