External/common/Timer/TWTimer.hpp

#include <chrono>
#include <functional>
#include <list>
#include <mutex>
#include <thread>
#include <vector>

#include "ThreadPool/ThreadPool.h"
#include "FmtLog/fmtlog.h"

/**
 * @brief 这是一个时间轮定时器的实现，将任务添加到时间轮中，时间轮会在指定的时间点执行任务
 *        这个定时器的最大定时时间是1小时，最小时间间隔是1ms
 *        
 *
 * 设计说明:
 *        1、时间轮的时间点是固定的，每个时间点的时间间隔是1ms，最大时间点是1000 * 60 * 60个，也就是1小时
 *        2、任务是挂在一个个时间点上的，同一个时间点可以有多个任务
 *        3、时间轮的执行是在一个新的线程中执行的，每隔1ms检测一次是否有任务需要执行
 *        4、当前时间点是下一个将要执行的时间点，新加入的立即执行的任务就会加入到当前时间点上
 *        5、时间轮的时间点是循环的，当到达最大时间点时，会重新从0开始
 *        6、管理线程是100us运行一次，在运行时会进行时间累积，相差1ms就会执行一次任务
 * 
 * 缺点：
 *        1、时间轮的时间点是固定的，如果任务的时间点不在时间轮的时间点上，就会被延迟执行
 */


/** ==================================================================
 * ************************** 任务结构体 ******************************
 * ================================================================== */

/**
 * 任务时间点结构体，用来标识任务在时间轮中的位置
 */
struct TaskTimerPoint
{
    long firstLevelIndex = 0;
    long secondLevelIndex = 0;
    long thirdLevelIndex = 0;

    TaskTimerPoint() = default;
    TaskTimerPoint& operator=(const TaskTimerPoint& other) {
        firstLevelIndex = other.firstLevelIndex;
        secondLevelIndex = other.secondLevelIndex;
        thirdLevelIndex = other.thirdLevelIndex;
        return *this;
    }
};

/* 定义函数指针 */
using TaskFunc = std::function<void()>;
/* 定义任务结构体 */
struct Task {
    bool is_loop;           /* 是否循环定时 */
    long interval;          /* 定时间隔 */
    TaskFunc func;          /* 任务函数 */
    TaskTimerPoint timerPoint;   /* 任务时间点，在时间轮中的位置 */
};





/** ==================================================================
 * ***************************** 任务类 ******************************
 * ================================================================== */

class TimerWheel {
public:
    
    /* 构造函数，第一个参数是任务时间点容器最大数量，第二个参数是最低检测时间单位 */
    explicit TimerWheel()
        : m_currentFirstIndex(0),
          m_currentSecondIndex(0),
          m_currentThirdIndex(0),
          m_currentIndex(0)
    {
        m_wheel_size = 1000 * 60 * 60;      /* 1小时定时间隔 */
        m_interval_ms = 1;                  /* 1ms */
        m_firstLevelWheel.resize(1000);
        m_secondLevelWheel.resize(60);
        m_thirdLevelWheel.resize(60);
    }

    ~TimerWheel() {
        Stop();
    }
    /* 开启时间轮 */
    void Start() 
    {
        if (m_running) {
            return;
        }
        m_running = true;
        /* 开启新线程，定时检测队列 */
        m_thread = std::thread([this]() {
            std::chrono::steady_clock::time_point lastTime = std::chrono::steady_clock::now();
            while (m_running) {
                /* 100us计算一次距离上次执行的间隔时间，超过1ms就执行一次 */
                std::this_thread::sleep_for(std::chrono::microseconds(m_tick_us));
                std::chrono::steady_clock::time_point nowTime = std::chrono::steady_clock::now();
                auto duration = std::chrono::duration_cast<std::chrono::microseconds>(nowTime - lastTime);
                if(duration.count() >= 900) {
                    Tick();
                    lastTime = nowTime;
                }
                
            }
            FMTLOG_INFO("TimerWheel thread exit");
        });
        
    }

    void Stop() 
    {
        if (!m_running) {
            return;
        }
        m_running = false;
        if (m_thread.joinable()) {
            m_thread.join();
        }
    }
    /* 绑定任务函数和参数 */
    template<typename F, typename... Args>
    std::function<void()> bindTask(int timeout_ms, F&& f, Args&&... args) 
    {
        return std::bind(std::forward<F>(f), std::forward<Args>(args)...);
    }
    
