CMS 3D CMS Logo

SerialTaskQueue.cc
Go to the documentation of this file.
1 // -*- C++ -*-
2 //
3 // Package: Concurrency
4 // Class : SerialTaskQueue
5 //
6 // Implementation:
7 // [Notes on implementation]
8 //
9 // Original Author: Chris Jones
10 // Created: Thu Feb 21 11:31:52 CST 2013
11 // $Id$
12 //
13 
14 // system include files
15 #include "oneapi/tbb/task_group.h"
16 
17 // user include files
19 
21 
22 using namespace edm;
23 
24 //
25 // member functions
26 //
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  oneapi::tbb::task_group g;
33  tbb::task_handle last{g.defer([]() {})};
34  push(g, [&g, &last]() { g.run(std::move(last)); });
35  g.wait();
36  }
37 }
38 
40  auto pTask = &iTask;
41  iTask.group()->run([pTask, this]() {
42  TaskBase* t = pTask;
43  auto g = pTask->group();
44  do {
45  t->execute();
46  delete t;
47  t = finishedTask();
48  if (t and t->group() != g) {
49  spawn(*t);
50  t = nullptr;
51  }
52  } while (t != nullptr);
53  });
54 }
55 
57  if (0 == --m_pauseCount) {
58  auto t = pickNextTask();
59  if (nullptr != t) {
60  spawn(*t);
61  }
62  return true;
63  }
64  return false;
65 }
66 
68  auto t = pushAndGetNextTask(iTask);
69  if (nullptr != t) {
70  spawn(*t);
71  }
72 }
73 
75  TaskBase* returnValue{nullptr};
76  if LIKELY (nullptr != iTask) {
77  m_tasks.push(iTask);
78  returnValue = pickNextTask();
79  }
80  return returnValue;
81 }
82 
84  m_taskChosen.store(false);
85  return pickNextTask();
86 }
87 
89  bool expect = false;
90  if LIKELY (0 == m_pauseCount and m_taskChosen.compare_exchange_strong(expect, true)) {
91  TaskBase* t = nullptr;
92  if LIKELY (m_tasks.try_pop(t)) {
93  return t;
94  }
95  //no task was actually pulled
96  m_taskChosen.store(false);
97 
98  //was a new entry added after we called 'try_pop' but before we did the clear?
99  expect = false;
100  if (not m_tasks.empty() and m_taskChosen.compare_exchange_strong(expect, true)) {
101  t = nullptr;
102  if (m_tasks.try_pop(t)) {
103  return t;
104  }
105  //no task was still pulled since a different thread beat us to it
106  m_taskChosen.store(false);
107  }
108  }
109  return nullptr;
110 }
111 
112 //
113 // const member functions
114 //
115 
116 //
117 // static member functions
118 //
TaskBase * pushAndGetNextTask(TaskBase *)
std::atomic< unsigned long > m_pauseCount
void push(oneapi::tbb::task_group &, const T &iAction)
asynchronously pushes functor iAction into queue
#define LIKELY(x)
Definition: Likely.h:20
void pushTask(TaskBase *)
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 g
Definition: Activities.doc:4
bool resume()
Resumes processing if the queue was paused.
TaskBase * finishedTask()
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.
TaskBase * pickNextTask()
std::atomic< bool > m_taskChosen
oneapi::tbb::concurrent_queue< TaskBase * > m_tasks
void spawn(TaskBase &)
HLT enums.
oneapi::tbb::task_group * group()
def move(src, dest)
Definition: eostools.py:511