定时器模块
定时器模块
基础知识
- 非活跃:指客户端与服务器建立连接后,长时间不交换数据,一直占用服务器的文件描述符,造成连接资源的浪费
- 定时事件:指固定一段时间之后触发某段代码,由该段代码处理一个事件(举例:从内核事件表删除事件,并关闭文件描述符,释放连接资源)
- 定时器:指利用结构体或其他形式,将多种定时事件进行封装起来(本项目中只涉及一种定时事件,即定时检查非活跃连接,将定时事件与连接资源封装为一个结构体定时器)
- 定时器容器:指使用某种容器类数据结构,将上述多个定时器组合起来,便于对定时事件统一管理(本项目中使用升序链表将所有定时器串联组织起来)
- 统一事件源:将信号事件与其他事件一起处理。具体地,信号处理函数使用管道将信号传递给主循环,信号处理函数往管道的写端写入信号值,主循环则从管道的读端读出信号值,使用I/O复用系统调用来监听管道读端的可读事件,这样信号事件与其他文件描述符都可以通过epoll来监测,从而实现统一处理
模块功能
本项目中,服务器主循环为每一个连接创建一个定时器,并对每个连接进行定时。另外,利用升序时间链表容器将所有定时器串联起来,若主循环接收到定时通知,则在链表中依次执行定时任务。
三种定时的方法
Linux提供了三种定时的方法:
- socket选项SO_RECVTIMEO和SO_SNDTIMEO
- socket选项SO_SNDTIMEO:用来设置socket发送数据的超时时间,若该时间内未接收到任何数据,则socket返回一个错误码
- socket选项SO_SNDTIMEO:用来设置socket接收数据的超时时间,若该时间内无法发送完所有数据,则发送操作失败并返回一个错误
使用SO_SNDTIMEO选项时,需要先将socket设置为非阻塞模式,否则会忽略超时设置并阻塞发送操作
通过函数setsockopt()可以设置SO_SNDTIMEO和SO_RECVTIMEO选项
int setsockopt( int socket, int level, int option_name,const void *option_value, size_t ,ption_len)
//socket:socket描述符
//level:被设置的选项的级别。如果要在socket级别上设置选项,则必须将level设置为SOL_SOCKET
- SIGALRM信号
- SIGALRM是一种处理定时器信号的信号,代表在已定义的时间间隔后发生的警告信号,可用于定时器、计时器和轮询器等
- 通常使用系统调用alarm()来设置发生SIGALRM的时间间隔
- I/O复用系统调用的超时参数
- I/O复用系统调用自带的超时参数,既能统一处理信号和I/O事件,也能统一处理定时事件
- I/O复用系统调用可能在超时时间到期之前就返回(有I/O事件发生),如果要利用它们来定时,就需要不断更新定时参数以反映剩余的时间
三种方法没有一劳永逸的应用场景,没有绝对的优劣。在本项目中主要使用的是SIGALRM信号
具体地,利用alarm函数周期性地触发SIGALRM信号,信号处理函数利用管道通知主循环,主循环接收到该信号后对升序链表上所有定时器进行处理,若该段时间内没有交换数据,则将该连接关闭,释放所占用的资源。该模块主要分为两部分,其一为定时方法与信号通知流程,其二为定时器及其容器设计与定时任务的处理
基础API
SIGALRM、SIGTERM信号
SIGALRM:由alarm系统调用产生的timer时钟信号
SIGTERM:终端发送的终止信号
sigaction结构体、检查或修改与指定信号相关联的处理动作——sigaction()函数
struct sigaction {
void (*sa_handler)(int); //函数指针,指向旧的信号处理函数
void (*sa_sigaction)(int, siginfo_t *, void *); //新的信号处理函数
sigset_t sa_mask; //信号阻塞集,指定在信号处理函数执行期间需要被屏蔽的信号
int sa_flags; //信号的处理方式
void (*sa_restorer)(void); //已弃用
}
#include <signal.h>
int sigaction(int signum, const struct sigaction *act, struct sigaction *oldact)
// signum:要操作的信号
// act:要设置的对信号的新处理方式
// oldact:原来对信号的处理方式
// return:成功返回0,出现错误则返回-1
sa_flags指定信号的处理方式,可以是以下几种的“按位或”组合
SA_RESTART:使被信号打断的系统调用自动重新发起
SA_NOCLDSTOP:使父进程在它的子进程暂停或继续运行时不会收到 SIGCHLD 信号
SA_NOCLDWAIT:使父进程在它的子进程退出时不会收到 SIGCHLD 信号,这时子进程如果退出也不会成为僵尸进程
SA_NODEFER:使对信号的屏蔽无效,即在信号处理函数执行期间仍能发出这个信号
SA_RESETHAND:信号处理之后重新设置为默认的处理方式
SA_SIGINFO:使用 sa_sigaction 成员而不是 sa_handler 作为信号处理函数
- 初始化自定义信号集——sigfillset()函数
- 实际使用中,常用sigemptyset()将信号集清空,再使用sigaddset()想信号集中添加所需的特定信号以达到更精细的控制信号的目的
#include <signal.h>
int sigfillset(sigset_t *set)
//将参数set信号集初始化(信号集中所有标志位置1),然后把所有的信号加入到此信号集中
//set:指向信号集合的指针
//return:成功返回0,出现错误则返回-1
- 设置信号传送闹钟——alarm()函数
- 如果未设置信号SIGALRM的处理函数,则alarm()默认终止进程
#include <unisted.h>
unsigned int alarm(unsigned int seconds)
//设置信号SIGALRM在经过参数seconds秒数后发送给目前的进程
//seconds:要设定的定时时间,以秒为单位。在alarm调用后开始计时,超过该时间将触发SIGALRM信号
//return:返回当前进程之前设置的定时器剩余秒数
- 创建套接字对用于通信——socketpair()函数
- 用SOCK_STREAM建立的套接字对是管道流,建立的通道是双向的,支持全双工通信
#include <sys/types.h>
#include <sys/socket.h>
int socketpair(int domain, int type, int protocol, int sv[2]) //创建一对无名的、相互连接的套接字
//domain:协议族,只能为PF_UNIX或者AF_UNIX
//type:协议,可以是SOCK_STREAM(基于TCP)或SOCK_DGRAM(基于UDP)
//protocol:类型,只能为0
//sv[2]:套接字柄对,该两个句柄作用相同,均能进行读写双向操作
//return:创建成功返回0,失败返回-1
信号处理机制
Linux下的信号采用异步处理机制,信号处理函数与当前进程是两条不同的执行路线。当进程收到信号时,操作系统会中断进程当前的正常流程,转而进入信号处理函数执行操作,完成后再返回中断的地方继续执行
为避免信号竞态现象发生,信号处理期间系统不会再次触发它。所以,为确保该信号不被屏蔽太久,信号处理函数需要尽可能快地执行完毕。一般的信号处理函数需要处理该信号对应的逻辑,当该逻辑比较复杂时,信号处理函数执行时间过长,会导致信号屏蔽太久。为了解决该问题,本项目中信号处理函数仅发送信号通知程序主循环,将信号对应的处理逻辑放在程序主循环中,由主循环执行信号对应的逻辑代码。
信号的接收
接收信号的任务不是由用户进程来完成的,而是由内核代理的。当内核接收到信号后,会将其放到对应进程的信号队列中,同时向进程发送一个中断,使其陷入内核态。注意,此时信号还只是在队列中,对进程来说暂时是不知道有信号到来的
信号的检测
进程陷入内核态后,有两种场景会对信号进行检测:
进程从内核态返回到用户态前进行信号检测
进程在内核态中,从睡眠状态被唤醒的时候进行信号检测
当发现有新信号时,便会进入下一步,信号的处理
信号的处理
信号处理函数是在用户态上的
(内核态)调用处理函数前,内核将当前内核栈的内容备份拷贝到用户栈上,并且修改指令寄存器(eip)将其指向信号处理函数
(用户态)进程返回到用户态中,执行信号处理函数
(内核态)信号处理函数执行完成后,需要返回内核态检查是否有其他信号未处理
(用户态)所有信号都处理完成,则将内核栈恢复(从步骤1中的用户栈备份拷贝回来),同时恢复指令寄存器(eip)将其指向中断前的运行位置,最后回到用户态继续执行进程
信号处理函数
//信号处理函数中仅仅通过管道发送信号值,不处理信号对应的逻辑,缩短异步执行时间,减少对主程序的影响
void Utils::sig_handler(int sig) {
//为保证函数的可重入性,保留原来的errno
//可重入性表示中断后再次进入该函数,环境变量与之前相同,不会丢失数据
int save_errno = errno;
int msg = sig;
send(u_pipefd[1], (char*)&msg, 1, 0); //将信号值从管道写端写入,传输字符类型,而非整型
errno = save_errno; //将原来的errno赋值为当前的errno
}
信号处理函数中仅通过管道发送信号值,不处理信号对应的逻辑,缩短异步执行时间,减少对主程序的影响
//设置信号函数
void Utils::addsig(int sig, void(handler)(int), bool restart) { //项目中设置信号函数,仅关注SIGTERM和SIGALRM两个信号
struct sigaction sa; //创建sigaction结构体变量
memset(&sa, '\0', sizeof(sa));
sa.sa_handler = handler; //信号处理函数中仅仅发送信号值,不做对应逻辑处理
if (restart)
sa.sa_flags |= SA_RESTART;
sigfillset(&sa.sa_mask); //将所有信号添加到信号集中
assert(sigaction(sig, &sa, NULL) != -1); //执行sigaction函数
}
项目中设置信号函数,仅关注SIGTERM和AIGALRM两个信号
信号通知逻辑
- 创建管道,其中管道写端写入信号值,管道读端通过I/O复用系统监测读事件
- 设置信号处理函数SIGALRM(时间到了触发)和SIGTERM(kill会触发)
- 通过struct sigaction结构体和sigaction函数注册信号捕捉函数
- 在结构体的handler参数设置信号处理函数,具体地,从管道写端写入信号的名字
- 利用I/O复用系统监听管道读端文件描述符的可读事件
- 信息值传递给主循环,主循环再根据接收到的信号值执行目标信号对应的逻辑代码
定时器设计
定时器类的定义
项目中将连接资源、定时事件和超时时间封装为定时类
- 连接资源包括客户端套接字地址、文件描述符和定时器
- 定时事件为回调函数,将其封装起来由用户自定义,这里是删除非活动socket上的注册事件,并关闭
- 定时器超时时间 = 浏览器和服务器连接时刻 + 固定时间(TIMESLOT),这里定时器使用绝对时间作为超时值
//连接资源结构体成员需要用到定时器类,需要前向声明
class util_timer;
struct client_data { //开辟用户socket结构 对应于最大处理fd
sockaddr_in address; //客户端socket地址
int sockfd; //socket文件描述符
util_timer* timer; //定时器
};
class util_timer //定时器类
{
public:
util_timer() : prev(NULL), next(NULL) {}
public:
time_t expire; //超时时间
void (*cb_func)(client_data*); //回调函数
client_data* user_data; //连接资源
util_timer* prev; //前向定时器
util_timer* next; //后继定时器
};
定时事件,具体地,从内核事件表删除事件,关闭文件描述符,释放连接资源
//定时器回调函数
void cb_func(client_data* user_data){
epoll_ctl(Utils::u_epollfd, EPOLL_CTL_DEL, user_data->sockfd, 0); //删除非活动连接在socket上的注册事件
assert(user_data);
close(user_data->sockfd); //关闭文件描述符
http_conn::m_user_count--; //减少连接数
}
定时器的创建与销毁
该项目的定时器采用升序双向链表作为容器,具体地,为每个连接创建一个定时器,将其添加到链表中,并按照超时时间升序排序。
在本项目中,定时器的创建与销毁采用了RAII思想,在构造函数与析构函数中完成定时器容器的创建与销毁
sort_timer_lst::sort_timer_lst(){
head = NULL;
tail = NULL;
}
//常规销毁链表
sort_timer_lst::~sort_timer_lst() {
util_timer* tmp = head;
while (tmp) {
head = tmp->next;
delete tmp;
tmp = head;
}
}
添加定时任务
添加定时任务即将新连接的定时器添加到链表中。若当前链表中只有头尾结点(链表为空),则直接使头尾结点指向该定时器
- 若新的定时器(当前正在添加到链表中的定时器)的超时时间小于当前头部结点,则直接将新的定时器结点作为头部结点
void sort_timer_lst::add_timer(util_timer* timer) { //添加定时器,内部调用私有成员add_timer
if (!timer){
return;
}
if (!head){
head = tail = timer;
return;
}
if (timer->expire < head->expire) { //如果新的定时器超时时间小于当前头部结点,则直接将当前定时器结点作为头部结点
timer->next = head;
head->prev = timer;
head = timer;
return;
}
add_timer(timer, head); //否则调用私有成员,调整内部结点
}
- 若新的定时器结点的超时时间大于等于当前头部结点,则需要插入到链表中,采用双向链表的插入操作
//私有成员,被公有成员add_timer和adjust_time调用,主要用于调整链表内部结点
void sort_timer_lst::add_timer(util_timer* timer, util_timer* lst_head) { //主要用于调整链表内部结点
util_timer* prev = lst_head;
util_timer* tmp = prev->next;
while (tmp) { //遍历当前结点之后的链表,按照超时时间找到目标定时器对应的位置,常规双向链表插入操作
if (timer->expire < tmp->expire){
prev->next = timer;
timer->next = tmp;
tmp->prev = timer;
timer->prev = prev;
break;
}
prev = tmp;
tmp = tmp->next;
}
if (!tmp) { //遍历完发现,目标定时器需要放到尾结点处
prev->next = timer;
timer->prev = prev;
timer->next = NULL;
tail = timer;
}
}
任务超时时间调整
当定时任务发生变化,调整对应定时器在链表中的位置
- 客户端在设定时间内有数据收发,则当前时刻对该定时器重新设定时间
- 被调整的目标定时器在尾部,或定时器新的超时时间仍然小于下一个定时器的超时,则不用调整;否则先将定时器从链表取出,重新插入链表
void sort_timer_lst::adjust_timer(util_timer* timer){ //调整定时器,任务发生变化时,调整定时器在链表中的位置
if (!timer){
return;
}
util_timer* tmp = timer->next;
if (!tmp || (timer->expire < tmp->expire)){
return;
}
if (timer == head) { //被调整定时器是链表头结点,将定时器取出,重新插入
head = head->next;
head->prev = NULL;
timer->next = NULL;
add_timer(timer, head);
}
else { //被调整定时器在内部,将定时器取出,重新插入
timer->prev->next = timer->next;
timer->next->prev = timer->prev;
add_timer(timer, timer->next);
}
}
删除定时任务
定时器超时,则将其从链表中删除
void sort_timer_lst::del_timer(util_timer* timer) { //删除定时器
if (!timer){
return;
}
if ((timer == head) && (timer == tail)) { //链表中只有一个定时器,需要删除该定时器
delete timer;
head = NULL;
tail = NULL;
return;
}
if (timer == head) { //被删除的定时器为头结点
head = head->next;
head->prev = NULL;
delete timer;
return;
}
if (timer == tail){ //被删除的定时器为尾结点
tail = tail->prev;
tail->next = NULL;
delete timer;
return;
}
timer->prev->next = timer->next; //被删除的定时器在链表内部,常规链表结点删除
timer->next->prev = timer->prev;
delete timer;
}
定时任务处理函数
使用统一事件源,SIGALRM信号每次被触发,主循环中调用一次定时任务处理函数,处理链表容器中到期的定时器
具体逻辑如下:
- 遍历定时器升序链表容器,从头结点开始依次处理每个定时器,直到遇到尚未到期的定时器
- 若当前时间小于定时器超时时间,跳出循环,即未找到到期的定时器;若当前时间大于定时器超时时间,即找到了到期的定时器,执行回调函数,然后将它从链表中删除,然后继续遍历
void sort_timer_lst::tick(){
if (!head){
return;
}
time_t cur = time(NULL); //获取当前时间
util_timer* tmp = head;
while (tmp) { //遍历定时器链表
if (cur < tmp->expire) { //链表容器为升序排列,当前时间小于定时器的超时时间,后面的定时器也没有到期
break;
}
tmp->cb_func(tmp->user_data); //当前定时器到期,则调用回调函数,执行定时事件
head = tmp->next; //将处理后的定时器从链表容器中删除,并重置头结点
if (head){
head->prev = NULL;
}
delete tmp;
tmp = head;
}
}
定时器的使用(webserver.cpp)
客户端与服务器连接时,创建该连接对应的定时器,并将定时器添加到链表上
void WebServer::timer(int connfd, struct sockaddr_in client_address){
users[connfd].init(connfd, client_address, m_root, m_CONNTrigmode, m_close_log, m_user, m_passWord, m_databaseName);
//初始化client_data数据
//创建定时器,设置回调函数和超时时间,绑定用户数据,将定时器添加到链表中
users_timer[connfd].address = client_address;
users_timer[connfd].sockfd = connfd;
util_timer *timer = new util_timer; //创建定时器
timer->user_data = &users_timer[connfd]; //绑定用户数据
timer->cb_func = cb_func; //设置回调函数
time_t cur = time(NULL);
timer->expire = cur + 3 * TIMESLOT; //设置超时时间
users_timer[connfd].timer = timer;
utils.m_timer_lst.add_timer(timer); //添加定时器到链表中
}
处理异常事件时,执行定时事件,服务器关闭连接,并将定时器从链表中移除
void WebServer::eventLoop(){
bool timeout = false; //超时默认为false
bool stop_server = false;
while (!stop_server){
int number = epoll_wait(m_epollfd, events, MAX_EVENT_NUMBER, -1); //监测发生事件的文件描述符
if (number < 0 && errno != EINTR){
LOG_ERROR("%s", "epoll failure");
break;
}
for (int i = 0; i < number; i++) { //轮询事件描述符
int sockfd = events[i].data.fd;
//处理新到的客户连接
if (sockfd == m_listenfd){
bool flag = dealclinetdata();
if (false == flag)
continue;
}
else if (events[i].events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR)) { //处理异常事件
//服务器端关闭连接,移除对应的定时器
util_timer* timer = users_timer[sockfd].timer;
deal_timer(timer, sockfd);
}
//处理定时器信号
else if ((sockfd == m_pipefd[0]) && (events[i].events & EPOLLIN)) { //管道读端对应文件描述符发生读事件
bool flag = dealwithsignal(timeout, stop_server);
if (false == flag)
LOG_ERROR("%s", "dealclientdata failure");
}
//处理客户连接上接收到的数据
else if (events[i].events & EPOLLIN){
dealwithread(sockfd);
}
else if (events[i].events & EPOLLOUT){
dealwithwrite(sockfd);
}
}
if (timeout){ //处理定时器为非必须事件,收到信号并不是立马处理,而是完成读写事件后再进行处理
utils.timer_handler();
LOG_INFO("%s", "timer tick");
timeout = false;
}
}
}
其中,处理定时器信号部分中的dealwithsignal()内容如下
bool WebServer::dealwithsignal(bool& timeout, bool& stop_server){
int ret = 0;
int sig;
char signals[1024];
ret = recv(m_pipefd[0], signals, sizeof(signals), 0); //从管道读端读出信号值,成功返回字节数,失败返回-1
if (ret == -1){
return false;
}
else if (ret == 0){
return false;
}
else{
for (int i = 0; i < ret; ++i){
switch (signals[i]){
case SIGALRM: //接收到SIGALRM信号,timeout设置为true
{
timeout = true;
break;
}
case SIGTERM: //接收到SIGTERM信号,终止进程运行,stop_server设置为true
{
stop_server = true;
break;
}
}
}
}
return true;
}
处理客户连接上接收到的数据,其函数内容为
void WebServer::dealwithread(int sockfd){
util_timer* timer = users_timer[sockfd].timer;
//reactor
if (1 == m_actormodel){
if (timer){
adjust_timer(timer);
}
//若监测到读事件,将该事件放入请求队列
m_pool->append(users + sockfd, 0);
while (true) {
if (1 == users[sockfd].improv){
if (1 == users[sockfd].timer_flag){
deal_timer(timer, sockfd);
users[sockfd].timer_flag = 0;
}
users[sockfd].improv = 0;
break;
}
}
}
else{
//proactor
if (users[sockfd].read_once()){
LOG_INFO("deal with the client(%s)", inet_ntoa(users[sockfd].get_address()->sin_addr));
//若监测到读事件,将该事件放入请求队列
m_pool->append_p(users + sockfd);
if (timer){
adjust_timer(timer); //有数据传输,则将定时器向后延迟3个单位,并对新的定时器在链表上的位置进行调整
}
}
else {
deal_timer(timer, sockfd); //服务器端关闭连接,移除对应的定时器
}
}
}
void WebServer::dealwithwrite(int sockfd){
util_timer* timer = users_timer[sockfd].timer;
//reactor
if (1 == m_actormodel){
if (timer) {
adjust_timer(timer);
}
m_pool->append(users + sockfd, 1);
while (true){
if (1 == users[sockfd].improv){
if (1 == users[sockfd].timer_flag){
deal_timer(timer, sockfd);
users[sockfd].timer_flag = 0;
}
users[sockfd].improv = 0;
break;
}
}
}
else{
//proactor
if (users[sockfd].write()){
LOG_INFO("send data to the client(%s)", inet_ntoa(users[sockfd].get_address()->sin_addr));
if (timer){
adjust_timer(timer); //有数据传输,则将定时器向后延迟3个单位,并对新的定时器在链表上的位置进行调整
}
}
else{
deal_timer(timer, sockfd); //服务器端关闭连接,移除对应的定时器
}
}
}