Athena::DelayedConditionsCleanerSvc Class Reference

Clean conditions containers after a delay. More...

#include <DelayedConditionsCleanerSvc.h>

Inheritance diagram for Athena::DelayedConditionsCleanerSvc:

Collaboration diagram for Athena::DelayedConditionsCleanerSvc:

Classes
class	CondContInfo
	Information that we maintain about each conditions container. More...
struct	QueueItem
	Item in the work queue. More...

Public Types
typedef CondContBase::key_type	key_type
	Packed key type.

Public Member Functions
	DelayedConditionsCleanerSvc (const std::string &name, ISvcLocator *svc)
	Standard Gaudi constructor.
	~DelayedConditionsCleanerSvc ()
	Standard destructor.
virtual StatusCode	initialize () override
	Standard Gaudi initialize method.
virtual StatusCode	finalize () override
	Standard Gaudi finalize method.
virtual StatusCode	event (const EventContext &ctx, bool allowAsync) override
	Called at the start of each event.
virtual StatusCode	condObjAdded (const EventContext &ctx, CondContBase &cc) override
	Called after a conditions object has been added.
virtual StatusCode	printStats () const override
	Print some statistics about the garbage collection.
virtual StatusCode	reset () override
	Clear the internal state of the service.

Private Types
typedef CxxUtils::Ring< key_type >	Ring
	Ring buffer holding most recent IOV keys of a given type.
using	KeyType = CondContBase::KeyType
	Run+LBN or timestamp key?
typedef std::array< std::vector< key_type >, 2 >	twoKeys_t
typedef std::unordered_map< CondContBase *, CondContInfo >	CCInfoMap_t
	Map of information, indexed by the conditions container.
typedef std::mutex	mutex_t
	Serialize access to m_ccinfo and m_work.
typedef std::lock_guard< mutex_t >	lock_t

Private Member Functions
twoKeys_t	getKeys (const Ring &runLBRing, const Ring &TSRing) const
void	scheduleClean (std::vector< CondContInfo * > &&cis, twoKeys_t &&twoKeys, bool allowAsync)
	Do cleaning for a set of containers.
void	cleanContainers (std::vector< CondContInfo * > &&cis, twoKeys_t &&twoKeys)
	Clean a set of containers.
bool	cleanContainer (CondContInfo *ci, const twoKeys_t &keys) const
	Clean a single container.

Private Attributes
Ring	m_runlbn
	Two ring buffers for recent IOV keys, one for run+LBN and one for timestamp.
Ring	m_timestamp
std::vector< key_type >	m_slotLBN
	IOV keys currently in use for each slot.
std::vector< key_type >	m_slotTimestamp
CCInfoMap_t	m_ccinfo
std::priority_queue< QueueItem >	m_work
	Priority queue of pending cleaning requests.
mutex_t	m_workMutex
size_t	m_nEvents = 0
	Priority queue statistics.
size_t	m_queueSum = 0
size_t	m_workRemoved = 0
size_t	m_maxQueue = 0
std::atomic< int >	m_cleanTasks {0}
	Number of active asynchronous cleaning tasks.
std::unique_ptr< DelayedConditionsCleanerSvcProps >	m_props
	Component properties.

Friends
class	DelayedConditionsCleanerTask

Detailed Description

Clean conditions containers after a delay.

This is an implementation of IConditionsCleanerSvc, for doing garbage collection of conditions objects.

Briefly, it works like this.

When a conditions object is added (condObjAdded interface), we put an entry in a priority queue, saying that we want to clean this container CleanDelay events later.

On each event (event interface) we put the current IOV keys (run+LBN and timestamp) into ring buffers (of size RingSize). If the topmost entry of the priority queue has come due, then we pull off that entry and other entries due up to LookAhead events later. We then do a trim operation on each of the conditions containers, removing conditions objects from the oldest first that do not match any of the IOV keys in the ring buffer.

The cleaning can optionally be done as an asynchronous TBB job if Async is true and allowAsync=true is passed to event.

Definition at line 57 of file DelayedConditionsCleanerSvc.h.

Member Typedef Documentation

CCInfoMap_t

typedef std::unordered_map<CondContBase*, CondContInfo> Athena::DelayedConditionsCleanerSvc::CCInfoMap_t

private

Map of information, indexed by the conditions container.

Definition at line 214 of file DelayedConditionsCleanerSvc.h.

key_type

typedef CondContBase::key_type Athena::DelayedConditionsCleanerSvc::key_type

Packed key type.

