^-^
后端技术_定时器方案红黑树和时间轮和最小堆
最近更新：2024-09-23   |   字数总计：3.8k   |   阅读估时：21分钟   |   阅读量：次
  1. 定时器
    1. 单线程
      1. 容器
        1. STL实现
      2. 检测触发机制
        1. 使用timerfd实现
      3. 定时器常见优化点
    2. 多线程
      1. 时间轮
        1. 模拟时间表盘 (要看)
        2. 多层级时间轮 (要看)

定时器

单线程

  • 组织异步任务
    1
    2
    3
    4
    5
    struct TimerNode{
        int expire;
        Callback func;
        void *ctx;
    };
  1. 容器，可以按触发时间排序的结构, 查找要快，能很快找到最小值
  2. 检测触发机制
  3. 行业：红黑树ngix、workflow；最小堆libevent、libev、go、boost.asio；

容器

  • Why红黑树
  1. 平衡二叉搜索树，平衡黑节点的高度，查找时间O(logN)
  2. 找最小值，永远都在最左侧O(1)
  • Why最小堆
  1. 完全二叉树，一个节点总是小于等于它的子节点（父<子），任意一个节点的子树都是最小堆
  2. 可以使用数组去描述，父节点索引k，左子2k+1，右子2k+2
  3. 0索引位置就是最小值，增加< o(logN)(添加在最后面，看上升)，删除（和最后一个交换，上升再下沉）O（logN）
  4. 比红黑树快一点
  • 面试问题：手写高性能定时器
  1. 封装（不建议最小堆，priority_queue只有构建堆的api，没有办法高性能实现上升下降）；触发时间可能一样，所以使用set和map，不能用expire作为key，需要自定义task的排序规则。
  2. 实现的问题 容器 + 检测触发
  3. 测试代码
  4. 考虑优化点（optional）
STL实现
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
##include <sys/epoll.h>
##include <functional>
##include <chrono>
##include <set>
##include <memory>
##include <iostream>
using namespace std;
struct TimerNodeBase {
    time_t expire;
    int64_t id; // 用id来区别不同的任务
    // ！将函数对象本身和参与比较的内容进行了分离，因为函数对象的拷贝构造代价很大，而比较的时候可能会产生拷贝
};
struct TimerNode : public TimerNodeBase {
    using Callback = std::function<void(const TimerNode &node)>; 
    Callback func;
    TimerNode(int64_t id, time_t expire, Callback func) : func(func) {
        this->expire = expire;
        this->id = id;
    }
};
// 重载比较操作符，引用具备多态的
bool operator < (const TimerNodeBase &lhd, const TimerNodeBase &rhd) { 
    if (lhd.expire < rhd.expire)
        return true;
    else if (lhd.expire > rhd.expire) 
        return false;
    return lhd.id < rhd.id; // ！假设时间一致时，返回更先创建的
}
class Timer {
public:
    static time_t GetTick() {
        auto sc = chrono::time_point_cast<chrono::milliseconds>(chrono::steady_clock::now());
        auto temp = chrono::duration_cast<chrono::milliseconds>(sc.time_since_epoch());
        return temp.count();
    }

    TimerNodeBase AddTimer(time_t msec, TimerNode::Callback func) {
        time_t expire = GetTick() + msec;
        if (timeouts.empty() || expire <= timeouts.crbegin()->expire) {
            auto pairs = timeouts.emplace(GenID(), expire, std::move(func));
            return static_cast<TimerNodeBase>(*pairs.first);
        }
        auto ele = timeouts.emplace_hint(timeouts.crbegin().base(), GenID(), expire, std::move(func));
        return static_cast<TimerNodeBase>(*ele);
    }
    
    bool DelTimer(TimerNodeBase &node) {
        auto iter = timeouts.find(node);
        if (iter != timeouts.end()) {
            timeouts.erase(iter);
            return true;
        }
        return false;
    }

