高性能定时器概述(包含时间堆的实现)

ゝ一世哀愁。 2022-03-29 20:16 394阅读 0赞

### 文章目录 ###

*   *   *  网络编程处理的事件主要有I/O，信号和定时器！！！
         *  socket选项SO\_RCVTIMEO 和 SO\_SNDTIMEO
         *  SIGALRM 信号(在多线程程序中处理信号是非常麻烦的事情，应当尽量避免，原因有时间再来谈)
         *  I/O 复用系统调用的超时参数（精度太低，毫秒）
         *  timerfd\_ \*系列函数（新内核使用，现在都使用这个，原因精度高(纳秒),能统一到复用框架中，统一处理I/O事件和超时事件）
         *  时间堆的实现原理：（std::priority\_queue

### 网络编程处理的事件主要有I/O，信号和定时器！！！ ###

[为什么网络库会带有定时器的功能？][Link 1]

玩家下线

其中第三类事件就是定时事件，比如：定期检测一个客户连接的活动状态．`server`通常管理着众多的定时事件，因此有效的管理这些事件，**使之能在预期的时间点触发且不影响`server`的主要逻辑**，对于`server`来讲，至关重要．那么如何来解决这个问题呐？

我们可以将每个定时事件封装成定时器，并使用某种容器类数据结构，比如：`链表，排序链表，时间轮，时间堆`来实现将所有的定时器串联起来，实现对定时事件的统一管理．

各个实现方式比较如下（摘自欢神讲座）：

<table> 
 <thead> 
  <tr> 
   <th>实现方式</th> 
   <th>AddTimer</th> 
   <th>DelTimer</th> 
   <th>ToDoTimer</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>基于链表</td> 
   <td>O(1)</td> 
   <td>O(n)</td> 
   <td>O(n)</td> 
  </tr> 
  <tr> 
   <td>基于排序链表</td> 
   <td>O(n)</td> 
   <td>O(1)</td> 
   <td>O(1)</td> 
  </tr> 
  <tr> 
   <td>基于最小堆</td> 
   <td>O(lgn)</td> 
   <td>O(1)</td> 
   <td>O(1)</td> 
  </tr> 
  <tr> 
   <td>基于时间轮</td> 
   <td>O(1)</td> 
   <td>O(1)</td> 
   <td>O(1)(前提是使用多个轮子来实现，不过单个轮子也要比O(n)快得多)</td> 
  </tr> 
 </tbody> 
</table>

**一般情况下最小堆(或者红黑树)的定时器已经足够用了**。如果在需要创建很多定时器的场景下，比如异步客户端，大量并发请求发出去的情况下，每个请求都需要一个 Timer 做超时处理，这样的场景，时间轮 (Timing-Wheel) 是最好的选择(Emmm，1987年提出来的) 。

时间轮实现方式比较多，可以很简单也可以很复杂。  
`io.netty.util.HashedWheelTimer` 是个不错的实现。  
如果不需要特别多的定时器，C++的话大多数情况下，下面的实现足够了:

typedef std::multimap<timer_key_t, timer_value_t> timer_map_t;

**还可以配合 timerfd 把时间处理统一到事件循环里面去 (since Linux 2.6.27)**

关于链表的定时器可能用的会比较少，因为毕竟效率有点低下，不过还是要分具体情况具体对待，比如`redis`就用的是链表来实现的定时器，可是人家还是那么强．

在本文，我们主要讨论的是两种比较高效的管理定时器的容器：**时间堆和时间轮**

首先，在介绍如何实现定时器之前，我们先来说一下`定时`的方法（定时是指在一段时间后触发某段代码的机制）主要有：

1.  socket选项SO\_RCVTIMEO 和 SO\_SNDTIMEO
2.  SIGALRM 信号
3.  I/O 复用系统调用的超时参数
4.  timerfd\_ \*系列函数

### socket选项SO\_RCVTIMEO 和 SO\_SNDTIMEO ###