Definition at line 62 of file DelayedConditionsCleanerSvc.h.

KeyType

using Athena::DelayedConditionsCleanerSvc::KeyType = CondContBase::KeyType

private

Run+LBN or timestamp key?

Definition at line 133 of file DelayedConditionsCleanerSvc.h.

lock_t

typedef std::lock_guard<mutex_t> Athena::DelayedConditionsCleanerSvc::lock_t

private

Definition at line 241 of file DelayedConditionsCleanerSvc.h.

mutex_t

typedef std::mutex Athena::DelayedConditionsCleanerSvc::mutex_t

private

Serialize access to m_ccinfo and m_work.

Definition at line 240 of file DelayedConditionsCleanerSvc.h.

Ring

typedef CxxUtils::Ring<key_type> Athena::DelayedConditionsCleanerSvc::Ring

private

Ring buffer holding most recent IOV keys of a given type.

Definition at line 129 of file DelayedConditionsCleanerSvc.h.

twoKeys_t

typedef std::array<std::vector<key_type>,2> Athena::DelayedConditionsCleanerSvc::twoKeys_t

private

Definition at line 135 of file DelayedConditionsCleanerSvc.h.

Constructor & Destructor Documentation

DelayedConditionsCleanerSvc()

Athena::DelayedConditionsCleanerSvc::DelayedConditionsCleanerSvc	(	const std::string &	name,
		ISvcLocator *	svc )

Standard Gaudi constructor.

Parameters

name	Service name.
svc	Service locator.

Definition at line 155 of file DelayedConditionsCleanerSvc.cxx.

  : base_class (name, svc),
    m_props (std::make_unique<DelayedConditionsCleanerSvcProps> (this))
{
}

~DelayedConditionsCleanerSvc()

Athena::DelayedConditionsCleanerSvc::~DelayedConditionsCleanerSvc

(

)

Standard destructor.

Needed to avoid problems with unique_ptr

Definition at line 493 of file DelayedConditionsCleanerSvc.cxx.

{}

Member Function Documentation

cleanContainer()

bool Athena::DelayedConditionsCleanerSvc::cleanContainer ( CondContInfo * ci, const twoKeys_t & twoKeys ) const

private

Clean a single container.

Parameters

ci	The container to clean.
keys	Set of IOV keys for recent events.

Returns true if anything was removed from the container,

Parameters

ci	The container to clean.
keyType	Run+LBN or timestamp keys?
keys	Set of IOV keys for recent events.

Returns true if anything was removed from the container,

Definition at line 468 of file DelayedConditionsCleanerSvc.cxx.

{
  size_t n = ci->m_cc.trim (twoKeys[0],twoKeys[1]);
 
  ++ci->m_nClean;
  ci->m_nRemoved += n;
  switch (n) {
  case 0:
    ++ci->m_removed0;
    break;
  case 1:
    ++ci->m_removed1;
    break;
  default:
    ++ci->m_removed2plus;
  }
 
  return n > 0;
}

cleanContainers()

void Athena::DelayedConditionsCleanerSvc::cleanContainers ( std::vector< CondContInfo * > && cis, twoKeys_t && twoKeys )

private

Clean a set of containers.

Parameters

cis	Set of containers to clean.
keys	Set of IOV keys for recent events.

Definition at line 423 of file DelayedConditionsCleanerSvc.cxx.

{
  // FIXME: Some conditions objects have pointers to parts of other
  // conditions objects, which violates the lifetime guarantees of
  // conditions containers (a pointer you get from a conditions container
  // is guaranteed to be valid until the end of the current event, but
  // not past that).  In some cases, this can be dealt with by replacing
  // the pointers with CondLink, but that is sometimes rather inconvenient.
  // Try to work around this for now by ensuring that when we delete an object,
  // we also try to clean the containers of other objects that may depend
  // on it.
  std::vector<CondContInfo*> toclean = std::move (cis);
  std::unordered_set<CondContInfo*> cleaned (toclean.begin(), toclean.end());
  while (!toclean.empty()) {
    std::vector<CondContInfo*> newclean;
    for (CondContInfo* ci : toclean) {
      if (cleanContainer (ci, twoKeys)) {
        lock_t lock (m_workMutex);
        for (CondContBase* dep : ci->m_cc.getDeps()) {
          CCInfoMap_t::iterator it = m_ccinfo.find (dep);
          // If we don't find it, then dep must have no conditions objects.
          if (it != m_ccinfo.end()) {
            CondContInfo* ci_dep = &it->second;
            if (cleaned.insert (ci_dep).second) {
              newclean.push_back (ci_dep);
            }
          }
        }
      }
    }
    toclean = std::move (newclean);
  }
}

