PileUpStream Class Reference

a triple selector/context/store defines a stream More...

#include <PileUpStream.h>

Inheritance diagram for PileUpStream:

Collaboration diagram for PileUpStream:

Public Types
typedef IEvtSelector::Context	EvtIterator

Public Member Functions
bool	msgLvl (const MSG::Level lvl) const
	Test the output level. More...
MsgStream &	msg () const
	The standard message stream. More...
MsgStream &	msg (const MSG::Level lvl) const
	The standard message stream. More...
void	setLevel (MSG::Level lvl)
	Change the current logging level. More...
Structors
	PileUpStream ()
	Structors. More...
	PileUpStream (const PileUpStream &rhs)=delete
PileUpStream &	operator= (const PileUpStream &rhs)=delete
	PileUpStream (PileUpStream &&rhs)
PileUpStream &	operator= (PileUpStream &&rhs)
	PileUpStream (const std::string &name, IEvtSelector *sel, StoreGateSvc *store)
	PileUpStream (const std::string &name, ISvcLocator *svcLoc, const std::string &selecName)
	PileUpStream (const std::string &name, ISvcLocator *svcLoc, IEvtSelector *sel)
virtual	~PileUpStream ()

Private Member Functions
void	initMessaging () const
	Initialize our message level and MessageSvc. More...

Private Attributes
std::string	m_nm
	Message source name. More...
boost::thread_specific_ptr< MsgStream >	m_msg_tls
	MsgStream instance (a std::cout like with print-out levels) More...
std::atomic< IMessageSvc * >	m_imsg { nullptr }
	MessageSvc pointer. More...
std::atomic< MSG::Level >	m_lvl { MSG::NIL }
	Current logging level. More...
std::atomic_flag m_initialized	ATLAS_THREAD_SAFE = ATOMIC_FLAG_INIT
	Messaging initialized (initMessaging) More...

accessors
std::string	m_name
	Stream name. More...
ISvcLocator *	p_svcLoc
	ServiceLocator. More...
IEvtSelector *	p_sel
	Selector. More...
StoreGateSvc *	p_SG
	StoreGateSvc;. More...
EvtIterator *	p_iter
	Input Iterators. More...
PileUpMergeSvc *	p_mergeSvc
bool	m_ownEvtIterator
bool	m_neverLoaded
	do we own p_iter? More...
bool	m_ownStore
	has an event been loaded into this stream? More...
bool	m_used
	is p_SG a store we cloned from the master one? More...
bool	m_hasRing
	has this stream already been used? (for the current event) More...
unsigned int	m_iOriginalRing
const std::string &	name ()
IEvtSelector &	selector ()
EvtIterator &	iterator ()
StoreGateSvc &	store ()
const IEvtSelector &	selector () const
const EvtIterator &	iterator () const
const StoreGateSvc &	store () const
const xAOD::EventInfo *	nextEventPre (bool readRecord=true)
	return next Event, load store with next Event More...
bool	nextEventPre_Passive (bool readRecord)
	like nextEventPre, but doesn't actually load anything More...
bool	setupStore ()
	setup input and overlay selectors and iters More...
bool	setupStore (bool)
	backward compatibility More...
StatusCode	finalize ()
	finalize and release store. To be called on ... finalize() More...
bool	used () const
	has this stream already be used in this event? More...
void	setUsed ()
void	resetUsed ()
bool	hasRing () const
unsigned int	originalIRing () const
	index of the ring for which this stream was used originally More...
void	setOriginalIRing (unsigned int iR)
ISvcLocator *	serviceLocator ()
StatusCode	nextRecordPre ()
	increment event iterator before loading store More...
StatusCode	nextRecordPre_Passive ()
	increment event iterator More...
bool	loadStore ()
	clear store and load new proxies More...
void	setActiveStore ()
	set ActiveStore More...

Detailed Description

a triple selector/context/store defines a stream

Definition at line 31 of file PileUpStream.h.

Member Typedef Documentation

EvtIterator

