edm::WaitingTaskList Class Reference

#include <WaitingTaskList.h>

Classes
struct	WaitNode

Public Member Functions
void	add (oneapi::tbb::task_group *, WaitingTask *)
	Adds task to the waiting list. More...
void	add (WaitingTaskHolder)
	Adds task to the waiting list. More...
void	doneWaiting (std::exception_ptr iPtr)
	Signals that the resource is now available and tasks should be spawned. More...
const WaitingTaskList &	operator= (const WaitingTaskList &)=delete
void	presetTaskAsFailed (std::exception_ptr iExcept)
void	reset ()
	Resets access to the resource so that added tasks will wait. More...
	WaitingTaskList (unsigned int iInitialSize=2)
	Constructor. More...
	WaitingTaskList (const WaitingTaskList &)=delete
	~WaitingTaskList ()=default

Private Member Functions
void	announce ()
WaitNode *	createNode (oneapi::tbb::task_group *iGroup, WaitingTask *iTask)

Private Attributes
std::exception_ptr	m_exceptionPtr
std::atomic< WaitNode * >	m_head
std::atomic< unsigned int >	m_lastAssignedCacheIndex
std::unique_ptr< WaitNode[]>	m_nodeCache
unsigned int	m_nodeCacheSize
std::atomic< bool >	m_waiting

Detailed Description

Definition at line 84 of file WaitingTaskList.h.

Constructor & Destructor Documentation

WaitingTaskList() [1/2]

WaitingTaskList::WaitingTaskList ( unsigned int  iInitialSize = 2 )

explicit

Constructor.

The WaitingTaskList is initial set to waiting.

Parameters

[in]

iInitialSize

specifies the initial size of the cache used to hold waiting tasks. The value is only useful for optimization as the object can resize itself.

Definition at line 38 of file WaitingTaskList.cc.

    : m_head{nullptr},
      m_nodeCache{new WaitNode[iInitialSize]},
      m_nodeCacheSize{iInitialSize},
      m_lastAssignedCacheIndex{0},
      m_waiting{true} {
  auto nodeCache = m_nodeCache.get();
  for (auto it = nodeCache, itEnd = nodeCache + m_nodeCacheSize; it != itEnd; ++it) {
    it->m_fromCache = true;
  }
}

WaitingTaskList() [2/2]

edm::WaitingTaskList::WaitingTaskList ( const WaitingTaskList &  )

delete

~WaitingTaskList()

edm::WaitingTaskList::~WaitingTaskList ( )

default

Member Function Documentation

add() [1/2]

void WaitingTaskList::add	(	oneapi::tbb::task_group *	iGroup,
		WaitingTask *	iTask
	)

Adds task to the waiting list.

If doneWaiting() has already been called then the added task will immediately be run. If that is not the case then the task will be held until doneWaiting() is called and will then be run. Calls to add() and doneWaiting() can safely be done concurrently.

Definition at line 125 of file WaitingTaskList.cc.

References announce(), createNode(), edm::TaskBase::decrement_ref_count(), edm::WaitingTask::dependentTaskFailed(), edm::TaskBase::execute(), edm::TaskBase::increment_ref_count(), m_exceptionPtr, m_head, m_waiting, alignCSCRings::s, edm::WaitingTaskList::WaitNode::setNextNode(), and UNLIKELY.

