TaskScheduler.hpp

Go to the documentation of this file.

 
#ifndef HORIZON_CORE_UTILITY_TASKSCHEDULER_HPP
#define HORIZON_CORE_UTILITY_TASKSCHEDULER_HPP
 
#include <chrono>
#include <cstdint>
#include <functional>
#include <memory>
#include <set>
#include <queue>
#include <mutex>
#include <random>
 
typedef std::chrono::microseconds Microseconds;
typedef std::chrono::milliseconds Milliseconds;
 
typedef std::chrono::seconds Seconds;
 
typedef std::chrono::minutes Minutes;
 
typedef std::chrono::hours Hours;
 
class TaskContext;
 
class TaskScheduler
{
    friend class TaskContext;
 
    // Time definitions (use steady clock)
    typedef std::chrono::system_clock clock_t;
    typedef clock_t::time_point timepoint_t;
 
    // Duration calculator
    typedef std::function<clock_t::duration()> duration_calculator_t;
 
    // Static time
    template<typename _Rep, typename _Period>
    static duration_calculator_t MakeDurationCalculator(std::chrono::duration<_Rep, _Period> const& duration)
    {
        return [duration]
        {
            return duration;
        };
    }
 
    // Random time between min and max
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    static duration_calculator_t MakeDurationCalculator(
                                                        std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                                                        std::chrono::duration<_RepRight, _PeriodRight> const& max)
    {
        return std::bind(RandomDurationBetween<_RepLeft, _PeriodLeft, _RepRight, _PeriodRight>, min, max);
    }
 
    // Task group type
    typedef uint64_t group_t;
    // Task repeated type
    typedef unsigned int repeated_t;
 
    // Task handle type
    typedef std::function<void(TaskContext)> task_handler_t;
    // Predicate type
    typedef std::function<bool()> predicate_t;
    // Success handle type
    typedef std::function<void()> success_t;
 
    // C++11 std::make_unique workarround
    template<typename T, typename... Args>
    static std::unique_ptr<T> MakeUnique(Args&&... args)
    {
        return std::unique_ptr<T>(new T(std::forward<Args>(args)...));
    }
 
    class Task
    {
        friend class TaskContext;
        friend class TaskScheduler;
 
        timepoint_t _end;
        duration_calculator_t _duration_calculator;
        std::unique_ptr<group_t> _group;
        repeated_t _repeated;
        task_handler_t _task;
 
    public:
        // All Argument construct
        Task(timepoint_t const& end, duration_calculator_t&& duration_calculator,
             group_t const group,
             repeated_t const repeated, task_handler_t const& task)
        : _end(end), _duration_calculator(std::move(duration_calculator)),
        _group(TaskScheduler::MakeUnique<group_t>(group)),
        _repeated(repeated), _task(task) { }
 
        // Minimal Argument construct
        Task(timepoint_t const& end, duration_calculator_t&& duration_calculator,
             task_handler_t const& task)
        : _end(end), _duration_calculator(std::move(duration_calculator)),
        _group(nullptr), _repeated(0), _task(task) { }
 
        // Copy construct
        Task(Task const&) = delete;
        // Move construct
        Task(Task&&) = delete;
        // Copy Assign
        Task& operator= (Task const&) = delete;
        // Move Assign
        Task& operator= (Task&& right) = delete;
 
        // Order tasks by its end
        inline bool operator< (Task const& other) const
        {
            return _end < other._end;
        }
 
        inline bool operator> (Task const& other) const
        {
            return _end > other._end;
        }
 
        // Compare tasks with its end
        inline bool operator== (Task const& other)
        {
            return _end == other._end;
        }
 
        // Returns true if the task is in the given group
        inline bool IsInGroup(group_t const group) const
        {
            return _group && (*_group == group);
        }
    };
 
    typedef std::shared_ptr<Task> TaskContainer;
 
    struct Compare
    {
        bool operator() (TaskContainer const& left, TaskContainer const& right) const
        {
            return (*left.get()) < (*right.get());
        };
    };
 