typedef IEvtSelector::Context PileUpStream::EvtIterator

Definition at line 35 of file PileUpStream.h.

Constructor & Destructor Documentation

PileUpStream() [1/6]

PileUpStream::PileUpStream

(

)

Structors.

Definition at line 27 of file PileUpStream.cxx.

                          :
  AthMessaging ("PileUpStream"),
  m_name("INVALID"), p_svcLoc(0), p_sel(0), p_SG(0), p_iter(0), 
  p_mergeSvc(nullptr), m_ownEvtIterator(false),
  m_neverLoaded(true), m_ownStore(false),
  m_used(false), m_hasRing(false), m_iOriginalRing(0)
{ 
}

PileUpStream() [2/6]

PileUpStream::PileUpStream ( const PileUpStream &  rhs )

delete

PileUpStream() [3/6]

PileUpStream::PileUpStream

(

PileUpStream &&

rhs

)

Definition at line 36 of file PileUpStream.cxx.

                                            :
  AthMessaging (rhs.m_name),
  m_name(rhs.m_name), p_svcLoc(rhs.p_svcLoc), p_sel(rhs.p_sel), 
  p_SG(rhs.p_SG), p_iter(rhs.p_iter), p_mergeSvc(rhs.p_mergeSvc), m_ownEvtIterator(rhs.m_ownEvtIterator),
  m_neverLoaded(rhs.m_neverLoaded), m_ownStore(rhs.m_ownStore),
  m_used(rhs.m_used), m_hasRing(rhs.m_hasRing), m_iOriginalRing(rhs.m_iOriginalRing)
{ 
  //transferred ownership
  rhs.m_ownEvtIterator=false;
  rhs.m_ownStore=false;
}

PileUpStream() [4/6]

PileUpStream::PileUpStream	(	const std::string &	name,
		IEvtSelector *	sel,
		StoreGateSvc *	store
	)

PileUpStream() [5/6]

PileUpStream::PileUpStream	(	const std::string &	name,
		ISvcLocator *	svcLoc,
		const std::string &	selecName
	)

Definition at line 89 of file PileUpStream.cxx.

                                          :
  AthMessaging (name),
  m_name(name), p_svcLoc(svcLoc), p_sel(0), p_SG(0), p_iter(0),
  m_ownEvtIterator(false), 
  m_neverLoaded(true), m_ownStore(false),
  m_used(false), m_hasRing(false), m_iOriginalRing(0)
 
{
  assert(p_svcLoc);
  if (!(serviceLocator()->service(selecName, p_sel).isSuccess() &&
        serviceLocator()->service("PileUpMergeSvc", p_mergeSvc, true).isSuccess() &&
    p_sel->createContext(p_iter).isSuccess() )) {
    const std::string errMsg("PileUpStream: can not create stream");
    ATH_MSG_ERROR ( errMsg );
    throw std::runtime_error(errMsg);
  } else  m_ownEvtIterator=true;
}

PileUpStream() [6/6]

PileUpStream::PileUpStream	(	const std::string &	name,
		ISvcLocator *	svcLoc,
		IEvtSelector *	sel
	)

Definition at line 69 of file PileUpStream.cxx.

                                 :
  AthMessaging (name),
  m_name(name), p_svcLoc(svcLoc), p_sel(sel), p_SG(0), p_iter(0), 
  m_ownEvtIterator(false), 
  m_neverLoaded(true), m_ownStore(false),
  m_used(false), m_hasRing(false), m_iOriginalRing(0)
{ 
  assert(p_sel);
  assert(p_svcLoc);
  if( !( p_sel->createContext(p_iter).isSuccess() &&
         serviceLocator()->service("PileUpMergeSvc", p_mergeSvc, true).isSuccess() ) ) {
 
    const std::string errMsg("PileUpStream:: can not create stream");
    ATH_MSG_ERROR ( errMsg );
    throw std::runtime_error(errMsg);
  } else  m_ownEvtIterator=true;
}

~PileUpStream()

