Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : */
4 :
5 : #include <assert.h>
6 : #include <fstream>
7 : #include <map>
8 : #include <iostream>
9 : #include <atomic>
10 : #include <boost/intrusive/set.hpp>
11 : #include <boost/optional.hpp>
12 : #include <oneapi/tbb/enumerable_thread_specific.h>
13 : #include <oneapi/tbb/global_control.h>
14 :
15 : #include "base/logging.h"
16 : #include "base/task.h"
17 : #include "base/task_annotations.h"
18 : #include "base/task_tbbkeepawake.h"
19 : #include "base/task_monitor.h"
20 :
21 : #include <base/sandesh/task_types.h>
22 :
23 : using namespace std;
24 :
25 : int TaskScheduler::ThreadAmpFactor_ = 1;
26 : class TaskEntry;
27 : struct TaskDeferEntryCmp;
28 :
29 : typedef oneapi::tbb::enumerable_thread_specific<Task *> TaskInfo;
30 :
31 : static TaskInfo task_running;
32 :
33 : // Vector of Task entries
34 : typedef std::vector<TaskEntry *> TaskEntryList;
35 :
36 : boost::scoped_ptr<TaskScheduler> TaskScheduler::singleton_;
37 :
38 : #define TASK_TRACE(scheduler, task, msg, delay)\
39 : do {\
40 : scheduler->Log(__FILE__, __LINE__, task, msg, delay);\
41 : } while (false)
42 :
43 : /// @brief A private class used to implement tbb::task
44 : /// An object is created when task is ready for execution and
45 : /// registered with tbb::task
46 : class TaskFunctor {
47 : public:
48 :
49 : /// @brief Creates a new instance of TaskFunctor using the provided
50 : /// implementaton (a parent).
51 7234883 : TaskFunctor(Task *t) : parent_(t) {};
52 :
53 : /// @brief Creates a copy of the object.
54 : TaskFunctor(const TaskFunctor& tf)
55 : : parent_(tf.parent_) {}
56 :
57 7234883 : TaskFunctor(TaskFunctor &&tf)
58 7234883 : : parent_(tf.parent_) {}
59 :
60 : /// @brief Destructor is called when a task execution is compeleted.
61 : /// Invoked implicitly by tbb::task.
62 : /// Invokes OnTaskExit to schedule tasks pending tasks.
63 : virtual ~TaskFunctor();
64 :
65 : /// @brief Method called from TBB to execute the task.
66 : /// Invokes Run() method of the parent (an implementation).
67 : /// Supports task continuation when Run() returns false.
68 : void operator ()() const;
69 :
70 : private:
71 :
72 : /// @brief
73 : mutable Task *parent_;
74 :
75 : /// @brief
76 : const TaskFunctor & operator = (const TaskFunctor) = delete;
77 : };
78 :
79 : /// @brief A class maintaning information for every <task, instance>
80 : ///
81 : /// policyq_ : contains,
82 : /// - Policies configured for a task
83 : /// - Complementary policies for a task.
84 : /// Example, if a policy is of form <tid0> => <tid1, inst1> <tid2, -1>
85 : /// <tid1, inst1> cannot run if <tid0, inst1> is running
86 : /// <tid2, *> cannot run when <tid0, *> are running. These become
87 : /// complementary rule
88 : ///
89 : /// waitq_ : Tasks of this instance created and waiting to be executed. Tasks
90 : /// are stored and executed in order of their creation
91 : /// Task can be added here on Enqueue if policy conditions are not
92 : /// met. Its taken out from waitq_ only when its about to Run
93 : ///
94 : /// deferq_ : Tree of TaskEntry waiting on this task instance. The TaskEntry.
95 : /// This tree is populated if all conditions are met
96 : /// - This TaskEntry has tasks created
97 : /// - The TaskEntry in deferq_ has tasks created
98 : /// - The Tree is sorted on task seqno_
99 : ///
100 : /// run_task_ : Task running in context of this TaskEntry. Only entries in
101 : /// task_entry_db_ have this set. Entries in task_db_ will always
102 : /// have this as NULL
103 : /// Running task is not in waitq_ or deferq_
104 : ///
105 : /// run_count_: Number of running tasks for this TaskEntry
106 : class TaskEntry {
107 : public:
108 : TaskEntry(int task_id);
109 : TaskEntry(int task_id, int task_instance);
110 : ~TaskEntry();
111 :
112 : void AddPolicy(TaskEntry *entry);
113 62502448 : size_t WaitQSize() const { return waitq_.size(); };
114 : void AddToWaitQ(Task *t);
115 : bool DeleteFromWaitQ(Task *t);
116 :
117 : /// @brief Adds a task to deferq_.
118 : /// Only one task of a given instance goes into deferq_ for its policies.
119 : void AddToDeferQ(TaskEntry *entry);
120 :
121 : /// @brief Deletes a task from deferq_.
122 : void DeleteFromDeferQ(TaskEntry &entry);
123 :
124 : TaskEntry *ActiveEntryInPolicy();
125 : bool DeferOnPolicyFail(Task *t);
126 :
127 : /// @brief Starts a task.
128 : /// If there are more entries in waitq_ add them to deferq_.
129 : void RunTask(Task *t);
130 :
131 : /// @brief Starts executing tasks from deferq_ of a TaskEntry.
132 : void RunDeferQ();
133 :
134 : /// @brief Starts executing tasks from deferq_ of TaskEntry and
135 : /// TaskGroup in the temporal order
136 : void RunCombinedDeferQ();
137 : void RunWaitQ();
138 : void RunDeferEntry();
139 :
140 : /// @brief Starts executing tasks from deferq_ of TaskEntries which are
141 : /// enabled.
142 : void RunDeferQForGroupEnable();
143 :
144 : void TaskExited(Task *t, TaskGroup *group);
145 : TaskStats *GetTaskStats();
146 : void ClearTaskStats();
147 : void ClearQueues();
148 :
149 : /// @brief Addition/deletion of TaskEntry in the deferq_ is based on the
150 : /// seqno.
151 : /// seqno of the first Task in the waitq_ is used as the key. This function
152 : /// would be invoked by the comparison function during addition/deletion
153 : /// of TaskEntry in the deferq_.
154 : boost::optional<uint64_t> GetTaskDeferEntrySeqno() const;
155 :
156 : /// @brief Returns the code ID of this task entry
157 : int task_code_id() const { return task_code_id_; }
158 :
159 : /// @brief Returns the data ID of this task entry
160 : int task_data_id() const { return task_data_id_; }
161 :
162 : /// @brief Returns the count of runs for this task entry
163 : int GetRunCount() const { return run_count_; }
164 :
165 : /// @brief Disables this task entry
166 0 : void SetDisable(bool disable) { disable_ = disable; }
167 35405347 : bool IsDisabled() { return disable_; }
168 : void GetSandeshData(SandeshTaskEntry *resp) const;
169 :
170 : private:
171 : friend class TaskGroup;
172 : friend class TaskScheduler;
173 :
174 : /// @brief List of Task's in waitq_
175 : typedef boost::intrusive::member_hook<Task,
176 : boost::intrusive::list_member_hook<>, &Task::waitq_hook_> WaitQHook;
177 : typedef boost::intrusive::list<Task, WaitQHook> TaskWaitQ;
178 :
179 : boost::intrusive::set_member_hook<> task_defer_node;
180 : typedef boost::intrusive::member_hook<TaskEntry,
181 : boost::intrusive::set_member_hook<>,
182 : &TaskEntry::task_defer_node> TaskDeferListOption;
183 :
184 : /// @brief It is a tree of TaskEntries deferred and waiting on the
185 : /// containing task to exit. The tree is sorted by seqno_ of first
186 : /// task in the TaskEntry
187 : typedef boost::intrusive::set<TaskEntry, TaskDeferListOption,
188 : boost::intrusive::compare<TaskDeferEntryCmp> > TaskDeferList;
189 :
190 : int task_code_id_;
191 : int task_data_id_;
192 :
193 : /// @brief No. of tasks running
194 : int run_count_;
195 :
196 : /// @brief Task currently running
197 : Task *run_task_;
198 :
199 : /// @brief Tasks waiting to run on some condition
200 : TaskWaitQ waitq_;
201 :
202 : /// @brief Policy rules for a task
203 : TaskEntryList policyq_;
204 :
205 : /// @brief Tasks deferred for this to exit
206 : TaskDeferList *deferq_;
207 : TaskEntry *deferq_task_entry_;
208 : TaskGroup *deferq_task_group_;
209 : bool disable_;
210 :
211 : /// @brief Cummulative Maintenance stats
212 : TaskStats stats_;
213 :
214 : DISALLOW_COPY_AND_ASSIGN(TaskEntry);
215 : };
216 :
217 : /// @brief Comparison routine for the TaskDeferList
218 : struct TaskDeferEntryCmp {
219 14639143 : bool operator() (const TaskEntry &lhs, const TaskEntry &rhs) const {
220 14639143 : return (lhs.GetTaskDeferEntrySeqno() <
221 29278286 : rhs.GetTaskDeferEntrySeqno());
222 : }
223 : };
224 :
225 : /// @brief TaskGroup maintains per <task-id> information including,
226 : ///
227 : /// polic_set_ : Boolean used to ensure policy is set only once per task
228 : /// Task policy change is not yet supported
229 : ///
230 : /// policy_ : List of policy rules for the task
231 : ///
232 : /// run_count_ : Number of tasks running in context of this task-group
233 : ///
234 : /// deferq_ : Tasks deferred till run_count_ on this task becomes 0
235 : ///
236 : /// task_entry_ : Default TaskEntry used for task without an instance
237 : ///
238 : /// disable_entry_ : TaskEntry which maintains a deferQ for tasks enqueued
239 : /// while TaskGroup is disabled
240 : class TaskGroup {
241 : public:
242 : TaskGroup(int task_id);
243 : ~TaskGroup();
244 :
245 : TaskEntry *QueryTaskEntry(int task_instance) const;
246 : TaskEntry *GetTaskEntry(int task_instance);
247 : void AddPolicy(TaskGroup *group);
248 :
249 : /// @brief Add task to deferq_
250 : /// Only one task of a given instance goes into deferq_ for its policies.
251 : void AddToDeferQ(TaskEntry *entry);
252 :
253 : /// @brief Enqueue TaskEntry in disable_entry's deferQ
254 : void AddToDisableQ(TaskEntry *entry);
255 :
256 : /// @brief Add TaskEntries to disable_entry_ which have tasks enqueued and
257 : /// are already disabled.
258 : void AddEntriesToDisableQ();
259 :
260 524 : TaskEntry *GetDisableEntry() { return disable_entry_; }
261 :
262 : /// @brief Delete task from deferq_
263 : void DeleteFromDeferQ(TaskEntry &entry);
264 : TaskGroup *ActiveGroupInPolicy();
265 : bool DeferOnPolicyFail(TaskEntry *entry, Task *t);
266 :
267 : /// @brief Returns true, if the waiq_ of all the tasks in the group are
268 : /// empty.
269 : ///
270 : /// Note: This function is invoked from TaskScheduler::IsEmpty() for each
271 : /// task group and is intended to be invoked only in the test code. If this
272 : /// function needs to be used outside test code, then we may want to
273 : /// consider storing the waitq_ count for performance reason.
274 : bool IsWaitQEmpty();
275 :
276 20715073 : int TaskRunCount() const {return run_count_;};
277 :
278 : /// @brief Starts executing tasks from deferq_ of a TaskGroup
279 : void RunDeferQ();
280 :
281 : /// @brief Run tasks that maybe suspended. Schedule tasks only for
282 : /// TaskEntries which are enabled.
283 : void RunDisableEntries();
284 : void TaskExited(Task *t);
285 : void PolicySet();
286 7234883 : void TaskStarted() {run_count_++;};
287 0 : void IncrementTotalRunTime(int64_t rtime) { total_run_time_ += rtime; }
288 : TaskStats *GetTaskGroupStats();
289 : TaskStats *GetTaskStats();
290 : TaskStats *GetTaskStats(int task_instance);
291 : void ClearTaskGroupStats();
292 : void ClearTaskStats();
293 : void ClearTaskStats(int instance_id);
294 8 : void SetDisable(bool disable) { disable_ = disable; }
295 22005162 : bool IsDisabled() { return disable_; }
296 : void GetSandeshData(SandeshTaskGroup *resp, bool summary) const;
297 :
298 0 : int task_id() const { return task_code_id_; }
299 0 : size_t deferq_size() const { return deferq_.size(); }
300 0 : size_t num_tasks() const {
301 0 : size_t count = 0;
302 0 : for (TaskEntryList::const_iterator it = task_entry_db_.begin();
303 0 : it != task_entry_db_.end(); ++it) {
304 0 : if (*it != NULL) {
305 0 : count++;
306 : }
307 : }
308 0 : return count;
309 : }
310 :
311 7234883 : oneapi::tbb::task_group &tbb_group() {
312 7234883 : return tbb_group_;
313 : }
314 :
315 : private:
316 : friend class TaskEntry;
317 : friend class TaskScheduler;
318 :
319 : /// @brief Vector of Task Group policies
320 : typedef std::vector<TaskGroup *> TaskGroupPolicyList;
321 : typedef boost::intrusive::member_hook<TaskEntry,
322 : boost::intrusive::set_member_hook<>,
323 : &TaskEntry::task_defer_node> TaskDeferListOption;
324 :
325 : /// @brief It is a tree of TaskEntries deferred and waiting on the
326 : /// containing task to exit. The tree is sorted by seqno_ of first
327 : /// task in the TaskEntry
328 : typedef boost::intrusive::set<TaskEntry, TaskDeferListOption,
329 : boost::intrusive::compare<TaskDeferEntryCmp> > TaskDeferList;
330 :
331 : static const int kVectorGrowSize = 16;
332 : int task_code_id_;
333 :
334 : /// @brief A TBB object to store executing tasks.
335 : oneapi::tbb::task_group tbb_group_;
336 :
337 : /// @brief Specifies if policy is already set
338 : bool policy_set_;
339 :
340 : /// @brief No. of tasks running in the group
341 : int run_count_;
342 : std::atomic<uint64_t> total_run_time_;
343 :
344 : /// @brief Policy rules for the group
345 : TaskGroupPolicyList policy_;
346 :
347 : /// @brief Tasks deferred till run_count_ is 0
348 : TaskDeferList deferq_;
349 :
350 : /// @brief Tasks deferred till run_count_ is 0
351 : TaskEntry *task_entry_;
352 :
353 : /// @brief Task entry for disabled group
354 : TaskEntry *disable_entry_;
355 :
356 : /// @brief task-entries in this group
357 : TaskEntryList task_entry_db_;
358 : uint32_t execute_delay_;
359 : uint32_t schedule_delay_;
360 : bool disable_;
361 :
362 : TaskStats stats_;
363 : DISALLOW_COPY_AND_ASSIGN(TaskGroup);
364 : };
365 :
366 : ////////////////////////////////////////////////////////////////////////////
367 : // Implementation for class TaskImpl
368 : ////////////////////////////////////////////////////////////////////////////
369 :
370 7234076 : void TaskFunctor::operator ()() const {
371 7234076 : TaskInfo::reference running = task_running.local();
372 7231212 : running = parent_;
373 7231212 : parent_->tbb_state(Task::TBB_EXEC);
374 : try {
375 7230991 : uint64_t t = 0;
376 7230991 : if (parent_->enqueue_time() != 0) {
377 0 : t = ClockMonotonicUsec();
378 0 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
379 0 : if ((t - parent_->enqueue_time()) >
380 0 : scheduler->schedule_delay(parent_)) {
381 0 : TASK_TRACE(scheduler, parent_, "TBB schedule time(in usec) ",
382 : (t - parent_->enqueue_time()));
383 : }
384 7230823 : } else if (TaskScheduler::GetInstance()->track_run_time()) {
385 0 : t = ClockMonotonicUsec();
386 : }
387 7230683 : bool is_complete = parent_->Run();
388 7231861 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
389 7231995 : if (t != 0) {
390 0 : int64_t delay = ClockMonotonicUsec() - t;
391 0 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
392 0 : uint32_t execute_delay = scheduler->execute_delay(parent_);
393 0 : if (execute_delay && delay > execute_delay) {
394 0 : TASK_TRACE(scheduler, parent_, "Run time(in usec) ", delay);
395 : }
396 0 : if (scheduler->track_run_time()) {
397 : TaskGroup *group =
398 0 : scheduler->QueryTaskGroup(parent_->task_code_id());
399 0 : group->IncrementTotalRunTime(delay);
400 : }
401 : }
402 7231827 : running = NULL;
403 7231827 : if (is_complete == true) {
404 5817992 : parent_->set_task_complete();
405 : } else {
406 1413835 : parent_->set_task_recycle();
407 : }
408 7231224 : scheduler->OnTaskExit(parent_);
409 0 : } catch (std::exception &e) {
410 :
411 : // Store exception information statically, to easily read exception
412 : // information from the core.
413 0 : static std::string what = e.what();
414 :
415 0 : LOG(ERROR, "!!!! ERROR !!!! Task caught fatal exception: " << what
416 : << " TaskImpl: " << this);
417 0 : assert(0);
418 0 : } catch (...) {
419 0 : LOG(ERROR, "!!!! ERROR !!!! Task caught fatal unknown exception"
420 : << " TaskImpl: " << this);
421 0 : assert(0);
422 0 : }
423 7234713 : }
424 :
425 14469334 : TaskFunctor::~TaskFunctor() {
426 14469334 : }
427 :
428 15932 : int TaskScheduler::GetThreadCount(int thread_count) {
429 : static bool init_;
430 : static int num_cores_;
431 :
432 15932 : if (init_) {
433 15682 : return num_cores_ * ThreadAmpFactor_;
434 : }
435 :
436 250 : char *num_cores_str = getenv("TBB_THREAD_COUNT");
437 250 : if (!num_cores_str) {
438 250 : if (thread_count == 0) {
439 250 : num_cores_ = oneapi::tbb::info::default_concurrency();
440 : } else {
441 0 : num_cores_ = thread_count;
442 : }
443 : } else {
444 0 : num_cores_ = strtol(num_cores_str, NULL, 0);
445 : }
446 :
447 250 : init_ = true;
448 250 : return num_cores_ * ThreadAmpFactor_;
449 : }
450 :
451 0 : int TaskScheduler::GetDefaultThreadCount() {
452 0 : return oneapi::tbb::this_task_arena::max_concurrency();
453 : }
454 :
455 250 : bool TaskScheduler::ShouldUseSpawn() {
456 250 : if (getenv("TBB_USE_SPAWN"))
457 0 : return true;
458 :
459 250 : return false;
460 : }
461 :
462 : ////////////////////////////////////////////////////////////////////////////
463 : // Implementation for class TaskScheduler
464 : ////////////////////////////////////////////////////////////////////////////
465 :
466 250 : TaskScheduler::TaskScheduler(int task_count) :
467 250 : use_spawn_(ShouldUseSpawn()),
468 500 : tbb_global_control_(
469 : oneapi::tbb::global_control::max_allowed_parallelism,
470 250 : GetThreadCount(task_count) + 1),
471 250 : task_scheduler_(GetThreadCount(task_count) + 1),
472 250 : running_(true), seqno_(0), id_max_(0), log_fn_(), track_run_time_(false),
473 250 : measure_delay_(false), schedule_delay_(0), execute_delay_(0),
474 250 : enqueue_count_(0), done_count_(0), cancel_count_(0), evm_(NULL),
475 250 : tbb_awake_task_(NULL), task_monitor_(NULL) {
476 250 : hw_thread_count_ = GetThreadCount(task_count);
477 250 : task_group_db_.resize(TaskScheduler::kVectorGrowSize);
478 250 : stop_entry_ = new TaskEntry(-1);
479 250 : }
480 :
481 250 : TaskScheduler::~TaskScheduler() {
482 : TaskGroup *group;
483 :
484 250 : for (TaskGroupDb::iterator iter = task_group_db_.begin();
485 8129 : iter != task_group_db_.end(); ++iter) {
486 7879 : if ((group = *iter) == NULL) {
487 3647 : continue;
488 : }
489 4232 : *iter = NULL;
490 4232 : delete group;
491 : }
492 :
493 4595 : for (TaskIdMap::iterator loc = id_map_.begin(); loc != id_map_.end();
494 4345 : id_map_.erase(loc++)) {
495 : }
496 :
497 250 : delete stop_entry_;
498 250 : stop_entry_ = NULL;
499 250 : task_group_db_.clear();
500 :
501 250 : return;
502 250 : }
503 :
504 0 : void TaskScheduler::Initialize(uint32_t thread_count, EventManager *evm) {
505 0 : assert(singleton_.get() == NULL);
506 0 : singleton_.reset(new TaskScheduler((int)thread_count));
507 :
508 0 : if (evm) {
509 0 : singleton_.get()->evm_ = evm;
510 0 : singleton_.get()->tbb_awake_task_ = new TaskTbbKeepAwake();
511 0 : assert(singleton_.get()->tbb_awake_task_);
512 :
513 0 : singleton_.get()->tbb_awake_task_->StartTbbKeepAwakeTask(
514 : singleton_.get(), evm,
515 : "TaskScheduler::TbbKeepAwake");
516 : }
517 0 : }
518 :
519 0 : void TaskScheduler::set_event_manager(EventManager *evm) {
520 0 : assert(evm);
521 0 : evm_ = evm;
522 0 : if (tbb_awake_task_ == NULL) {
523 0 : tbb_awake_task_ = new TaskTbbKeepAwake();
524 0 : assert(tbb_awake_task_);
525 :
526 0 : tbb_awake_task_->StartTbbKeepAwakeTask(this, evm,
527 : "TaskScheduler::TbbKeepAwake");
528 : }
529 0 : }
530 :
531 4 : void TaskScheduler::ModifyTbbKeepAwakeTimeout(uint32_t timeout) {
532 4 : if (tbb_awake_task_) {
533 0 : tbb_awake_task_->ModifyTbbKeepAwakeTimeout(timeout);
534 : }
535 4 : }
536 :
537 4 : void TaskScheduler::EnableMonitor(EventManager *evm,
538 : uint64_t tbb_keepawake_time_msec,
539 : uint64_t inactivity_time_msec,
540 : uint64_t poll_interval_msec) {
541 4 : if (task_monitor_ != NULL)
542 0 : return;
543 :
544 4 : task_monitor_ = new TaskMonitor(this, tbb_keepawake_time_msec,
545 4 : inactivity_time_msec, poll_interval_msec);
546 4 : task_monitor_->Start(evm);
547 : }
548 :
549 0 : void TaskScheduler::Log(const char *file_name, uint32_t line_no,
550 : const Task *task, const char *description,
551 : uint64_t delay) {
552 0 : if (log_fn_.empty() == false) {
553 0 : log_fn_(file_name, line_no, task, description, delay);
554 : }
555 0 : }
556 :
557 4 : void TaskScheduler::RegisterLog(LogFn fn) {
558 4 : log_fn_ = fn;
559 4 : }
560 :
561 0 : uint32_t TaskScheduler::schedule_delay(Task *task) const {
562 0 : if (task->schedule_delay() > schedule_delay_)
563 0 : return task->schedule_delay();
564 0 : return schedule_delay_;
565 : }
566 :
567 0 : uint32_t TaskScheduler::execute_delay(Task *task) const {
568 0 : if (task->execute_delay() > execute_delay_)
569 0 : return task->execute_delay();
570 0 : return execute_delay_;
571 : }
572 :
573 101715380 : TaskScheduler *TaskScheduler::GetInstance() {
574 101715380 : if (singleton_.get() == NULL) {
575 250 : singleton_.reset(new TaskScheduler());
576 : }
577 101695019 : return singleton_.get();
578 : }
579 :
580 38510405 : TaskGroup *TaskScheduler::GetTaskGroup(int task_id) {
581 38510405 : assert(task_id >= 0);
582 38510405 : int size = task_group_db_.size();
583 38510405 : if (size <= task_id) {
584 187 : task_group_db_.resize(task_id + TaskScheduler::kVectorGrowSize);
585 : }
586 :
587 38510405 : TaskGroup *group = task_group_db_[task_id];
588 38510405 : if (group == NULL) {
589 4232 : group = new TaskGroup(task_id);
590 4232 : task_group_db_[task_id] = group;
591 : }
592 :
593 38510405 : return group;
594 : }
595 :
596 20306097 : TaskGroup *TaskScheduler::QueryTaskGroup(int task_id) {
597 20306097 : return task_group_db_[task_id];
598 : }
599 :
600 5132171 : bool TaskScheduler::IsTaskGroupEmpty(int task_id) const {
601 5132171 : CHECK_CONCURRENCY("bgp::Config");
602 5132171 : std::scoped_lock lock(mutex_);
603 5132171 : TaskGroup *group = task_group_db_[task_id];
604 5132171 : assert(group);
605 5132171 : assert(group->TaskRunCount() == 0);
606 10264342 : return group->IsWaitQEmpty();
607 5132171 : }
608 :
609 7240127 : TaskEntry *TaskScheduler::GetTaskEntry(int task_id, int task_instance) {
610 7240127 : TaskGroup *group = GetTaskGroup(task_id);
611 7240127 : return group->GetTaskEntry(task_instance);
612 : }
613 :
614 7235407 : TaskEntry *TaskScheduler::QueryTaskEntry(int task_id, int task_instance) {
615 7235407 : TaskGroup *group = QueryTaskGroup(task_id);
616 7235407 : if (group == NULL)
617 0 : return NULL;
618 7235407 : return group->QueryTaskEntry(task_instance);
619 : }
620 :
621 8 : void TaskScheduler::EnableLatencyThresholds(uint32_t execute,
622 : uint32_t schedule) {
623 8 : execute_delay_ = execute;
624 8 : schedule_delay_ = schedule;
625 8 : measure_delay_ = (execute_delay_ != 0 || schedule_delay_ != 0);
626 8 : }
627 :
628 24 : void TaskScheduler::SetLatencyThreshold(const std::string &name,
629 : uint32_t execute, uint32_t schedule) {
630 24 : int task_id = GetTaskId(name);
631 24 : TaskGroup *group = GetTaskGroup(task_id);
632 24 : group->execute_delay_ = execute;
633 24 : group->schedule_delay_ = schedule;
634 24 : }
635 :
636 3066 : void TaskScheduler::SetPolicy(int task_id, TaskPolicy &policy) {
637 3066 : std::scoped_lock lock(mutex_);
638 :
639 3066 : TaskGroup *group = GetTaskGroup(task_id);
640 3066 : TaskEntry *group_entry = group->GetTaskEntry(-1);
641 3066 : group->PolicySet();
642 :
643 30402 : for (const auto& pol_item: policy) {
644 27336 : if (pol_item.match_data_id == -1) {
645 24976 : TaskGroup *policy_group = GetTaskGroup(pol_item.match_code_id);
646 24976 : group->AddPolicy(policy_group);
647 24976 : policy_group->AddPolicy(group);
648 : } else {
649 2360 : TaskEntry *entry = GetTaskEntry(task_id, pol_item.match_data_id);
650 4720 : TaskEntry *policy_entry = GetTaskEntry(pol_item.match_code_id,
651 2360 : pol_item.match_data_id);
652 2360 : entry->AddPolicy(policy_entry);
653 2360 : policy_entry->AddPolicy(entry);
654 :
655 2360 : group_entry->AddPolicy(policy_entry);
656 2360 : policy_entry->AddPolicy(group_entry);
657 : }
658 : }
659 3066 : }
660 :
661 5818253 : void TaskScheduler::Enqueue(Task *t) {
662 5818253 : std::scoped_lock lock(mutex_);
663 5820546 : EnqueueUnLocked(t);
664 5820546 : }
665 :
666 7235407 : void TaskScheduler::EnqueueUnLocked(Task *t) {
667 7235407 : if (measure_delay_) {
668 0 : t->enqueue_time_ = ClockMonotonicUsec();
669 : }
670 : // Ensure that task is enqueued only once.
671 7235407 : assert(t->seqno() == 0);
672 7235407 : enqueue_count_++;
673 7235407 : t->seqno(++seqno_);
674 7235407 : TaskGroup *group = GetTaskGroup(t->task_code_id());
675 7235407 : t->schedule_delay_ = group->schedule_delay_;
676 7235407 : t->execute_delay_ = group->execute_delay_;
677 7235407 : group->stats_.enqueue_count_++;
678 :
679 7235407 : TaskEntry *entry = GetTaskEntry(t->task_code_id(), t->task_data_id());
680 7235407 : entry->stats_.enqueue_count_++;
681 : // If either TaskGroup or TaskEntry is disabled for Unit-Test purposes,
682 : // enqueue new task in waitq and update TaskGroup if needed.
683 :
684 7235407 : if (group->IsDisabled() || entry->IsDisabled()) {
685 16 : entry->AddToWaitQ(t);
686 16 : if (group->IsDisabled()) {
687 16 : group->AddToDisableQ(entry);
688 : }
689 16 : return;
690 : }
691 :
692 : // Add task to waitq_ if its already populated
693 7235391 : if (entry->WaitQSize() != 0) {
694 2849506 : entry->AddToWaitQ(t);
695 2849506 : return;
696 : }
697 :
698 : // Is scheduler stopped? Dont add task to deferq_ if scheduler is stopped.
699 : // TaskScheduler::Start() will run tasks from waitq_
700 4385885 : if (!running_) {
701 5309 : entry->AddToWaitQ(t);
702 5309 : stop_entry_->AddToDeferQ(entry);
703 5309 : return;
704 : }
705 :
706 : // Check Task Group policy. On policy violation, DeferOnPolicyFail()
707 : // adds the Task to the TaskEntry's waitq_ and the TaskEntry will be
708 : // added to deferq_ of the matching TaskGroup.
709 4380576 : if (group->DeferOnPolicyFail(entry, t)) {
710 1622396 : return;
711 : }
712 :
713 : // Check Task Entry policy. On policy violation, DeferOnPolicyFail()
714 : // adds the Task to the TaskEntry's waitq_ and the TaskEntry will be
715 : // added to deferq_ of the matching TaskEntry.
716 2758180 : if (entry->DeferOnPolicyFail(t)) {
717 1468486 : return;
718 : }
719 :
720 :
721 1289694 : entry->RunTask(t);
722 :
723 1289694 : return;
724 : }
725 :
726 527 : TaskScheduler::CancelReturnCode TaskScheduler::Cancel(Task *t) {
727 527 : std::scoped_lock lock(mutex_);
728 :
729 : // If the task is in RUN state, mark the task for cancellation and return.
730 527 : if (t->state_ == Task::RUN) {
731 3 : t->task_cancel_ = true;
732 524 : } else if (t->state_ == Task::WAIT) {
733 524 : TaskEntry *entry = QueryTaskEntry(t->task_code_id(), t->task_data_id());
734 524 : TaskGroup *group = QueryTaskGroup(t->task_code_id());
735 524 : assert(entry->WaitQSize());
736 : // Get the first entry in the waitq_
737 524 : Task *first_wait_task = &(*entry->waitq_.begin());
738 524 : TaskEntry *disable_entry = group->GetDisableEntry();
739 524 : assert(entry->DeleteFromWaitQ(t) == true);
740 : // If the waitq_ is empty, then remove the TaskEntry from the deferq.
741 524 : if (!entry->WaitQSize()) {
742 412 : if (entry->deferq_task_group_) {
743 361 : assert(entry->deferq_task_entry_ == NULL);
744 361 : entry->deferq_task_group_->DeleteFromDeferQ(*entry);
745 51 : } else if (entry->deferq_task_entry_) {
746 51 : entry->deferq_task_entry_->DeleteFromDeferQ(*entry);
747 0 : } else if (group->IsDisabled()) {
748 : // Remove TaskEntry from deferq of disable_entry
749 0 : disable_entry->DeleteFromDeferQ(*entry);
750 : } else {
751 0 : if (!entry->IsDisabled()) {
752 0 : assert(0);
753 : }
754 : }
755 112 : } else if (t == first_wait_task) {
756 : // TaskEntry is inserted in the deferq_ based on the Task seqno.
757 : // deferq_ comparison function uses the seqno of the first entry in
758 : // the waitq_. Therefore, if the task to be cancelled is the first
759 : // entry in the waitq_, then delete the entry from the deferq_ and
760 : // add it again.
761 31 : TaskGroup *deferq_tgroup = entry->deferq_task_group_;
762 31 : TaskEntry *deferq_tentry = entry->deferq_task_entry_;
763 31 : if (deferq_tgroup) {
764 16 : assert(deferq_tentry == NULL);
765 16 : deferq_tgroup->DeleteFromDeferQ(*entry);
766 16 : deferq_tgroup->AddToDeferQ(entry);
767 15 : } else if (deferq_tentry) {
768 15 : deferq_tentry->DeleteFromDeferQ(*entry);
769 15 : deferq_tentry->AddToDeferQ(entry);
770 0 : } else if (group->IsDisabled()) {
771 : // Remove TaskEntry from deferq of disable_entry and add back
772 0 : disable_entry->DeleteFromDeferQ(*entry);
773 0 : disable_entry->AddToDeferQ(entry);
774 : } else {
775 0 : if (!entry->IsDisabled()) {
776 0 : assert(0);
777 : }
778 : }
779 : }
780 524 : delete t;
781 524 : cancel_count_++;
782 524 : return CANCELLED;
783 : } else {
784 0 : return FAILED;
785 : }
786 3 : return QUEUED;
787 527 : }
788 :
789 7231265 : void TaskScheduler::OnTaskExit(Task *t) {
790 7231265 : std::scoped_lock lock(mutex_);
791 7234883 : done_count_++;
792 :
793 7234883 : t->tbb_state(Task::TBB_DONE);
794 7234883 : TaskEntry *entry = QueryTaskEntry(t->task_code_id(), t->task_data_id());
795 7234883 : entry->TaskExited(t, GetTaskGroup(t->task_code_id()));
796 :
797 : //
798 : // Delete the task it is not marked for recycling or already cancelled.
799 : //
800 7234883 : if ((t->task_recycle_ == false) || (t->task_cancel_ == true)) {
801 : // Delete the container Task object, if the
802 : // task is not marked to be recycled (or)
803 : // if the task is marked for cancellation
804 5820022 : if (t->task_cancel_ == true) {
805 3 : t->OnTaskCancel();
806 : }
807 5820022 : delete t;
808 5820022 : return;
809 : }
810 :
811 : // Task is being recycled, reset the state, seq_no and TBB task handle
812 1414861 : t->task_impl_ = oneapi::tbb::task_handle{};
813 1414861 : t->seqno(0);
814 1414861 : t->state(Task::INIT);
815 1414861 : t->tbb_state(Task::TBB_INIT);
816 1414861 : EnqueueUnLocked(t);
817 7234883 : }
818 :
819 37729 : void TaskScheduler::Stop() {
820 37729 : std::scoped_lock lock(mutex_);
821 :
822 37729 : running_ = false;
823 37729 : }
824 :
825 37729 : void TaskScheduler::Start() {
826 37729 : std::scoped_lock lock(mutex_);
827 :
828 37729 : running_ = true;
829 :
830 : // Run all tasks that may be suspended
831 37729 : stop_entry_->RunDeferQ();
832 75458 : return;
833 37729 : }
834 :
835 0 : void TaskScheduler::Print() {
836 0 : for (TaskGroupDb::iterator iter = task_group_db_.begin();
837 0 : iter != task_group_db_.end(); ++iter) {
838 0 : TaskGroup *group = *iter;
839 0 : if (group == NULL) {
840 0 : continue;
841 : }
842 :
843 0 : cout << "id: " << group->task_id() <<
844 0 : " run: " << group->TaskRunCount() << endl;
845 0 : cout << "deferq: " << group->deferq_size() <<
846 0 : " task count: " << group->num_tasks() << endl;
847 : }
848 0 : }
849 :
850 1581854 : bool TaskScheduler::IsEmpty(bool running_only) {
851 : TaskGroup *group;
852 :
853 1581854 : std::scoped_lock lock(mutex_);
854 :
855 1581854 : for (TaskGroupDb::iterator it = task_group_db_.begin();
856 22631242 : it != task_group_db_.end(); ++it) {
857 22281154 : if ((group = *it) == NULL) {
858 6698252 : continue;
859 : }
860 15582902 : if (group->TaskRunCount()) {
861 1231766 : return false;
862 : }
863 14769735 : if (group->IsDisabled()) {
864 0 : continue;
865 : }
866 14769735 : if ((false == running_only) && (false == group->IsWaitQEmpty())) {
867 418599 : return false;
868 : }
869 : }
870 :
871 350088 : return true;
872 1581854 : }
873 13 : std::string TaskScheduler::GetTaskName(int task_id) const {
874 405 : for (TaskIdMap::const_iterator it = id_map_.begin(); it != id_map_.end();
875 392 : it++) {
876 405 : if (task_id == it->second)
877 13 : return it->first;
878 : }
879 :
880 0 : return "ERROR";
881 : }
882 :
883 63604833 : int TaskScheduler::GetTaskId(const string &name) {
884 : {
885 : // Grab read-only lock first. Most of the time, task-id already exists
886 : // in the id_map_. Hence there should not be any contention for lock
887 : // aquisition.
888 63604833 : std::shared_lock<std::shared_mutex> lock(id_map_mutex_);
889 63738134 : TaskIdMap::iterator loc = id_map_.find(name);
890 63574579 : if (loc != id_map_.end()) {
891 63564753 : return loc->second;
892 : }
893 63560938 : }
894 :
895 : // Grab read-write lock to allocate a new task id and insert into the map.
896 4345 : std::unique_lock<std::shared_mutex> lock(id_map_mutex_);
897 4345 : int tid = ++id_max_;
898 4345 : id_map_.insert(make_pair(name, tid));
899 4345 : return tid;
900 4345 : }
901 :
902 0 : void TaskScheduler::ClearTaskGroupStats(int task_id) {
903 0 : TaskGroup *group = GetTaskGroup(task_id);
904 0 : if (group == NULL)
905 0 : return;
906 :
907 0 : group->ClearTaskGroupStats();
908 : }
909 :
910 27 : void TaskScheduler::ClearTaskStats(int task_id) {
911 27 : TaskGroup *group = GetTaskGroup(task_id);
912 27 : if (group == NULL)
913 0 : return;
914 :
915 27 : group->ClearTaskStats();
916 : }
917 :
918 0 : void TaskScheduler::ClearTaskStats(int task_id, int instance_id) {
919 0 : TaskGroup *group = GetTaskGroup(task_id);
920 0 : if (group == NULL)
921 0 : return;
922 :
923 0 : group->ClearTaskStats(instance_id);
924 : }
925 :
926 0 : TaskStats *TaskScheduler::GetTaskGroupStats(int task_id) {
927 0 : TaskGroup *group = GetTaskGroup(task_id);
928 0 : if (group == NULL)
929 0 : return NULL;
930 :
931 0 : return group->GetTaskGroupStats();
932 : }
933 :
934 4 : TaskStats *TaskScheduler::GetTaskStats(int task_id) {
935 4 : TaskGroup *group = GetTaskGroup(task_id);
936 4 : if (group == NULL)
937 0 : return NULL;
938 :
939 4 : return group->GetTaskStats();
940 : }
941 :
942 0 : TaskStats *TaskScheduler::GetTaskStats(int task_id, int instance_id) {
943 0 : TaskGroup *group = GetTaskGroup(task_id);
944 0 : if (group == NULL)
945 0 : return NULL;
946 :
947 0 : return group->GetTaskStats(instance_id);
948 : }
949 :
950 190 : void TaskScheduler::Terminate() {
951 190 : if (task_monitor_) {
952 4 : task_monitor_->Terminate();
953 4 : delete task_monitor_;
954 4 : task_monitor_ = NULL;
955 : }
956 :
957 190 : for (int i = 0; i < 10000; i++) {
958 190 : if (IsEmpty()) break;
959 0 : usleep(1000);
960 : }
961 190 : assert(IsEmpty());
962 190 : if (tbb_awake_task_) {
963 0 : tbb_awake_task_->ShutTbbKeepAwakeTask();
964 0 : delete tbb_awake_task_;
965 0 : tbb_awake_task_ = NULL;
966 : }
967 190 : evm_ = NULL;
968 190 : singleton_->task_scheduler_.terminate();
969 : oneapi::tbb::task_scheduler_handle handle =
970 190 : oneapi::tbb::task_scheduler_handle{oneapi::tbb::attach{}};
971 190 : oneapi::tbb::finalize(handle);
972 190 : singleton_.reset(NULL);
973 190 : }
974 :
975 57150 : void TaskScheduler::SetRunningTask(Task *unit_test) {
976 57150 : TaskInfo::reference running = task_running.local();
977 57150 : running = unit_test;
978 57150 : }
979 :
980 57150 : void TaskScheduler::ClearRunningTask() {
981 57150 : TaskInfo::reference running = task_running.local();
982 57150 : running = NULL;
983 57150 : }
984 :
985 44 : void TaskScheduler::SetThreadAmpFactor(int n) {
986 44 : ThreadAmpFactor_ = n;
987 44 : }
988 :
989 4 : void TaskScheduler::DisableTaskGroup(int task_id) {
990 4 : TaskGroup *group = GetTaskGroup(task_id);
991 4 : if (!group->IsDisabled()) {
992 : // Add TaskEntries(that contain enqueued tasks) which are already
993 : // disabled to disable_ entry maintained at TaskGroup.
994 4 : group->SetDisable(true);
995 4 : group->AddEntriesToDisableQ();
996 : }
997 4 : }
998 :
999 4 : void TaskScheduler::EnableTaskGroup(int task_id) {
1000 4 : TaskGroup *group = GetTaskGroup(task_id);
1001 4 : group->SetDisable(false);
1002 : // Run tasks that maybe suspended
1003 4 : group->RunDisableEntries();
1004 4 : }
1005 :
1006 0 : void TaskScheduler::DisableTaskEntry(int task_id, int instance_id) {
1007 0 : TaskEntry *entry = GetTaskEntry(task_id, instance_id);
1008 0 : entry->SetDisable(true);
1009 0 : }
1010 :
1011 0 : void TaskScheduler::EnableTaskEntry(int task_id, int instance_id) {
1012 0 : TaskEntry *entry = GetTaskEntry(task_id, instance_id);
1013 0 : entry->SetDisable(false);
1014 0 : TaskGroup *group = GetTaskGroup(task_id);
1015 : // If group is still disabled, do not schedule the task. Task will be
1016 : // scheduled for run when TaskGroup is enabled.
1017 0 : if (group->IsDisabled()) {
1018 0 : return;
1019 : }
1020 : // Run task instances that maybe suspended
1021 0 : if (entry->WaitQSize() != 0) {
1022 0 : entry->RunDeferEntry();
1023 : }
1024 : }
1025 :
1026 : ////////////////////////////////////////////////////////////////////////////
1027 : // Implementation for class TaskGroup
1028 : ////////////////////////////////////////////////////////////////////////////
1029 :
1030 4232 : TaskGroup::TaskGroup(int task_id) : task_code_id_(task_id), tbb_group_(),
1031 8464 : policy_set_(false), run_count_(0), execute_delay_(0), schedule_delay_(0),
1032 8464 : disable_(false) {
1033 4232 : total_run_time_ = 0;
1034 4232 : task_entry_db_.resize(TaskGroup::kVectorGrowSize);
1035 4232 : task_entry_ = new TaskEntry(task_id);
1036 4232 : memset(&stats_, 0, sizeof(stats_));
1037 4232 : disable_entry_ = new TaskEntry(task_id);
1038 4232 : }
1039 :
1040 4232 : TaskGroup::~TaskGroup() {
1041 4232 : policy_.clear();
1042 4232 : deferq_.clear();
1043 :
1044 4232 : delete task_entry_;
1045 4232 : task_entry_ = NULL;
1046 :
1047 71944 : for (size_t i = 0; i < task_entry_db_.size(); i++) {
1048 67712 : if (task_entry_db_[i] != NULL) {
1049 4901 : delete task_entry_db_[i];
1050 4901 : task_entry_db_[i] = NULL;
1051 : }
1052 : }
1053 :
1054 4232 : delete disable_entry_;
1055 4232 : disable_entry_ = NULL;
1056 4232 : task_entry_db_.clear();
1057 4232 : }
1058 :
1059 7243193 : TaskEntry *TaskGroup::GetTaskEntry(int task_instance) {
1060 7243193 : if (task_instance == -1)
1061 240626 : return task_entry_;
1062 :
1063 7002567 : int size = task_entry_db_.size();
1064 7002567 : if (size <= task_instance) {
1065 0 : task_entry_db_.resize(task_instance + TaskGroup::kVectorGrowSize);
1066 : }
1067 :
1068 7002567 : TaskEntry *entry = task_entry_db_.at(task_instance);
1069 7002567 : if (entry == NULL) {
1070 4901 : entry = new TaskEntry(task_code_id_, task_instance);
1071 4901 : task_entry_db_[task_instance] = entry;
1072 : }
1073 :
1074 7002567 : return entry;
1075 : }
1076 :
1077 7235408 : TaskEntry *TaskGroup::QueryTaskEntry(int task_instance) const {
1078 7235408 : if (task_instance == -1) {
1079 237561 : return task_entry_;
1080 : }
1081 :
1082 6997847 : if (task_instance >= (int)task_entry_db_.size())
1083 0 : return NULL;
1084 :
1085 6997847 : return task_entry_db_[task_instance];
1086 : }
1087 :
1088 49952 : void TaskGroup::AddPolicy(TaskGroup *group) {
1089 49952 : policy_.push_back(group);
1090 49952 : }
1091 :
1092 15206746 : TaskGroup *TaskGroup::ActiveGroupInPolicy() {
1093 15206746 : for (TaskGroupPolicyList::iterator it = policy_.begin();
1094 160860269 : it != policy_.end(); ++it) {
1095 149725883 : if ((*it)->run_count_ != 0) {
1096 4072360 : return (*it);
1097 : }
1098 : }
1099 11134386 : return NULL;
1100 : }
1101 :
1102 15206746 : bool TaskGroup::DeferOnPolicyFail(TaskEntry *entry, Task *task) {
1103 : TaskGroup *group;
1104 15206746 : if ((group = ActiveGroupInPolicy()) != NULL) {
1105 : // TaskEntry is inserted in the deferq_ based on the Task seqno.
1106 : // deferq_ comparison function uses the seqno of the first Task queued
1107 : // in the waitq_. Therefore, add the Task to waitq_ before adding
1108 : // TaskEntry in the deferq_.
1109 4072360 : if (0 == entry->WaitQSize()) {
1110 1622396 : entry->AddToWaitQ(task);
1111 : }
1112 4072360 : group->AddToDeferQ(entry);
1113 4072360 : return true;
1114 : }
1115 11134386 : return false;
1116 : }
1117 :
1118 4072376 : void TaskGroup::AddToDeferQ(TaskEntry *entry) {
1119 4072376 : stats_.defer_count_++;
1120 4072376 : deferq_.insert(*entry);
1121 4072376 : assert(entry->deferq_task_group_ == NULL);
1122 4072376 : entry->deferq_task_group_ = this;
1123 4072376 : }
1124 :
1125 4072376 : void TaskGroup::DeleteFromDeferQ(TaskEntry &entry) {
1126 4072376 : assert(this == entry.deferq_task_group_);
1127 8144752 : deferq_.erase(deferq_.iterator_to(entry));
1128 4072376 : entry.deferq_task_group_ = NULL;
1129 4072376 : }
1130 :
1131 16 : void TaskGroup::AddToDisableQ(TaskEntry *entry) {
1132 16 : disable_entry_->AddToDeferQ(entry);
1133 16 : }
1134 :
1135 3066 : void TaskGroup::PolicySet() {
1136 3066 : assert(policy_set_ == false);
1137 3066 : policy_set_ = true;
1138 3066 : }
1139 :
1140 1818144 : void TaskGroup::RunDeferQ() {
1141 : TaskDeferList::iterator it;
1142 :
1143 1818144 : it = deferq_.begin();
1144 9804772 : while (it != deferq_.end()) {
1145 3084242 : TaskEntry &entry = *it;
1146 3084242 : TaskDeferList::iterator it_work = it++;
1147 3084242 : DeleteFromDeferQ(*it_work);
1148 3084242 : entry.RunDeferEntry();
1149 : }
1150 :
1151 3636288 : return;
1152 : }
1153 :
1154 7234883 : inline void TaskGroup::TaskExited(Task *t) {
1155 7234883 : run_count_--;
1156 7234883 : stats_.total_tasks_completed_++;
1157 7234883 : }
1158 :
1159 4 : void TaskGroup::RunDisableEntries() {
1160 4 : disable_entry_->RunDeferQForGroupEnable();
1161 4 : }
1162 :
1163 4 : void TaskGroup::AddEntriesToDisableQ() {
1164 : TaskEntry *entry;
1165 4 : if (task_entry_->WaitQSize()) {
1166 0 : AddToDisableQ(task_entry_);
1167 : }
1168 :
1169 : // Walk thru the task_entry_db_ and add if waitq is non-empty
1170 4 : for (TaskEntryList::iterator it = task_entry_db_.begin();
1171 68 : it != task_entry_db_.end(); ++it) {
1172 64 : if ((entry = *it) == NULL) {
1173 48 : continue;
1174 : }
1175 16 : if (entry->WaitQSize()) {
1176 0 : AddToDisableQ(entry);
1177 : }
1178 : }
1179 4 : }
1180 :
1181 19124115 : bool TaskGroup::IsWaitQEmpty() {
1182 : TaskEntry *entry;
1183 :
1184 : // Check the waitq_ of the instance -1
1185 19124115 : if (task_entry_->WaitQSize()) {
1186 5 : return false;
1187 : }
1188 :
1189 : // Walk thru the task_entry_db_ until waitq_ of any of the task is non-zero
1190 19124110 : for (TaskEntryList::iterator it = task_entry_db_.begin();
1191 315589543 : it != task_entry_db_.end(); ++it) {
1192 297081375 : if ((entry = *it) == NULL) {
1193 268911435 : continue;
1194 : }
1195 28169940 : if (entry->IsDisabled()) {
1196 0 : continue;
1197 : }
1198 28169940 : if (entry->WaitQSize()) {
1199 615942 : return false;
1200 : }
1201 : }
1202 :
1203 : // Well, no task has been enqueued in this task group
1204 18508168 : return true;
1205 : }
1206 :
1207 0 : void TaskGroup::ClearTaskGroupStats() {
1208 0 : memset(&stats_, 0, sizeof(stats_));
1209 0 : }
1210 :
1211 27 : void TaskGroup::ClearTaskStats() {
1212 27 : task_entry_->ClearTaskStats();
1213 27 : }
1214 :
1215 0 : void TaskGroup::ClearTaskStats(int task_instance) {
1216 0 : TaskEntry *entry = QueryTaskEntry(task_instance);
1217 0 : if (entry != NULL)
1218 0 : entry->ClearTaskStats();
1219 0 : }
1220 :
1221 0 : TaskStats *TaskGroup::GetTaskGroupStats() {
1222 0 : return &stats_;
1223 : }
1224 :
1225 4 : TaskStats *TaskGroup::GetTaskStats() {
1226 4 : return task_entry_->GetTaskStats();
1227 : }
1228 :
1229 0 : TaskStats *TaskGroup::GetTaskStats(int task_instance) {
1230 0 : TaskEntry *entry = QueryTaskEntry(task_instance);
1231 0 : return entry->GetTaskStats();
1232 : }
1233 :
1234 : ////////////////////////////////////////////////////////////////////////////
1235 : // Implementation for class TaskEntry
1236 : ////////////////////////////////////////////////////////////////////////////
1237 :
1238 9802 : TaskEntry::TaskEntry(int task_id, int task_instance) : task_code_id_(task_id),
1239 4901 : task_data_id_(task_instance), run_count_(0), run_task_(NULL),
1240 4901 : waitq_(), deferq_task_entry_(NULL), deferq_task_group_(NULL),
1241 9802 : disable_(false) {
1242 : // When a new TaskEntry is created, adds an implicit rule into policyq_ to
1243 : // ensure that only one Task of an instance is run at a time
1244 4901 : if (task_instance != -1) {
1245 4901 : policyq_.push_back(this);
1246 : }
1247 4901 : memset(&stats_, 0, sizeof(stats_));
1248 : // allocate memory for deferq
1249 4901 : deferq_ = new TaskDeferList;
1250 4901 : }
1251 :
1252 17428 : TaskEntry::TaskEntry(int task_id) : task_code_id_(task_id),
1253 8714 : task_data_id_(-1), run_count_(0), run_task_(NULL),
1254 17428 : deferq_task_entry_(NULL), deferq_task_group_(NULL), disable_(false) {
1255 8714 : memset(&stats_, 0, sizeof(stats_));
1256 : // allocate memory for deferq
1257 8714 : deferq_ = new TaskDeferList;
1258 8714 : }
1259 :
1260 13615 : TaskEntry::~TaskEntry() {
1261 13615 : policyq_.clear();
1262 :
1263 13615 : assert(0 == deferq_->size());
1264 13615 : delete deferq_;
1265 13615 : }
1266 :
1267 9440 : void TaskEntry::AddPolicy(TaskEntry *entry) {
1268 9440 : policyq_.push_back(entry);
1269 9440 : }
1270 :
1271 11134386 : TaskEntry *TaskEntry::ActiveEntryInPolicy() {
1272 33776574 : for (TaskEntryList::iterator it = policyq_.begin(); it != policyq_.end();
1273 22642188 : ++it) {
1274 26541762 : if ((*it)->run_count_ != 0) {
1275 3899574 : return (*it);
1276 : }
1277 : }
1278 :
1279 7234812 : return NULL;
1280 : }
1281 :
1282 11134386 : bool TaskEntry::DeferOnPolicyFail(Task *task) {
1283 : TaskEntry *policy_entry;
1284 :
1285 11134386 : if ((policy_entry = ActiveEntryInPolicy()) != NULL) {
1286 : // TaskEntry is inserted in the deferq_ based on the Task seqno.
1287 : // deferq_ comparison function uses the seqno of the first Task queued
1288 : // in the waitq_. Therefore, add the Task to waitq_ before adding
1289 : // TaskEntry in the deferq_.
1290 3899574 : if (0 == WaitQSize()) {
1291 1468486 : AddToWaitQ(task);
1292 : }
1293 3899574 : policy_entry->AddToDeferQ(this);
1294 3899574 : return true;
1295 : }
1296 7234812 : return false;
1297 : }
1298 :
1299 5945713 : void TaskEntry::AddToWaitQ(Task *t) {
1300 5945713 : t->state(Task::WAIT);
1301 5945713 : stats_.wait_count_++;
1302 5945713 : waitq_.push_back(*t);
1303 :
1304 5945713 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
1305 5945713 : TaskGroup *group = scheduler->GetTaskGroup(task_code_id_);
1306 5945713 : group->stats_.wait_count_++;
1307 5945713 : }
1308 :
1309 5945713 : bool TaskEntry::DeleteFromWaitQ(Task *t) {
1310 5945713 : TaskWaitQ::iterator it = waitq_.iterator_to(*t);
1311 11891426 : waitq_.erase(it);
1312 5945713 : return true;
1313 : }
1314 :
1315 6754237 : void TaskEntry::AddToDeferQ(TaskEntry *entry) {
1316 6754237 : stats_.defer_count_++;
1317 6754237 : deferq_->insert(*entry);
1318 6754237 : assert(entry->deferq_task_entry_ == NULL);
1319 6754237 : entry->deferq_task_entry_ = this;
1320 6754237 : }
1321 :
1322 6754237 : void TaskEntry::DeleteFromDeferQ(TaskEntry &entry) {
1323 6754237 : assert(this == entry.deferq_task_entry_);
1324 13508474 : deferq_->erase(deferq_->iterator_to(entry));
1325 6754237 : entry.deferq_task_entry_ = NULL;
1326 6754237 : }
1327 :
1328 7234883 : void TaskEntry::RunTask(Task *t) {
1329 7234883 : stats_.run_count_++;
1330 7234883 : if (t->task_data_id() != -1) {
1331 6997324 : assert(run_task_ == NULL);
1332 6997324 : assert (run_count_ == 0);
1333 6997324 : run_task_ = t;
1334 : }
1335 :
1336 7234883 : run_count_++;
1337 7234883 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
1338 7234883 : TaskGroup *group = scheduler->QueryTaskGroup(t->task_code_id());
1339 7234883 : group->TaskStarted();
1340 :
1341 7234883 : t->StartTask(scheduler, group);
1342 7234883 : }
1343 :
1344 5945118 : void TaskEntry::RunWaitQ() {
1345 5945118 : if (waitq_.size() == 0)
1346 0 : return;
1347 :
1348 5945118 : TaskWaitQ::iterator it = waitq_.begin();
1349 :
1350 5945118 : if (task_data_id_ != -1) {
1351 5939058 : Task *t = &(*it);
1352 5939058 : DeleteFromWaitQ(t);
1353 5939058 : RunTask(t);
1354 : // If there are more tasks in waitq_, put them in deferq_
1355 5939058 : if (waitq_.size() != 0) {
1356 2849323 : AddToDeferQ(this);
1357 : }
1358 : } else {
1359 : // Run all instances in waitq_
1360 12262 : while (it != waitq_.end()) {
1361 6131 : Task *t = &(*it);
1362 6131 : DeleteFromWaitQ(t);
1363 6131 : RunTask(t);
1364 6131 : if (waitq_.size() == 0)
1365 6060 : break;
1366 142 : it = waitq_.begin();
1367 : }
1368 : }
1369 : }
1370 :
1371 10826170 : void TaskEntry::RunDeferEntry() {
1372 10826170 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
1373 10826170 : TaskGroup *group = scheduler->GetTaskGroup(task_code_id_);
1374 :
1375 : // Sanity check
1376 10826170 : assert(waitq_.size());
1377 10826170 : Task *task = &(*waitq_.begin());
1378 :
1379 : // Check Task group policies
1380 10826170 : if (group->DeferOnPolicyFail(this, task)) {
1381 2449964 : return;
1382 : }
1383 :
1384 : // Check Task entry policies
1385 8376206 : if (DeferOnPolicyFail(task)) {
1386 2431088 : return;
1387 : }
1388 :
1389 5945118 : RunWaitQ();
1390 5945118 : return;
1391 : }
1392 :
1393 4407098 : void TaskEntry::RunDeferQ() {
1394 : TaskDeferList::iterator it;
1395 :
1396 4407098 : it = deferq_->begin();
1397 21112452 : while (it != deferq_->end()) {
1398 6149128 : TaskEntry &entry = *it;
1399 6149128 : TaskDeferList::iterator it_work = it++;
1400 6149128 : DeleteFromDeferQ(*it_work);
1401 6149128 : entry.RunDeferEntry();
1402 : }
1403 :
1404 8814196 : return;
1405 : }
1406 :
1407 4 : void TaskEntry::RunDeferQForGroupEnable() {
1408 : TaskDeferList::iterator it;
1409 :
1410 4 : it = deferq_->begin();
1411 40 : while (it != deferq_->end()) {
1412 16 : TaskEntry &entry = *it;
1413 16 : TaskDeferList::iterator it_work = it++;
1414 16 : DeleteFromDeferQ(*it_work);
1415 16 : if (!entry.IsDisabled()) {
1416 16 : entry.RunDeferEntry();
1417 : }
1418 : }
1419 :
1420 8 : return;
1421 : }
1422 :
1423 5835282 : void TaskEntry::RunCombinedDeferQ() {
1424 5835282 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
1425 5835282 : TaskGroup *group = scheduler->QueryTaskGroup(task_code_id_);
1426 : TaskDeferEntryCmp defer_entry_compare;
1427 :
1428 5835282 : TaskDeferList::iterator group_it = group->deferq_.begin();
1429 5835282 : TaskDeferList::iterator entry_it = deferq_->begin();
1430 :
1431 : // Loop thru the deferq_ of TaskEntry and TaskGroup in the temporal order.
1432 : // Exit the loop when any of the queues become empty.
1433 18267060 : while ((group_it != group->deferq_.end()) &&
1434 14249922 : (entry_it != deferq_->end())) {
1435 1592784 : TaskEntry &g_entry = *group_it;
1436 1592784 : TaskEntry &t_entry = *entry_it;
1437 :
1438 1592784 : if (defer_entry_compare(g_entry, t_entry)) {
1439 987757 : TaskDeferList::iterator group_it_work = group_it++;
1440 1975514 : group->DeleteFromDeferQ(*group_it_work);
1441 987757 : g_entry.RunDeferEntry();
1442 : } else {
1443 605027 : TaskDeferList::iterator entry_it_work = entry_it++;
1444 605027 : DeleteFromDeferQ(*entry_it_work);
1445 605027 : t_entry.RunDeferEntry();
1446 : }
1447 : }
1448 :
1449 : // Now, walk thru the non-empty deferq_
1450 11670564 : if (group_it != group->deferq_.end()) {
1451 1818144 : group->RunDeferQ();
1452 8034276 : } else if (entry_it != deferq_->end()) {
1453 3183374 : RunDeferQ();
1454 : }
1455 5835282 : }
1456 :
1457 7234883 : void TaskEntry::TaskExited(Task *t, TaskGroup *group) {
1458 7234883 : if (task_data_id_ != -1) {
1459 6997324 : assert(run_task_ == t);
1460 6997324 : run_task_ = NULL;
1461 6997324 : assert(run_count_ == 1);
1462 : }
1463 :
1464 7234883 : run_count_--;
1465 7234883 : stats_.total_tasks_completed_++;
1466 7234883 : stats_.last_exit_time_ = UTCTimestampUsec();
1467 7234883 : group->TaskExited(t);
1468 :
1469 7234883 : if (!group->run_count_ && !run_count_) {
1470 5835282 : RunCombinedDeferQ();
1471 1399601 : } else if (!group->run_count_) {
1472 0 : group->RunDeferQ();
1473 1399601 : } else if (!run_count_) {
1474 1185995 : RunDeferQ();
1475 : }
1476 7234883 : }
1477 :
1478 0 : void TaskEntry::ClearQueues() {
1479 0 : deferq_->clear();
1480 0 : policyq_.clear();
1481 0 : waitq_.clear();
1482 0 : }
1483 :
1484 27 : void TaskEntry::ClearTaskStats() {
1485 27 : memset(&stats_, 0, sizeof(stats_));
1486 27 : }
1487 :
1488 4 : TaskStats *TaskEntry::GetTaskStats() {
1489 4 : return &stats_;
1490 : }
1491 :
1492 29278286 : boost::optional<uint64_t> TaskEntry::GetTaskDeferEntrySeqno() const {
1493 29278286 : if(waitq_.size()) {
1494 29278286 : const Task *task = &(*waitq_.begin());
1495 29278286 : return task->seqno();
1496 : }
1497 :
1498 0 : return boost::none;
1499 : }
1500 :
1501 : ////////////////////////////////////////////////////////////////////////////
1502 : // Implementation for class Task
1503 : ////////////////////////////////////////////////////////////////////////////
1504 5819006 : Task::Task(int task_id, int task_instance) : task_code_id_(task_id),
1505 11637015 : task_data_id_(task_instance), task_impl_(), state_(INIT),
1506 5818009 : tbb_state_(TBB_INIT), seqno_(0), task_recycle_(false), task_cancel_(false),
1507 5819006 : enqueue_time_(0), schedule_time_(0), execute_delay_(0), schedule_delay_(0) {
1508 5818006 : }
1509 :
1510 57156 : Task::Task(int task_id) : task_code_id_(task_id),
1511 114312 : task_data_id_(-1), task_impl_(), state_(INIT), tbb_state_(TBB_INIT),
1512 57156 : seqno_(0), task_recycle_(false), task_cancel_(false), enqueue_time_(0),
1513 57156 : schedule_time_(0), execute_delay_(0), schedule_delay_(0) {
1514 57156 : }
1515 :
1516 :
1517 7234883 : void Task::StartTask(TaskScheduler *scheduler, TaskGroup *group) {
1518 7234883 : if (enqueue_time_ != 0) {
1519 0 : schedule_time_ = ClockMonotonicUsec();
1520 0 : if ((schedule_time_ - enqueue_time_) >
1521 0 : scheduler->schedule_delay(this)) {
1522 0 : TASK_TRACE(scheduler, this, "Schedule delay(in usec) ",
1523 : (schedule_time_ - enqueue_time_));
1524 : }
1525 : }
1526 7234883 : state(RUN);
1527 7234883 : tbb_state(TBB_ENQUEUED);
1528 7234883 : task_impl_ = group->tbb_group().defer(TaskFunctor(this));
1529 7234883 : scheduler->tbb_arena().enqueue(std::move(task_impl_));
1530 7234883 : }
1531 :
1532 48437749 : Task *Task::Running() {
1533 48437749 : TaskInfo::reference running = task_running.local();
1534 48320635 : return running;
1535 : }
1536 :
1537 0 : ostream& operator<<(ostream& out, const Task &t) {
1538 0 : out << "Task <" << t.task_code_id_ << "," << t.task_data_id_ << ":"
1539 0 : << t.seqno_ << "> ";
1540 0 : return out;
1541 : }
1542 :
1543 : ////////////////////////////////////////////////////////////////////////////
1544 : // Implementation for sandesh APIs for Task
1545 : ////////////////////////////////////////////////////////////////////////////
1546 1 : void TaskEntry::GetSandeshData(SandeshTaskEntry *resp) const {
1547 1 : resp->set_instance_id(task_data_id_);
1548 1 : resp->set_tasks_created(stats_.enqueue_count_);
1549 1 : resp->set_total_tasks_completed(stats_.total_tasks_completed_);
1550 1 : resp->set_tasks_running(run_count_);
1551 1 : resp->set_waitq_size(waitq_.size());
1552 1 : resp->set_deferq_size(deferq_->size());
1553 1 : resp->set_last_exit_time(stats_.last_exit_time_);
1554 1 : }
1555 1 : void TaskGroup::GetSandeshData(SandeshTaskGroup *resp, bool summary) const {
1556 1 : if (total_run_time_)
1557 0 : resp->set_total_run_time(duration_usecs_to_string(total_run_time_));
1558 :
1559 1 : std::vector<SandeshTaskEntry> list;
1560 1 : TaskEntry *task_entry = QueryTaskEntry(-1);
1561 1 : if (task_entry) {
1562 1 : SandeshTaskEntry entry_resp;
1563 1 : task_entry->GetSandeshData(&entry_resp);
1564 1 : list.push_back(entry_resp);
1565 1 : }
1566 1 : for (TaskEntryList::const_iterator it = task_entry_db_.begin();
1567 17 : it != task_entry_db_.end(); ++it) {
1568 16 : task_entry = *it;
1569 16 : if (task_entry) {
1570 0 : SandeshTaskEntry entry_resp;
1571 0 : task_entry->GetSandeshData(&entry_resp);
1572 0 : list.push_back(entry_resp);
1573 0 : }
1574 : }
1575 1 : resp->set_task_entry_list(list);
1576 :
1577 1 : if (summary)
1578 0 : return;
1579 :
1580 1 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
1581 1 : std::vector<SandeshTaskPolicyEntry> policy_list;
1582 1 : for (TaskGroupPolicyList::const_iterator it = policy_.begin();
1583 14 : it != policy_.end(); ++it) {
1584 13 : SandeshTaskPolicyEntry policy_entry;
1585 13 : policy_entry.set_task_name(scheduler->GetTaskName((*it)->task_code_id_));
1586 13 : policy_entry.set_tasks_running((*it)->run_count_);
1587 13 : policy_list.push_back(policy_entry);
1588 13 : }
1589 1 : resp->set_task_policy_list(policy_list);
1590 1 : }
1591 :
1592 0 : void TaskScheduler::GetSandeshData(SandeshTaskScheduler *resp, bool summary) {
1593 0 : std::scoped_lock lock(mutex_);
1594 :
1595 0 : resp->set_running(running_);
1596 0 : resp->set_use_spawn(use_spawn_);
1597 0 : resp->set_total_count(seqno_);
1598 0 : resp->set_thread_count(hw_thread_count_);
1599 :
1600 0 : std::vector<SandeshTaskGroup> list;
1601 0 : for (TaskIdMap::const_iterator it = id_map_.begin(); it != id_map_.end();
1602 0 : it++) {
1603 0 : SandeshTaskGroup resp_group;
1604 0 : TaskGroup *group = QueryTaskGroup(it->second);
1605 0 : resp_group.set_task_id(it->second);
1606 0 : resp_group.set_name(it->first);
1607 0 : if (group)
1608 0 : group->GetSandeshData(&resp_group, summary);
1609 0 : list.push_back(resp_group);
1610 0 : }
1611 0 : resp->set_task_group_list(list);
1612 0 : }
|