    class TaskQueue
    {
        std::multiset<TaskContainer, Compare> container;
 
    public:
        // Pushes the task in the container
        void Push(TaskContainer&& task);
 
        TaskContainer Pop();
 
        TaskContainer const& First() const;
 
        void Clear();
 
        void RemoveIf(std::function<bool(TaskContainer const&)> const& filter);
 
        void ModifyIf(std::function<bool(TaskContainer const&)> const& filter);
 
        std::size_t Count(group_t const &group);
 
        bool IsEmpty() const;
    };
 
    std::shared_ptr<TaskScheduler> self_reference;
 
    timepoint_t _now;
 
    TaskQueue _task_holder;
 
    typedef std::queue<std::function<void()>> AsyncHolder;
 
    AsyncHolder _asyncHolder;
 
    predicate_t _predicate;
 
    static bool EmptyValidator()
    {
        return true;
    }
 
    static void EmptyCallback()
    {
    }
 
public:
    TaskScheduler()
    : self_reference(this, [](TaskScheduler const*) { }),
    _now(clock_t::now()), _predicate(EmptyValidator) { }
 
    template<typename P>
    TaskScheduler(P&& predicate)
    : self_reference(this, [](TaskScheduler const*) { }),
    _now(clock_t::now()), _predicate(std::forward<P>(predicate)) { }
 
    TaskScheduler(TaskScheduler const&) = delete;
    TaskScheduler(TaskScheduler&&) = delete;
    TaskScheduler& operator= (TaskScheduler const&) = delete;
    TaskScheduler& operator= (TaskScheduler&&) = delete;
 
    template<typename P>
    TaskScheduler& SetValidator(P&& predicate)
    {
        _predicate = std::forward<P>(predicate);
        return *this;
    }
 
    TaskScheduler& ClearValidator();
 
    TaskScheduler& Update(success_t const& callback = EmptyCallback);
 
    TaskScheduler& Update(size_t const microseconds, success_t const& callback = EmptyCallback);
 
    template<typename _Rep, typename _Period>
    TaskScheduler& Update(std::chrono::duration<_Rep, _Period> const& difftime,
                          success_t const& callback = EmptyCallback)
    {
        _now += difftime;
        Dispatch(callback);
        return *this;
    }
 
    TaskScheduler& Async(std::function<void()> const& callable);
 
    template<typename _Rep, typename _Period>
    TaskScheduler& Schedule(std::chrono::duration<_Rep, _Period> const& time,
                            task_handler_t const& task)
    {
        return ScheduleAt(_now, MakeDurationCalculator(time), task);
    }
 
    std::size_t Count(group_t const &group);
 
