#include <SerialTaskQueue.h>

Classes
class	QueuedTask

class	TaskBase

Public Member Functions
bool	isPaused () const
	Checks to see if the queue has been paused. More...

bool	pause ()
	Pauses processing of additional tasks from the queue. More...

template<typename T >
void	push (const T &iAction)
	asynchronously pushes functor iAction into queue More...

template<typename T >
tbb::task *	pushAndGetNextTaskToRun (const T &iAction)
	asynchronously pushes functor iAction into queue and finds next task to execute More...

template<typename T >
void	pushAndWait (const T &iAction)
	synchronously pushes functor iAction into queue More...

bool	resume ()
	Resumes processing if the queue was paused. More...

	SerialTaskQueue ()

	SerialTaskQueue (SerialTaskQueue &&iOther)

	~SerialTaskQueue ()

Private Member Functions
tbb::task *	finishedTask ()

const SerialTaskQueue &	operator= (const SerialTaskQueue &)=delete

TaskBase *	pickNextTask ()

tbb::task *	pushAndGetNextTask (TaskBase *)

void	pushAndWait (tbb::empty_task iWait, TaskBase )

void	pushTask (TaskBase *)

	SerialTaskQueue (const SerialTaskQueue &)=delete

Private Attributes
std::atomic< unsigned long >	m_pauseCount

std::atomic< bool >	m_taskChosen

tbb::concurrent_queue< TaskBase * >	m_tasks

Friends
class	TaskBase

Detailed Description

Definition at line 67 of file SerialTaskQueue.h.

Constructor & Destructor Documentation

◆ SerialTaskQueue() [1/3]

edm::SerialTaskQueue::SerialTaskQueue ( )

inline

Definition at line 69 of file SerialTaskQueue.h.

69 : m_taskChosen(false), m_pauseCount{0} {}

◆ SerialTaskQueue() [2/3]

edm::SerialTaskQueue::SerialTaskQueue ( SerialTaskQueue && iOther )

inline

Definition at line 71 of file SerialTaskQueue.h.

         : m_tasks(std::move(iOther.m_tasks)),
           m_taskChosen(iOther.m_taskChosen.exchange(false)),
           m_pauseCount(iOther.m_pauseCount.exchange(0)) {
       assert(m_tasks.empty() and m_taskChosen == false);
     }

References cms::cuda::assert(), m_taskChosen, and m_tasks.

◆ ~SerialTaskQueue()

SerialTaskQueue::~SerialTaskQueue ( )

Definition at line 26 of file SerialTaskQueue.cc.

                                   {
   //be certain all tasks have completed
   bool isEmpty = m_tasks.empty();
   bool isTaskChosen = m_taskChosen;
   if ((not isEmpty and not isPaused()) or isTaskChosen) {
     pushAndWait([]() { return; });
   }
 }

References isPaused(), m_taskChosen, m_tasks, or, and pushAndWait().

◆ SerialTaskQueue() [3/3]

edm::SerialTaskQueue::SerialTaskQueue ( const SerialTaskQueue & )

privatedelete

Member Function Documentation

◆ finishedTask()

tbb::task * SerialTaskQueue::finishedTask ( )

private

Definition at line 63 of file SerialTaskQueue.cc.

                                        {
   m_taskChosen.store(false);
   return pickNextTask();
 }

References m_taskChosen, and pickNextTask().

Referenced by edm::SerialTaskQueue::QueuedTask< T >::execute(), and edm::SerialTaskQueue::TaskBase::finishedTask().

◆ isPaused()

bool edm::SerialTaskQueue::isPaused ( ) const

inline

Checks to see if the queue has been paused.

Returns: true if the queue is paused

See also: pause(), resume()

Definition at line 84 of file SerialTaskQueue.h.

84 { return m_pauseCount.load() != 0; }

References m_pauseCount.

Referenced by ~SerialTaskQueue().

◆ operator=()

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

privatedelete

◆ pause()

bool edm::SerialTaskQueue::pause ( )

inline

Pauses processing of additional tasks from the queue.

Any task already running will not be paused however once that running task finishes no further tasks will be started. Multiple calls to pause() are allowed, however each call to pause() must be balanced by a call to resume().

Returns: false if queue was already paused.

See also: resume(), isPaused()

Definition at line 96 of file SerialTaskQueue.h.