PileUpStream::~PileUpStream ( )

virtual

Definition at line 109 of file PileUpStream.cxx.

{ 
}

Member Function Documentation

finalize()

StatusCode PileUpStream::finalize

(

)

finalize and release store. To be called on ... finalize()

Definition at line 262 of file PileUpStream.cxx.

                                  {
  StatusCode sc(StatusCode::SUCCESS);
  if (m_ownEvtIterator) delete p_iter;
  //we own and manage our cloned SG instance
#ifdef GAUDIKERNEL_STATEMACHINE_H_
  if (m_ownStore && Gaudi::StateMachine::INITIALIZED == store().FSMState()) {
    sc = this->store().sysFinalize();
  }
#else
  if (m_ownStore && this->store().state() == IService::INITIALIZED) sc = this->store().sysFinalize();
#endif
  this->store().release();
  return sc;
}

hasRing()

bool PileUpStream::hasRing ( ) const

inline

Definition at line 85 of file PileUpStream.h.

{ return m_hasRing; }

initMessaging()

void AthMessaging::initMessaging ( ) const

privateinherited

Initialize our message level and MessageSvc.

This method should only be called once.

Definition at line 39 of file AthMessaging.cxx.

{
  m_imsg = Athena::getMessageSvc();
  m_lvl = m_imsg ?
    static_cast<MSG::Level>( m_imsg.load()->outputLevel(m_nm) ) :
    MSG::INFO;
}

iterator() [1/2]

EvtIterator& PileUpStream::iterator ( )

inline

Definition at line 60 of file PileUpStream.h.

{ return *p_iter; }

iterator() [2/2]

const EvtIterator& PileUpStream::iterator ( ) const

inline

Definition at line 63 of file PileUpStream.h.

{ return *p_iter; }

loadStore()

bool PileUpStream::loadStore ( )

private

clear store and load new proxies

Definition at line 209 of file PileUpStream.cxx.

{
  m_neverLoaded=false;
 
  IOpaqueAddress* paddr(0);
  bool rc = (this->selector().createAddress(iterator(), paddr)).isSuccess();
  if ( 0 != paddr) rc &= this->store().recordAddress(paddr).isSuccess();
 
  // load store proxies
  rc &= this->store().loadEventProxies().isSuccess();
  return rc;
}

msg() [1/2]

MsgStream & AthMessaging::msg ( ) const

inlineinherited

The standard message stream.

Returns a reference to the default message stream May not be invoked before sysInitialize() has been invoked.

Definition at line 164 of file AthMessaging.h.

{
  MsgStream* ms = m_msg_tls.get();
  if (!ms) {
    if (!m_initialized.test_and_set()) initMessaging();
    ms = new MsgStream(m_imsg,m_nm);
    m_msg_tls.reset( ms );
  }
 
  ms->setLevel (m_lvl);
  return *ms;
}

msg() [2/2]

MsgStream & AthMessaging::msg ( const MSG::Level  lvl ) const

inlineinherited

The standard message stream.

Returns a reference to the default message stream May not be invoked before sysInitialize() has been invoked.

Definition at line 179 of file AthMessaging.h.

{ return msg() << lvl; }

msgLvl()

bool AthMessaging::msgLvl ( const MSG::Level  lvl ) const

inlineinherited

Test the output level.

Parameters

lvl	The message level to test against

Returns

boolean Indicating if messages at given level will be printed

Return values

true	Messages at level "lvl" will be printed

Definition at line 151 of file AthMessaging.h.

{
  if (!m_initialized.test_and_set()) initMessaging();
  if (m_lvl <= lvl) {
    msg() << lvl;
    return true;
  } else {
    return false;
  }
}

name()

const std::string& PileUpStream::name ( )

inline

Definition at line 58 of file PileUpStream.h.

{ return m_name; }

nextEventPre()

const xAOD::EventInfo * PileUpStream::nextEventPre

(

bool

readRecord = true

)

return next Event, load store with next Event