#include <sys/types.h> /* See NOTES */
     #include <sys/socket.h>
    /*专门用来设置socket文件描述符属性的函数*/
     int setsockopt(int sockfd, int level, int optname,
                          const void *optval, socklen_t optlen);

第三个参数`optname`可用来指定超时接受(`SO_RCVTIMEO`)或者超时发送(`SO_SNDTIMEO`)，与其关联的第四个参数此时为`timeout`类型，指定具体的超时时间。如果是`connect()`函数，超时对应的`errno`会是`EINPROGRESS`。检测到此`errno`则关闭连接。这就是**根据系统调用的返回值来判断超时时间是否已到，据此处理定时任务即关闭连接**这两个选项分别用来设置socket接收数据和发送数据的超时时间,因此仅对于数据接收和发送相关的socket专用系统调用有效,这些系统调用包括

send,sendmsg,recv,recvmsg ,accept 和connect.

几个函数的返回值分别是：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXNoZW5neGlfcm9vdA_size_16_color_FFFFFF_t_70]

示例代码：

#include <sys/types.h>
    #include <sys/socket.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <stdlib.h>
    #include <assert.h>
    #include <stdio.h>
    #include <errno.h>
    #include <fcntl.h>
    #include <unistd.h>
    #include <string.h>
    
    int timeout_connect(const char *ip, int port, int time) // 5
    { 
    	int ret = 0;
    	struct sockaddr_in address;
    	bzero(&address, sizeof(address));
    	address.sin_family = AF_INET;
    	inet_pton(AF_INET, ip, &address.sin_addr);
    	address.sin_port = htons(port);
    
    	int sockfd = socket(PF_INET, SOCK_STREAM, 0);
    	assert(sockfd >= 0);
    
    	struct timeval timeout;
    	timeout.tv_sec = time;
    	timeout.tv_usec = 0;
    	socklen_t len = sizeof(timeout);
    	ret = setsockopt(sockfd, SOL_SOCKET, SO_SNDTIMEO, &timeout, len);
    	assert(ret != -1);
    
    	ret = connect(sockfd, (struct sockaddr *)&address, sizeof(address));
    	if (ret == -1)
    	{ 
    		if (errno == EINPROGRESS)
    		{ 
    			printf("connecting timeout\n");
    			return -1;
    		}
    		printf("error occur when connecting to server\n");
    		return -1;
    	}
    
    	return sockfd;
    }
    
    int main(int argc, char *argv[])
    { 
    	if (argc <= 2)
    	{ 
    		printf("usage: %s ip_address port_number\n", basename(argv[0]));
    		return 1;
    	}
    	const char *ip = argv[1];
    	int port = atoi(argv[2]);
    
    	int sockfd = timeout_connect(ip, port, 5);
    	if (sockfd < 0)
    	{ 
    		return 1;
    	}
    	return 0;
    }