condObjAdded()

StatusCode Athena::DelayedConditionsCleanerSvc::condObjAdded ( const EventContext & ctx, CondContBase & cc )

overridevirtual

Called after a conditions object has been added.

Parameters

ctx	The current event context.
cc	The container to which the object was added.

Definition at line 256 of file DelayedConditionsCleanerSvc.cxx.

{
  // Add this container to the priority queue.
  lock_t lock (m_workMutex);
  CCInfoMap_t::iterator it = m_ccinfo.find (&cc);
  if (it == m_ccinfo.end()) {
    it = m_ccinfo.emplace (&cc, CondContInfo (cc)).first;
  }
 
  EventContext::ContextEvt_t evt = ctx.evt();
  m_work.emplace (evt + m_props->m_cleanDelay, it->second);
  return StatusCode::SUCCESS;
}

event()

StatusCode Athena::DelayedConditionsCleanerSvc::event ( const EventContext & ctx, bool allowAsync )

overridevirtual

Called at the start of each event.

Parameters

ctx	The current event context.
allowAsync	If true, then cleaning may be run in an asynchronous TBB task.

Definition at line 188 of file DelayedConditionsCleanerSvc.cxx.

{
  // Push the IOV key for the current event into the ring buffers.
  // Also save in the per-slot arrays.
  key_type key_lbn = CondContBase::keyFromRunLBN (ctx.eventID());
  key_type key_ts  = CondContBase::keyFromTimestamp (ctx.eventID());
  m_runlbn.push (key_lbn);
  m_timestamp.push (key_ts);
  EventContext::ContextID_t slot = ctx.slot();
  if (slot != EventContext::INVALID_CONTEXT_ID) {
    m_slotLBN[slot] = key_lbn;
    m_slotTimestamp[slot] = key_ts;
  }
 
  // Return now if an asynchronous cleaning task is still running ---
  // we don't want to start a new one yet.  We'll check pending work
  // on the next call.
  if (m_cleanTasks > 0) {
    return StatusCode::SUCCESS;
  }
 
  // Collect conditions containers in need of cleaning.
  std::vector<CondContInfo*> ci_vec;
  {
    lock_t lock (m_workMutex);
    // Is it time to clean the container at the top of the work queue?
    if (!m_work.empty() && m_work.top().m_evt <= ctx.evt()) {
      ++m_nEvents;
      size_t sz = m_work.size();
      m_queueSum += sz;
      m_maxQueue = std::max (m_maxQueue, sz);
 
      // Yes.  Put it on the correct list.  Also look ahead in the queue
      // a bit; if there are other containers that we want to clean soon,
      // go ahead and do them now.
      do {
        CondContInfo* ci = m_work.top().m_ci;
        switch (ci->m_cc.keyType()) {
        case KeyType::SINGLE:
          break;
        case KeyType::RUNLBN:
        case KeyType::MIXED:
        case KeyType::TIMESTAMP:
          ci_vec.push_back (ci);
          break;
        default:
          std::abort();
        }
        m_work.pop();
        ++m_workRemoved;
      } while (!m_work.empty() && m_work.top().m_evt <= ctx.evt() + m_props->m_lookAhead);
    }
  }
 
  // Clean the containers.
  if (!ci_vec.empty()) {
    scheduleClean (std::move (ci_vec), getKeys(m_runlbn,m_timestamp),
                   allowAsync);
  }
  return StatusCode::SUCCESS;
}

finalize()

StatusCode Athena::DelayedConditionsCleanerSvc::finalize ( )

overridevirtual

Standard Gaudi finalize method.

Definition at line 498 of file DelayedConditionsCleanerSvc.cxx.

{
  ATH_CHECK( printStats() );
  return StatusCode::SUCCESS;
}

getKeys()

DelayedConditionsCleanerSvc::twoKeys_t Athena::DelayedConditionsCleanerSvc::getKeys ( const Ring & runLBRing, const Ring & TSRing ) const

private

Sort the key array and remove duplicates. We expect that the key array is probably ‘almost’ sorted. std::sort, at least in the gcc implementation, is designed to perform well in such cases.