    void HandleTimer(time_t now) {
        auto iter = timeouts.begin();
        while (iter != timeouts.end() && iter->expire <= now) {
            iter->func(*iter);
            iter = timeouts.erase(iter);
        }
    }

    time_t TimeToSleep() {
        auto iter = timeouts.begin();
        if (iter == timeouts.end()) {
            return -1;
        }
        time_t diss = iter->expire - GetTick();
        return diss > 0 ? diss : 0;
    }
private:
    static int64_t GenID() {
        return gid++;
    }
    static int64_t gid;
    set<TimerNode, std::less<>> timeouts;
};
int64_t Timer::gid = 0;

int main() {
    int epfd = epoll_create(1);
    unique_ptr<Timer> timer = make_unique<Timer>();
    int i =0;
    timer->AddTimer(1000, [&](const TimerNode &node) {
        cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl;
    });
    timer->AddTimer(1000, [&](const TimerNode &node) {
        cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl;
    });
    timer->AddTimer(3000, [&](const TimerNode &node) {
        cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl;
    });
    auto node = timer->AddTimer(2100, [&](const TimerNode &node) {
        cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl;
    });
    timer->DelTimer(node);
    cout << "now time:" << Timer::GetTick() << endl;
    epoll_event ev[64] = {0};
    while (true) {
        int n = epoll_wait(epfd, ev, 64, timer->TimeToSleep());
        time_t now = Timer::GetTick();
        for (int i = 0; i < n; i++) {
            /**/
        }
        /* 处理定时事件*/
        timer->HandleTimer(now);
    }
    return 0;
}

检测触发机制

  1. select\poll\epoll 最后一个参数 阻塞时长
    1
    2
    3
    4
    // 找到最近一个触发需要等待的时长
    int timeout = find_nearest_timer() - now(); 
    // 让epoll_wait阻塞该时长
    int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
  2. timerfd
    1
    2
    3
    4
    5
    ##include <sys/timefd.h>
    // 创建定时器文件描述符，可以像网络io一样，将这个fd交由io多路复用来管理
    int timerfd_create(int clockid, int flags);
    // 设置一个触发时间，io多路复用将会检测这个过期事件，然后通知应用程序定时事件就绪
    int timefd_settime(int fd, int flags, const struct itimerspec *new_value, struct itimerspec *old_value);
使用timerfd实现
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
##include <sys/epoll.h>
##include <sys/timerfd.h>
##include <time.h> // for timespec itimerspec
##include <unistd.h> // for close

##include <functional>
##include <chrono>
##include <set>
##include <memory>
##include <iostream>

using namespace std;

struct TimerNodeBase {
    time_t expire;
    uint64_t id; 
};

struct TimerNode : public TimerNodeBase {
    using Callback = std::function<void(const TimerNode &node)>;
    Callback func;
    TimerNode(int64_t id, time_t expire, Callback func) : func(func) {
        this->expire = expire;
        this->id = id;
    }
};

bool operator < (const TimerNodeBase &lhd, const TimerNodeBase &rhd) {
    if (lhd.expire < rhd.expire) {
        return true;
    } else if (lhd.expire > rhd.expire) {
        return false;
    } else return lhd.id < rhd.id;
}

class Timer {
public:
    static inline time_t GetTick() {
        return chrono::duration_cast<chrono::milliseconds>(chrono::steady_clock::now().time_since_epoch()).count();
    }
    
    TimerNodeBase AddTimer(int msec, TimerNode::Callback func) {
        time_t expire = GetTick() + msec;
        if (timeouts.empty() || expire <= timeouts.crbegin()->expire) {
            auto pairs = timeouts.emplace(GenID(), expire, std::move(func));
            return static_cast<TimerNodeBase>(*pairs.first);
        }
        // !!! 优化点1. 如果时间大于最大的那个节点，在它后面插入，复杂度O(1)
        auto ele = timeouts.emplace_hint(timeouts.crbegin().base(), GenID(), expire, std::move(func));
        return static_cast<TimerNodeBase>(*ele);
    }
    