Referenced by edm::Worker::callWhenDoneAsync(), edm::eventsetup::EventSetupRecordIOVQueue::checkForNewIOVs(), edm::Worker::doWorkAsync(), edm::Worker::doWorkNoPrefetchingAsync(), edm::eventsetup::EventSetupRecordIOVQueue::endIOVAsync(), edm::eventsetup::ESSourceDataProxyBase::needToPrefetch(), edm::eventsetup::Callback< T, TReturn, TRecord, TDecorator >::prefetchAsync(), edm::DelayedReaderInputProductResolver::prefetchAsync_(), edm::PuttableProductResolver::prefetchAsync_(), edm::UnscheduledProductResolver::prefetchAsync_(), edm::SwitchProducerProductResolver::prefetchAsync_(), edm::SwitchAliasProductResolver::prefetchAsync_(), edm::NoProcessProductResolver::prefetchAsync_(), edm::eventsetup::DataProxyTemplate< RecordT, DataT >::prefetchAsyncImpl(), edm::Path::processOneOccurrenceAsync(), counter.Counter::register(), SequenceTypes._TaskBase::remove(), SequenceTypes._TaskBase::replace(), and edm::eventsetup::EventSetupRecordIOVQueue::startNewIOVAsync().

                                                                         {
  iTask->increment_ref_count();
  if (!m_waiting) {
    if (UNLIKELY(bool(m_exceptionPtr))) {
      iTask->dependentTaskFailed(m_exceptionPtr);
    }
    if (0 == iTask->decrement_ref_count()) {
      iGroup->run([iTask]() {
        TaskSentry s{iTask};
        iTask->execute();
      });
    }
  } else {
    WaitNode* newHead = createNode(iGroup, iTask);
    //This exchange is sequentially consistent thereby
    // ensuring ordering between it and setNextNode
    WaitNode* oldHead = m_head.exchange(newHead);
    newHead->setNextNode(oldHead);

    //For the case where oldHead != nullptr,
    // even if 'm_waiting' changed, we don't
    // have to recheck since we beat 'announce()' in
    // the ordering of 'm_head.exchange' call so iTask
    // is guaranteed to be in the link list

    if (nullptr == oldHead) {
      if (!m_waiting) {
        //if finished waiting right before we did the
        // exchange our task will not be run. Also,
        // additional threads may be calling add() and swapping
        // heads and linking us to the new head.
        // It is safe to call announce from multiple threads
        announce();
      }
    }
  }
}

add() [2/2]

void WaitingTaskList::add

(

WaitingTaskHolder

iTask

)

Adds task to the waiting list.

Calls to add() and doneWaiting() can safely be done concurrently.

Definition at line 92 of file WaitingTaskList.cc.

References announce(), createNode(), edm::WaitingTaskHolder::doneWaiting(), edm::WaitingTaskHolder::group(), m_exceptionPtr, m_head, m_waiting, edm::WaitingTaskHolder::release_no_decrement(), edm::WaitingTaskList::WaitNode::setNextNode(), and TrackValidation_cff::task.

Referenced by counter.Counter::register(), SequenceTypes._TaskBase::remove(), and SequenceTypes._TaskBase::replace().

                                                 {
  if (!m_waiting) {
    if (m_exceptionPtr) {
      iTask.doneWaiting(m_exceptionPtr);
    }
  } else {
    auto task = iTask.release_no_decrement();
    WaitNode* newHead = createNode(iTask.group(), task);
    //This exchange is sequentially consistent thereby
    // ensuring ordering between it and setNextNode
    WaitNode* oldHead = m_head.exchange(newHead);
    newHead->setNextNode(oldHead);

    //For the case where oldHead != nullptr,
    // even if 'm_waiting' changed, we don't
    // have to recheck since we beat 'announce()' in
    // the ordering of 'm_head.exchange' call so iTask
    // is guaranteed to be in the link list

    if (nullptr == oldHead) {
      newHead->setNextNode(nullptr);
      if (!m_waiting) {
        //if finished waiting right before we did the
        // exchange our task will not be run. Also,
        // additional threads may be calling add() and swapping
        // heads and linking us to the new head.
        // It is safe to call announce from multiple threads
        announce();
      }
    }
  }
}

announce()

void WaitingTaskList::announce ( )

private

Handles running the tasks, safe to call from multiple threads

Definition at line 177 of file WaitingTaskList.cc.

References g, m_exceptionPtr, m_head, dqmiodumpmetadata::n, GetRecoTauVFromDQM_MC_cff::next, alignCSCRings::s, submitPVValidationJobs::t, and UNLIKELY.