Definition at line 345 of file DelayedConditionsCleanerSvc.cxx.

                                                                                    {
  
  // Get a copy of the contents of the ring buffer holding runLumi and time-stamp keys
  std::vector<key_type> runLBKeys=runLBRing.getKeysDedup();
  std::vector<key_type> TSKeys=TSRing.getKeysDedup();
 
  // Add in the keys for the currently-executing slots.
  // These are very likely to already be in the ring, but that's
  // not absolutely guaranteed.
  // FIXME: This probably does another memory allocation, due to
  // growing the buffer.  Would be nice to avoid that.
  runLBKeys.insert (runLBKeys.end(), m_slotLBN.begin(), m_slotLBN.end());
  TSKeys.insert(TSKeys.end(), m_slotTimestamp.begin(), m_slotTimestamp.end());
 
  twoKeys_t result{std::move(runLBKeys), std::move(TSKeys)};

  for ( auto& keys : result ) {
    std::sort (keys.begin(), keys.end());
    auto end = std::unique (keys.begin(), keys.end());
    keys.resize (end - keys.begin());
  }
 
 
  return result;
}

initialize()

StatusCode Athena::DelayedConditionsCleanerSvc::initialize ( )

overridevirtual

Standard Gaudi initialize method.

Definition at line 166 of file DelayedConditionsCleanerSvc.cxx.

{
  // Set the ring buffer sizes.
  m_runlbn.reset (m_props->m_ringSize);
  m_timestamp.reset (m_props->m_ringSize);
 
  ATH_CHECK( m_props->m_rcu.retrieve() );
  size_t nslots = m_props->m_rcu->getNumSlots();
  m_slotLBN.resize (nslots);
  m_slotTimestamp.resize (nslots);
 
  return StatusCode::SUCCESS;
}

printStats()

StatusCode Athena::DelayedConditionsCleanerSvc::printStats ( ) const

overridevirtual

Print some statistics about the garbage collection.

Would generally be called in finalize(), but broken out as a separate interface for testing/debugging purposes.

Definition at line 277 of file DelayedConditionsCleanerSvc.cxx.

{
  // Suppress output if we didn't actually do anything.
  if (m_nEvents == 0) {
    return StatusCode::SUCCESS;
  }
 
  ATH_MSG_INFO( "Conditions container statistics" );
  ATH_MSG_INFO( CxxUtils::strformat ("  Work q: Max size: %zu  (%zu queries) ",
                                     m_maxQueue, m_nEvents) );
  size_t den = std::max (m_nEvents, 1lu);
  ATH_MSG_INFO( CxxUtils::strformat ("          Avg size: %.2f / Avg removed: %.2f",
                                     static_cast<float>(m_queueSum)/den,
                                     static_cast<float>(m_workRemoved)/den) );
 
  std::vector<const CondContInfo*> infos;
  for (const auto& p : m_ccinfo) {
    infos.push_back (&p.second);
  }
  std::sort (infos.begin(), infos.end(),
             [](const CondContInfo* a, const CondContInfo* b)
             { return a->m_cc.id().key() < b->m_cc.id().key(); });
 
  for (const CondContInfo* ci : infos) {
    ATH_MSG_INFO( CxxUtils::strformat ("  %-20s nInserts %6zu maxSize %3zu",
                                       ci->m_cc.id().key().c_str(),
                                       ci->m_cc.nInserts(),
                                       ci->m_cc.maxSize()) );
    den = std::max (ci->m_nClean, 1lu);
    ATH_MSG_INFO( CxxUtils::strformat ("    nClean %zu avgRemoved %.2f 0/1/2+ %zu/%zu/%zu",
                                       ci->m_nClean,
                                       static_cast<float> (ci->m_nRemoved) / den,
                                       ci->m_removed0,
                                       ci->m_removed1,
                                       ci->m_removed2plus) );
  }
                                     
  return StatusCode::SUCCESS;
}

reset()

StatusCode Athena::DelayedConditionsCleanerSvc::reset ( )

overridevirtual

Clear the internal state of the service.

Only for testing. Don't call if any other thread may be touching the service.

Definition at line 322 of file DelayedConditionsCleanerSvc.cxx.

{
  m_runlbn.reset (m_props->m_ringSize);
  m_timestamp.reset (m_props->m_ringSize);
 
  std::fill (m_slotLBN.begin(), m_slotLBN.end(), 0);
  std::fill (m_slotTimestamp.begin(), m_slotTimestamp.end(), 0);
 
  m_ccinfo.clear();
  std::priority_queue<QueueItem> tmp;
  m_work.swap (tmp);
 
  m_nEvents = 0;
  m_queueSum = 0;
  m_workRemoved = 0;
  m_maxQueue = 0;
  m_cleanTasks = 0;
 
  return StatusCode::SUCCESS;
}