测试见：[https://blog.csdn.net/liushengxi\_root/article/details/86062661][https_blog.csdn.net_liushengxi_root_article_details_86062661]

### SIGALRM 信号(在多线程程序中处理信号是非常麻烦的事情，应当尽量避免，原因有时间再来谈) ###

### I/O 复用系统调用的超时参数（精度太低，毫秒） ###

### timerfd\_ \*系列函数（新内核使用，现在都使用这个，原因精度高(纳秒),能统一到复用框架中，统一处理I/O事件和超时事件） ###

先给个不统一的看看：

void Event::run()
    { 
        while (true)
        { 
            /*注意：该处等待时间不能填-1，否则下面定时器无法周期性触发*/
            epoll_wait(....);
    
            /*处理网络事件*/
            do something ...
    
            now = get_now_time();
    
            /*all_timer管理所有需要触发的timer，按照触发时间，从小至大排序*/
            for item in all_timer:
                if item.expire_time <= now:
                    /*调用timer的处理函数*/
                    item.call_back()
                else
                    /*由于定时器按照触发时间排序，该item之后的timer均未到触发时间*/
                    break;
        }
    }
    该方案设计简单、容易实现，但是与事件驱动框架不协调，显得比较怪异。

然后就是`timerfd`的使用：

#include <sys/timerfd.h>
    #include <sys/epoll.h>
    #include <unistd.h>
    #include <stdint.h>
    #include <iostream>
    using namespace std;
    
    const int EPOLL_SIZE = 10;
    
    int main(int argc, char *argv[])
    { 
        int tfd, epfd, nfds;
        struct epoll_event event;
        struct epoll_event events[EPOLL_SIZE];
    
        //创建timerfd， CLOCK_REALTIME为绝对时间，TFD_NONBLOCK为非阻塞
        tfd = timerfd_create(CLOCK_REALTIME, TFD_NONBLOCK);
        if (tfd < 0)
        { 
            cerr << "timerfd_create error!" << endl;
            return -1;
        }
        struct timespec startTime, intervalTime; //　intervalTime　间隔时间
        startTime.tv_sec = 0;
        startTime.tv_nsec = 1; //相当于立即到达超时时间
    
        intervalTime.tv_sec = 3; //首次超时后，每三秒超时一次,周期时间
        intervalTime.tv_nsec = 0;
        struct itimerspec newValue;
        //it_value 是首次超时时间，一般需要填写从clock_gettime获取的时间，并加上要超时的时间。
        newValue.it_value = startTime;
        newValue.it_interval = intervalTime;
    
        //设置超时时间，且为相对时间
        if (timerfd_settime(tfd, 0, &newValue, NULL) < 0)
        { 
            cerr << "timerfd_settime error!" << endl;
            return -1;
        }
        //用epoll来监听描述符
        epfd = epoll_create(EPOLL_SIZE);
        if (epfd < 0)
        { 
            cerr << "epoll_create error!" << endl;
            return -1;
        }
    
        event.data.fd = tfd;
        event.events = EPOLLIN;
        if (epoll_ctl(epfd, EPOLL_CTL_ADD, tfd, &event) < 0)
        { 
            cerr << "epoll_ctl error!" << endl;
            return -1;
        }
    
        uint64_t count = 0;
        while (1)
        { 
            //非阻塞等待
            nfds = epoll_wait(epfd, events, EPOLL_SIZE, 0);
            if (nfds == 0)
                continue;
            for (int i = 0; i < nfds; i++)
            { 
                if (events[i].events & EPOLLIN)
                { 
                    uint64_t data;
                    read(events[i].data.fd, &data, sizeof(uint64_t));
                    count += data;
                    cout << "read: " << data << ", timer count: " << count << endl;
                    //在这里就可以设置　callback　函数
                }
            }
        }
        return 0;
    }

统一起来了

### 时间堆的实现原理：（std::priority\_queue<TimerPtr, std::vector, std::greater> que） ###

`timer.h` (client\_data类和定时器类的实现)

#ifndef _TIMER_H
    #define _TIMER_H
    #include <time.h>
    #include <sys/socket.h>
    #include <sys/types.h>
    #include <arpa/inet.h>
    const int BUFFER_SIZE = 64;
    
    class Timer;
    /*用户数据结构*/
    class client_data
    { 
      public:
    	struct sockaddr_in address;
    	int sockfd;
    	char buf[BUFFER_SIZE];
    	Timer *timer;
    };
    /*定时器类*/
    class Timer
    { 
      public:
    	Timer(/* args */)
    	{ 
    	}
    	~Timer()
    	{ 
    	}
    	bool operator>(const Timer a) const //重载　＞　
    	{ 
    		return expire > a.expire;
    	}
    
    	time_t expire;					//任务的超时事件
    	void (*cb_func)(client_data *); // 回调函数
    	client_data *user_data;			//回调函数处理的客户数据，由定时器的执行者传递给回调函数
    };
    #endif

`lst_timer.h`(时间堆的实现)

#ifndef _LST_TIMER_H
    #define _LST_TIMER_H
    #include "timer.h"
    #include <queue>
    #include <memory>
    #include <vector>
    #include <iostream>
    
    class ListTimer
    { 
      public:
    	using TimerPtr = std::shared_ptr<class Timer>;
    	ListTimer(/* args */)
    	{ 
    	}
    	~ListTimer()
    	{ 
    		auto size = que.size();
    		for (int i = 0; i < size; i++)
    		{ 
    			que.pop();
    		}
    	}
    	/*添加定时器*/
    	void AddTimer(const TimerPtr &timer)
    	{ 
    		if (!timer)
    			return;
    		que.push(timer);
    	}
    	/*删除定时器（肯定是优先队列的第一个元素）想想为什么？？*/
    	void DelTimer(Timer *timer)
    	{ 
    		if (!timer)
    			return;
    		que.pop();
    	}
    	/*SIGALRM 信号被触发，就在信号处理函数中执行一次tick函数，以处理到期的任务*/
    	void tick()
    	{ 
    		if (que.size() == 0)
    			return;
    		std::cout << "time tick " << std::endl;
    		time_t curr = time(NULL);
    		/*从头处理到期的定时器，直到遇到一个尚未到期的定时器*/
    		while (!que.empty())
    		{ 
    			auto tt = que.top();
    			if (curr < tt->expire)
    			{ 
    				break;
    			}
    			//堆顶元素时间到期　
    			else
    			{ 
    				tt->cb_func(tt->user_data);
    				/*执行完就把他从队列中删除*/
    				que.pop();
    			}
    		}
    	}
    	time_t TopTime()
    	{ 
    		if (que.size() == 0)
    			return 5;
    		auto curr = time(NULL);
    		return que.top()->expire - curr;
    	}
    
      private:
    	/* data */
    	std::priority_queue<TimerPtr, std::vector<TimerPtr>, std::greater<TimerPtr>> que;
    };
    
    #endif

`main.cpp`

#include <sys/types.h>
    #include <sys/socket.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <assert.h>
    #include <stdio.h>
    #include <signal.h>
    #include <unistd.h>
    #include <errno.h>
    #include <string.h>
    #include <fcntl.h>
    #include <stdlib.h>
    #include <sys/epoll.h>
    #include <pthread.h>
    #include "lst_timer.h"
    
    #define FD_LIMIT 65535
    #define MAX_EVENT_NUMBER 1024
    #define TIMESLOT 5
    
    static int pipefd[2];
    static class ListTimer timer_lst;
    static int epollfd = 0;
    
    int setnonblocking(int fd)
    { 
    	int old_option = fcntl(fd, F_GETFL);
    	int new_option = old_option | O_NONBLOCK;
    	fcntl(fd, F_SETFL, new_option);
    	return old_option;
    }
    
    void addfd(int epollfd, int fd)
    { 
    	epoll_event event;
    	event.data.fd = fd;
    	event.events = EPOLLIN | EPOLLET;
    	epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);
    	setnonblocking(fd);
    }
    
    void sig_handler(int sig)
    { 
    	int save_errno = errno;
    	int msg = sig;
    	send(pipefd[1], (char *)&msg, 1, 0);
    	errno = save_errno;
    }
    
    void addsig(int sig)
    { 
    	struct sigaction sa;
    	memset(&sa, '\0', sizeof(sa));
    	sa.sa_handler = sig_handler;
    	sa.sa_flags |= SA_RESTART;
    	sigfillset(&sa.sa_mask);
    	assert(sigaction(sig, &sa, NULL) != -1);
    }
    
    void timer_handler()
    { 
    	timer_lst.tick();
    	alarm(TIMESLOT);
    }
    
    void cb_func(client_data *user_data)
    { 
    	epoll_ctl(epollfd, EPOLL_CTL_DEL, user_data->sockfd, 0);
    	assert(user_data);
    	close(user_data->sockfd);
    	printf("close fd %d\n", user_data->sockfd);
    }
    
    int main(int argc, char *argv[])
    { 
    	if (argc <= 2)
    	{ 
    		printf("usage: %s ip_address port_number\n", basename(argv[0]));
    		return 1;
    	}
    	const char *ip = argv[1];
    	int port = atoi(argv[2]);
    
    	int ret = 0;
    	struct sockaddr_in address;
    	bzero(&address, sizeof(address));
    	address.sin_family = AF_INET;
    	inet_pton(AF_INET, ip, &address.sin_addr);
    	address.sin_port = htons(port);
    
    	int listenfd = socket(PF_INET, SOCK_STREAM, 0);
    	assert(listenfd >= 0);
    
    	ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address));
    	assert(ret != -1);
    
    	ret = listen(listenfd, 5);
    	assert(ret != -1);
    
    	epoll_event events[MAX_EVENT_NUMBER];
    	int epollfd = epoll_create(5);
    	assert(epollfd != -1);
    	addfd(epollfd, listenfd);
    
    	ret = socketpair(PF_UNIX, SOCK_STREAM, 0, pipefd);
    	assert(ret != -1);
    	setnonblocking(pipefd[1]);
    	addfd(epollfd, pipefd[0]);
    
    	// add all the interesting signals here
    	addsig(SIGALRM);
    	addsig(SIGTERM);
    	bool stop_server = false;
    
    	client_data *users = new client_data[FD_LIMIT];
    	bool timeout = false;
    	alarm(timer_lst.TopTime());
    
    	while (!stop_server)
    	{ 
    		int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);
    		if ((number < 0) && (errno != EINTR))
    		{ 
    			printf("epoll failure\n");
    			break;
    		}
    
    		for (int i = 0; i < number; i++)
    		{ 
    			int sockfd = events[i].data.fd;
    			if (sockfd == listenfd)
    			{ 
    				struct sockaddr_in client_address;
    				socklen_t client_addrlength = sizeof(client_address);
    				int connfd = accept(listenfd, (struct sockaddr *)&client_address, &client_addrlength);
    				addfd(epollfd, connfd);
    				users[connfd].address = client_address;
    				users[connfd].sockfd = connfd;
    				// Timer *timer = new Timer;
    				std::shared_ptr<Timer> timer = std::make_shared<Timer>();
    				timer->user_data = &users[connfd];
    				timer->cb_func = cb_func;
    				time_t cur = time(NULL);
    				timer->expire = cur + 3 * 5;
    				users[connfd].timer = timer.get();
    				timer_lst.AddTimer(timer);
    			}
    			else if ((sockfd == pipefd[0]) && (events[i].events & EPOLLIN))
    			{ 
    				int sig;
    				char signals[1024];
    				ret = recv(pipefd[0], signals, sizeof(signals), 0);
    				if (ret == -1)
    				{ 
    					// handle the error
    					continue;
    				}
    				else if (ret == 0)
    				{ 
    					continue;
    				}
    				else
    				{ 
    					for (int i = 0; i < ret; ++i)
    					{ 
    						switch (signals[i])
    						{ 
    						case SIGALRM:
    						{ 
    							timeout = true;
    							break;
    						}
    						case SIGTERM:
    						{ 
    							stop_server = true;
    						}
    						}
    					}
    				}
    			}
    			else if (events[i].events & EPOLLIN)
    			{ 
    				memset(users[sockfd].buf, '\0', BUFFER_SIZE);
    				ret = recv(sockfd, users[sockfd].buf, BUFFER_SIZE - 1, 0);
    				printf("get %d bytes of client data %s from %d\n", ret, users[sockfd].buf, sockfd);
    				Timer *timer = users[sockfd].timer;
    				if (ret < 0)
    				{ 
    					if (errno != EAGAIN)
    					{ 
    						cb_func(&users[sockfd]);
    						if (timer)
    						{ 
    							timer_lst.DelTimer(timer);
    						}
    					}
    				}
    				else if (ret == 0)
    				{ 
    					cb_func(&users[sockfd]);
    					if (timer)
    					{ 
    						timer_lst.DelTimer(timer);
    					}
    				}
    				else
    				{ 
    					//send( sockfd, users[sockfd].buf, BUFFER_SIZE-1, 0 );
    					if (timer)
    					{ 
    						time_t cur = time(NULL);
    						timer->expire = cur + 3 * 5;
    						printf("adjust timer once\n");
    						// timer_lst.adjust_timer(timer);
    					}
    				}
    			}
    			else
    			{ 
    				// others
    			}
    		}
    
    		if (timeout)
    		{ 
    			timer_handler();
    			timeout = false;
    		}
    	}
    
    	close(listenfd);
    	close(pipefd[1]);
    	close(pipefd[0]);
    	delete[] users;
    	return 0;
    }