Referenced by add(), and doneWaiting().

                               {
  //Need a temporary storage since one of these tasks could
  // cause the next event to start processing which would refill
  // this waiting list after it has been reset
  WaitNode* n = m_head.exchange(nullptr);
  WaitNode* next;
  while (n) {
    //it is possible that 'WaitingTaskList::add' is running in a different
    // thread and we have a new 'head' but the old head has not yet been
    // attached to the new head (we identify this since 'nextNode' will return itself).
    //  In that case we have to wait until the link has been established before going on.
    while (n == (next = n->nextNode())) {
      hardware_pause();
    }
    auto t = n->m_task;
    auto g = n->m_group;
    if (UNLIKELY(bool(m_exceptionPtr))) {
      t->dependentTaskFailed(m_exceptionPtr);
    }
    if (!n->m_fromCache) {
      delete n;
    }
    n = next;

    //the task may indirectly call WaitingTaskList::reset
    // so we need to call spawn after we are done using the node.
    if (0 == t->decrement_ref_count()) {
      g->run([t]() {
        TaskSentry s{t};
        t->execute();
      });
    }
  }
}

createNode()

WaitingTaskList::WaitNode * WaitingTaskList::createNode ( oneapi::tbb::task_group *  iGroup, WaitingTask *  iTask  )

private

Definition at line 72 of file WaitingTaskList.cc.

References edm::WaitingTaskList::WaitNode::m_fromCache, edm::WaitingTaskList::WaitNode::m_group, m_lastAssignedCacheIndex, edm::WaitingTaskList::WaitNode::m_next, m_nodeCache, m_nodeCacheSize, and edm::WaitingTaskList::WaitNode::m_task.

Referenced by add().

                                                                                                    {
  unsigned int index = m_lastAssignedCacheIndex++;

  WaitNode* returnValue;
  if (index < m_nodeCacheSize) {
    returnValue = m_nodeCache.get() + index;
  } else {
    returnValue = new WaitNode;
    returnValue->m_fromCache = false;
  }
  returnValue->m_task = iTask;
  returnValue->m_group = iGroup;
  //No other thread can see m_next yet. The caller to create node
  // will be doing a synchronization operation anyway which will
  // make sure m_task and m_next are synched across threads
  returnValue->m_next.store(returnValue, std::memory_order_relaxed);

  return returnValue;
}

doneWaiting()

void WaitingTaskList::doneWaiting

(

std::exception_ptr

iPtr

)

Signals that the resource is now available and tasks should be spawned.

The owner of the resource calls this function to allow the waiting tasks to start accessing it. If the task fails, a non 'null' std::exception_ptr should be used. To have tasks wait again one must call reset(). Calls to add() and doneWaiting() can safely be done concurrently.

Definition at line 212 of file WaitingTaskList.cc.

References announce(), m_exceptionPtr, and m_waiting.

Referenced by edm::Worker::doWorkNoPrefetchingAsync(), edm::Path::finished(), edm::DelayedReaderInputProductResolver::prefetchAsync_(), edm::UnscheduledProductResolver::prefetchAsync_(), edm::SwitchProducerProductResolver::prefetchAsync_(), edm::SwitchAliasProductResolver::prefetchAsync_(), edm::eventsetup::DataProxyTemplate< RecordT, DataT >::prefetchAsyncImpl(), edm::eventsetup::ESSourceDataProxyBase::prefetchAsyncImplTemplate(), edm::Worker::prePrefetchSelectionAsync(), edm::SwitchProducerProductResolver::putProduct(), edm::LuminosityBlockProcessingStatus::resetResources(), edm::Worker::runModuleAfterAsyncPrefetch(), edm::eventsetup::Callback< T, TReturn, TRecord, TDecorator >::runProducerAsync(), edm::NoProcessProductResolver::setCache(), edm::Worker::skipOnPath(), edm::eventsetup::EventSetupRecordIOVQueue::startNewIOVAsync(), edm::eventsetup::synchronousEventSetupForInstance(), and edm::LuminosityBlockProcessingStatus::~LuminosityBlockProcessingStatus().

                                                       {
  m_exceptionPtr = iPtr;
  m_waiting = false;
  announce();
}

operator=()

const WaitingTaskList& edm::WaitingTaskList::operator= ( const WaitingTaskList &  )

delete

presetTaskAsFailed()

void WaitingTaskList::presetTaskAsFailed

(

std::exception_ptr

iExcept

)

