DPVS的定时器

一，时间轮

DPDK的定时器使用跳表实现的，在处理大量比如会话时间时，session老化啥的，效率不高，DPVS的使用的是复杂度 O(1)的时间轮算法。

二，DPVS的定时器

1.数据结构

struct timer_scheduler {
    /* wheels and cursors */
    rte_spinlock_t      lock;
    uint32_t            cursors[LEVEL_DEPTH];
    struct list_head    *hashs[LEVEL_DEPTH];
 
    /* leverage dpdk rte_timer to drive us */
    struct rte_timer    rte_tim;
};
static struct timer_scheduler g_timer_sched;

lock是锁，增删改查都要锁。

rte_tim负责时间滴答，也就是tick++。

cursors和hashs 配合使用，构成时间轮。

从 0 到 LEVEL_SIZE (2<<18), LEVEL_DEPTH 值为 2，也就是说 cursors[0] 可以保存 2<<18 个嘀嗒，如果 DPVS_TIMER_HZ = 1000，那么就是 524s. cursors[0] 驱动 cursors[1] 时间轮，两个轮一共 8.7 年时间。hashs 是一个长度为 2 的数组，每个元素是一个链表数组，长度是 LEVEL_SIZE (2<<18), 链表成员就是具体的定时器消息。

2.初始化

int dpvs_timer_init(void)
{
    lcoreid_t cid;
    int err;
 
    /* per-lcore timer */
    rte_eal_mp_remote_launch(timer_lcore_init, NULL, SKIP_MAIN);
    RTE_LCORE_FOREACH_WORKER(cid) {
        err = rte_eal_wait_lcore(cid);
        if (err < 0) {
            RTE_LOG(ERR, DTIMER, "%s: lcore %d: %s.\n",
                    __func__, cid, dpvs_strerror(err));
            return err;
        }
    }
 
    /* global timer */
    return timer_init_schedler(&g_timer_sched, rte_get_main_lcore());
}

每个slave lcore 都要调 timer_lcore_init初始化自己的，然后初始化全局的_timer_sched。

3.timer_lcore_init

static int timer_lcore_init(void *arg)
{
    if (!rte_lcore_is_enabled(rte_lcore_id()))
        return EDPVS_DISABLED;
 
    return timer_init_schedler(&RTE_PER_LCORE(timer_sched), rte_lcore_id());
}

4.timer_init_schedler

static int timer_init_schedler(struct timer_scheduler *sched, lcoreid_t cid)
{