Definition at line 223 of file PileUpStream.cxx.

{
   if (m_neverLoaded) readRecord=true;
   else if (readRecord) {
      //do not reset these the first time we call nextEventPre
      this->resetUsed();
      m_hasRing=false;
   }
   //  if (isNotEmpty()) {
   if (readRecord && this->nextRecordPre().isFailure()) {
      ATH_MSG_INFO ( "nextEventPre(): end of the loop. No more events in the selection" );
      return nullptr;
   }
 
   const xAOD::EventInfo* xAODEventInfo = p_mergeSvc->getPileUpEvent( p_SG, "" );
   if (readRecord and xAODEventInfo) {
      ATH_MSG_DEBUG ( "nextEventPre(): read new event " 
              <<  xAODEventInfo->eventNumber() 
              << " run " << xAODEventInfo->runNumber()
              << " into store " << this->store().name() );
   }
 
   return xAODEventInfo;
}

nextEventPre_Passive()

bool PileUpStream::nextEventPre_Passive

(

bool

readRecord

)

like nextEventPre, but doesn't actually load anything

Definition at line 248 of file PileUpStream.cxx.

                                                       {
  if (m_neverLoaded) readRecord=true;
  else if (readRecord) {
    //do not reset these the first time we call nextEventPre_Passive
    this->resetUsed();
    m_hasRing=false;
  }
  if (readRecord && this->nextRecordPre_Passive().isFailure()) {
    ATH_MSG_INFO ( "nextEventPre_Passive(): end of the loop. No more events in the selection" );
    return false;
  } 
  return true;
}    

nextRecordPre()

StatusCode PileUpStream::nextRecordPre ( )

private

increment event iterator before loading store

Definition at line 194 of file PileUpStream.cxx.

{
  // Clear the store, move to next event and load the store
  return (this->nextRecordPre_Passive().isSuccess() && 
      this->loadStore()) ?
    StatusCode::SUCCESS :
    StatusCode::FAILURE;
}

nextRecordPre_Passive()

StatusCode PileUpStream::nextRecordPre_Passive ( )

private

increment event iterator

Definition at line 184 of file PileUpStream.cxx.

{
  this->setActiveStore();
  // Clear the store, move to next event
  return (this->store().clearStore().isSuccess() && 
      this->selector().next(iterator()).isSuccess() ) ?
    StatusCode::SUCCESS :
    StatusCode::FAILURE;
}

operator=() [1/2]

PileUpStream& PileUpStream::operator= ( const PileUpStream &  rhs )

delete

operator=() [2/2]

PileUpStream & PileUpStream::operator=

(

PileUpStream &&

rhs

)

Definition at line 49 of file PileUpStream.cxx.

                                          {
  if (this != &rhs) {
    m_name=rhs.m_name;
    p_svcLoc=rhs.p_svcLoc;
    p_sel=rhs.p_sel; 
    p_SG=rhs.p_SG; 
    p_iter=rhs.p_iter;
    p_mergeSvc = rhs.p_mergeSvc;
    m_ownEvtIterator=rhs.m_ownEvtIterator;
    rhs.m_ownEvtIterator=false;
    m_neverLoaded=rhs.m_neverLoaded;
    m_ownStore=rhs.m_ownStore;
    rhs.m_ownStore=false;
    m_used=rhs.m_used;
    m_hasRing=rhs.m_hasRing;
    m_iOriginalRing=rhs.m_iOriginalRing;
  }  
  return *this;
}

originalIRing()

unsigned int PileUpStream::originalIRing ( ) const

inline

index of the ring for which this stream was used originally

Definition at line 87 of file PileUpStream.h.

{ return m_iOriginalRing; }

resetUsed()

void PileUpStream::resetUsed ( )

inline

Definition at line 84 of file PileUpStream.h.

{m_used=false;}

selector() [1/2]

IEvtSelector& PileUpStream::selector ( )

inline

Definition at line 59 of file PileUpStream.h.

{ return *p_sel; }

selector() [2/2]