    template<typename _Rep, typename _Period>
    TaskScheduler& Schedule(std::chrono::duration<_Rep, _Period> const& time,
                            group_t const group, task_handler_t const& task)
    {
        return ScheduleAt(_now, MakeDurationCalculator(time), group, task);
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskScheduler& Schedule(std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                            std::chrono::duration<_RepRight, _PeriodRight> const& max, task_handler_t const& task)
    {
        return ScheduleAt(_now, MakeDurationCalculator(min, max), task);
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskScheduler& Schedule(std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                            std::chrono::duration<_RepRight, _PeriodRight> const& max, group_t const group,
                            task_handler_t const& task)
    {
        return ScheduleAt(_now, MakeDurationCalculator(min, max), group, task);
    }
 
    TaskScheduler& CancelAll();
 
    TaskScheduler& CancelGroup(group_t const group);
 
    TaskScheduler& CancelGroupsOf(std::vector<group_t> const& groups);
 
    template<typename _Rep, typename _Period>
    TaskScheduler& DelayAll(std::chrono::duration<_Rep, _Period> const& duration)
    {
        _task_holder.ModifyIf([&duration](TaskContainer const& task) -> bool
                              {
                                  task->_end += duration;
                                  return true;
                              });
        return *this;
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskScheduler& DelayAll(std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                            std::chrono::duration<_RepRight, _PeriodRight> const& max)
    {
        return DelayAll(RandomDurationBetween(min, max));
    }
 
    template<typename _Rep, typename _Period>
    TaskScheduler& DelayGroup(group_t const group, std::chrono::duration<_Rep, _Period> const& duration)
    {
        _task_holder.ModifyIf([&duration, group](TaskContainer const& task) -> bool
                              {
                                  if (task->IsInGroup(group))
                                  {
                                      task->_end += duration;
                                      return true;
                                  }
                                  else
                                      return false;
                              });
        return *this;
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskScheduler& DelayGroup(group_t const group,
                              std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                              std::chrono::duration<_RepRight, _PeriodRight> const& max)
    {
        return DelayGroup(group, RandomDurationBetween(min, max));
    }
 
    template<typename _Rep, typename _Period>
    TaskScheduler& RescheduleAll(std::chrono::duration<_Rep, _Period> const& duration)
    {
        return RescheduleAt(_now + duration);
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskScheduler& RescheduleAll(std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                                 std::chrono::duration<_RepRight, _PeriodRight> const& max)
    {
        return RescheduleAll(RandomDurationBetween(min, max));
    }
 
    template<typename _Rep, typename _Period>
    TaskScheduler& RescheduleGroup(group_t const group, std::chrono::duration<_Rep, _Period> const& duration)
    {
        return RescheduleAtWithPredicate(_now + duration, [&](TaskContainer const& task)
                                         {
                                             return task->IsInGroup(group);
                                         });
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskScheduler& RescheduleGroup(group_t const group,
                                   std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                                   std::chrono::duration<_RepRight, _PeriodRight> const& max)
    {
        return RescheduleGroup(group, RandomDurationBetween(min, max));
    }
 
private:
    TaskScheduler& InsertTask(TaskContainer task);
 
    TaskScheduler& ScheduleAt(timepoint_t const& end,
                              duration_calculator_t&& duration_calculator, task_handler_t const& task)
    {
        return InsertTask(TaskContainer(new Task(end + duration_calculator(), std::move(duration_calculator), task)));
    }
 
    TaskScheduler& ScheduleAt(timepoint_t const& end,
                              duration_calculator_t&& duration_calculator,
                              group_t const group, task_handler_t const& task)
    {
        // FIXME Reuse std::unique_ptr
        static repeated_t const DEFAULT_REPEATED = 0;
        return InsertTask(TaskContainer(new Task(end + duration_calculator(), std::move(duration_calculator), group, DEFAULT_REPEATED, task)));
    }
 
    static bool AlwaysTruePredicate(TaskContainer const&)
    {
        return true;
    }
 
    TaskScheduler& RescheduleAt(timepoint_t const& end)
    {
        return RescheduleAtWithPredicate(end, AlwaysTruePredicate);
    }
 
    TaskScheduler& RescheduleAtWithPredicate(timepoint_t const& end, std::function<bool(TaskContainer const&)> const& predicate)
    {
        _task_holder.ModifyIf([&](TaskContainer const& task) -> bool
                              {
                                  if (predicate(task))
                                  {
                                      task->_end = end;
                                      return true;
                                  }
                                  else
                                      return false;
                              });
        return *this;
    }
 
    // Returns a random duration between min and max
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    std::chrono::duration<_RepLeft, _PeriodLeft>
    static RandomDurationBetween(std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                                 std::chrono::duration<_RepRight, _PeriodRight> const& max)
    {
        if (!min.count() && !max.count())
            return std::chrono::duration<_RepLeft, _PeriodLeft>(0);
 
        using normalized_t = std::chrono::duration<_RepLeft, _PeriodLeft>;
        auto normalized = std::chrono::duration_cast<normalized_t>(max);
 
        if (min.count() > normalized.count())
            throw std::logic_error("min > max");
 
        static std::mutex _lock;
        std::lock_guard<std::mutex> guard(_lock);
 
        // Random engine
        static std::random_device rd;
        static std::mt19937 rng(rd());
 
        std::uniform_int_distribution<typename std::chrono::duration<_RepLeft, _PeriodLeft>::rep>
        _distribution(min.count(), normalized.count());
 
        // Distribute
        return std::chrono::duration<_RepLeft, _PeriodLeft>(_distribution(rng));
    }
 
    void Dispatch(success_t const& callback);
};
 
class TaskContext
{
    friend class TaskScheduler;
 
    TaskScheduler::TaskContainer _task;
 
    std::weak_ptr<TaskScheduler> _owner;
 
    std::shared_ptr<bool> _consumed;
 
    TaskContext& Dispatch(std::function<TaskScheduler&(TaskScheduler&)> const& apply);
 
    // Copy assign
    // TODO Make TaskContext move only
    TaskContext& operator= (TaskContext const& right)
    {
        _task = right._task;
        _owner = right._owner;
        _consumed = right._consumed;
        return *this;
    }
 
    // Copy construct
    // TODO Make TaskContext move only
    TaskContext(TaskContext const& right)
    : _task(right._task), _owner(right._owner), _consumed(right._consumed) { }
 
public:
    // Empty constructor
    TaskContext()
    : _task(), _owner(), _consumed(std::make_shared<bool>(true)) { }
 
    // Construct from task and owner
    explicit TaskContext(TaskScheduler::TaskContainer&& task, std::weak_ptr<TaskScheduler>&& owner)
    : _task(task), _owner(owner), _consumed(std::make_shared<bool>(false)) { }
 
    // Move construct
    TaskContext(TaskContext&& right)
    : _task(std::move(right._task)), _owner(std::move(right._owner)),
    _consumed(std::move(right._consumed)) { }
 
    // Move assign
    TaskContext& operator= (TaskContext&& right)
    {
        _task = std::move(right._task);
        _owner = std::move(right._owner);
        _consumed = std::move(right._consumed);
        return *this;
    }
 
    bool IsExpired() const;
 
    bool IsInGroup(TaskScheduler::group_t const group) const;
 
    TaskContext& SetGroup(TaskScheduler::group_t const group);
 
    TaskContext& ClearGroup();
 
    TaskScheduler::repeated_t GetRepeatCounter() const;
 
    template<typename _Rep, typename _Period>
    TaskContext& Repeat(std::chrono::duration<_Rep, _Period> const& duration)
    {
        _task->_duration_calculator = TaskScheduler::MakeDurationCalculator(duration);
        return Repeat();
    }
 
    TaskContext& Repeat()
    {
        ThrowOnConsumed();
 
        // Set new duration, in-context timing and increment repeat counter
        _task->_end += _task->_duration_calculator();
        _task->_repeated += 1;
        (*_consumed) = true;
        return Dispatch(std::bind(&TaskScheduler::InsertTask, std::placeholders::_1, std::cref(_task)));
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskContext& Repeat(std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                        std::chrono::duration<_RepRight, _PeriodRight> const& max)
    {
        _task->_duration_calculator = TaskScheduler::MakeDurationCalculator(min, max);
        return Repeat();
    }
 
    TaskContext& Async(std::function<void()> const& callable);
 
    template<typename _Rep, typename _Period>
    TaskContext& Schedule(std::chrono::duration<_Rep, _Period> const& time,
                          TaskScheduler::task_handler_t const& task)
    {
        return Dispatch([&](TaskScheduler& scheduler) -> TaskScheduler&
                        {
                            return scheduler.ScheduleAt(_task->_end, TaskScheduler::MakeDurationCalculator(time), task);
                        });
    }
 
    template<typename _Rep, typename _Period>
    TaskContext& Schedule(std::chrono::duration<_Rep, _Period> const& time,
                          TaskScheduler::group_t const group, TaskScheduler::task_handler_t const& task)
    {
        return Dispatch([&](TaskScheduler& scheduler) -> TaskScheduler&
                        {
                            return scheduler.ScheduleAt(_task->_end, TaskScheduler::MakeDurationCalculator(time), group, task);
                        });
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskContext& Schedule(std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                          std::chrono::duration<_RepRight, _PeriodRight> const& max, TaskScheduler::task_handler_t const& task)
    {
        return Dispatch([&](TaskScheduler& scheduler) -> TaskScheduler&
                        {
                            return scheduler.ScheduleAt(_task->_end, TaskScheduler::MakeDurationCalculator(min, max), task);
                        });
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskContext& Schedule(std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                          std::chrono::duration<_RepRight, _PeriodRight> const& max, TaskScheduler::group_t const group,
                          TaskScheduler::task_handler_t const& task)
    {
        return Dispatch([&](TaskScheduler& scheduler) -> TaskScheduler&
                        {
                            return scheduler.ScheduleAt(_task->_end, TaskScheduler::MakeDurationCalculator(min, max), group, task);
                        });
    }
 
    TaskContext& CancelAll();
 
    TaskContext& CancelGroup(TaskScheduler::group_t const group);
 
    TaskContext& CancelGroupsOf(std::vector<TaskScheduler::group_t> const& groups);
 
    template<typename _Rep, typename _Period>
    TaskContext& DelayAll(std::chrono::duration<_Rep, _Period> const& duration)
    {
        return Dispatch(std::bind(&TaskScheduler::DelayAll<_Rep, _Period>, std::placeholders::_1, duration));
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskContext& DelayAll(std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                          std::chrono::duration<_RepRight, _PeriodRight> const& max)
    {
        return DelayAll(TaskScheduler::RandomDurationBetween(min, max));
    }
 
    template<typename _Rep, typename _Period>
    TaskContext& DelayGroup(TaskScheduler::group_t const group, std::chrono::duration<_Rep, _Period> const& duration)
    {
        return Dispatch(std::bind(&TaskScheduler::DelayGroup<_Rep, _Period>, std::placeholders::_1, group, duration));
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskContext& DelayGroup(TaskScheduler::group_t const group,
                            std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                            std::chrono::duration<_RepRight, _PeriodRight> const& max)
    {
        return DelayGroup(group, TaskScheduler::RandomDurationBetween(min, max));
    }
 
    template<typename _Rep, typename _Period>
    TaskContext& RescheduleAll(std::chrono::duration<_Rep, _Period> const& duration)
    {
        return Dispatch(std::bind(&TaskScheduler::RescheduleAt, std::placeholders::_1, _task->_end + duration));
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskContext& RescheduleAll(std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                               std::chrono::duration<_RepRight, _PeriodRight> const& max)
    {
        return RescheduleAll(TaskScheduler::RandomDurationBetween(min, max));
    }
 
    template<typename _Rep, typename _Period>
    TaskContext& RescheduleGroup(TaskScheduler::group_t const group, std::chrono::duration<_Rep, _Period> const& duration)
    {
        return Dispatch(std::bind(&TaskScheduler::RescheduleAtWithPredicate, std::placeholders::_1, _task->_end + duration,
                                  [&](TaskScheduler::TaskContainer const& task)
                                  {
                                      return task->IsInGroup(group);
                                  }));
    }
 
    template<typename _RepLeft, typename _PeriodLeft, typename _RepRight, typename _PeriodRight>
    TaskContext& RescheduleGroup(TaskScheduler::group_t const group,
                                 std::chrono::duration<_RepLeft, _PeriodLeft> const& min,
                                 std::chrono::duration<_RepRight, _PeriodRight> const& max)
    {
        return RescheduleGroup(group, TaskScheduler::RandomDurationBetween(min, max));
    }
 
private:
    void ThrowOnConsumed() const;
 
    void Invoke();
};
 
#endif /* HORIZON_CORE_UTILITY_TASKSCHEDULER_HPP */