96 { return 1 == ++m_pauseCount; }

References m_pauseCount.

Referenced by edm::EventProcessor::beginLumiAsync(), and edm::LimitedTaskQueue::Resumer::Resumer().

◆ pickNextTask()

SerialTaskQueue::TaskBase * SerialTaskQueue::pickNextTask ( )

private

Definition at line 68 of file SerialTaskQueue.cc.

                                                        {
   bool expect = false;
   if
     LIKELY(0 == m_pauseCount and m_taskChosen.compare_exchange_strong(expect, true)) {
       TaskBase* t = nullptr;
       if
         LIKELY(m_tasks.try_pop(t)) { return t; }
       //no task was actually pulled
       m_taskChosen.store(false);
  
       //was a new entry added after we called 'try_pop' but before we did the clear?
       expect = false;
       if (not m_tasks.empty() and m_taskChosen.compare_exchange_strong(expect, true)) {
         t = nullptr;
         if (m_tasks.try_pop(t)) {
           return t;
         }
         //no task was still pulled since a different thread beat us to it
         m_taskChosen.store(false);
       }
     }
   return nullptr;
 }

References LIKELY, m_pauseCount, m_taskChosen, m_tasks, and OrderedSet::t.

Referenced by finishedTask(), pushAndGetNextTask(), and resume().

◆ push()

template<typename T >

void SerialTaskQueue::push ( const T & iAction )

asynchronously pushes functor iAction into queue

The function will return immediately and iAction will either process concurrently with the calling thread or wait until the protected resource becomes available or until a CPU becomes available.

Parameters

[in] iAction Must be a functor that takes no arguments and return no values.

Definition at line 187 of file SerialTaskQueue.h.

                                              {
     QueuedTask<T>* pTask{new (tbb::task::allocate_root()) QueuedTask<T>{iAction}};
     pTask->setQueue(this);
     pushTask(pTask);
   }

References pushTask(), and edm::SerialTaskQueue::TaskBase::setQueue().

Referenced by edm::EventProcessor::beginLumiAsync().

◆ pushAndGetNextTask()

tbb::task * SerialTaskQueue::pushAndGetNextTask ( TaskBase * iTask )

private

Definition at line 53 of file SerialTaskQueue.cc.

                                                             {
   tbb::task* returnValue{nullptr};
   if
     LIKELY(nullptr != iTask) {
       m_tasks.push(iTask);
       returnValue = pickNextTask();
     }
   return returnValue;
 }

References LIKELY, m_tasks, pickNextTask(), and TrackValidation_cff::task.

Referenced by pushAndGetNextTaskToRun(), pushAndWait(), and pushTask().

◆ pushAndGetNextTaskToRun()

template<typename T >

tbb::task * SerialTaskQueue::pushAndGetNextTaskToRun ( const T & iAction )

asynchronously pushes functor iAction into queue and finds next task to execute

This function is useful if you are accessing the SerialTaskQueue for the execute() method of a TBB task and want to efficiently schedule the next task from the queue. In that case you can take the return value and return it directly from your execute() method. The function will return immediately and not wait for iAction to run.

Parameters

[in] iAction Must be a functor that takes no arguments and return no values.

Returns: Returns either the next task that the user must schedule with TBB or a nullptr.

Definition at line 203 of file SerialTaskQueue.h.

                                                                     {
     QueuedTask<T>* pTask{new (tbb::task::allocate_root()) QueuedTask<T>{iAction}};
     pTask->setQueue(this);
     return pushAndGetNextTask(pTask);
   }

References pushAndGetNextTask(), and edm::SerialTaskQueue::TaskBase::setQueue().

◆ pushAndWait() [1/2]

template<typename T >

void SerialTaskQueue::pushAndWait ( const T & iAction )

synchronously pushes functor iAction into queue

The function will wait until iAction has completed before returning. If another task is already running on the queue, the system is allowed to find another TBB task to execute while waiting for the iAction to finish. In that way the core is not idled while waiting.

Parameters

[in] iAction Must be a functor that takes no arguments and return no values.

Definition at line 194 of file SerialTaskQueue.h.

                                                     {
     tbb::empty_task* waitTask = new (tbb::task::allocate_root()) tbb::empty_task;
     waitTask->set_ref_count(2);
     QueuedTask<T>* pTask{new (waitTask->allocate_child()) QueuedTask<T>{iAction}};
     pTask->setQueue(this);
     pushAndWait(waitTask, pTask);
   }