scheduleClean()

void Athena::DelayedConditionsCleanerSvc::scheduleClean ( std::vector< CondContInfo * > && cis, twoKeys_t && twoKeys, bool allowAsync )

private

Do cleaning for a set of containers.

Parameters

cis	Set of containers to clean.
ring	Ring buffer with recent IOV keys.
slotKeys	Vector of current keys for all slots.
allowAsync	Can this task run asynchronously?

This will either run cleaning directly, or submit it as a TBB task.

Definition at line 386 of file DelayedConditionsCleanerSvc.cxx.

{
  // Remove any duplicates from the list of containers.
  std::sort (cis.begin(), cis.end());
  auto pos = std::unique (cis.begin(), cis.end());
  cis.resize (pos - cis.begin());
 
  if (allowAsync && m_props->m_async) {
#if USE_ASYNC_TASK
    // Queue cleaning as a TBB task.
    // Count that we have another executing task.
    ++m_cleanTasks;
 
    // Create the TBB task and queue it.
    // TBB will delete the task object after it completes.
    tbb::task* t = new (tbb::task::allocate_root())
      DelayedConditionsCleanerTask (*this, std::move (cis),
                                    std::move (twoKeys));
    tbb::task::enqueue (*t);
#endif
  }
  else
  {
    // Call cleaning directly.
    cleanContainers (std::move (cis), std::move (twoKeys));
  }
}

Friends And Related Symbol Documentation

DelayedConditionsCleanerTask

friend class DelayedConditionsCleanerTask

friend

Definition at line 125 of file DelayedConditionsCleanerSvc.h.

Member Data Documentation

m_ccinfo

CCInfoMap_t Athena::DelayedConditionsCleanerSvc::m_ccinfo

private

Definition at line 215 of file DelayedConditionsCleanerSvc.h.

m_cleanTasks

std::atomic<int> Athena::DelayedConditionsCleanerSvc::m_cleanTasks {0}

private

Number of active asynchronous cleaning tasks.

Definition at line 253 of file DelayedConditionsCleanerSvc.h.

{0};

m_maxQueue

size_t Athena::DelayedConditionsCleanerSvc::m_maxQueue = 0

private

Definition at line 249 of file DelayedConditionsCleanerSvc.h.

m_nEvents

size_t Athena::DelayedConditionsCleanerSvc::m_nEvents = 0

private

Priority queue statistics.

Definition at line 246 of file DelayedConditionsCleanerSvc.h.

m_props

std::unique_ptr<DelayedConditionsCleanerSvcProps> Athena::DelayedConditionsCleanerSvc::m_props

private

Component properties.

Definition at line 256 of file DelayedConditionsCleanerSvc.h.

m_queueSum

size_t Athena::DelayedConditionsCleanerSvc::m_queueSum = 0

private

Definition at line 247 of file DelayedConditionsCleanerSvc.h.

m_runlbn

Ring Athena::DelayedConditionsCleanerSvc::m_runlbn

private

Two ring buffers for recent IOV keys, one for run+LBN and one for timestamp.

We only access these from event(), which is called from the event loop, so no locking is needed.

Definition at line 205 of file DelayedConditionsCleanerSvc.h.

m_slotLBN

std::vector<key_type> Athena::DelayedConditionsCleanerSvc::m_slotLBN

private

IOV keys currently in use for each slot.

Definition at line 209 of file DelayedConditionsCleanerSvc.h.

m_slotTimestamp

std::vector<key_type> Athena::DelayedConditionsCleanerSvc::m_slotTimestamp

private

Definition at line 210 of file DelayedConditionsCleanerSvc.h.

m_timestamp

Ring Athena::DelayedConditionsCleanerSvc::m_timestamp

private

Definition at line 206 of file DelayedConditionsCleanerSvc.h.

m_work

std::priority_queue<QueueItem> Athena::DelayedConditionsCleanerSvc::m_work

private

Priority queue of pending cleaning requests.

Definition at line 237 of file DelayedConditionsCleanerSvc.h.

m_workMutex

mutex_t Athena::DelayedConditionsCleanerSvc::m_workMutex

mutableprivate

Definition at line 242 of file DelayedConditionsCleanerSvc.h.

m_workRemoved

size_t Athena::DelayedConditionsCleanerSvc::m_workRemoved = 0

private

Definition at line 248 of file DelayedConditionsCleanerSvc.h.

---

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

• DelayedConditionsCleanerSvc.h
• DelayedConditionsCleanerSvc.cxx