    /* 添加任务函数 */
    void AddTask(long timeout_ms, std::function<void()> func, bool is_loop = false) 
    {
        /* 这里需要将当前时间点一起上锁，防止立即执行的任务会错过 */
        std::lock_guard<std::mutex> lock(m_mutex);
        /* 计算出定时的时间点数目 */
        long ticks = timeout_ms / m_interval_ms;
        if(ticks > m_wheel_size) {
            FMTLOG_WARN("timeout is too long");
            return;
        }
        Task task;
        task.is_loop = is_loop;
        task.interval = timeout_ms;
        task.func = func;
        task.timerPoint.firstLevelIndex = 0;
        task.timerPoint.secondLevelIndex = 0;
        task.timerPoint.thirdLevelIndex = 0;
        /* 加上当前时间点的位置偏移，计算出需要插入的时间点 */
        // long index = m_currentFirstIndex + m_currentSecondIndex * m_firstLevelCount + m_currentThirdIndex * m_secondLevelCount * m_firstLevelCount + ticks;
        long index = m_currentIndex + ticks;
        /* 计算出需要的位置，如果超出了最大范围就放到过去的时间点中 */
        index = index % m_wheel_size;
        /* 计算出时间点的位置 */
        task.timerPoint.firstLevelIndex = index % m_firstLevelCount;
        task.timerPoint.secondLevelIndex = (index / m_firstLevelCount) % m_secondLevelCount;
        task.timerPoint.thirdLevelIndex = (index / m_firstLevelCount / m_secondLevelCount) % m_thirdLevelCount;
        /* 加入到时间轮的任务队列中 */
        if(task.timerPoint.thirdLevelIndex != m_currentThirdIndex)
        {
            /* 不在当前正在轮训的第三层时间点内，直接加入到对应的时间点中即可 */
            m_thirdLevelWheel[task.timerPoint.thirdLevelIndex].push_back(task);
        }
        /* 在当前正在轮训的第三层时间点内，继续判断是否已经是过去的时间点 */
        else 
        {
            if(task.timerPoint.secondLevelIndex < m_currentSecondIndex || 
            (task.timerPoint.secondLevelIndex == m_currentSecondIndex && task.timerPoint.firstLevelIndex < m_currentFirstIndex))
            {
                /* 在当前正在轮训的第三层时间点内，但是已经是过去的时间点，加入到当前第三层的队列中 */
                m_thirdLevelWheel[task.timerPoint.thirdLevelIndex].push_back(task);
            }
            /* 在当前第三层还未执行的时间点内 */
            else if(task.timerPoint.secondLevelIndex > m_currentFirstIndex)
            {
                /* 在第二轮未执行的点中，加入到第二轮队列中 */
                m_secondLevelWheel[task.timerPoint.secondLevelIndex].push_back(task);
            }
            else
            {
                /* 在当前正在轮训的第二层时间点内，且是未来的时间点，加入到对应的第一层时间点中 */
                m_firstLevelWheel[task.timerPoint.firstLevelIndex].push_back(task);
            }
        }
        
    }

private:
    /* 时间片 */
    void Tick() 
    {
        m_mutex.lock();
        /* 取出这个时间点的函数，循环执行 */
        auto listTask = std::move(m_firstLevelWheel[m_currentFirstIndex]);
        m_firstLevelWheel[m_currentFirstIndex].clear();
        m_mutex.unlock();
        /* 将任务加入到线程池中执行 */
        for (const auto& task : listTask) {
            CPPTP.add_task(task.func);
        }
        /* 将时间点向前 + 1 */
        m_mutex.lock();
        m_currentFirstIndex = (m_currentFirstIndex + 1) % m_firstLevelCount;
        /* 判断第一层时间轮是否走完了 */
        if(m_currentFirstIndex == 0)
        {
            /* 第一个轮走完，取出第二层的下一个时间点 */
            m_currentSecondIndex = (m_currentSecondIndex + 1) % m_secondLevelCount;
            if(m_currentSecondIndex == 0)
            {
                /* 第二层时间轮也走完了，就取出第三个时间轮的下一个时间点 */
                m_currentThirdIndex = (m_currentThirdIndex + 1) % m_thirdLevelCount;
                /* 清空第二层时间轮 */
                m_secondLevelWheel.clear();
                for (auto& task : m_thirdLevelWheel[m_currentThirdIndex]) 
                {
                    m_secondLevelWheel[task.timerPoint.secondLevelIndex].push_back(task);
                }
                m_thirdLevelWheel[m_currentThirdIndex].clear();
            }

            /* 清空第一层时间轮 */
            m_firstLevelWheel.clear();
            for (auto& task : m_secondLevelWheel[m_currentSecondIndex]) 
            {
                m_firstLevelWheel[task.timerPoint.firstLevelIndex].push_back(task);
            }
            m_secondLevelWheel[m_currentSecondIndex].clear();
        }
        m_mutex.unlock();
        /* 将循环定时的任务重新加入到时间轮中 */
        for(auto& task : listTask) 
        {
            if(task.is_loop)
            {
                AddTask(task.interval, task.func, true);
            }
        }
        m_currentIndex = (m_currentIndex + 1) % m_wheel_size;
        
    }

private:
    long m_max_interval = 0;                /* 最大的定时时间长度，等于下面两个之和，超过就报错 */
    long m_wheel_size;                      /* 时间轮最大的轮训次数 */
    long m_interval_ms;                     /* 每个时间点的间隔秒数，这里是1ms */
    const long m_tick_us = 100;                         /* 管理线程跳动时间是100us */
    const long m_firstLevelCount = 1000;                /* 第一层级的时间轮的个数 */
    const long m_secondLevelCount = 60;                 /* 第二层级的时间轮的个数 */
    const long m_thirdLevelCount = 60;                  /* 第三层级的时间轮的个数 */
    std::vector<std::list<Task>> m_firstLevelWheel;     /* 第一层级的时间轮，这里是ms等级，总共1000个 */
    std::vector<std::list<Task>> m_secondLevelWheel;    /* 第二层级的时间轮，这里是秒的等级，总共60个 */
    std::vector<std::list<Task>> m_thirdLevelWheel;     /* 第三层级的时间轮，这里是分钟的等级，总共60个 */
    long m_currentFirstIndex;               /* 现在的第一层时间点 */
    long m_currentSecondIndex;              /* 现在的第二层位置 */
    long m_currentThirdIndex;               /* 现在的第三层位置 */
    long m_currentIndex;                    /* 现在的时间点 */
    bool m_running = false;                 /* 运行标志位 */
    std::thread m_thread;
    std::mutex m_mutex;
    
};