EventSelectorAthenaPoolSharedIO Class Reference

This class is the EventSelector for event data. More...

#include <EventSelectorAthenaPoolSharedIO.h>

Inheritance diagram for EventSelectorAthenaPoolSharedIO:

Collaboration diagram for EventSelectorAthenaPoolSharedIO:

Public Member Functions
	EventSelectorAthenaPoolSharedIO (const std::string &name, ISvcLocator *pSvcLocator)
	Standard Service Constructor.
virtual	~EventSelectorAthenaPoolSharedIO ()=default
	Destructor.
virtual StatusCode	initialize () override
	Required of all Gaudi Services.
virtual StatusCode	start () override
virtual StatusCode	stop () override
virtual StatusCode	finalize () override
virtual StatusCode	next (IEvtSelector::Context &ctxt) const override
virtual StatusCode	next (IEvtSelector::Context &ctxt, int jump) const override
virtual StatusCode	makeServer (int num) override
	Make this a server.
virtual StatusCode	makeClient (int num) override
	Make this a client.
virtual StatusCode	share (int evtnum) override
	Request to share a given event number.
virtual StatusCode	readEvent (int maxevt) override
	Read the next maxevt events.
virtual StatusCode	io_reinit () override
	Callback method to reinitialize the internal state of the component for I/O purposes (e.g. upon fork(2)).

Protected Member Functions
virtual StatusCode	recordAttributeList () const override
	Record AttributeList in StoreGate.

Private Attributes
ToolHandle< IAthenaIPCTool >	m_eventStreamingTool {this, "SharedMemoryTool", "", ""}
Gaudi::Property< int >	m_makeStreamingToolClient {this, "MakeStreamingToolClient", 0}
	Make this instance a Streaming Client during first iteration automatically.

Detailed Description

This class is the EventSelector for event data.

Definition at line 21 of file EventSelectorAthenaPoolSharedIO.h.

Constructor & Destructor Documentation

EventSelectorAthenaPoolSharedIO()

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

Standard Service Constructor.

Definition at line 36 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                                                                                :
    base_class(name, pSvcLocator) {
}

~EventSelectorAthenaPoolSharedIO()

virtual EventSelectorAthenaPoolSharedIO::~EventSelectorAthenaPoolSharedIO ( )

virtualdefault

Destructor.

Member Function Documentation

finalize()

StatusCode EventSelectorAthenaPoolSharedIO::finalize ( )

overridevirtual

Definition at line 75 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                     {
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      delete m_endIter;   m_endIter   = nullptr;
      return ::AthService::finalize();
   }
   return EventSelectorAthenaPool::finalize();
}

initialize()

StatusCode EventSelectorAthenaPoolSharedIO::initialize ( )

overridevirtual

Required of all Gaudi Services.

Definition at line 40 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                       {
   if (!EventSelectorAthenaPool::initialize().isSuccess()) {
      return StatusCode::FAILURE;
   }
   // Get SharedMemoryTool (if configured)
   if (!m_eventStreamingTool.empty() && !m_eventStreamingTool.retrieve().isSuccess()) {
      ATH_MSG_FATAL("Cannot get " << m_eventStreamingTool.typeAndName() << "");
      return StatusCode::FAILURE;
   } else if (m_makeStreamingToolClient.value() == -1) {
      std::string dummyStr;
      ATH_CHECK(m_eventStreamingTool->makeClient(m_makeStreamingToolClient.value(), dummyStr));
   }
   // Don't listen to the Event Processing incidents
   if (!m_eventStreamingTool.empty()) {
      m_incidentSvc->removeListener(this, IncidentType::BeginProcessing);
      m_incidentSvc->removeListener(this, IncidentType::EndProcessing);
   }
   return StatusCode::SUCCESS;
}

io_reinit()

StatusCode EventSelectorAthenaPoolSharedIO::io_reinit ( )

overridevirtual

Callback method to reinitialize the internal state of the component for I/O purposes (e.g. upon fork(2)).

