CMS 3D CMS Logo

List of all members | Classes | Public Member Functions | Private Member Functions | Private Attributes | Friends
edm::SerialTaskQueue Class Reference

#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 ()
 

Private Member Functions

tbb::task * finishedTask ()
 
const SerialTaskQueueoperator= (const SerialTaskQueue &)=delete
 
TaskBasepickNextTask ()
 
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 65 of file SerialTaskQueue.h.

Constructor & Destructor Documentation

edm::SerialTaskQueue::SerialTaskQueue ( )
inline

Definition at line 68 of file SerialTaskQueue.h.

References ~SerialTaskQueue().

Referenced by pause().

68  :
69  m_taskChosen(false),
70  m_pauseCount{0}
71  { }
std::atomic< unsigned long > m_pauseCount
std::atomic< bool > m_taskChosen
SerialTaskQueue::~SerialTaskQueue ( )

Definition at line 26 of file SerialTaskQueue.cc.

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

Referenced by SerialTaskQueue().

27 {
28  //be certain all tasks have completed
29  bool isEmpty = m_tasks.empty();
30  bool isTaskChosen = m_taskChosen;
31  if ( (not isEmpty and not isPaused()) or isTaskChosen) {
32  pushAndWait([]() {return;});
33  }
34 }
void pushAndWait(const T &iAction)
synchronously pushes functor iAction into queue
The Signals That Services Can Subscribe To This is based on ActivityRegistry and is current per Services can connect to the signals distributed by the ActivityRegistry in order to monitor the activity of the application Each possible callback has some defined which we here list in angle e< void, edm::EventID const &, edm::Timestamp const & > We also list in braces which AR_WATCH_USING_METHOD_ is used for those or
Definition: Activities.doc:12
bool isPaused() const
Checks to see if the queue has been paused.
std::atomic< bool > m_taskChosen
tbb::concurrent_queue< TaskBase * > m_tasks
edm::SerialTaskQueue::SerialTaskQueue ( const SerialTaskQueue )
privatedelete

Member Function Documentation

tbb::task * SerialTaskQueue::finishedTask ( )
private

Definition at line 68 of file SerialTaskQueue.cc.

References m_taskChosen, and pickNextTask().

Referenced by edm::SerialTaskQueue::TaskBase::finishedTask().

68  {
69  m_taskChosen.store(false);
70  return pickNextTask();
71 }
TaskBase * pickNextTask()
std::atomic< bool > m_taskChosen
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 80 of file SerialTaskQueue.h.

References m_pauseCount.

Referenced by ~SerialTaskQueue().

80 { return m_pauseCount.load()==0;}
std::atomic< unsigned long > m_pauseCount
const SerialTaskQueue& edm::SerialTaskQueue::operator= ( const SerialTaskQueue )
privatedelete

Referenced by 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 92 of file SerialTaskQueue.h.

References m_pauseCount, operator=(), push(), pushAndGetNextTaskToRun(), pushAndWait(), resume(), and SerialTaskQueue().

92  {
93  return 1 == ++m_pauseCount;
94  }
std::atomic< unsigned long > m_pauseCount
SerialTaskQueue::TaskBase * SerialTaskQueue::pickNextTask ( )
private

Definition at line 74 of file SerialTaskQueue.cc.

References likely, m_pauseCount, m_taskChosen, m_tasks, and lumiQTWidget::t.

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

74  {
75 
76  bool expect = false;
77  if likely(0 == m_pauseCount and m_taskChosen.compare_exchange_strong(expect,true)) {
78  TaskBase* t=0;
79  if likely(m_tasks.try_pop(t)) {
80  return t;
81  }
82  //no task was actually pulled
83  m_taskChosen.store(false);
84 
85  //was a new entry added after we called 'try_pop' but before we did the clear?
86  expect = false;
87  if(not m_tasks.empty() and m_taskChosen.compare_exchange_strong(expect,true)) {
88  TaskBase* t=0;
89  if(m_tasks.try_pop(t)) {
90  return t;
91  }
92  //no task was still pulled since a different thread beat us to it
93  m_taskChosen.store(false);
94 
95  }
96  }
97  return 0;
98 }
std::atomic< unsigned long > m_pauseCount
#define likely(x)
std::atomic< bool > m_taskChosen
tbb::concurrent_queue< TaskBase * > m_tasks
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]iActionMust be a functor that takes no arguments and return no values.