### 时间轮的实现原理 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXNoZW5neGlfcm9vdA_size_16_color_FFFFFF_t_70 1]  
如上图所示，就是一个时间轮的基本轮廓。一个轮子上有很多槽slot,每一个槽指向一个定时器链表，这个链表是无序的。时间轮每转动一步就会指向下一个槽，其实也可以理解为一个滴答时间成为时间轮的槽间隔si (slot interval)。它实际上就是心跳时间。如果该时间轮有n个槽，因此它运转一周的时间是n\*si.

如果现在指针指向槽cs,我们此时要添加一个定时器时间为ti的定时器，则该定时器将被插入槽ts对应的链表

ts = (cs+ti/si)%N

基于排序链表的定时器使用唯一的链表来管理所有定时器，所以插入定时器的数目越多，效率就会越低，而时间轮则是利用哈希表的思想，将定时器散列到不同的链表中，这样每条链表上的数据就会显著少于原来排序链表的数目。插入操作的效率基本不受定时器数目的影响(不需要排序，直接散列到一个链表上)。

显然要提高时间轮的精度，就要使si(slot interval)足够小，要提高其执行效率则要N要足够大

去看Linux高性能服务器编程

void tick()
        { 
            tw_timer* tmp = slots[cur_slot];
            printf( "current slot is %d\n", cur_slot );
            while( tmp )
            { 
                printf( "tick the timer once\n" );
                if( tmp->rotation > 0 )　// 判断转动的圈数是否到达，如果到达，就执行，如果没有到达就减一即可
                { 
                    tmp->rotation--;
                    tmp = tmp->next;
                }
                else
                { 
                    tmp->cb_func( tmp->user_data );
                    if( tmp == slots[cur_slot] )
                    { 
                        printf( "delete header in cur_slot\n" );
                        slots[cur_slot] = tmp->next;
                        delete tmp;
                        if( slots[cur_slot] )
                        { 
                            slots[cur_slot]->prev = NULL;
                        }
                        tmp = slots[cur_slot];
                    }
                    else
                    { 
                        tmp->prev->next = tmp->next;
                        if( tmp->next )
                        { 
                            tmp->next->prev = tmp->prev;
                        }
                        tw_timer* tmp2 = tmp->next;
                        delete tmp;
                        tmp = tmp2;
                    }
                }
            }
            cur_slot = ++cur_slot % N;
        }

### 内核时间轮的实现 ###

ing…

[Link 1]: https://www.zhihu.com/question/27364219
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXNoZW5neGlfcm9vdA_size_16_color_FFFFFF_t_70]: /images/20220330/375898a559ba48248afa2461f88bcf63.png
[https_blog.csdn.net_liushengxi_root_article_details_86062661]: https://blog.csdn.net/liushengxi_root/article/details/86062661
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXNoZW5neGlfcm9vdA_size_16_color_FFFFFF_t_70 1]: /images/20220330/58b032f1462d4429bddc610c0640900b.png