    void DelTimer(TimerNodeBase &node) {
        auto iter = timeouts.find(node);
        if (iter != timeouts.end())
            timeouts.erase(iter);
    }

    void HandleTimer(time_t now) {
        auto iter = timeouts.begin();
        while (iter != timeouts.end() && iter->expire <= now) {
            iter->func(*iter);
            iter = timeouts.erase(iter);
        }
    }

public:
    virtual void UpdateTimerfd(const int fd) {
        struct timespec abstime;
        auto iter = timeouts.begin();
        if (iter != timeouts.end()) {
            abstime.tv_sec = iter->expire / 1000;
            abstime.tv_nsec = (iter->expire % 1000) * 1000000;
        } else {
            abstime.tv_sec = 0;
            abstime.tv_nsec = 0;
        }
        struct itimerspec its = {
            .it_interval = {},
            .it_value = abstime
        };
        timerfd_settime(fd, TFD_TIMER_ABSTIME, &its, nullptr);
    }

private:
    static inline uint64_t GenID() {
        return gid++;
    }
    static uint64_t gid; 

    set<TimerNode, std::less<>> timeouts;
};
uint64_t Timer::gid = 0;

int main() {
    int epfd = epoll_create(1);
    int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
    struct epoll_event ev = {.events=EPOLLIN | EPOLLET};
    epoll_ctl(epfd, EPOLL_CTL_ADD, timerfd, &ev);
    
    unique_ptr<Timer> timer = make_unique<Timer>();
    int i = 0;
    timer->AddTimer(1000, [&](const TimerNode &node) {
        cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl;
    });

    timer->AddTimer(1000, [&](const TimerNode &node) {
        cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl;
    });

    timer->AddTimer(3000, [&](const TimerNode &node) {
        cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl;
    });

    auto node = timer->AddTimer(2100, [&](const TimerNode &node) {
        cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl;
    });
    timer->DelTimer(node);

    cout << "now time:" << Timer::GetTick() << endl;

    struct epoll_event evs[64] = {0};
    while (true) {
        timer->UpdateTimerfd(timerfd);
        int n = epoll_wait(epfd, evs, 64, -1);
        time_t now = Timer::GetTick();
        for (int i = 0; i < n; i++) {
            // for network event handle
        }
        timer->HandleTimer(now);
    }
    epoll_ctl(epfd, EPOLL_CTL_DEL, timerfd, &ev);
    close(timerfd);
    close(epfd);
    return 0;
}

定时器常见优化点

  1. 业务中常见添加间隔时间相同的任务
    1
    2
    3
    4
    5
    AddTimer(1000, func);
    AddTimer(1000, func);
    AddTimer(1000, func);
    AddTimer(1000, func);
    ...
  • 总是给最后一个最大的节点插入后面的节点
  • 优化：使用emplace_hint（给定一个迭代器，在该迭代器后面添加节点，时间复杂度O(1)）
  1. 模块拆分的思路
  • 如nginx，最大只能有10240个定时器任务，说明红黑树有性能问题
  • 优化：分模块构建单独的定时器，使每个定时器的个数都比较小
  1. 多reactor模式，将定时任务拆分到多个线程中去检测，每个reactor对应一个定时器，主线程发布定时任务时负载均衡到每个reactor中

多线程

时间轮

  • 多线程按照执行序排序
  • 行业：linux、kafka、skynet、netty
  • 删除、添加都是O(1)
  1. 容器：指针数组，每次指针移动最小时间精度为一格，每个格子存储相关任务的链表
  2. 触发机制：时间指针移动到哪儿，哪个格子的任务就到期了
  • 单层级时间轮（如果用秒做单位，会有大量空踏现象）
    18
  • 多层次时间轮（添加的时候模糊映射，秒针走一轮，分针层级重新映射）
    19
