edm::WaitingTaskList Class Reference

#include <WaitingTaskList.h>

Classes
struct	WaitNode

Public Member Functions
void	add (tbb::task *)
	Adds task to the waiting list. More...
void	doneWaiting ()
	Signals that the resource is now available and tasks should be spawned. More...
void	reset ()
	Resets access to the resource so that added tasks will wait. More...
	WaitingTaskList (unsigned int iInitialSize=2)
	Constructor. More...
	~WaitingTaskList ()

Private Member Functions
void	announce ()
WaitNode *	createNode (tbb::task *iTask)
const WaitingTaskList &	operator= (const WaitingTaskList &)=delete
	WaitingTaskList (const WaitingTaskList &)=delete

Private Attributes
std::atomic< WaitNode * >	m_head
std::atomic< unsigned int >	m_lastAssignedCacheIndex
WaitNode *	m_nodeCache
unsigned int	m_nodeCacheSize
std::atomic< bool >	m_waiting

Detailed Description

Definition at line 79 of file WaitingTaskList.h.

Constructor & Destructor Documentation

WaitingTaskList::WaitingTaskList

(

unsigned int

iInitialSize = 2

)

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 35 of file WaitingTaskList.cc.

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

WaitingTaskList::~WaitingTaskList

(

)

Definition at line 52 of file WaitingTaskList.cc.

References m_nodeCache.

{
  delete [] m_nodeCache;
}

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

privatedelete

Member Function Documentation

void WaitingTaskList::add

(

tbb::task *

iTask

)

Adds task to the waiting list.

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

Definition at line 111 of file WaitingTaskList.cc.

References announce(), createNode(), m_head, m_waiting, and edm::WaitingTaskList::WaitNode::setNextNode().

                                   {
  iTask->increment_ref_count();
  if(!m_waiting) {
    if(0==iTask->decrement_ref_count()) {
      tbb::task::spawn(*iTask);
    }
  } else {
    WaitNode* newHead = createNode(iTask);
    WaitNode* oldHead = m_head.exchange(newHead);
    if(oldHead) {
      newHead->setNextNode(oldHead);
      //NOTE: 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
    } else {
      newHead->setNextNode(0);
      if(!m_waiting) {
        //if finished waiting right before we did the
        // exchange our task will not be spawned. 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();
      }
      
    }
  }
}

void WaitingTaskList::announce ( )

private

Handles spawning the tasks, safe to call from multiple threads

Definition at line 142 of file WaitingTaskList.cc.

References edm::WaitingTaskList::WaitNode::m_fromCache, m_head, edm::WaitingTaskList::WaitNode::m_task, n, GetRecoTauVFromDQM_MC_cff::next, edm::WaitingTaskList::WaitNode::nextNode(), and edmStreamStallGrapher::t.

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(0);
  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;
    if(0==t->decrement_ref_count()){
      tbb::task::spawn(*t);
    }
    if(!n->m_fromCache ) {
      delete n;
    }
    n=next;
  }
}

WaitingTaskList::WaitNode * WaitingTaskList::createNode ( tbb::task *  iTask )

private

Definition at line 92 of file WaitingTaskList.cc.

References cmsHarvester::index, edm::WaitingTaskList::WaitNode::m_fromCache, 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+index;
  } else {
    returnValue = new WaitNode;
    returnValue->m_fromCache=false;
  }
  returnValue->m_task = iTask;
  returnValue->m_next = returnValue;
  
  return returnValue;
}

void WaitingTaskList::doneWaiting

(

)

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. To have tasks wait again one must call reset(). Calls to add() and doneWaiting() can safely be done concurrently.

Definition at line 169 of file WaitingTaskList.cc.

References announce(), and m_waiting.

{
  m_waiting=false;
  announce();
}

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

privatedelete

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() call be called concurrently with reset().

Definition at line 73 of file WaitingTaskList.cc.

References edm::WaitingTaskList::WaitNode::m_fromCache, m_head, m_lastAssignedCacheIndex, m_nodeCache, m_nodeCacheSize, and m_waiting.

{
  m_waiting = true;
  unsigned int nSeenTasks = m_lastAssignedCacheIndex;
  m_lastAssignedCacheIndex = 0;
  assert(m_head == 0);
  if (nSeenTasks > m_nodeCacheSize) {
    //need to expand so next time we don't have to do any
    // memory requests
    delete [] m_nodeCache;
    m_nodeCacheSize = nSeenTasks;
    m_nodeCache = new WaitNode[nSeenTasks];
    for(auto it = m_nodeCache, itEnd = m_nodeCache+m_nodeCacheSize; it!=itEnd; ++it) {
      it->m_fromCache=true;
    }
  }
}

Member Data Documentation

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

private

Definition at line 144 of file WaitingTaskList.h.

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

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

private

Definition at line 147 of file WaitingTaskList.h.

Referenced by createNode(), and reset().

WaitNode* edm::WaitingTaskList::m_nodeCache

private

Definition at line 145 of file WaitingTaskList.h.

Referenced by createNode(), reset(), and ~WaitingTaskList().

unsigned int edm::WaitingTaskList::m_nodeCacheSize

private

Definition at line 146 of file WaitingTaskList.h.

Referenced by createNode(), and reset().

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

private

Definition at line 148 of file WaitingTaskList.h.

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