GroupBlog

文章目录

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

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

玩家下线

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

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

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

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

一般情况下最小堆(或者红黑树)的定时器已经足够用了。如果在需要创建很多定时器的场景下，比如异步客户端，大量并发请求发出去的情况下，每个请求都需要一个 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就用的是链表来实现的定时器，可是人家还是那么强．

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

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

socket选项SO_RCVTIMEO 和 SO_SNDTIMEO
SIGALRM 信号
I/O 复用系统调用的超时参数
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.

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

在这里插入图片描述

示例代码：

#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

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;
}

时间轮的实现原理

在这里插入图片描述
如上图所示，就是一个时间轮的基本轮廓。一个轮子上有很多槽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…