Definition at line 84 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                      {
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      m_guid = Guid::null();
      m_evtCount = 0;
   }
   return EventSelectorAthenaPool::io_reinit();
}

makeClient()

StatusCode EventSelectorAthenaPoolSharedIO::makeClient ( int num )

overridevirtual

Make this a client.

Definition at line 178 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                              {
   IDataShare* ds = dynamic_cast<IDataShare*>(m_athenaPoolCnvSvc.get());
   if (ds == nullptr) {
      ATH_MSG_ERROR("Cannot cast AthenaPoolCnvSvc to DataShare");
      return StatusCode::FAILURE;
   }
   if (ds->makeClient(num + 1).isFailure()) {
      ATH_MSG_ERROR("Failed to switch AthenaPoolCnvSvc to DataStreaming client");
      return StatusCode::FAILURE;
   }
   if (m_eventStreamingTool.empty()) {
      return StatusCode::SUCCESS;
   }
   ATH_MSG_DEBUG("makeClient: " << m_eventStreamingTool << " = " << num);
   std::string dummyStr;
   return(m_eventStreamingTool->makeClient(0, dummyStr));
}

makeServer()

StatusCode EventSelectorAthenaPoolSharedIO::makeServer ( int num )

overridevirtual

Make this a server.

Definition at line 153 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                              {
   IDataShare* ds = dynamic_cast<IDataShare*>(m_athenaPoolCnvSvc.get());
   if (ds == nullptr) {
      ATH_MSG_ERROR("Cannot cast AthenaPoolCnvSvc to DataShare");
      return StatusCode::FAILURE;
   }
   if (num < 0) {
      if (ds->makeServer(num - 1).isFailure()) {
         ATH_MSG_ERROR("Failed to switch AthenaPoolCnvSvc to output DataStreaming server");
      }
      return StatusCode::SUCCESS;
   }
   if (ds->makeServer(num + 1).isFailure()) {
      ATH_MSG_ERROR("Failed to switch AthenaPoolCnvSvc to input DataStreaming server");
      return StatusCode::FAILURE;
   }
   if (m_eventStreamingTool.empty()) {
      return StatusCode::SUCCESS;
   }
   m_processMetadata = false;
   ATH_MSG_DEBUG("makeServer: " << m_eventStreamingTool << " = " << num);
   return(m_eventStreamingTool->makeServer(1, ""));
}

next() [1/2]

StatusCode EventSelectorAthenaPoolSharedIO::next ( IEvtSelector::Context & ctxt ) const

overridevirtual

Parameters

ctxt	[IN/OUT] current event context is interated to next event.

Definition at line 93 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                                                {
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      if (m_makeStreamingToolClient.value() == -1) {
         StatusCode sc = m_eventStreamingTool->lockEvent(m_evtCount);
         while (sc.isRecoverable()) {
            usleep(1000);
            sc = m_eventStreamingTool->lockEvent(m_evtCount);
         }
      }
      // Increase event count
      ++m_evtCount;
      void* tokenStrPtr = nullptr;
      unsigned int status = 0;
      StatusCode sc = m_eventStreamingTool->getLockedEvent(&tokenStrPtr, status);
      std::unique_ptr<const char[]> tokenStr{static_cast<const char*>(tokenStrPtr)};
      if (sc.isRecoverable()) {
         // Return end iterator
         ctxt = *m_endIter;
         // This is not a real failure but a Gaudi way of handling "end of job"
         return StatusCode::FAILURE;
      }
      if (sc.isFailure()) {
         ATH_MSG_FATAL("Cannot get NextEvent from AthenaSharedMemoryTool");
         return StatusCode::FAILURE;
      }
      if (!eventStore()->clearStore().isSuccess()) {
         ATH_MSG_WARNING("Cannot clear Store");
      }
      std::unique_ptr<AthenaAttributeList> athAttrList = std::make_unique<AthenaAttributeList>();
      athAttrList->extend("eventRef", "string");
      (*athAttrList)["eventRef"].data<std::string>() = tokenStr.get();
      SG::WriteHandle<AthenaAttributeList> wh(m_attrListKey, eventStore()->name());
      if (!wh.record(std::move(athAttrList)).isSuccess()) {
         ATH_MSG_ERROR("Cannot record AttributeList to StoreGate " << StoreID::storeName(eventStore()->storeID()));
         return StatusCode::FAILURE;
      }
      Token token;
      token.fromString(tokenStr.get());
      Guid guid = token.dbID();
      if (guid != m_guid && m_processMetadata.value()) {
         InputFileIncidentGuard::transition(m_inputFileGuard, *m_incidentSvc, name(),
                                          "FID:" + guid.toString(), guid.toString(),
                                          /*endFileName=*/{});
         m_guid = guid;
      }
      return StatusCode::SUCCESS;
   }
   return EventSelectorAthenaPool::next(ctxt);
}

