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