task.h

Go to the documentation of this file.

/*
 * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
 */
 
#ifndef ctrlplane_task_h
#define ctrlplane_task_h
 
#include <boost/scoped_ptr.hpp>
#include <boost/intrusive/list.hpp>
#include <map>
#include <shared_mutex>
#include <vector>
#include <mutex>
#define TBB_SUPPRESS_DEPRECATED_MESSAGES 1
#include <oneapi/tbb/task.h>
#include <oneapi/tbb/task_arena.h>
#include <oneapi/tbb/task_group.h>
#include <oneapi/tbb/global_control.h>
 
#include "base/util.h"
 
class TaskGroup;
class TaskEntry;
class SandeshTaskScheduler;
class TaskTbbKeepAwake;
class EventManager;
class TaskMonitor;
class TaskScheduler;
 
class Task {
public:
 
    enum State {
 
        INIT,
 
        WAIT,
 
        RUN
    };
 
    enum TbbState {
        TBB_INIT,
        TBB_ENQUEUED,
        TBB_EXEC,
        TBB_DONE
    };
 
    const static int kTaskInstanceAny = -1;
 
    Task(int task_id, int task_data_id);
 
 
    Task(int task_id);
 
    virtual ~Task() { };
 
    virtual bool Run() = 0;
 
    virtual void OnTaskCancel() { };
 
    // Accessor methods
 
    State state() const { return state_; };
 
    int task_code_id() const { return task_code_id_; };
 
    int task_data_id() const { return task_data_id_; };
 
    uint64_t seqno() const { return seqno_; };
 
    friend std::ostream& operator<<(std::ostream& out, const Task &task);
 
    static Task *Running();
 
    bool task_cancelled() const { return task_cancel_; };
 
    virtual std::string Description() const  = 0;
 
    uint64_t enqueue_time() const { return enqueue_time_; }
    
    uint64_t schedule_time() const { return schedule_time_; }
 
    uint32_t execute_delay() const { return execute_delay_; }
 
    uint32_t schedule_delay() const { return schedule_delay_; }
 
private:
 
    friend class TaskEntry;
 
    friend class TaskScheduler;
 
    friend class TaskFunctor;
 
    void seqno(uint64_t seqno) {seqno_ = seqno;};
 
    void tbb_state(TbbState s) { tbb_state_ = s; };
 
    void state(State s) { state_ = s; };
 
    void set_task_recycle() { task_recycle_ = true; };
 
    void set_task_complete() { task_recycle_ = false; };
 
    void StartTask(TaskScheduler *scheduler, TaskGroup *group);
 
    int                 task_code_id_;
 
    int                 task_data_id_;
 
    oneapi::tbb::task_handle   task_impl_;
 
    State               state_;
 
    TbbState            tbb_state_;
 
    uint64_t            seqno_;
 
    bool                task_recycle_;
 
    bool                task_cancel_;
 
    uint64_t            enqueue_time_;
 
    uint64_t            schedule_time_;
 
    uint32_t            execute_delay_;
 
    uint32_t            schedule_delay_;
 
    // Hook in intrusive list for TaskEntry::waitq_
    boost::intrusive::list_member_hook<> waitq_hook_;
 
    DISALLOW_COPY_AND_ASSIGN(Task);
};
 
struct TaskExclusion {
 
    TaskExclusion(int task_code_id)
        : match_code_id(task_code_id), match_data_id(-1) {}
 
    TaskExclusion(int task_code_id, int task_data_id)
        : match_code_id(task_code_id), match_data_id(task_data_id) {
    }
 
    int match_code_id;
 
    int match_data_id;
};
 
typedef std::vector<TaskExclusion> TaskPolicy;
 
struct TaskStats {
 
    int wait_count_;
 
    int run_count_;
 
    int defer_count_;
 
    uint64_t enqueue_count_;
 
    uint64_t total_tasks_completed_;
 
    uint64_t last_exit_time_;
};
 
class TaskScheduler {
public:
    typedef boost::function<void(const char *file_name, uint32_t line_no,
                                 const Task *task, const char *description,
                                 uint64_t delay)> LogFn;
 
    TaskScheduler(int thread_count = 0);
 
    ~TaskScheduler();
 
    static void Initialize(uint32_t thread_count = 0, EventManager *evm = NULL);
    static TaskScheduler *GetInstance();
 
    void Enqueue(Task *task);
 
    void EnqueueUnLocked(Task *task);
 
    enum CancelReturnCode {
        CANCELLED,
        FAILED,
        QUEUED,
    };
 
    CancelReturnCode Cancel(Task *task);
 
    void SetPolicy(int task_id, TaskPolicy &policy);
 
