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