const IEvtSelector& PileUpStream::selector ( ) const

inline

Definition at line 62 of file PileUpStream.h.

{ return *p_sel; }

serviceLocator()

ISvcLocator* PileUpStream::serviceLocator ( )

inlineprivate

Definition at line 97 of file PileUpStream.h.

{ return p_svcLoc; }

setActiveStore()

void PileUpStream::setActiveStore ( )

private

set ActiveStore

Definition at line 179 of file PileUpStream.cxx.

{
  store().makeCurrent();
}

setLevel()

void AthMessaging::setLevel ( MSG::Level  lvl )

inherited

Change the current logging level.

Use this rather than msg().setLevel() for proper operation with MT.

Definition at line 28 of file AthMessaging.cxx.

{
  m_lvl = lvl;
}

setOriginalIRing()

void PileUpStream::setOriginalIRing ( unsigned int  iR )

inline

Definition at line 88 of file PileUpStream.h.

                                         {
    if (!m_hasRing) {
      m_hasRing=true;
      m_iOriginalRing=iR; 
    }
  }

setupStore() [1/2]

bool PileUpStream::setupStore

(

)

setup input and overlay selectors and iters

Definition at line 113 of file PileUpStream.cxx.

{
  assert( p_iter );
  assert( p_sel );
  bool rc(true);
  std::string storeName(name() + "_SG");
 
  //start by looking for the store directly: in overlay jobs it may already be there
  const bool DONOTCREATE(false);
  rc = (serviceLocator()->service(storeName, p_SG, DONOTCREATE)).isSuccess();
  if (rc) {
    m_ownStore = false;
  } else {
    //not there, create one cloning the master store
    Service *child;
    //if the parent store is not there barf
    //remember the clone function also initializes the service if needed
    IService* pIS(0);
    rc = ((serviceLocator()->getService("StoreGateSvc", pIS)).isSuccess() &&
          CloneService::clone(pIS, storeName, child).isSuccess() &&
          0 != (p_SG = dynamic_cast<StoreGateSvc*>(child)));
    if ( rc )  {
      m_ownStore = true;
      // further initialization of the cloned service 
      rc = (p_SG->sysInitialize()).isSuccess();      
      p_SG->setStoreID(StoreID::PILEUP_STORE); //needed by ProxyProviderSvc
    } //clones
  }
  if (rc) {
    //if the selector is an address provider like the AthenaPool one, 
    //create a dedicated ProxyProviderSvc and associate it to the store
    IAddressProvider* pIAP(dynamic_cast<IAddressProvider*>(p_sel));
    if (0 != pIAP) {
      IProxyProviderSvc* pPPSvc(0);
      std::string PPSName(name() + "_PPS");
      ISvcManager* pISM(dynamic_cast<ISvcManager*>(serviceLocator()));
      if ( 0 != pISM &&  
       (pISM->declareSvcType(PPSName, "ProxyProviderSvc")).isSuccess() &&
       //check the service is not there then create it
       (serviceLocator()->service(PPSName,
                      pPPSvc,
                      true)).isSuccess()) {
    pPPSvc->addProvider(pIAP);
        IService* pSAthPoolAddProv(0);
        IAddressProvider* pAthPoolAddProv(0);
        if (serviceLocator()->service("AthenaPoolAddressProviderSvc", pSAthPoolAddProv).isSuccess() &&
            0 != (pAthPoolAddProv = dynamic_cast<IAddressProvider*>(pSAthPoolAddProv))) {
          pPPSvc->addProvider(pAthPoolAddProv);
        } else {
          ATH_MSG_WARNING ( "could not add AthenaPoolAddressProviderSvc as AddresssProvider for "<< PPSName );
        }
    IService* pSAddrRemap(0);
    IAddressProvider* pAddrRemap(0);
    if (serviceLocator()->service("AddressRemappingSvc", pSAddrRemap).isSuccess() &&
        0 != (pAddrRemap = dynamic_cast<IAddressProvider*>(pSAddrRemap))) {
      pPPSvc->addProvider(pAddrRemap);
    } else {
      ATH_MSG_WARNING ( "could not add AddressRemappingSvc as AddresssProvider for "<< PPSName );
    }
    p_SG->setProxyProviderSvc(pPPSvc);
      } 
    } //valid address provider
  }
  return rc;
}

