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 499269 : 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 499269 : TaskFunctor(TaskFunctor &&tf)
58 499269 : : 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 4301687 : 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 3214582 : 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 163063 : bool operator() (const TaskEntry &lhs, const TaskEntry &rhs) const {
220 163063 : return (lhs.GetTaskDeferEntrySeqno() <
221 326126 : 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 27 : 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 1241893 : 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 499269 : 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 0 : void SetDisable(bool disable) { disable_ = disable; }
295 1395517 : 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 499269 : oneapi::tbb::task_group &tbb_group() {
312 499269 : 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 499012 : void TaskFunctor::operator ()() const {
371 499012 : TaskInfo::reference running = task_running.local();
372 498410 : running = parent_;
373 498410 : parent_->tbb_state(Task::TBB_EXEC);
374 : try {
375 498328 : uint64_t t = 0;
376 498328 : 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 498342 : } else if (TaskScheduler::GetInstance()->track_run_time()) {
385 0 : t = ClockMonotonicUsec();
386 : }
387 498213 : bool is_complete = parent_->Run();
388 496893 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
389 497862 : 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 497810 : running = NULL;
403 497810 : if (is_complete == true) {
404 491367 : parent_->set_task_complete();
405 : } else {
406 6443 : parent_->set_task_recycle();
407 : }
408 497729 : 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 499202 : }
424 :
425 998158 : TaskFunctor::~TaskFunctor() {
426 998158 : }
427 :
428 4203 : int TaskScheduler::GetThreadCount(int thread_count) {
429 : static bool init_;
430 : static int num_cores_;
431 :
432 4203 : if (init_) {
433 4112 : return num_cores_ * ThreadAmpFactor_;
434 : }
435 :
436 91 : char *num_cores_str = getenv("TBB_THREAD_COUNT");
437 91 : if (!num_cores_str) {
438 91 : if (thread_count == 0) {
439 91 : 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 91 : init_ = true;
448 91 : return num_cores_ * ThreadAmpFactor_;
449 : }
450 :
451 0 : int TaskScheduler::GetDefaultThreadCount() {
452 0 : return oneapi::tbb::this_task_arena::max_concurrency();
453 : }
454 :
455 91 : bool TaskScheduler::ShouldUseSpawn() {
456 91 : if (getenv("TBB_USE_SPAWN"))
457 0 : return true;
458 :
459 91 : return false;
460 : }
461 :
462 : ////////////////////////////////////////////////////////////////////////////
463 : // Implementation for class TaskScheduler
464 : ////////////////////////////////////////////////////////////////////////////
465 :
466 91 : TaskScheduler::TaskScheduler(int task_count) :
467 91 : use_spawn_(ShouldUseSpawn()),
468 182 : tbb_global_control_(
469 : oneapi::tbb::global_control::max_allowed_parallelism,
470 91 : GetThreadCount(task_count) + 1),
471 91 : task_scheduler_(GetThreadCount(task_count) + 1),
472 91 : running_(true), seqno_(0), id_max_(0), log_fn_(), track_run_time_(false),
473 91 : measure_delay_(false), schedule_delay_(0), execute_delay_(0),
474 91 : enqueue_count_(0), done_count_(0), cancel_count_(0), evm_(NULL),
475 91 : tbb_awake_task_(NULL), task_monitor_(NULL) {
476 91 : hw_thread_count_ = GetThreadCount(task_count);
477 91 : task_group_db_.resize(TaskScheduler::kVectorGrowSize);
478 91 : stop_entry_ = new TaskEntry(-1);
479 91 : }
480 :
481 91 : TaskScheduler::~TaskScheduler() {
482 : TaskGroup *group;
483 :
484 91 : for (TaskGroupDb::iterator iter = task_group_db_.begin();
485 2048 : iter != task_group_db_.end(); ++iter) {
486 1957 : if ((group = *iter) == NULL) {
487 1137 : continue;
488 : }
489 820 : *iter = NULL;
490 820 : delete group;
491 : }
492 :
493 1000 : for (TaskIdMap::iterator loc = id_map_.begin(); loc != id_map_.end();
494 909 : id_map_.erase(loc++)) {
495 : }
496 :
497 91 : delete stop_entry_;
498 91 : stop_entry_ = NULL;
499 91 : task_group_db_.clear();
500 :
501 91 : return;
502 91 : }
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 2 : void TaskScheduler::ModifyTbbKeepAwakeTimeout(uint32_t timeout) {
532 2 : if (tbb_awake_task_) {
533 0 : tbb_awake_task_->ModifyTbbKeepAwakeTimeout(timeout);
534 : }
535 2 : }
536 :
537 2 : 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 2 : if (task_monitor_ != NULL)
542 0 : return;
543 :
544 2 : task_monitor_ = new TaskMonitor(this, tbb_keepawake_time_msec,
545 2 : inactivity_time_msec, poll_interval_msec);
546 2 : 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 2 : void TaskScheduler::RegisterLog(LogFn fn) {
558 2 : log_fn_ = fn;
559 2 : }
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 2847135 : TaskScheduler *TaskScheduler::GetInstance() {
574 2847135 : if (singleton_.get() == NULL) {
575 91 : singleton_.reset(new TaskScheduler());
576 : }
577 2848399 : return singleton_.get();
578 : }
579 :
580 1791818 : TaskGroup *TaskScheduler::GetTaskGroup(int task_id) {
581 1791818 : assert(task_id >= 0);
582 1791818 : int size = task_group_db_.size();
583 1791818 : if (size <= task_id) {
584 25 : task_group_db_.resize(task_id + TaskScheduler::kVectorGrowSize);
585 : }
586 :
587 1791818 : TaskGroup *group = task_group_db_[task_id];
588 1791818 : if (group == NULL) {
589 820 : group = new TaskGroup(task_id);
590 820 : task_group_db_[task_id] = group;
591 : }
592 :
593 1791818 : return group;
594 : }
595 :
596 1135420 : TaskGroup *TaskScheduler::QueryTaskGroup(int task_id) {
597 1135420 : return task_group_db_[task_id];
598 : }
599 :
600 50559 : bool TaskScheduler::IsTaskGroupEmpty(int task_id) const {
601 50559 : CHECK_CONCURRENCY("bgp::Config");
602 50559 : std::scoped_lock lock(mutex_);
603 50559 : TaskGroup *group = task_group_db_[task_id];
604 50559 : assert(group);
605 50559 : assert(group->TaskRunCount() == 0);
606 101118 : return group->IsWaitQEmpty();
607 50559 : }
608 :
609 500200 : TaskEntry *TaskScheduler::GetTaskEntry(int task_id, int task_instance) {
610 500200 : TaskGroup *group = GetTaskGroup(task_id);
611 500200 : return group->GetTaskEntry(task_instance);
612 : }
613 :
614 499296 : TaskEntry *TaskScheduler::QueryTaskEntry(int task_id, int task_instance) {
615 499296 : TaskGroup *group = QueryTaskGroup(task_id);
616 499296 : if (group == NULL)
617 0 : return NULL;
618 499296 : return group->QueryTaskEntry(task_instance);
619 : }
620 :
621 4 : void TaskScheduler::EnableLatencyThresholds(uint32_t execute,
622 : uint32_t schedule) {
623 4 : execute_delay_ = execute;
624 4 : schedule_delay_ = schedule;
625 4 : measure_delay_ = (execute_delay_ != 0 || schedule_delay_ != 0);
626 4 : }
627 :
628 12 : void TaskScheduler::SetLatencyThreshold(const std::string &name,
629 : uint32_t execute, uint32_t schedule) {
630 12 : int task_id = GetTaskId(name);
631 12 : TaskGroup *group = GetTaskGroup(task_id);
632 12 : group->execute_delay_ = execute;
633 12 : group->schedule_delay_ = schedule;
634 12 : }
635 :
636 492 : void TaskScheduler::SetPolicy(int task_id, TaskPolicy &policy) {
637 492 : std::scoped_lock lock(mutex_);
638 :
639 492 : TaskGroup *group = GetTaskGroup(task_id);
640 492 : TaskEntry *group_entry = group->GetTaskEntry(-1);
641 492 : group->PolicySet();
642 :
643 4095 : for (const auto& pol_item: policy) {
644 3603 : if (pol_item.match_data_id == -1) {
645 3151 : TaskGroup *policy_group = GetTaskGroup(pol_item.match_code_id);
646 3151 : group->AddPolicy(policy_group);
647 3151 : policy_group->AddPolicy(group);
648 : } else {
649 452 : TaskEntry *entry = GetTaskEntry(task_id, pol_item.match_data_id);
650 904 : TaskEntry *policy_entry = GetTaskEntry(pol_item.match_code_id,
651 452 : pol_item.match_data_id);
652 452 : entry->AddPolicy(policy_entry);
653 452 : policy_entry->AddPolicy(entry);
654 :
655 452 : group_entry->AddPolicy(policy_entry);
656 452 : policy_entry->AddPolicy(group_entry);
657 : }
658 : }
659 492 : }
660 :
661 492501 : void TaskScheduler::Enqueue(Task *t) {
662 492501 : std::scoped_lock lock(mutex_);
663 492853 : EnqueueUnLocked(t);
664 492853 : }
665 :
666 499296 : void TaskScheduler::EnqueueUnLocked(Task *t) {
667 499296 : if (measure_delay_) {
668 0 : t->enqueue_time_ = ClockMonotonicUsec();
669 : }
670 : // Ensure that task is enqueued only once.
671 499296 : assert(t->seqno() == 0);
672 499296 : enqueue_count_++;
673 499296 : t->seqno(++seqno_);
674 499296 : TaskGroup *group = GetTaskGroup(t->task_code_id());
675 499296 : t->schedule_delay_ = group->schedule_delay_;
676 499296 : t->execute_delay_ = group->execute_delay_;
677 499296 : group->stats_.enqueue_count_++;
678 :
679 499296 : TaskEntry *entry = GetTaskEntry(t->task_code_id(), t->task_data_id());
680 499296 : 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 499296 : if (group->IsDisabled() || entry->IsDisabled()) {
685 0 : entry->AddToWaitQ(t);
686 0 : if (group->IsDisabled()) {
687 0 : group->AddToDisableQ(entry);
688 : }
689 0 : return;
690 : }
691 :
692 : // Add task to waitq_ if its already populated
693 499296 : if (entry->WaitQSize() != 0) {
694 11091 : entry->AddToWaitQ(t);
695 11091 : 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 488205 : if (!running_) {
701 9 : entry->AddToWaitQ(t);
702 9 : stop_entry_->AddToDeferQ(entry);
703 9 : 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 488196 : if (group->DeferOnPolicyFail(entry, t)) {
710 50392 : 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 437804 : if (entry->DeferOnPolicyFail(t)) {
717 72961 : return;
718 : }
719 :
720 :
721 364843 : entry->RunTask(t);
722 :
723 364843 : return;
724 : }
725 :
726 29 : TaskScheduler::CancelReturnCode TaskScheduler::Cancel(Task *t) {
727 29 : std::scoped_lock lock(mutex_);
728 :
729 : // If the task is in RUN state, mark the task for cancellation and return.
730 29 : if (t->state_ == Task::RUN) {
731 2 : t->task_cancel_ = true;
732 27 : } else if (t->state_ == Task::WAIT) {
733 27 : TaskEntry *entry = QueryTaskEntry(t->task_code_id(), t->task_data_id());
734 27 : TaskGroup *group = QueryTaskGroup(t->task_code_id());
735 27 : assert(entry->WaitQSize());
736 : // Get the first entry in the waitq_
737 27 : Task *first_wait_task = &(*entry->waitq_.begin());
738 27 : TaskEntry *disable_entry = group->GetDisableEntry();
739 27 : assert(entry->DeleteFromWaitQ(t) == true);
740 : // If the waitq_ is empty, then remove the TaskEntry from the deferq.
741 27 : if (!entry->WaitQSize()) {
742 27 : if (entry->deferq_task_group_) {
743 21 : assert(entry->deferq_task_entry_ == NULL);
744 21 : entry->deferq_task_group_->DeleteFromDeferQ(*entry);
745 6 : } else if (entry->deferq_task_entry_) {
746 6 : 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 0 : } 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 0 : TaskGroup *deferq_tgroup = entry->deferq_task_group_;
762 0 : TaskEntry *deferq_tentry = entry->deferq_task_entry_;
763 0 : if (deferq_tgroup) {
764 0 : assert(deferq_tentry == NULL);
765 0 : deferq_tgroup->DeleteFromDeferQ(*entry);
766 0 : deferq_tgroup->AddToDeferQ(entry);
767 0 : } else if (deferq_tentry) {
768 0 : deferq_tentry->DeleteFromDeferQ(*entry);
769 0 : 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 27 : delete t;
781 27 : cancel_count_++;
782 27 : return CANCELLED;
783 : } else {
784 0 : return FAILED;
785 : }
786 2 : return QUEUED;
787 29 : }
788 :
789 497446 : void TaskScheduler::OnTaskExit(Task *t) {
790 497446 : std::scoped_lock lock(mutex_);
791 499269 : done_count_++;
792 :
793 499269 : t->tbb_state(Task::TBB_DONE);
794 499269 : TaskEntry *entry = QueryTaskEntry(t->task_code_id(), t->task_data_id());
795 499269 : 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 499269 : 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 492826 : if (t->task_cancel_ == true) {
805 2 : t->OnTaskCancel();
806 : }
807 492826 : delete t;
808 492826 : return;
809 : }
810 :
811 : // Task is being recycled, reset the state, seq_no and TBB task handle
812 6443 : t->task_impl_ = oneapi::tbb::task_handle{};
813 6443 : t->seqno(0);
814 6443 : t->state(Task::INIT);
815 6443 : t->tbb_state(Task::TBB_INIT);
816 6443 : EnqueueUnLocked(t);
817 499269 : }
818 :
819 466 : void TaskScheduler::Stop() {
820 466 : std::scoped_lock lock(mutex_);
821 :
822 466 : running_ = false;
823 466 : }
824 :
825 466 : void TaskScheduler::Start() {
826 466 : std::scoped_lock lock(mutex_);
827 :
828 466 : running_ = true;
829 :
830 : // Run all tasks that may be suspended
831 466 : stop_entry_->RunDeferQ();
832 932 : return;
833 466 : }
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 351366 : bool TaskScheduler::IsEmpty(bool running_only) {
851 : TaskGroup *group;
852 :
853 351366 : std::scoped_lock lock(mutex_);
854 :
855 351366 : for (TaskGroupDb::iterator it = task_group_db_.begin();
856 1790328 : it != task_group_db_.end(); ++it) {
857 1767243 : if ((group = *it) == NULL) {
858 575909 : continue;
859 : }
860 1191334 : if (group->TaskRunCount()) {
861 328281 : return false;
862 : }
863 896221 : if (group->IsDisabled()) {
864 0 : continue;
865 : }
866 896221 : if ((false == running_only) && (false == group->IsWaitQEmpty())) {
867 33168 : return false;
868 : }
869 : }
870 :
871 23085 : return true;
872 351366 : }
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 437397 : 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 437397 : std::shared_lock<std::shared_mutex> lock(id_map_mutex_);
889 437485 : TaskIdMap::iterator loc = id_map_.find(name);
890 437326 : if (loc != id_map_.end()) {
891 436417 : return loc->second;
892 : }
893 437318 : }
894 :
895 : // Grab read-write lock to allocate a new task id and insert into the map.
896 909 : std::unique_lock<std::shared_mutex> lock(id_map_mutex_);
897 909 : int tid = ++id_max_;
898 909 : id_map_.insert(make_pair(name, tid));
899 909 : return tid;
900 909 : }
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 33 : void TaskScheduler::Terminate() {
951 33 : if (task_monitor_) {
952 2 : task_monitor_->Terminate();
953 2 : delete task_monitor_;
954 2 : task_monitor_ = NULL;
955 : }
956 :
957 33 : for (int i = 0; i < 10000; i++) {
958 33 : if (IsEmpty()) break;
959 0 : usleep(1000);
960 : }
961 33 : assert(IsEmpty());
962 33 : if (tbb_awake_task_) {
963 0 : tbb_awake_task_->ShutTbbKeepAwakeTask();
964 0 : delete tbb_awake_task_;
965 0 : tbb_awake_task_ = NULL;
966 : }
967 33 : evm_ = NULL;
968 33 : singleton_->task_scheduler_.terminate();
969 : oneapi::tbb::task_scheduler_handle handle =
970 33 : oneapi::tbb::task_scheduler_handle{oneapi::tbb::attach{}};
971 33 : oneapi::tbb::finalize(handle);
972 33 : singleton_.reset(NULL);
973 33 : }
974 :
975 410 : void TaskScheduler::SetRunningTask(Task *unit_test) {
976 410 : TaskInfo::reference running = task_running.local();
977 410 : running = unit_test;
978 410 : }
979 :
980 410 : void TaskScheduler::ClearRunningTask() {
981 410 : TaskInfo::reference running = task_running.local();
982 410 : running = NULL;
983 410 : }
984 :
985 44 : void TaskScheduler::SetThreadAmpFactor(int n) {
986 44 : ThreadAmpFactor_ = n;
987 44 : }
988 :
989 0 : void TaskScheduler::DisableTaskGroup(int task_id) {
990 0 : TaskGroup *group = GetTaskGroup(task_id);
991 0 : if (!group->IsDisabled()) {
992 : // Add TaskEntries(that contain enqueued tasks) which are already
993 : // disabled to disable_ entry maintained at TaskGroup.
994 0 : group->SetDisable(true);
995 0 : group->AddEntriesToDisableQ();
996 : }
997 0 : }
998 :
999 0 : void TaskScheduler::EnableTaskGroup(int task_id) {
1000 0 : TaskGroup *group = GetTaskGroup(task_id);
1001 0 : group->SetDisable(false);
1002 : // Run tasks that maybe suspended
1003 0 : group->RunDisableEntries();
1004 0 : }
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 820 : TaskGroup::TaskGroup(int task_id) : task_code_id_(task_id), tbb_group_(),
1031 1640 : policy_set_(false), run_count_(0), execute_delay_(0), schedule_delay_(0),
1032 1640 : disable_(false) {
1033 820 : total_run_time_ = 0;
1034 820 : task_entry_db_.resize(TaskGroup::kVectorGrowSize);
1035 820 : task_entry_ = new TaskEntry(task_id);
1036 820 : memset(&stats_, 0, sizeof(stats_));
1037 820 : disable_entry_ = new TaskEntry(task_id);
1038 820 : }
1039 :
1040 820 : TaskGroup::~TaskGroup() {
1041 820 : policy_.clear();
1042 820 : deferq_.clear();
1043 :
1044 820 : delete task_entry_;
1045 820 : task_entry_ = NULL;
1046 :
1047 13940 : for (size_t i = 0; i < task_entry_db_.size(); i++) {
1048 13120 : if (task_entry_db_[i] != NULL) {
1049 776 : delete task_entry_db_[i];
1050 776 : task_entry_db_[i] = NULL;
1051 : }
1052 : }
1053 :
1054 820 : delete disable_entry_;
1055 820 : disable_entry_ = NULL;
1056 820 : task_entry_db_.clear();
1057 820 : }
1058 :
1059 500692 : TaskEntry *TaskGroup::GetTaskEntry(int task_instance) {
1060 500692 : if (task_instance == -1)
1061 232335 : return task_entry_;
1062 :
1063 268357 : int size = task_entry_db_.size();
1064 268357 : if (size <= task_instance) {
1065 0 : task_entry_db_.resize(task_instance + TaskGroup::kVectorGrowSize);
1066 : }
1067 :
1068 268357 : TaskEntry *entry = task_entry_db_.at(task_instance);
1069 268357 : if (entry == NULL) {
1070 776 : entry = new TaskEntry(task_code_id_, task_instance);
1071 776 : task_entry_db_[task_instance] = entry;
1072 : }
1073 :
1074 268357 : return entry;
1075 : }
1076 :
1077 499297 : TaskEntry *TaskGroup::QueryTaskEntry(int task_instance) const {
1078 499297 : if (task_instance == -1) {
1079 231844 : return task_entry_;
1080 : }
1081 :
1082 267453 : if (task_instance >= (int)task_entry_db_.size())
1083 0 : return NULL;
1084 :
1085 267453 : return task_entry_db_[task_instance];
1086 : }
1087 :
1088 6302 : void TaskGroup::AddPolicy(TaskGroup *group) {
1089 6302 : policy_.push_back(group);
1090 6302 : }
1091 :
1092 643110 : TaskGroup *TaskGroup::ActiveGroupInPolicy() {
1093 643110 : for (TaskGroupPolicyList::iterator it = policy_.begin();
1094 2414408 : it != policy_.end(); ++it) {
1095 1835448 : if ((*it)->run_count_ != 0) {
1096 64150 : return (*it);
1097 : }
1098 : }
1099 578960 : return NULL;
1100 : }
1101 :
1102 643110 : bool TaskGroup::DeferOnPolicyFail(TaskEntry *entry, Task *task) {
1103 : TaskGroup *group;
1104 643110 : 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 64150 : if (0 == entry->WaitQSize()) {
1110 50392 : entry->AddToWaitQ(task);
1111 : }
1112 64150 : group->AddToDeferQ(entry);
1113 64150 : return true;
1114 : }
1115 578960 : return false;
1116 : }
1117 :
1118 64150 : void TaskGroup::AddToDeferQ(TaskEntry *entry) {
1119 64150 : stats_.defer_count_++;
1120 64150 : deferq_.insert(*entry);
1121 64150 : assert(entry->deferq_task_group_ == NULL);
1122 64150 : entry->deferq_task_group_ = this;
1123 64150 : }
1124 :
1125 64150 : void TaskGroup::DeleteFromDeferQ(TaskEntry &entry) {
1126 64150 : assert(this == entry.deferq_task_group_);
1127 128300 : deferq_.erase(deferq_.iterator_to(entry));
1128 64150 : entry.deferq_task_group_ = NULL;
1129 64150 : }
1130 :
1131 0 : void TaskGroup::AddToDisableQ(TaskEntry *entry) {
1132 0 : disable_entry_->AddToDeferQ(entry);
1133 0 : }
1134 :
1135 492 : void TaskGroup::PolicySet() {
1136 492 : assert(policy_set_ == false);
1137 492 : policy_set_ = true;
1138 492 : }
1139 :
1140 25019 : void TaskGroup::RunDeferQ() {
1141 : TaskDeferList::iterator it;
1142 :
1143 25019 : it = deferq_.begin();
1144 167364 : while (it != deferq_.end()) {
1145 58663 : TaskEntry &entry = *it;
1146 58663 : TaskDeferList::iterator it_work = it++;
1147 58663 : DeleteFromDeferQ(*it_work);
1148 58663 : entry.RunDeferEntry();
1149 : }
1150 :
1151 50038 : return;
1152 : }
1153 :
1154 499269 : inline void TaskGroup::TaskExited(Task *t) {
1155 499269 : run_count_--;
1156 499269 : stats_.total_tasks_completed_++;
1157 499269 : }
1158 :
1159 0 : void TaskGroup::RunDisableEntries() {
1160 0 : disable_entry_->RunDeferQForGroupEnable();
1161 0 : }
1162 :
1163 0 : void TaskGroup::AddEntriesToDisableQ() {
1164 : TaskEntry *entry;
1165 0 : 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 0 : for (TaskEntryList::iterator it = task_entry_db_.begin();
1171 0 : it != task_entry_db_.end(); ++it) {
1172 0 : if ((entry = *it) == NULL) {
1173 0 : continue;
1174 : }
1175 0 : if (entry->WaitQSize()) {
1176 0 : AddToDisableQ(entry);
1177 : }
1178 : }
1179 0 : }
1180 :
1181 943178 : bool TaskGroup::IsWaitQEmpty() {
1182 : TaskEntry *entry;
1183 :
1184 : // Check the waitq_ of the instance -1
1185 943178 : if (task_entry_->WaitQSize()) {
1186 0 : return false;
1187 : }
1188 :
1189 : // Walk thru the task_entry_db_ until waitq_ of any of the task is non-zero
1190 943178 : for (TaskEntryList::iterator it = task_entry_db_.begin();
1191 15508825 : it != task_entry_db_.end(); ++it) {
1192 14598817 : if ((entry = *it) == NULL) {
1193 11883531 : continue;
1194 : }
1195 2715286 : if (entry->IsDisabled()) {
1196 0 : continue;
1197 : }
1198 2715286 : if (entry->WaitQSize()) {
1199 33170 : return false;
1200 : }
1201 : }
1202 :
1203 : // Well, no task has been enqueued in this task group
1204 910008 : 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 1552 : TaskEntry::TaskEntry(int task_id, int task_instance) : task_code_id_(task_id),
1239 776 : task_data_id_(task_instance), run_count_(0), run_task_(NULL),
1240 776 : waitq_(), deferq_task_entry_(NULL), deferq_task_group_(NULL),
1241 1552 : 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 776 : if (task_instance != -1) {
1245 776 : policyq_.push_back(this);
1246 : }
1247 776 : memset(&stats_, 0, sizeof(stats_));
1248 : // allocate memory for deferq
1249 776 : deferq_ = new TaskDeferList;
1250 776 : }
1251 :
1252 3462 : TaskEntry::TaskEntry(int task_id) : task_code_id_(task_id),
1253 1731 : task_data_id_(-1), run_count_(0), run_task_(NULL),
1254 3462 : deferq_task_entry_(NULL), deferq_task_group_(NULL), disable_(false) {
1255 1731 : memset(&stats_, 0, sizeof(stats_));
1256 : // allocate memory for deferq
1257 1731 : deferq_ = new TaskDeferList;
1258 1731 : }
1259 :
1260 2507 : TaskEntry::~TaskEntry() {
1261 2507 : policyq_.clear();
1262 :
1263 2507 : assert(0 == deferq_->size());
1264 2507 : delete deferq_;
1265 2507 : }
1266 :
1267 1808 : void TaskEntry::AddPolicy(TaskEntry *entry) {
1268 1808 : policyq_.push_back(entry);
1269 1808 : }
1270 :
1271 578960 : TaskEntry *TaskEntry::ActiveEntryInPolicy() {
1272 1596348 : for (TaskEntryList::iterator it = policyq_.begin(); it != policyq_.end();
1273 1017388 : ++it) {
1274 1097111 : if ((*it)->run_count_ != 0) {
1275 79723 : return (*it);
1276 : }
1277 : }
1278 :
1279 499237 : return NULL;
1280 : }
1281 :
1282 578960 : bool TaskEntry::DeferOnPolicyFail(Task *task) {
1283 : TaskEntry *policy_entry;
1284 :
1285 578960 : 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 79723 : if (0 == WaitQSize()) {
1291 72961 : AddToWaitQ(task);
1292 : }
1293 79723 : policy_entry->AddToDeferQ(this);
1294 79723 : return true;
1295 : }
1296 499237 : return false;
1297 : }
1298 :
1299 134453 : void TaskEntry::AddToWaitQ(Task *t) {
1300 134453 : t->state(Task::WAIT);
1301 134453 : stats_.wait_count_++;
1302 134453 : waitq_.push_back(*t);
1303 :
1304 134453 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
1305 134453 : TaskGroup *group = scheduler->GetTaskGroup(task_code_id_);
1306 134453 : group->stats_.wait_count_++;
1307 134453 : }
1308 :
1309 134453 : bool TaskEntry::DeleteFromWaitQ(Task *t) {
1310 134453 : TaskWaitQ::iterator it = waitq_.iterator_to(*t);
1311 268906 : waitq_.erase(it);
1312 134453 : return true;
1313 : }
1314 :
1315 90791 : void TaskEntry::AddToDeferQ(TaskEntry *entry) {
1316 90791 : stats_.defer_count_++;
1317 90791 : deferq_->insert(*entry);
1318 90791 : assert(entry->deferq_task_entry_ == NULL);
1319 90791 : entry->deferq_task_entry_ = this;
1320 90791 : }
1321 :
1322 90791 : void TaskEntry::DeleteFromDeferQ(TaskEntry &entry) {
1323 90791 : assert(this == entry.deferq_task_entry_);
1324 181582 : deferq_->erase(deferq_->iterator_to(entry));
1325 90791 : entry.deferq_task_entry_ = NULL;
1326 90791 : }
1327 :
1328 499269 : void TaskEntry::RunTask(Task *t) {
1329 499269 : stats_.run_count_++;
1330 499269 : if (t->task_data_id() != -1) {
1331 267427 : assert(run_task_ == NULL);
1332 267427 : assert (run_count_ == 0);
1333 267427 : run_task_ = t;
1334 : }
1335 :
1336 499269 : run_count_++;
1337 499269 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
1338 499269 : TaskGroup *group = scheduler->QueryTaskGroup(t->task_code_id());
1339 499269 : group->TaskStarted();
1340 :
1341 499269 : t->StartTask(scheduler, group);
1342 499269 : }
1343 :
1344 134394 : void TaskEntry::RunWaitQ() {
1345 134394 : if (waitq_.size() == 0)
1346 0 : return;
1347 :
1348 134394 : TaskWaitQ::iterator it = waitq_.begin();
1349 :
1350 134394 : if (task_data_id_ != -1) {
1351 134074 : Task *t = &(*it);
1352 134074 : DeleteFromWaitQ(t);
1353 134074 : RunTask(t);
1354 : // If there are more tasks in waitq_, put them in deferq_
1355 134074 : if (waitq_.size() != 0) {
1356 11059 : AddToDeferQ(this);
1357 : }
1358 : } else {
1359 : // Run all instances in waitq_
1360 704 : while (it != waitq_.end()) {
1361 352 : Task *t = &(*it);
1362 352 : DeleteFromWaitQ(t);
1363 352 : RunTask(t);
1364 352 : if (waitq_.size() == 0)
1365 320 : break;
1366 64 : it = waitq_.begin();
1367 : }
1368 : }
1369 : }
1370 :
1371 154914 : void TaskEntry::RunDeferEntry() {
1372 154914 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
1373 154914 : TaskGroup *group = scheduler->GetTaskGroup(task_code_id_);
1374 :
1375 : // Sanity check
1376 154914 : assert(waitq_.size());
1377 154914 : Task *task = &(*waitq_.begin());
1378 :
1379 : // Check Task group policies
1380 154914 : if (group->DeferOnPolicyFail(this, task)) {
1381 13758 : return;
1382 : }
1383 :
1384 : // Check Task entry policies
1385 141156 : if (DeferOnPolicyFail(task)) {
1386 6762 : return;
1387 : }
1388 :
1389 134394 : RunWaitQ();
1390 134394 : return;
1391 : }
1392 :
1393 187934 : void TaskEntry::RunDeferQ() {
1394 : TaskDeferList::iterator it;
1395 :
1396 187934 : it = deferq_->begin();
1397 548552 : while (it != deferq_->end()) {
1398 86342 : TaskEntry &entry = *it;
1399 86342 : TaskDeferList::iterator it_work = it++;
1400 86342 : DeleteFromDeferQ(*it_work);
1401 86342 : entry.RunDeferEntry();
1402 : }
1403 :
1404 375868 : return;
1405 : }
1406 :
1407 0 : void TaskEntry::RunDeferQForGroupEnable() {
1408 : TaskDeferList::iterator it;
1409 :
1410 0 : it = deferq_->begin();
1411 0 : while (it != deferq_->end()) {
1412 0 : TaskEntry &entry = *it;
1413 0 : TaskDeferList::iterator it_work = it++;
1414 0 : DeleteFromDeferQ(*it_work);
1415 0 : if (!entry.IsDisabled()) {
1416 0 : entry.RunDeferEntry();
1417 : }
1418 : }
1419 :
1420 0 : return;
1421 : }
1422 :
1423 136827 : void TaskEntry::RunCombinedDeferQ() {
1424 136827 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
1425 136827 : TaskGroup *group = scheduler->QueryTaskGroup(task_code_id_);
1426 : TaskDeferEntryCmp defer_entry_compare;
1427 :
1428 136827 : TaskDeferList::iterator group_it = group->deferq_.begin();
1429 136827 : 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 328400 : while ((group_it != group->deferq_.end()) &&
1434 216592 : (entry_it != deferq_->end())) {
1435 9909 : TaskEntry &g_entry = *group_it;
1436 9909 : TaskEntry &t_entry = *entry_it;
1437 :
1438 9909 : if (defer_entry_compare(g_entry, t_entry)) {
1439 5466 : TaskDeferList::iterator group_it_work = group_it++;
1440 10932 : group->DeleteFromDeferQ(*group_it_work);
1441 5466 : g_entry.RunDeferEntry();
1442 : } else {
1443 4443 : TaskDeferList::iterator entry_it_work = entry_it++;
1444 4443 : DeleteFromDeferQ(*entry_it_work);
1445 4443 : t_entry.RunDeferEntry();
1446 : }
1447 : }
1448 :
1449 : // Now, walk thru the non-empty deferq_
1450 273654 : if (group_it != group->deferq_.end()) {
1451 25019 : group->RunDeferQ();
1452 223616 : } else if (entry_it != deferq_->end()) {
1453 39536 : RunDeferQ();
1454 : }
1455 136827 : }
1456 :
1457 499269 : void TaskEntry::TaskExited(Task *t, TaskGroup *group) {
1458 499269 : if (task_data_id_ != -1) {
1459 267427 : assert(run_task_ == t);
1460 267427 : run_task_ = NULL;
1461 267427 : assert(run_count_ == 1);
1462 : }
1463 :
1464 499269 : run_count_--;
1465 499269 : stats_.total_tasks_completed_++;
1466 499269 : stats_.last_exit_time_ = UTCTimestampUsec();
1467 499269 : group->TaskExited(t);
1468 :
1469 499269 : if (!group->run_count_ && !run_count_) {
1470 136827 : RunCombinedDeferQ();
1471 362442 : } else if (!group->run_count_) {
1472 0 : group->RunDeferQ();
1473 362442 : } else if (!run_count_) {
1474 147932 : RunDeferQ();
1475 : }
1476 499269 : }
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 326126 : boost::optional<uint64_t> TaskEntry::GetTaskDeferEntrySeqno() const {
1493 326126 : if(waitq_.size()) {
1494 326126 : const Task *task = &(*waitq_.begin());
1495 326126 : return task->seqno();
1496 : }
1497 :
1498 0 : return boost::none;
1499 : }
1500 :
1501 : ////////////////////////////////////////////////////////////////////////////
1502 : // Implementation for class Task
1503 : ////////////////////////////////////////////////////////////////////////////
1504 492153 : Task::Task(int task_id, int task_instance) : task_code_id_(task_id),
1505 984300 : task_data_id_(task_instance), task_impl_(), state_(INIT),
1506 492147 : tbb_state_(TBB_INIT), seqno_(0), task_recycle_(false), task_cancel_(false),
1507 492153 : enqueue_time_(0), schedule_time_(0), execute_delay_(0), schedule_delay_(0) {
1508 492163 : }
1509 :
1510 416 : Task::Task(int task_id) : task_code_id_(task_id),
1511 832 : task_data_id_(-1), task_impl_(), state_(INIT), tbb_state_(TBB_INIT),
1512 416 : seqno_(0), task_recycle_(false), task_cancel_(false), enqueue_time_(0),
1513 416 : schedule_time_(0), execute_delay_(0), schedule_delay_(0) {
1514 416 : }
1515 :
1516 :
1517 499269 : void Task::StartTask(TaskScheduler *scheduler, TaskGroup *group) {
1518 499269 : 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 499269 : state(RUN);
1527 499269 : tbb_state(TBB_ENQUEUED);
1528 499269 : task_impl_ = group->tbb_group().defer(TaskFunctor(this));
1529 499269 : scheduler->tbb_arena().enqueue(std::move(task_impl_));
1530 499269 : }
1531 :
1532 289090 : Task *Task::Running() {
1533 289090 : TaskInfo::reference running = task_running.local();
1534 289056 : 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 : }
|
