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 <vector>
#include <tbb/mutex.h>
#include <tbb/reader_writer_lock.h>
#include <tbb/task.h>
#include <tbb/task_scheduler_init.h>
#include "base/util.h"

class TaskGroup;
class TaskEntry;
class SandeshTaskScheduler;
class TaskTbbKeepAwake;
class EventManager;
class TaskMonitor;
class TaskScheduler;

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_;
};

struct TaskExclusion {
    TaskExclusion(int task_id) : match_id(task_id), match_instance(-1) {}
    TaskExclusion(int task_id, int instance_id)
        : match_id(task_id), match_instance(instance_id) {
    }

    int match_id;

    int match_instance;
};
typedef std::vector<TaskExclusion> TaskPolicy;

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_instance);
    Task(int task_id);
    virtual ~Task() { };

    virtual bool Run() = 0;

    virtual void OnTaskCancel() { };

    // Accessor methods
    State GetState() const { return state_; };
    int GetTaskId() const { return task_id_; };
    int GetTaskInstance() const { return task_instance_; };
    uint64_t GetSeqno() 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 TaskImpl;
    void SetSeqNo(uint64_t seqno) {seqno_ = seqno;};
    void SetTbbState(TbbState s) { tbb_state_ = s; };
    void SetState(State s) { state_ = s; };
    void SetTaskRecycle() { task_recycle_ = true; };
    void SetTaskComplete() { task_recycle_ = false; };

    void StartTask(TaskScheduler *scheduler);

    int                 task_id_;

    int                 task_instance_;
    tbb::task           *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);
};

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);

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();
    void WaitForTerminateCompletion();

    int CountThreadsPerPid(pid_t pid);

    bool                    use_spawn_;
    TaskEntry               *stop_entry_;

    tbb::task_scheduler_init task_scheduler_;
    mutable tbb::mutex      mutex_;
    bool                    running_;
    uint64_t                seqno_;
    TaskGroupDb             task_group_db_;

    tbb::reader_writer_lock 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