setupStore() [2/2]

bool PileUpStream::setupStore ( bool  )

inline

backward compatibility

Definition at line 76 of file PileUpStream.h.

{ return setupStore();}

setUsed()

void PileUpStream::setUsed ( )

inline

Definition at line 83 of file PileUpStream.h.

{m_used=true;}

store() [1/2]

StoreGateSvc& PileUpStream::store ( )

inline

Definition at line 61 of file PileUpStream.h.

{ return *p_SG; }

store() [2/2]

const StoreGateSvc& PileUpStream::store ( ) const

inline

Definition at line 64 of file PileUpStream.h.

{ return *p_SG; }

used()

bool PileUpStream::used ( ) const

inline

has this stream already be used in this event?

Definition at line 82 of file PileUpStream.h.

{return m_used;}

Member Data Documentation

ATLAS_THREAD_SAFE

std::atomic_flag m_initialized AthMessaging::ATLAS_THREAD_SAFE = ATOMIC_FLAG_INIT

mutableprivateinherited

Messaging initialized (initMessaging)

Definition at line 141 of file AthMessaging.h.

m_hasRing

bool PileUpStream::m_hasRing

private

has this stream already been used? (for the current event)

Definition at line 134 of file PileUpStream.h.

m_imsg

std::atomic<IMessageSvc*> AthMessaging::m_imsg { nullptr }

mutableprivateinherited

MessageSvc pointer.

Definition at line 135 of file AthMessaging.h.

m_iOriginalRing

unsigned int PileUpStream::m_iOriginalRing

private

Definition at line 135 of file PileUpStream.h.

m_lvl

std::atomic<MSG::Level> AthMessaging::m_lvl { MSG::NIL }

mutableprivateinherited

Current logging level.

Definition at line 138 of file AthMessaging.h.

m_msg_tls

boost::thread_specific_ptr<MsgStream> AthMessaging::m_msg_tls

mutableprivateinherited

MsgStream instance (a std::cout like with print-out levels)

Definition at line 132 of file AthMessaging.h.

m_name

std::string PileUpStream::m_name

private

Stream name.

Definition at line 115 of file PileUpStream.h.

m_neverLoaded

bool PileUpStream::m_neverLoaded

private

do we own p_iter?

Private message stream member

Definition at line 130 of file PileUpStream.h.

m_nm

std::string AthMessaging::m_nm

privateinherited

Message source name.

Definition at line 129 of file AthMessaging.h.

m_ownEvtIterator

bool PileUpStream::m_ownEvtIterator

private

Definition at line 128 of file PileUpStream.h.

m_ownStore

bool PileUpStream::m_ownStore

private

has an event been loaded into this stream?

Definition at line 131 of file PileUpStream.h.

m_used

bool PileUpStream::m_used

private

is p_SG a store we cloned from the master one?

Definition at line 133 of file PileUpStream.h.

p_iter

EvtIterator* PileUpStream::p_iter

private

Input Iterators.

Definition at line 123 of file PileUpStream.h.

p_mergeSvc

PileUpMergeSvc* PileUpStream::p_mergeSvc

private

Definition at line 125 of file PileUpStream.h.

p_sel

IEvtSelector* PileUpStream::p_sel

private

Selector.

Definition at line 119 of file PileUpStream.h.

p_SG

StoreGateSvc* PileUpStream::p_SG

private

StoreGateSvc;.

Definition at line 121 of file PileUpStream.h.

p_svcLoc

ISvcLocator* PileUpStream::p_svcLoc

private

ServiceLocator.

Definition at line 117 of file PileUpStream.h.

---

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

• PileUpStream.h
• PileUpStream.cxx