Use in the case where you need to inform the parent task of a failure before some other child task which may be run later reports a different, but related failure. You must later call doneWaiting with same exception later in the same thread.

Definition at line 163 of file WaitingTaskList.cc.

References edm::WaitingTask::dependentTaskFailed(), m_head, edm::WaitingTaskList::WaitNode::m_task, m_waiting, GetRecoTauVFromDQM_MC_cff::next, and edm::WaitingTaskList::WaitNode::nextNode().

Referenced by edm::Path::workerFinished().

                                                                 {
  if (iExcept and m_waiting) {
    WaitNode* node = m_head.load();
    while (node) {
      WaitNode* next;
      while (node == (next = node->nextNode())) {
        hardware_pause();
      }
      node->m_task->dependentTaskFailed(iExcept);
      node = next;
    }
  }
}

reset()

void WaitingTaskList::reset

(

void

)

Resets access to the resource so that added tasks will wait.

The owner of the resouce calls reset() to make tasks wait. Calling reset() is NOT thread safe. The system must guarantee that no tasks are using the resource when reset() is called and neither add() nor doneWaiting() can be called concurrently with reset().

Definition at line 53 of file WaitingTaskList.cc.

References cms::cuda::assert(), m_exceptionPtr, m_head, m_lastAssignedCacheIndex, m_nodeCache, m_nodeCacheSize, and m_waiting.

Referenced by edm::eventsetup::EventSetupRecordIOVQueue::endIOVAsync(), edm::eventsetup::ESSourceDataProxyBase::invalidateCache(), edm::eventsetup::DataProxyTemplate< RecordT, DataT >::invalidateCache(), edm::eventsetup::Callback< T, TReturn, TRecord, TDecorator >::newRecordComing(), edm::Path::processOneOccurrenceAsync(), edm::Worker::reset(), edm::DelayedReaderInputProductResolver::resetProductData_(), edm::UnscheduledProductResolver::resetProductData_(), edm::SwitchBaseProductResolver::resetProductData_(), and edm::NoProcessProductResolver::resetProductData_().

                            {
  m_exceptionPtr = std::exception_ptr{};
  unsigned int nSeenTasks = m_lastAssignedCacheIndex;
  m_lastAssignedCacheIndex = 0;
  assert(m_head == nullptr);
  if (nSeenTasks > m_nodeCacheSize) {
    //need to expand so next time we don't have to do any
    // memory requests
    m_nodeCacheSize = nSeenTasks;
    m_nodeCache = std::make_unique<WaitNode[]>(nSeenTasks);
    auto nodeCache = m_nodeCache.get();
    for (auto it = nodeCache, itEnd = nodeCache + m_nodeCacheSize; it != itEnd; ++it) {
      it->m_fromCache = true;
    }
  }
  //this will make sure all cores see the changes
  m_waiting = true;
}

Member Data Documentation

m_exceptionPtr

std::exception_ptr edm::WaitingTaskList::m_exceptionPtr

private

Definition at line 157 of file WaitingTaskList.h.

Referenced by add(), announce(), doneWaiting(), and reset().

m_head

std::atomic<WaitNode*> edm::WaitingTaskList::m_head

private

Definition at line 155 of file WaitingTaskList.h.

Referenced by add(), announce(), presetTaskAsFailed(), and reset().

m_lastAssignedCacheIndex

std::atomic<unsigned int> edm::WaitingTaskList::m_lastAssignedCacheIndex

private

Definition at line 159 of file WaitingTaskList.h.

Referenced by createNode(), and reset().

m_nodeCache

std::unique_ptr<WaitNode[]> edm::WaitingTaskList::m_nodeCache

private

Definition at line 156 of file WaitingTaskList.h.

Referenced by createNode(), and reset().

m_nodeCacheSize

unsigned int edm::WaitingTaskList::m_nodeCacheSize

private

Definition at line 158 of file WaitingTaskList.h.

Referenced by createNode(), and reset().

m_waiting

std::atomic<bool> edm::WaitingTaskList::m_waiting

private

Definition at line 160 of file WaitingTaskList.h.

Referenced by add(), doneWaiting(), presetTaskAsFailed(), and reset().