    bool GetRunStatus() { return running_; };
    int GetTaskId(const std::string &name);
    std::string GetTaskName(int task_id) const;
 
    TaskStats *GetTaskGroupStats(int task_id);
    TaskStats *GetTaskStats(int task_id);
    TaskStats *GetTaskStats(int task_id, int instance_id);
    void ClearTaskGroupStats(int task_id);
    void ClearTaskStats(int task_id);
    void ClearTaskStats(int task_id, int instance_id);
 
    TaskGroup *GetTaskGroup(int task_id);
 
    TaskGroup *QueryTaskGroup(int task_id);
 
    bool IsTaskGroupEmpty(int task_id) const;
 
    TaskEntry *GetTaskEntry(int task_id, int instance_id);
 
    TaskEntry *QueryTaskEntry(int task_id, int instance_id);
 
    void OnTaskExit(Task *task);
 
    void Stop();
 
    void Start();
 
    void Print();
 
    bool IsEmpty(bool running_only = false);
 
    void Terminate();
 
    int HardwareThreadCount() { return hw_thread_count_; }
 
    static int GetThreadCount(int thread_count = 0);
    static bool ShouldUseSpawn();
 
    static int GetDefaultThreadCount();
 
    uint64_t enqueue_count() const { return enqueue_count_; }
    uint64_t done_count() const { return done_count_; }
    uint64_t cancel_count() const { return cancel_count_; }
    
    void SetMaxThreadCount(int n);
    void GetSandeshData(SandeshTaskScheduler *resp, bool summary);
    void Log(const char *file_name, uint32_t line_no, const Task *task,
             const char *description, uint64_t delay);
    void RegisterLog(LogFn fn);
 
    void SetTrackRunTime(bool value) { track_run_time_ = value; }
    bool track_run_time() const { return track_run_time_; }
 
    void EnableLatencyThresholds(uint32_t execute, uint32_t schedule);
    uint32_t schedule_delay() const { return schedule_delay_; }
    uint32_t execute_delay() const { return execute_delay_; }
 
    bool measure_delay() const { return measure_delay_; }
    void SetLatencyThreshold(const std::string &name, uint32_t execute,
                             uint32_t schedule);
    uint32_t schedule_delay(Task *task) const;
    uint32_t execute_delay(Task *task) const;
    void set_event_manager(EventManager *evm);
 
    void DisableTaskGroup(int task_id);
    void EnableTaskGroup(int task_id);
    void DisableTaskEntry(int task_id, int instance_id);
    void EnableTaskEntry(int task_id, int instance_id);
 
    void ModifyTbbKeepAwakeTimeout(uint32_t timeout);
 
    void EnableMonitor(EventManager *evm, uint64_t tbb_keepawake_time_msec,
                       uint64_t inactivity_time_msec,
                       uint64_t poll_interval_msec);
    const TaskMonitor *task_monitor() const { return task_monitor_; }
    const TaskTbbKeepAwake *tbb_awake_task() const { return tbb_awake_task_; }
    bool use_spawn() const { return use_spawn_; }
 
    static void SetThreadAmpFactor(int n);
 
    oneapi::tbb::task_arena &tbb_arena() {
    return task_scheduler_;
    }
 
private:
    friend class ConcurrencyScope;
    typedef std::vector<TaskGroup *> TaskGroupDb;
    typedef std::map<std::string, int> TaskIdMap;
 
    static const int        kVectorGrowSize = 16;
    static boost::scoped_ptr<TaskScheduler> singleton_;
 
    // XXX
    // Following two methods are only for Unit Testing to control
    // current running task. Usage of this method would result in
    // unexpected behavior.
 
    void SetRunningTask(Task *);
    void ClearRunningTask();
 
    bool                    use_spawn_;
    TaskEntry               *stop_entry_;
 
    oneapi::tbb::global_control tbb_global_control_;
    oneapi::tbb::task_arena task_scheduler_;
    mutable std::mutex      mutex_;
    bool                    running_;
    uint64_t                seqno_;
    TaskGroupDb             task_group_db_;
 
    std::shared_mutex       id_map_mutex_;
    TaskIdMap               id_map_;
    int                     id_max_;
 
    LogFn                   log_fn_;
    int                     hw_thread_count_;
 
    bool                    track_run_time_;
    bool                    measure_delay_;
 
    uint32_t                schedule_delay_;
 
    uint32_t                execute_delay_;
 
    uint64_t                enqueue_count_;
    uint64_t                done_count_;
    uint64_t                cancel_count_;
    EventManager            *evm_;
 
    static int ThreadAmpFactor_;
 
    TaskTbbKeepAwake *tbb_awake_task_;
    TaskMonitor      *task_monitor_;
    DISALLOW_COPY_AND_ASSIGN(TaskScheduler);
};
 
#endif