模拟时间表盘 (要看)
  • clock-timer.h
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    ##ifndef _MARK_TIMEWHEEL_
    ##define _MARK_TIMEWHEEL_
    ##include <time.h>
    ##include <stdint.h>
    ##include <stddef.h>

    typedef struct timer_node timer_node_t;
    typedef void (*handler_pt) (struct timer_node *node);
    struct timer_node {
        struct timer_node *next;
        uint32_t expire;
        handler_pt callback;
        uint8_t cancel;
    };

    void init_timer(void);
    timer_node_t* add_timer(int time, handler_pt func);
    void del_timer(timer_node_t *node);
    void check_timer(int *stop);
    void clear_timer();
    time_t now_time();

    ##endif

  • clock-timer.c
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    190
    191
    192
    193
    194
    195
    196
    197
    198
    199
    200
    201
    202
    203
    204
    205
    206
    207
    208
    209
    210
    211
    212
    213
    214
    215
    216
    217
    218
    219
    220
    221
    222
    223
    224
    225
    226
    227
    228
    229
    230
    231
    232
    233
    234

    ##include <unistd.h>
    ##include <stdio.h>
    ##include <stdlib.h>
    ##include <string.h>
    ##include "spinlock.h"
    ##include "clock-timer.h"

    ##define SECONDS 60
    ##define MINUTES 60
    ##define HOURS   12
    ##define ONE_HOUR 3600
    ##define ONE_MINUTE 60
    ##define HALF_DAY 43200 // 12*3600

    typedef struct link_list {
        timer_node_t head;
        timer_node_t *tail;
    }link_list_t;

    typedef struct timer {
        link_list_t second[SECONDS];
        link_list_t minute[MINUTES];
        link_list_t hour[HOURS];
        spinlock_t lock;
        uint32_t time;
        time_t current_point;
    }timer_st;

    static timer_st * TI = NULL;

    static timer_node_t *
    link_clear(link_list_t *list) {
        timer_node_t * ret = list->head.next;
        list->head.next = 0;
        list->tail = &(list->head);

        return ret;
    }

    static void
    link_to(link_list_t *list, timer_node_t *node) {
        list->tail->next = node;
        list->tail = node;
        node->next=0;
    }

    static void
    add_node(timer_st *T, timer_node_t *node) {
        uint32_t time=node->expire;
        uint32_t current_time=T->time;
        uint32_t msec = time - current_time;
        if (msec < ONE_MINUTE) {
            link_to(&T->second[time % SECONDS], node);
        } else if (msec < ONE_HOUR) {
            link_to(&T->minute[(uint32_t)(time/ONE_MINUTE) % MINUTES], node);
        } else {
            link_to(&T->hour[(uint32_t)(time/ONE_HOUR) % HOURS], node);
        }
    }

    static void
    remap(timer_st *T, link_list_t *level, int idx) {
        timer_node_t *current = link_clear(&level[idx]);
        while (current) {
            timer_node_t *temp=current->next;
            add_node(T, current);
            current=temp;
        }
    }

    static void
    timer_shift(timer_st *T) {
        uint32_t ct = ++T->time % HALF_DAY;
        if (ct == 0) {
            remap(T, T->hour, 0);
        } else {
            if (ct % SECONDS == 0) {
                {
                    uint32_t idx = (uint32_t)(ct / ONE_MINUTE) % MINUTES;
                    if (idx != 0) {
                        remap(T, T->minute, idx);
                        return;
                    }
                }
                {
                    uint32_t idx = (uint32_t)(ct / ONE_HOUR) % HOURS;
                    if (idx != 0) {
                        remap(T, T->hour, idx);
                    }
                }
            }
        }
    }

    static void
    dispatch_list(timer_node_t *current) {
        do {
            timer_node_t * temp = current;
            current=current->next;
            if (temp->cancel == 0)
                temp->callback(temp);
            free(temp);
        } while (current);
    }

    static void
    timer_execute(timer_st *T) {
        uint32_t idx = T->time % SECONDS;

        while (T->second[idx].head.next) {
            timer_node_t *current = link_clear(&T->second[idx]);
            spinlock_unlock(&T->lock);
            dispatch_list(current);
            spinlock_lock(&T->lock);
        }
    }

    static void
    timer_update(timer_st *T) {
        spinlock_lock(&T->lock);
        timer_execute(T);
        timer_shift(T);
        timer_execute(T);
        spinlock_unlock(&T->lock);
    }

    static timer_st *
    create_timer() {
        timer_st *r = (timer_st *)malloc(sizeof(timer_st));
        memset(r,0,sizeof(*r));

        int i;
        for (i=0; i<SECONDS; i++) {
            link_clear(&r->second[i]);
        }
        for (i=0; i<MINUTES; i++) {
            link_clear(&r->minute[i]);
        }
        for (i=0; i<HOURS; i++) {
            link_clear(&r->hour[i]);
        }
        spinlock_init(&r->lock);
        return r;
    }

    void
    init_timer(void) {
        TI = create_timer();
        TI->current_point = now_time();
    }

    timer_node_t*
    add_timer(int time, handler_pt func) {
        timer_node_t *node = (timer_node_t *)malloc(sizeof(*node));
        spinlock_lock(&TI->lock);
        node->expire = time+TI->time;
        printf("add timer at %u, expire at %u, now_time at %lu\n", TI->time, node->expire, now_time());
        node->callback = func;
        node->cancel = 0;
        if (time <= 0) {
            spinlock_unlock(&TI->lock);
            node->callback(node);
            free(node);
            return NULL;
        }
        add_node(TI, node);
        spinlock_unlock(&TI->lock);
        return node;
    }

    void
    del_timer(timer_node_t *node) {
        node->cancel = 1;
    }

    void
    check_timer(int *stop) {
        while (*stop == 0) {
            time_t cp = now_time();
            if (cp != TI->current_point) {
                uint32_t diff = (uint32_t)(cp - TI->current_point);
                TI->current_point = cp;
                int i;
                for (i=0; i<diff; i++) {
                    timer_update(TI);
                }
            }
            usleep(200000);
        }
    }

    void
    clear_timer() {
        int i;
        for (i=0; i<SECONDS; i++) {
            link_list_t * list = &TI->second[i];
            timer_node_t* current = list->head.next;
            while(current) {
                timer_node_t * temp = current;
                current = current->next;
                free(temp);
            }
            link_clear(&TI->second[i]);
        }
        for (i=0; i<MINUTES; i++) {
            link_list_t * list = &TI->minute[i];
            timer_node_t* current = list->head.next;
            while(current) {
                timer_node_t * temp = current;
                current = current->next;
                free(temp);
            }
            link_clear(&TI->minute[i]);
        }
        for (i=0; i<HOURS; i++) {
            link_list_t * list = &TI->hour[i];
            timer_node_t* current = list->head.next;
            while(current) {
                timer_node_t * temp = current;
                current = current->next;
                free(temp);
            }
            link_clear(&TI->hour[i]);
        }
    }

    time_t
    now_time() {
        struct timespec ti;
        clock_gettime(CLOCK_MONOTONIC, &ti);
        // 1ns = 1/1000000000 s = 1/1000000 ms
        return ti.tv_sec;
    }
  • clock_main.c
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    ##include <stdio.h>
    ##include "clock-timer.h"

    void do_timer(timer_node_t *node) {
        (void)node;
        printf("do_timer expired now_time:%lu\n", now_time());
    }

    int stop = 0;
    int main() {
        init_timer();
        add_timer(3, do_timer);
        check_timer(&stop);
        clear_timer();
        return 0;
    }