next() [2/2]

StatusCode EventSelectorAthenaPoolSharedIO::next ( IEvtSelector::Context & ctxt, int jump ) const

overridevirtual

Parameters

ctxt	[IN/OUT] current event context is interated to next event.
jump	[IN] number of events to jump (currently not supported).

Definition at line 143 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                                                          {
   if (jump > 0) {
      for (int i = 0; i < jump; i++) {
         ATH_CHECK(next(ctxt));
      }
      return StatusCode::SUCCESS;
   }
   return StatusCode::FAILURE;
}

readEvent()

StatusCode EventSelectorAthenaPoolSharedIO::readEvent ( int maxevt )

overridevirtual

Read the next maxevt events.

Parameters

evtnum

[IN] The number of events to read.

Definition at line 230 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                                {
   IDataShare* ds = dynamic_cast<IDataShare*>(m_athenaPoolCnvSvc.get());
   if (ds == nullptr) {
      ATH_MSG_ERROR("Cannot cast AthenaPoolCnvSvc to DataShare");
      return StatusCode::FAILURE;
   }
   if (m_eventStreamingTool.empty()) {
      ATH_MSG_ERROR("No AthenaSharedMemoryTool configured for readEvent()");
      return StatusCode::FAILURE;
   }
   ATH_MSG_VERBOSE("Called read Event " << maxevt);
   std::unique_ptr<IEvtSelector::Context> ctxt = std::make_unique<EventContextAthenaPool>(this);
   for (int i = 0; i < maxevt || maxevt == -1; ++i) {
      if (!next(*ctxt).isSuccess()) {
         if (m_evtCount == -1) {
            ATH_MSG_VERBOSE("Called read Event and read last event from input: " << i);
            break;
         }
         ATH_MSG_ERROR("Cannot read Event " << m_evtCount - 1 << " into AthenaSharedMemoryTool");
         return StatusCode::FAILURE;
      } else {
         ATH_MSG_VERBOSE("Called next, read Event " << m_evtCount - 1);
      }
   }
   ctxt.reset();
   // End of file, wait for last event to be taken
   StatusCode sc;
   while ( (sc = putEvent_ST(*m_eventStreamingTool, 0, 0, 0, 0)).isRecoverable() ) {
      while (ds->readData().isSuccess()) {
         ATH_MSG_VERBOSE("Called last readData, while marking last event in readEvent()");
      }
      usleep(1000);
   }
   if (!sc.isSuccess()) {
      ATH_MSG_ERROR("Cannot put last Event marker to AthenaSharedMemoryTool");
      return StatusCode::FAILURE;
   } else {
      sc = ds->readData();
      while (sc.isSuccess() || sc.isRecoverable()) {
         sc = ds->readData();
      }
      ATH_MSG_DEBUG("Failed last readData -> Clients are stopped, after marking last event in readEvent()");
   }
   return StatusCode::SUCCESS;
}

recordAttributeList()

StatusCode EventSelectorAthenaPoolSharedIO::recordAttributeList ( ) const

overrideprotectedvirtual

Record AttributeList in StoreGate.