References edm::SerialTaskQueue::TaskBase::setQueue().

Referenced by ~SerialTaskQueue().

◆ pushAndWait() [2/2]

void SerialTaskQueue::pushAndWait	(	tbb::empty_task *	iWait,
		TaskBase *	iTask
	)

private

Definition at line 92 of file SerialTaskQueue.cc.

                                                                        {
   auto nextTask = pushAndGetNextTask(iTask);
   if
     LIKELY(nullptr != nextTask) {
       if
         LIKELY(nextTask == iTask) {
           //spawn and wait for all requires the task to have its parent set
           iWait->spawn_and_wait_for_all(*nextTask);
         }
       else {
         tbb::task::spawn(*nextTask);
         iWait->wait_for_all();
       }
     }
   else {
     //a task must already be running in this queue
     iWait->wait_for_all();
   }
   tbb::task::destroy(*iWait);
 }

References LIKELY, and pushAndGetNextTask().

◆ pushTask()

void SerialTaskQueue::pushTask ( TaskBase * iTask )

private

Definition at line 46 of file SerialTaskQueue.cc.

                                               {
   tbb::task* t = pushAndGetNextTask(iTask);
   if (nullptr != t) {
     tbb::task::spawn(*t);
   }
 }

References pushAndGetNextTask(), OrderedSet::t, and TrackValidation_cff::task.

Referenced by push().

◆ resume()

bool SerialTaskQueue::resume ( )

Resumes processing if the queue was paused.

Multiple calls to resume() are allowed if there were multiple calls to pause(). Only when we reach as many resume() calls as pause() calls will the queue restart.

Returns: true if the call really restarts the queue

See also: pause(), isPaused()

Definition at line 35 of file SerialTaskQueue.cc.

                              {
   if (0 == --m_pauseCount) {
     tbb::task* t = pickNextTask();
     if (nullptr != t) {
       tbb::task::spawn(*t);
     }
     return true;
   }
   return false;
 }

References m_pauseCount, pickNextTask(), OrderedSet::t, and TrackValidation_cff::task.

Referenced by edm::EventProcessor::globalEndLumiAsync(), edm::LimitedTaskQueue::Resumer::operator=(), and edm::Worker::RunModuleTask< T >::EnableQueueGuard::~EnableQueueGuard().

Friends And Related Function Documentation

◆ TaskBase

friend class TaskBase

friend

Definition at line 170 of file SerialTaskQueue.h.

Member Data Documentation

◆ m_pauseCount

std::atomic<unsigned long> edm::SerialTaskQueue::m_pauseCount

private

Definition at line 183 of file SerialTaskQueue.h.

Referenced by isPaused(), pause(), pickNextTask(), and resume().

◆ m_taskChosen

std::atomic<bool> edm::SerialTaskQueue::m_taskChosen

private

Definition at line 182 of file SerialTaskQueue.h.

Referenced by finishedTask(), pickNextTask(), SerialTaskQueue(), and ~SerialTaskQueue().

◆ m_tasks

tbb::concurrent_queue<TaskBase*> edm::SerialTaskQueue::m_tasks

private

Definition at line 181 of file SerialTaskQueue.h.

Referenced by pickNextTask(), pushAndGetNextTask(), SerialTaskQueue(), and ~SerialTaskQueue().

Classes

Public Member Functions

Private Member Functions

Private Attributes

Friends

Detailed Description

Constructor & Destructor Documentation

◆ SerialTaskQueue() [1/3]

◆ SerialTaskQueue() [2/3]

◆ ~SerialTaskQueue()

◆ SerialTaskQueue() [3/3]

Member Function Documentation

◆ finishedTask()

◆ isPaused()

◆ operator=()

◆ pause()

◆ pickNextTask()

◆ push()

◆ pushAndGetNextTask()

◆ pushAndGetNextTaskToRun()

◆ pushAndWait() [1/2]

◆ pushAndWait() [2/2]

◆ pushTask()

◆ resume()

Friends And Related Function Documentation

◆ TaskBase

Member Data Documentation

◆ m_pauseCount

◆ m_taskChosen

◆ m_tasks