多层级时间轮 (要看)
  • spinlock.h
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    ##ifndef SPINLOCK_H
    ##define SPINLOCK_H

    typedef struct spinlock {
    	int lock;
    }spinlock_t;

    void spinlock_init(spinlock_t *lock) {
    	lock->lock = 0;
    }

    void spinlock_lock(spinlock_t *lock) {
    	while (__sync_lock_test_and_set(&lock->lock, 1)) {}
    }

    int spinlock_trylock(spinlock_t *lock) {
    	return __sync_lock_test_and_set(&lock->lock, 1) == 0;
    }

    void spinlock_unlock(spinlock_t *lock) {
    	__sync_lock_release(&lock->lock);
    }

    void spinlock_destroy(spinlock_t *lock) {
    	(void) lock;
    }

    ##endif

  • timewheel.h
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    ##ifndef _MARK_TIMEWHEEL_
    ##define _MARK_TIMEWHEEL_

    ##include <stdint.h>

    ##define TIME_NEAR_SHIFT 8
    ##define TIME_NEAR (1 << TIME_NEAR_SHIFT)
    ##define TIME_LEVEL_SHIFT 6
    ##define TIME_LEVEL (1 << TIME_LEVEL_SHIFT)
    ##define TIME_NEAR_MASK (TIME_NEAR-1)
    ##define TIME_LEVEL_MASK (TIME_LEVEL-1)

    typedef struct timer_node timer_node_t;
    typedef void (*handler_pt) (struct timer_node *node);

    struct timer_node {
    	struct timer_node *next;
    	uint32_t expire;
        handler_pt callback;
        uint8_t cancel;
    	int id; // 此时携带参数
    };

    timer_node_t* add_timer(int time, handler_pt func, int threadid);

    void expire_timer(void);

    void del_timer(timer_node_t* node);

    void init_timer(void);

    void clear_timer();

    ##endif

  • timewheel.c
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    190
    191
    192
    193
    194
    195
    196
    197
    198
    199
    200
    201
    202
    203
    204
    205
    206
    207
    208
    209
    210
    211
    212
    213
    214
    215
    216
    217
    218
    219
    220
    221
    222
    223
    224
    225
    226
    227
    228
    229
    230
    231
    232
    233
    234
    235
    236
    ##include "spinlock.h"
    ##include "timewheel.h"
    ##include <string.h>
    ##include <stddef.h>
    ##include <stdlib.h>

    ##if defined(__APPLE__)
    ##include <AvailabilityMacros.h>
    ##include <sys/time.h>
    ##include <mach/task.h>
    ##include <mach/mach.h>
    ##else
    ##include <time.h>
    ##endif

    typedef struct link_list {
    	timer_node_t head;
    	timer_node_t *tail;
    }link_list_t;

    typedef struct timer {
    	link_list_t near[TIME_NEAR];
    	link_list_t t[4][TIME_LEVEL];
    	struct spinlock lock;
    	uint32_t time;
    	uint64_t current;
    	uint64_t current_point;
    }s_timer_t;

    static s_timer_t * TI = NULL;

    timer_node_t *
    link_clear(link_list_t *list) {
    	timer_node_t * ret = list->head.next;
    	list->head.next = 0;
    	list->tail = &(list->head);

    	return ret;
    }

    void
    link(link_list_t *list, timer_node_t *node) {
    	list->tail->next = node;
    	list->tail = node;
    	node->next=0;
    }

    void
    add_node(s_timer_t *T, timer_node_t *node) {
    	uint32_t time=node->expire;
    	uint32_t current_time=T->time;
    	uint32_t msec = time - current_time;
    	if (msec < TIME_NEAR) { //[0, 0x100)
    		link(&T->near[time&TIME_NEAR_MASK],node);
    	} else if (msec < (1 << (TIME_NEAR_SHIFT+TIME_LEVEL_SHIFT))) {//[0x100, 0x4000)
    		link(&T->t[0][((time>>TIME_NEAR_SHIFT) & TIME_LEVEL_MASK)],node);	
    	} else if (msec < (1 << (TIME_NEAR_SHIFT+2*TIME_LEVEL_SHIFT))) {//[0x4000, 0x100000)
    		link(&T->t[1][((time>>(TIME_NEAR_SHIFT + TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node);	
    	} else if (msec < (1 << (TIME_NEAR_SHIFT+3*TIME_LEVEL_SHIFT))) {//[0x100000, 0x4000000)
    		link(&T->t[2][((time>>(TIME_NEAR_SHIFT + 2*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node);	
    	} else {//[0x4000000, 0xffffffff]
    		link(&T->t[3][((time>>(TIME_NEAR_SHIFT + 3*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node);	
    	}
    }

    timer_node_t*
    add_timer(int time, handler_pt func, int threadid) {
    	timer_node_t *node = (timer_node_t *)malloc(sizeof(*node));
    	spinlock_lock(&TI->lock);
    	node->expire = time+TI->time;
    	node->callback = func;
    	node->id = threadid;
    	if (time <= 0) {
    		spinlock_unlock(&TI->lock);
    		node->callback(node);
    		free(node);
    		return NULL;
    	}
    	add_node(TI, node);
    	spinlock_unlock(&TI->lock);
    	return node;
    }

    void
    move_list(s_timer_t *T, int level, int idx) {
    	timer_node_t *current = link_clear(&T->t[level][idx]);
    	while (current) {
    		timer_node_t *temp=current->next;
    		add_node(T,current);
    		current=temp;
    	}
    }

    void
    timer_shift(s_timer_t *T) {
    	int mask = TIME_NEAR;
    	uint32_t ct = ++T->time;
    	if (ct == 0) {
    		move_list(T, 3, 0);
    	} else {
    		// ct / 256
    		uint32_t time = ct >> TIME_NEAR_SHIFT;
    		int i=0;
    		// ct % 256 == 0
    		while ((ct & (mask-1))==0) {
    			int idx=time & TIME_LEVEL_MASK;
    			if (idx!=0) {
    				move_list(T, i, idx);
    				break;				
    			}
    			mask <<= TIME_LEVEL_SHIFT;
    			time >>= TIME_LEVEL_SHIFT;
    			++i;
    		}
    	}
    }

    void
    dispatch_list(timer_node_t *current) {
    	do {
    		timer_node_t * temp = current;
    		current=current->next;
            if (temp->cancel == 0)
                temp->callback(temp);
    		free(temp);
    	} while (current);
    }

    void
    timer_execute(s_timer_t *T) {
    	int idx = T->time & TIME_NEAR_MASK;
    	
    	while (T->near[idx].head.next) {
    		timer_node_t *current = link_clear(&T->near[idx]);
    		spinlock_unlock(&T->lock);
    		dispatch_list(current);
    		spinlock_lock(&T->lock);
    	}
    }

    void 
    timer_update(s_timer_t *T) {
    	spinlock_lock(&T->lock);
    	timer_execute(T);
    	timer_shift(T);
    	timer_execute(T);
    	spinlock_unlock(&T->lock);
    }

    void
    del_timer(timer_node_t *node) {
        node->cancel = 1;
    }

    s_timer_t *
    timer_create_timer() {
    	s_timer_t *r=(s_timer_t *)malloc(sizeof(s_timer_t));
    	memset(r,0,sizeof(*r));
    	int i,j;
    	for (i=0;i<TIME_NEAR;i++) {
    		link_clear(&r->near[i]);
    	}
    	for (i=0;i<4;i++) {
    		for (j=0;j<TIME_LEVEL;j++) {
    			link_clear(&r->t[i][j]);
    		}
    	}
    	spinlock_init(&r->lock);
    	r->current = 0;
    	return r;
    }

    uint64_t
    gettime() {
    	uint64_t t;
    ##if !defined(__APPLE__) || defined(AVAILABLE_MAC_OS_X_VERSION_10_12_AND_LATER)
    	struct timespec ti;
    	clock_gettime(CLOCK_MONOTONIC, &ti);
    	t = (uint64_t)ti.tv_sec * 1000;
    	t += ti.tv_nsec / 1000000;
    	// 1ns = 1/1000000000 s = 1/1000000 ms
    ##else
    	struct timeval tv;
    	gettimeofday(&tv, NULL);
    	t = (uint64_t)tv.tv_sec * 100;
    	t += tv.tv_usec / 10000;
    ##endif
    	return t;
    }

    void
    expire_timer(void) {
    	uint64_t cp = gettime();
    	if (cp != TI->current_point) {
    		uint32_t diff = (uint32_t)(cp - TI->current_point);
    		TI->current_point = cp;
    		int i;
    		for (i=0; i<diff; i++) {
    			timer_update(TI);
    		}
    	}
    }

    void 
    init_timer(void) {
    	TI = timer_create_timer();
    	TI->current_point = gettime();
    }

    void
    clear_timer() {
    	int i,j;
    	for (i=0;i<TIME_NEAR;i++) {
    		link_list_t * list = &TI->near[i];
    		timer_node_t* current = list->head.next;
    		while(current) {
    			timer_node_t * temp = current;
    			current = current->next;
    			free(temp);
    		}
    		link_clear(&TI->near[i]);
    	}
    	for (i=0;i<4;i++) {
    		for (j=0;j<TIME_LEVEL;j++) {
    			link_list_t * list = &TI->t[i][j];
    			timer_node_t* current = list->head.next;
    			while (current) {
    				timer_node_t * temp = current;
    				current = current->next;
    				free(temp);
    			}
    			link_clear(&TI->t[i][j]);
    		}
    	}
    }

  • tw-timer.c
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    ##include <stdio.h>
    ##include <unistd.h>

    ##include <pthread.h>
    ##include <time.h>
    ##include <stdlib.h>
    ##include "timewheel.h"

    struct context {
    	int quit;
        int thread;
    };

    struct thread_param {
    	struct context *ctx;
    	int id;
    };

    static struct context ctx = {0};

    void do_timer(timer_node_t *node) {
        printf("do_timer expired:%d - thread-id:%d\n", node->expire, node->id);
        add_timer(100, do_timer, node->id);
    }

    void do_clock(timer_node_t *node) {
        static int time;
        time ++;
        printf("---time = %d ---\n", time);
        add_timer(100, do_clock, node->id);
    }

    void* thread_worker(void *p) {
    	struct thread_param *tp = p;
    	int id = tp->id;
        struct context *ctx = tp->ctx;
        int expire = rand() % 200; 
        add_timer(expire, do_timer, id);
    	while (!ctx->quit) {
            usleep(1000);
        }
        printf("thread_worker:%d exit!\n", id);
        return NULL;
    }

    void do_quit(timer_node_t * node) {
        ctx.quit = 1;
    }

    int main() {
        srand(time(NULL));
        ctx.thread = 2;
        pthread_t pid[ctx.thread];

        init_timer();
        add_timer(6000, do_quit, 100);
        add_timer(0, do_clock, 100);
        struct thread_param task_thread_p[ctx.thread];
        int i;
        for (i = 0; i < ctx.thread; i++) {
            task_thread_p[i].id = i;
            task_thread_p[i].ctx = &ctx;
            if (pthread_create(&pid[i], NULL, thread_worker, &task_thread_p[i])) {
                fprintf(stderr, "create thread failed\n");
                exit(1);
            }
        }

        while (!ctx.quit) {
            expire_timer();
            usleep(250);
        }
        clear_timer();
        for (i = 0; i < ctx.thread; i++) {
    		pthread_join(pid[i], NULL);
        }
        printf("all thread is closed\n");
        return 0;
    }

2024-03-04 该篇文章被 Cleofwine 归为分类: 服务端

上一篇：后端技术_手写死锁检测组件 下一篇：后端技术_原子操作CAS与锁实现（施工）
© 2023-2026 Cleofwine / Email / GitHub / 浙ICP备2024136127号
🌙赠我弯弯一枚月，也赠予我晚星⭐ 本站由Hexo驱动｜使用Hexo-theme-A4主题