Definition at line 276 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                                      {
   if (!m_eventStreamingTool.empty()) {
      if (m_eventStreamingTool->isServer()) {
         IDataShare* ds = dynamic_cast<IDataShare*>(m_athenaPoolCnvSvc.get());
         if (ds == nullptr) {
            ATH_MSG_ERROR("Cannot cast AthenaPoolCnvSvc to DataShare");
            return StatusCode::FAILURE;
         }
         std::string token = m_headerIterator->eventRef().toString();
         StatusCode sc;
         while ( (sc = putEvent_ST(*m_eventStreamingTool,
                                   m_evtCount - 1, token.c_str(),
                                   token.length() + 1, 0)).isRecoverable() ) {
            while (ds->readData().isSuccess()) {
               ATH_MSG_VERBOSE("Called last readData, while putting next event in next()");
            }
            // Nothing to do right now, trigger alternative (e.g. caching) here? Currently just fast loop.
         }
         if (!sc.isSuccess()) {
            ATH_MSG_ERROR("Cannot put Event " << m_evtCount - 1 << " to AthenaSharedMemoryTool");
            return StatusCode::FAILURE;
         }
      } else {
         return StatusCode::SUCCESS;
      }
   } else {
      return EventSelectorAthenaPool::recordAttributeList();
   }
   return StatusCode::SUCCESS;
}

share()

StatusCode EventSelectorAthenaPoolSharedIO::share ( int evtnum )

overridevirtual

Request to share a given event number.

Parameters

evtnum

[IN] The event number to share.

Definition at line 197 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                            {
   IDataShare* ds = dynamic_cast<IDataShare*>(m_athenaPoolCnvSvc.get());
   if (ds == nullptr) {
      ATH_MSG_ERROR("Cannot cast AthenaPoolCnvSvc to DataShare");
      return StatusCode::FAILURE;
   }
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      StatusCode sc = m_eventStreamingTool->lockEvent(evtnum);
      while (sc.isRecoverable()) {
         usleep(1000);
         sc = m_eventStreamingTool->lockEvent(evtnum);
      }
// Send stop client and wait for restart
      if (sc.isFailure()) {
         if (ds->makeClient(0).isFailure()) {
            return StatusCode::FAILURE;
         }
         sc = m_eventStreamingTool->lockEvent(evtnum);
         while (sc.isRecoverable() || sc.isFailure()) {
            usleep(1000);
            sc = m_eventStreamingTool->lockEvent(evtnum);
         }
//FIXME
         if (ds->makeClient(1).isFailure()) {
            return StatusCode::FAILURE;
         }
      }
      return(sc);
   }
   return StatusCode::FAILURE;
}

start()

StatusCode EventSelectorAthenaPoolSharedIO::start ( )

overridevirtual

Definition at line 60 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                  {
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      return StatusCode::SUCCESS;
   }
   return EventSelectorAthenaPool::start();
}

stop()

StatusCode EventSelectorAthenaPoolSharedIO::stop ( )

overridevirtual

Definition at line 67 of file EventSelectorAthenaPoolSharedIO.cxx.

                                                 {
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      return StatusCode::SUCCESS;
   }
   return EventSelectorAthenaPool::stop();
}

Member Data Documentation

m_eventStreamingTool

ToolHandle<IAthenaIPCTool> EventSelectorAthenaPoolSharedIO::m_eventStreamingTool {this, "SharedMemoryTool", "", ""}

private

Definition at line 73 of file EventSelectorAthenaPoolSharedIO.h.

{this, "SharedMemoryTool", "", ""};

m_makeStreamingToolClient

Gaudi::Property<int> EventSelectorAthenaPoolSharedIO::m_makeStreamingToolClient {this, "MakeStreamingToolClient", 0}

private

Make this instance a Streaming Client during first iteration automatically.

Definition at line 75 of file EventSelectorAthenaPoolSharedIO.h.

{this, "MakeStreamingToolClient", 0};

---

The documentation for this class was generated from the following files:

• EventSelectorAthenaPoolSharedIO.h
• EventSelectorAthenaPoolSharedIO.cxx