Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : */
4 :
5 : #ifndef ctrlplane_task_h
6 : #define ctrlplane_task_h
7 :
8 : #include <boost/scoped_ptr.hpp>
9 : #include <boost/intrusive/list.hpp>
10 : #include <map>
11 : #include <shared_mutex>
12 : #include <vector>
13 : #include <mutex>
14 : #define TBB_SUPPRESS_DEPRECATED_MESSAGES 1
15 : #include <oneapi/tbb/task.h>
16 : #include <oneapi/tbb/task_arena.h>
17 : #include <oneapi/tbb/task_group.h>
18 : #include <oneapi/tbb/global_control.h>
19 :
20 : #include "base/util.h"
21 :
22 : class TaskGroup;
23 : class TaskEntry;
24 : class SandeshTaskScheduler;
25 : class TaskTbbKeepAwake;
26 : class EventManager;
27 : class TaskMonitor;
28 : class TaskScheduler;
29 :
30 : /// @brief Task is a class to describe a computational task within OpenSDN
31 : /// control plane applications. A task is a labelled sequence of instructions
32 : /// (code) and data processed by them in a single thread. OpenSDN Task wraps
33 : /// over tbb::task. Tasks are labelled using a pair of numbers:
34 : /// - task code ID (or task ID), which corresponds to a version of code
35 : /// to run into the task;
36 : /// - task data ID, which corresponds to a version of data supplied
37 : /// to the task and processed by code.
38 : ///
39 : /// This labelling is used to apply execution policies determining
40 : /// which tasks are allowed to be executed in parallel with others and
41 : /// which tasks are forbidden to run in parallel. The labels are expressed as
42 : /// <tcid, tdid>, where *tcid* is a task code ID and *tdid* is a task
43 : /// data ID.
44 : ///
45 : /// If a task with *tcid* has *tdid* equal to -1, then any number of
46 : /// tasks with label <tcid,-1> can run at a time. If a task has task data ID
47 : /// larger or equal to 0, then only one task with this
48 : /// given label (i.e. <tcid,tdid>) can run at a time.
49 : ///
50 : /// When there are multiple tasks ready to run, they are scheduled in their
51 : /// order of enqueue.
52 : ///
53 : /// Additionaly, parallel execution of tasks can be managed using task
54 : /// execution policies.
55 : /// Task execution policies are specified per a task with the given code
56 : /// ID *tcid0* and arbitrary task data ID in a form of a list of task
57 : /// labels:
58 : /// <tcid0,-1> => <tcid1,tdid1>, <tcid2,tdid2>, ..., <tcidN,tdidN>.
59 : ///
60 : /// This list specifies which tasks can't be executed in parallel with
61 : /// a task having specified task code ID (tcid0).
62 : /// Each label <tcidN, tdidN> in a policy (i.e. the list) is called task
63 : /// exclusion because it specifies that the task with this task code ID and
64 : /// task data ID cannot run in parallel with <tcid0, -1>.
65 : /// When *tdid* is equal to -1 in a task exclusion, it corresponds to
66 : /// wildcard (*), i.e. all possible values of the task data ID *tdid*.
67 : ///
68 : /// For example, if we have a policy:
69 : /// - <tcid0,-1> => <tcid1, -1> <tcid2, 2> <tcid3, 3>
70 : ///
71 : /// The policy states that:
72 : /// - Task <tcid0,-1> cannot run as long as <tcid1, -1> is running;
73 : /// - Task <tcid0, 2> cannot run as long as task <tcid2, 2> is running;
74 : /// - Task <tcid0, 3> cannot run as long as task <tcid3, 3> is running.
75 : ///.
76 : /// Policy rules are symmetric. I.e., the previous example states also:
77 : /// - Task <tcid1,-1> cannot run as long as <tcid0,-1> is running;
78 : /// - Task <tcid2, 2> cannot run as long as task <tcid0, 2> is running;
79 : /// - Task <tcid3, 3> cannot run as long as task <tcid0, 3> is running.
80 : ///
81 : class Task {
82 : public:
83 :
84 : /// @brief Task states.
85 : enum State {
86 :
87 : /// @brief A task was initialized.
88 : INIT,
89 :
90 : /// @brief A task is waiting in a queue.
91 : WAIT,
92 :
93 : /// @brief A task is being run.
94 : RUN
95 : };
96 :
97 : /// @brief Describes states of a task according to TBB library.
98 : enum TbbState {
99 : TBB_INIT,
100 : TBB_ENQUEUED,
101 : TBB_EXEC,
102 : TBB_DONE
103 : };
104 :
105 : /// @brief Specifies value for wildcard (any or *) task data ID.
106 : const static int kTaskInstanceAny = -1;
107 :
108 : /// @brief Creates a new task with the given values of
109 : /// task code ID and task data ID.
110 : Task(int task_id, int task_data_id);
111 :
112 :
113 : /// @brief Creates a new task with the given value of
114 : /// task code ID and wildcard for task data ID.
115 : Task(int task_id);
116 :
117 : /// @brief Destroys a task
118 5886566 : virtual ~Task() { };
119 :
120 : /// @brief Code to execute in a task.
121 : /// Returns true if task is completed. Return false to reschedule the task.
122 : virtual bool Run() = 0;
123 :
124 : /// @brief Called on task exit, if it is marked for cancellation.
125 : /// If the user wants to do any cleanup on task cancellation,
126 : /// then he/she can overload this function.
127 2 : virtual void OnTaskCancel() { };
128 :
129 : // Accessor methods
130 :
131 : /// @brief Returns a state value of a task.
132 : State state() const { return state_; };
133 :
134 : /// @brief Returns the code ID of this task.
135 84455076 : int task_code_id() const { return task_code_id_; };
136 :
137 : /// @brief Returns the data ID of this task.
138 21705029 : int task_data_id() const { return task_data_id_; };
139 :
140 : /// @brief Returns the sequence number of this task.
141 36513693 : uint64_t seqno() const { return seqno_; };
142 :
143 : /// @brief Provides access to private members of a task for the
144 : /// output stream redirection operator.
145 : friend std::ostream& operator<<(std::ostream& out, const Task &task);
146 :
147 : /// @brief Returns a pointer to the current task the code is executing
148 : /// under.
149 : static Task *Running();
150 :
151 : /// @brief Returns true if the task has been canceled.
152 64632 : bool task_cancelled() const { return task_cancel_; };
153 :
154 : /// @brief Gives a description of the task.
155 : virtual std::string Description() const = 0;
156 :
157 : /// @brief Returns the time when the task was enqueued for execution.
158 7230891 : uint64_t enqueue_time() const { return enqueue_time_; }
159 :
160 : /// @brief Returns the time when the task execution was started.
161 : uint64_t schedule_time() const { return schedule_time_; }
162 :
163 : /// @brief Returns the threshold for the task execution duration.
164 0 : uint32_t execute_delay() const { return execute_delay_; }
165 :
166 : /// @brief Returns the time threshold for time difference between
167 : /// moments when the task was started and when it was enqueue.
168 0 : uint32_t schedule_delay() const { return schedule_delay_; }
169 :
170 : private:
171 :
172 : /// @brief Gives access to private members for TaskEntry class.
173 : friend class TaskEntry;
174 :
175 : /// @brief Gives access to private members for TaskScheduler class.
176 : friend class TaskScheduler;
177 :
178 : /// @brief Gives access to private members for TaskImpl class.
179 : friend class TaskFunctor;
180 :
181 : /// @brief Sets sequence number of the task
182 8650268 : void seqno(uint64_t seqno) {seqno_ = seqno;};
183 :
184 : /// @brief Sets a TBB state for the task
185 23109694 : void tbb_state(TbbState s) { tbb_state_ = s; };
186 :
187 : /// @brief Sets a state for this task
188 14595457 : void state(State s) { state_ = s; };
189 :
190 : /// @brief Marks this task for recycle
191 1413833 : void set_task_recycle() { task_recycle_ = true; };
192 :
193 : /// @brief Marks this task as completed (forbids recycling)
194 5817873 : void set_task_complete() { task_recycle_ = false; };
195 :
196 : /// @brief Starts execution of a task.
197 : void StartTask(TaskScheduler *scheduler, TaskGroup *group);
198 :
199 : /// @brief The code path executed by the task.
200 : int task_code_id_;
201 :
202 : /// @brief The dataset id within a code path.
203 : int task_data_id_;
204 :
205 : /// @brief A handle to a oneTBB object storing
206 : /// low-level information about the managed task.
207 : oneapi::tbb::task_handle task_impl_;
208 :
209 : /// @brief Stores a state of the task.
210 : State state_;
211 :
212 : /// @brief Stores a state of the TBB object.
213 : TbbState tbb_state_;
214 :
215 : /// @brief Stores the sequence number.
216 : uint64_t seqno_;
217 :
218 : /// @brief Determines if the task must be rescheduled (reused)
219 : /// after its completion.
220 : bool task_recycle_;
221 :
222 : /// @brief Determines if the task's execution was canceled.
223 : bool task_cancel_;
224 :
225 : /// @brief Contains the time when the task was enqueued
226 : /// for execution
227 : uint64_t enqueue_time_;
228 :
229 : /// @brief Contains the time when the task was started.
230 : uint64_t schedule_time_;
231 :
232 : /// @brief Sets threshold for the task's execution time.
233 : /// If the threshold is exceeded, the event is logged.
234 : uint32_t execute_delay_;
235 :
236 : /// @brief Sets threshold for delay between enqueueing and execution.
237 : /// If the threshold is exceeded, the event is logged.
238 : uint32_t schedule_delay_;
239 :
240 : // Hook in intrusive list for TaskEntry::waitq_
241 : boost::intrusive::list_member_hook<> waitq_hook_;
242 :
243 : DISALLOW_COPY_AND_ASSIGN(Task);
244 : };
245 :
246 : /// @brief The class is used to specify a Task label for formulating
247 : /// a task exclusion list (an execution policy).
248 : struct TaskExclusion {
249 :
250 : /// @brief Creates a new task exclusion from the given task code ID value
251 : /// and wildcard for task data ID.
252 22213 : TaskExclusion(int task_code_id)
253 22213 : : match_code_id(task_code_id), match_data_id(-1) {}
254 :
255 : /// @brief Creates a new task exclusion from the given task code ID and
256 : /// task data ID values.
257 1660 : TaskExclusion(int task_code_id, int task_data_id)
258 1660 : : match_code_id(task_code_id), match_data_id(task_data_id) {
259 1660 : }
260 :
261 : /// @brief Specifies task code ID (must be a valid id >= 0)
262 : /// for a task execution policy.
263 : int match_code_id;
264 :
265 : /// @brief Specifies task data ID for a task execution policy.
266 : /// The value of -1 corresponds to wildcard (any).
267 : int match_data_id;
268 : };
269 :
270 : /// @brief Defines a type to store an execution policy (a list of
271 : /// task exclusions).
272 : typedef std::vector<TaskExclusion> TaskPolicy;
273 :
274 : /// The class is used to store various statistics associated with a
275 : /// task or group of tasks
276 : struct TaskStats {
277 :
278 : /// @brief Number of entries in waitq
279 : int wait_count_;
280 :
281 : /// @brief Number of entries currently running
282 : int run_count_;
283 :
284 : /// @brief Number of entries in deferq
285 : int defer_count_;
286 :
287 : /// @brief Number of tasks enqueued
288 : uint64_t enqueue_count_;
289 :
290 : /// @brief Number of total tasks ran
291 : uint64_t total_tasks_completed_;
292 :
293 : ///@brief Number of time stamp of latest exist
294 : uint64_t last_exit_time_;
295 : };
296 :
297 : /// @brief The TaskScheduler keeps track of what tasks are currently
298 : /// schedulable.
299 : /// When a task is enqueued it is added to the run queue or the pending queue
300 : /// depending as to whether there is a runable or pending task ahead of it
301 : /// that violates the mutual exclusion policies.
302 : /// When tasks exit the scheduler re-examines the tasks on the pending queue
303 : /// which may now be runnable. It is important that this process is efficient
304 : /// such that exit events do not scan tasks that are not waiting on a
305 : /// particular task id or task instance to have a 0 count.
306 : class TaskScheduler {
307 : public:
308 : typedef boost::function<void(const char *file_name, uint32_t line_no,
309 : const Task *task, const char *description,
310 : uint64_t delay)> LogFn;
311 :
312 : /// @brief TaskScheduler constructor.
313 : /// TBB assumes it can use the "thread" invoking tbb::scheduler can be used
314 : /// for task scheduling. But, in our case we dont want "main" thread to be
315 : /// part of tbb. So, initialize TBB with one thread more than its default.
316 : TaskScheduler(int thread_count = 0);
317 :
318 : /// @brief Frees up the task_entry_db_ allocated for scheduler.
319 : ~TaskScheduler();
320 :
321 : static void Initialize(uint32_t thread_count = 0, EventManager *evm = NULL);
322 : static TaskScheduler *GetInstance();
323 :
324 : /// @brief Enqueues a task for running. Starts task if all policy rules
325 : /// are met else puts task in waitq. Enqueueing may may result in the
326 : /// task being immedietly
327 : /// added to the run queue or to a pending queue. Tasks may not be added
328 : /// to the run queue in violation of their exclusion policy.
329 : void Enqueue(Task *task);
330 :
331 : void EnqueueUnLocked(Task *task);
332 :
333 : enum CancelReturnCode {
334 : CANCELLED,
335 : FAILED,
336 : QUEUED,
337 : };
338 :
339 : /// @brief Cancels a Task that can be in RUN/WAIT state.
340 : /// The caller needs to ensure that the task exists when Cancel()
341 : /// is invoked.
342 : CancelReturnCode Cancel(Task *task);
343 :
344 : /// @brief Sets the task exclusion policy.
345 : /// Adds policy entries for the task
346 : /// Examples:
347 : /// - Policy <tid0> => <tid1, -1> <tid2, inst2> will result in following:
348 : /// - task_db_[tid0] : Rule <tid1, -1> is added to policyq
349 : /// - task_group_db_[tid0, inst2] : Rule <tid2, inst2> is added to policyq
350 : /// - The symmetry of policy will result in following additional rules,
351 : /// - task_db_[tid1] : Rule <tid0, -1> is added to policyq
352 : /// - task_group_db_[tid2, inst2] : Rule <tid0, inst2> is added to policyq
353 : void SetPolicy(int task_id, TaskPolicy &policy);
354 :
355 288 : bool GetRunStatus() { return running_; };
356 : int GetTaskId(const std::string &name);
357 : std::string GetTaskName(int task_id) const;
358 :
359 : TaskStats *GetTaskGroupStats(int task_id);
360 : TaskStats *GetTaskStats(int task_id);
361 : TaskStats *GetTaskStats(int task_id, int instance_id);
362 : void ClearTaskGroupStats(int task_id);
363 : void ClearTaskStats(int task_id);
364 : void ClearTaskStats(int task_id, int instance_id);
365 :
366 : /// @brief Get TaskGroup for a task_id. Grows task_entry_db_ if necessary
367 : TaskGroup *GetTaskGroup(int task_id);
368 :
369 : /// @brief Query TaskGroup for a task_id.Assumes valid entry is present for
370 : /// task_id
371 : TaskGroup *QueryTaskGroup(int task_id);
372 :
373 : /// @brief Check if there are any Tasks in the given TaskGroup.
374 : /// Assumes that all task ids are mutually exclusive with bgp::Config.
375 : bool IsTaskGroupEmpty(int task_id) const;
376 :
377 : /// @brief Get TaskGroup for a task_id. Grows task_entry_db_ if necessary
378 : TaskEntry *GetTaskEntry(int task_id, int instance_id);
379 :
380 : /// @brief Query TaskEntry for a task-id and task-instance
381 : TaskEntry *QueryTaskEntry(int task_id, int instance_id);
382 :
383 : /// @brief Method invoked on exit of a Task.
384 : /// Exit of a task can potentially start tasks in pendingq.
385 : void OnTaskExit(Task *task);
386 :
387 : /// @brief Stops scheduling of all tasks
388 : void Stop();
389 :
390 : /// @brief Starts scheduling of all tasks
391 : void Start();
392 :
393 : /// @brief Debug print routine
394 : void Print();
395 :
396 : /// @brief Returns true if there are no tasks running and/or enqueued
397 : /// If running_only is true, enqueued tasks are ignored i.e. return true if
398 : /// there are no running tasks. Ignore TaskGroup or TaskEntry if it is
399 : /// disabled.
400 : bool IsEmpty(bool running_only = false);
401 :
402 : void Terminate();
403 :
404 393013 : int HardwareThreadCount() { return hw_thread_count_; }
405 :
406 : /// @brief Get number of tbb worker threads.
407 : /// For testing purposes only. Limit the number of tbb worker threads.
408 : static int GetThreadCount(int thread_count = 0);
409 : static bool ShouldUseSpawn();
410 :
411 : static int GetDefaultThreadCount();
412 :
413 0 : uint64_t enqueue_count() const { return enqueue_count_; }
414 0 : uint64_t done_count() const { return done_count_; }
415 : uint64_t cancel_count() const { return cancel_count_; }
416 :
417 : /// @brief Force number of threads
418 : void SetMaxThreadCount(int n);
419 : void GetSandeshData(SandeshTaskScheduler *resp, bool summary);
420 : void Log(const char *file_name, uint32_t line_no, const Task *task,
421 : const char *description, uint64_t delay);
422 : void RegisterLog(LogFn fn);
423 :
424 : void SetTrackRunTime(bool value) { track_run_time_ = value; }
425 7230722 : bool track_run_time() const { return track_run_time_; }
426 :
427 : /// @brief Enable logging of tasks exceeding configured latency
428 : void EnableLatencyThresholds(uint32_t execute, uint32_t schedule);
429 0 : uint32_t schedule_delay() const { return schedule_delay_; }
430 0 : uint32_t execute_delay() const { return execute_delay_; }
431 :
432 : bool measure_delay() const { return measure_delay_; }
433 : void SetLatencyThreshold(const std::string &name, uint32_t execute,
434 : uint32_t schedule);
435 : uint32_t schedule_delay(Task *task) const;
436 : uint32_t execute_delay(Task *task) const;
437 : void set_event_manager(EventManager *evm);
438 :
439 : void DisableTaskGroup(int task_id);
440 : void EnableTaskGroup(int task_id);
441 : void DisableTaskEntry(int task_id, int instance_id);
442 : void EnableTaskEntry(int task_id, int instance_id);
443 :
444 : void ModifyTbbKeepAwakeTimeout(uint32_t timeout);
445 :
446 : /// @brief Enable Task monitoring
447 : void EnableMonitor(EventManager *evm, uint64_t tbb_keepawake_time_msec,
448 : uint64_t inactivity_time_msec,
449 : uint64_t poll_interval_msec);
450 0 : const TaskMonitor *task_monitor() const { return task_monitor_; }
451 : const TaskTbbKeepAwake *tbb_awake_task() const { return tbb_awake_task_; }
452 : bool use_spawn() const { return use_spawn_; }
453 :
454 : /// @brief following function allows one to increase max num of threads used by
455 : /// TBB
456 : static void SetThreadAmpFactor(int n);
457 :
458 : /// @brief returns current TBB arena.
459 7234883 : oneapi::tbb::task_arena &tbb_arena() {
460 7234883 : return task_scheduler_;
461 : }
462 :
463 : private:
464 : friend class ConcurrencyScope;
465 : typedef std::vector<TaskGroup *> TaskGroupDb;
466 : typedef std::map<std::string, int> TaskIdMap;
467 :
468 : static const int kVectorGrowSize = 16;
469 : static boost::scoped_ptr<TaskScheduler> singleton_;
470 :
471 : // XXX
472 : // Following two methods are only for Unit Testing to control
473 : // current running task. Usage of this method would result in
474 : // unexpected behavior.
475 :
476 : /// @brief This function should not be called in production code.
477 : /// It is only for unit testing to control current running task
478 : /// This function modifies the running task as specified by the input
479 : void SetRunningTask(Task *);
480 : void ClearRunningTask();
481 :
482 : /// @brief Use spawn() to run a tbb::task instead of enqueue()
483 : bool use_spawn_;
484 : TaskEntry *stop_entry_;
485 :
486 : oneapi::tbb::global_control tbb_global_control_;
487 : oneapi::tbb::task_arena task_scheduler_;
488 : mutable std::mutex mutex_;
489 : bool running_;
490 : uint64_t seqno_;
491 : TaskGroupDb task_group_db_;
492 :
493 : std::shared_mutex id_map_mutex_;
494 : TaskIdMap id_map_;
495 : int id_max_;
496 :
497 : LogFn log_fn_;
498 : int hw_thread_count_;
499 :
500 : bool track_run_time_;
501 : bool measure_delay_;
502 :
503 : /// @brief Log if time between enqueue and task-execute exceeds the delay
504 : uint32_t schedule_delay_;
505 :
506 : /// @brief Log if time taken to execute exceeds the delay
507 : uint32_t execute_delay_;
508 :
509 : uint64_t enqueue_count_;
510 : uint64_t done_count_;
511 : uint64_t cancel_count_;
512 : EventManager *evm_;
513 :
514 : /// @brief following variable allows one to increase max num of threads used by
515 : /// TBB
516 : static int ThreadAmpFactor_;
517 :
518 : TaskTbbKeepAwake *tbb_awake_task_;
519 : TaskMonitor *task_monitor_;
520 : DISALLOW_COPY_AND_ASSIGN(TaskScheduler);
521 : };
522 :
523 : #endif
|