Definition at line 186 of file SerialTaskQueue.h.

References pushTask().

Referenced by pause().

186  {
187  QueuedTask<T>* pTask{ new (tbb::task::allocate_root()) QueuedTask<T>{iAction} };
188  pTask->setQueue(this);
189  pushTask(pTask);
190  }
void pushTask(TaskBase *)
tbb::task * SerialTaskQueue::pushAndGetNextTask ( TaskBase iTask)
private

Definition at line 57 of file SerialTaskQueue.cc.

References likely, m_tasks, and pickNextTask().

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

57  {
58  tbb::task* returnValue{0};
59  if likely(0!=iTask) {
60  m_tasks.push(iTask);
61  returnValue = pickNextTask();
62  }
63  return returnValue;
64 }
#define likely(x)
TaskBase * pickNextTask()
tbb::concurrent_queue< TaskBase * > m_tasks
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]iActionMust 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 202 of file SerialTaskQueue.h.

References pushAndGetNextTask().

Referenced by pause().

202  {
203  QueuedTask<T>* pTask{ new (tbb::task::allocate_root()) QueuedTask<T>{iAction} };
204  pTask->setQueue(this);
205  return pushAndGetNextTask(pTask);
206  }
tbb::task * pushAndGetNextTask(TaskBase *)
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]iActionMust be a functor that takes no arguments and return no values.

Definition at line 193 of file SerialTaskQueue.h.

Referenced by pause(), and ~SerialTaskQueue().

193  {
194  tbb::empty_task* waitTask = new (tbb::task::allocate_root()) tbb::empty_task;
195  waitTask->set_ref_count(2);
196  QueuedTask<T>* pTask{ new (waitTask->allocate_child()) QueuedTask<T>{iAction} };
197  pTask->setQueue(this);
198  pushAndWait(waitTask,pTask);
199  }
void pushAndWait(const T &iAction)
synchronously pushes functor iAction into queue
void SerialTaskQueue::pushAndWait ( tbb::empty_task *  iWait,
TaskBase iTask 
)
private

Definition at line 100 of file SerialTaskQueue.cc.

References pyrootRender::destroy(), likely, and pushAndGetNextTask().

100  {
101  auto nextTask = pushAndGetNextTask(iTask);
102  if likely(nullptr != nextTask) {
103  if likely(nextTask == iTask) {
104  //spawn and wait for all requires the task to have its parent set
105  iWait->spawn_and_wait_for_all(*nextTask);
106  } else {
107  tbb::task::spawn(*nextTask);
108  iWait->wait_for_all();
109  }
110  } else {
111  //a task must already be running in this queue
112  iWait->wait_for_all();
113  }
114  tbb::task::destroy(*iWait);
115 }
def destroy(e)
Definition: pyrootRender.py:13
tbb::task * pushAndGetNextTask(TaskBase *)
#define likely(x)
void SerialTaskQueue::pushTask ( TaskBase iTask)
private

Definition at line 49 of file SerialTaskQueue.cc.

References pushAndGetNextTask(), and lumiQTWidget::t.

Referenced by push().

49  {
50  tbb::task* t = pushAndGetNextTask(iTask);
51  if(0!=t) {
52  tbb::task::spawn(*t);
53  }
54 }
tbb::task * pushAndGetNextTask(TaskBase *)
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 37 of file SerialTaskQueue.cc.

References m_pauseCount, pickNextTask(), and lumiQTWidget::t.

Referenced by pause().

37  {
38  if(0==--m_pauseCount) {
39  tbb::task* t = pickNextTask();
40  if(0 != t) {
41  tbb::task::spawn(*t);
42  }
43  return true;
44  }
45  return false;
46 }
std::atomic< unsigned long > m_pauseCount
TaskBase * pickNextTask()

Friends And Related Function Documentation

friend class TaskBase
friend

Definition at line 168 of file SerialTaskQueue.h.

Member Data Documentation

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

Definition at line 182 of file SerialTaskQueue.h.

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

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

Definition at line 181 of file SerialTaskQueue.h.

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

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

Definition at line 180 of file SerialTaskQueue.h.

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