【阅读】《Linux高性能服务器编程》——第十一章

• 11.1 socket选项SO_RCVTIMEO和SO_SNDTIMEO
• 11.2 SIGALRM信号
• • 11.2.1 基于升序链表的定时器
  • 11.2.2 处理非活动连接
• 11.3 I/O复用系统调用的超时参数
• 11.4 高性能定时器
• • 11.4.1 时间轮
  • 11.4.2 时间堆

  socket选项的SO_RCVTIMEO和SO_SNDTIMEO分别用来设置socket接收数据超时事件和发送数据的超时时间，仅对send、sendmsg、recv、recvmsg、accpet和connect有效。

实例：设置connect超时时间：

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 

int timeout_connect(const char* ip, int port, int time){
    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_STERAM, 0);
    assert(sockfd>0);
    // 通过选项设置超时时间类型为timeval
    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){
        // 超时错误号对应的EINPROGRESS
        if(errno == EINPROGRESS){
            printf("connecting timeout, process timeoutn")
            return -1;
        }
        printf("error occur when connectiong to servern");
        return -1;
    }
    return sockfd;
}

int main(int argc, char* argv[]){
    if(argc<=2){
        printf("usage: %s ip_adress port_numbern", basename(argv[0]));
        return 1;
    }
    const char* ip = argv[1];
    int port = atoi(argv[2]);

    int sockfd = timeout_connect(ip, port, 10);
    if(sockfd < 0){
        return 1;
    }
    return 0;
}

  由alarm和setitimer函数设置的闹钟一旦超时，将触发SIGALRM信号，可利用该信号的信号处理函数来处理定时任务。一般而言，SIGALRM信号按照固定的频率生成，即由alarm或setitimer函数设置的定时周期T保持不变。如果某个定时任务的超时时间不是T的整数倍，则其被执行的时间和预期的时间有偏差，可通过T反应定时精度。

  定时器通常至少包含两个成员：超时时间和任务回调函数。如果用链表作为容器来串联所有的定时器，则每个定时器还要包含指向下一个定时器的指针成员(单向链表)。

实例：升序定时器链表

#ifndef LST_TIMER
#define LST_TIMER

#include 
#include 

#define BUFFER_SIZE 64

class util_timer;       // 前向声明

// 用户数据结构：客户都安socket地址、socket文件描述符、读缓存和定时器
struct client_data{
    sockaddr_in address;
    int sockfd;
    char buf[BUFFER_SIZE];
    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;       // 指向下一个定时器
};

// 定时器链表。是一个升序、双向链表，带有头节点和尾节点
class sort_timer_lst{
public:
    sort_timer_lst() : head(NULL),tail(NULL){};
    // 析构函数，销毁时删除所有定时器
    ~sort_timer_lst(){
        util_timer* tmp = head;
        while(tmp){
            head = tmp->next;
            delete tmp;
            tmp = head;
        }
    }

    // 将目标定时器timer添加到链表中
    void add_timer(util_timer* timer){
        if(!timer){
            return;
        }
        if(!head){
            head = tail = timer;
            return;
        }
        // 如果目标定时器的时间小于当前链表中所有定时器的时间，则将该定时器插入链表头部，作为头节点
        // 否则调用重载函数add_timer，将它插入链表合适位置，保证链表的升序特性
        if(timer->expire < head->expire){
            timer->next = head;
            head->prev = timer;
            head = timer;
            return;
        }
        add_timer(timer, head);
    }

    // 当某个定时任务发生变化时，调整对应的定时器在链表中的位置
    // 只考虑被调整的定时器的超时时间延长的情况，即该定时器需要向链表尾部移动
    void 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 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 = timer->prev;
            tail->next = NULL;
            delete timer;
            return;
        }
        // 如果定时器在链表中间
        timer->prev->next = timer->next;
        timer->next->prev = timer->prev;
        delete timer;
    }

    // SIGALRM信号被触发一次及阻碍信号处理函数中执行一次tick函数，以处理链表上到期的任务
    // 相当于心搏函数，每隔一段时间就执行一次，检测并处理到期的任务
    void tick(){
        if(!head){
            return;
        }
        printf("timer tickn");
        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;
        }
    }
private:
    // 重载add_timer，将目标定时器添加到节点lst_head后面的链表部分
    void add_timer(util_timer* timer, util_timer* lst_head){
        util_timer* prev = lst_head;
        util_timer* tmp = prev->next;
        // 遍历lst_head后面的链表，将目标定时器插入
        while(tmp){
            if(timer->expireexpire){
                prev->next = timer;
                timer->next = tmp;
                tmp->prev = timer;
                timer->prev = prev;
                break;
            }
            prev = tmp;
            tmp = tmp->next;
        }
        // 如果遍历完tmp，则需要将目标定时器插入到尾部
        if(!tmp){
            prev->next = timer;
            timer->prev = prev;
            timer->next = NULL;
            tail = timer;
        }
    }
private:
    util_timer* head;
    util_timer* tail;
};

#endif

  服务器程序通常要定期处理非活动连接：给客户都安发送一个重连请求，或者关闭该连接，或者其他。Linux在内核中提供了对连接是否处于活动状态的定期检查机制，可通过socketKEEPALIVE进行激活，但此方法使得应用程序对连接的管理变得复杂。
  可考虑在应用层实现类似KEEPALIVE的机制，管理长时间处于非活动状态的链接。可利用alarm函数定期触发SIGALRM信号，该信号的信号处理函数利用管道通道通知主循环执行定时器链表上的定时任务————关闭非活动的连接。

实例：关闭非活动连接

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include "升序定时器链表.h"

#define FD_LIMIT 65535
#define MAX_EVENT_NUMBER 1024
#define TIMESLOT 5

static int pipefd[2];
// 利用升序链表管理定时器
static sort_timer_lst 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;
}

// 将文件描述符fd上的EPOLLIN注册到epollfd指示的epoll内核事件表中，参数enable_et指定是否对fd启用ET模式
void addfd(int epollfd, int fd, bool enable_et){
    epoll_event event;
    event.data.fd = fd;
    event.events = EPOLLIN;
    if(enable_et){
        event.events |= 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, '', 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(){
    // 定时处理任务，实际是调用tick函数
    timer_lst.tick();
    // alarm调用只会引起一次SIGALRM信号，需要重新定时，以不断触发SIGALRM信号
    alarm(TIMESLOT);
}

// 定时器回调函数，剔除非活动连接socket的注册时间，并关闭
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 %dn", 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_STERAM, 0);
    assert(sock>0);
    ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address));
    if(ret = -1){
        printf("errno is %dn", errno);
        return 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]);

    // 设置信号处理函数
    addsig(SIGALRM);
    addsig(SIGTERM);
    bool stop_server = false;

    client_data* users = new client_data[FD_LIMIT];
    bool timeout = false;
    alarm(TIMESLOT);        // 定时

    while(!stop_server){
        int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);
        if((number<0)&&(errno!=EINTR)){
            printf("epoll failuren");
            break;
        }
        for(int i=0;iuser_data = &users[connfd];
                timer->cb_func = cb_func;
                time_t cur = timer(NULL);
                timer->expire = cur + 3*TIMESLOT;
                users[connfd].timer = timer;
                timer_lst.add_timer(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;iexpire = cur + 3*TIMESLOT;
                        printf("adjust timer oncen");
                        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;
}

  Linux下3组I/O复用调用都带有超时参数，可以统一处理信号和I/O事件，也能统一处理定时事件。由于I/O复用系统调用可能在超时时间到期之前就返回，因此需要不断更新定时参数以返回剩余的时间。

实例：利用I/O复用系统调用定时

#define TIMEOUT 5000

int timeout = TIMEOUT;
time_t start = time(NULL);
time_t end = time(NULL);
while(1){
    printf("the timeout is now %d mil-secondsn", timeout);
    start = time(NULL);
    int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, timeout);
    if((number<0)&&(errno!=EINTR)){
        printf("epoll failuren");
        break;
    }
    // 如果epoll_wait成功返回0，则说梦超时时间到，可执行定时任务并重置
    if(number==0){
        timeout = TIMEOUT;
        continue;
    }
    end = time(NULL);
    // 如果epoll_wait返回值大于零，可获得本次持续时间，将定时时间timeout减去这段时间以获得下次epoll_wait调用的超时参数
    timeout -= (end-start)*1000;
    // 若timeout小于等于0则说明，本次不仅有文件描述符就绪，且超时时间刚好到达，则需要处理定时任务，并重置定时时间
    if(timeout<=0){
        timeout = TIMEOUT;
    }
    // handle connections
}

  如上图锁是，(实线)指针只想轮子上的一个槽(slot)，它以恒定的速度顺时针转动，每转动一步就指向下一个槽(虚线指针指向的槽)，每次转动称为一个滴答(tick)。一个滴答的时间就成为时间轮的槽间隔si(slot interval)，即心搏时间。该时间轮共N个槽，则运转一周的时间为N×si。每个槽指向一个定时器链表(每个链表上的定时器具有相同的特征)，它们的定时时间相差N×si的整数倍。利用此关系将定时器散列到不同链表中。
  假如现在指针指向槽cs，需要添加定时间为ti的定时器，则定时器被插入槽ts对应的链表中(ts=[cs+(ts/si)]%N)。
  基于排序链表的定时器使用唯一的一条链表来管理所有定时器。但时间轮使用哈希表思想，将定时器散列到不同的李娜表上，可以提升插入的效率。对时间轮而言，要提高定时精度，需要使si足够小，要提高执行效率，需要N足够大。

实例：时间轮

#ifndef TIME_WHEEL_TIMER
#define TIME_WHEEL_TIMER

#include 
#include 
#include 

#define BUFFER_SIZE 64
class tw_timer;
// 绑定socket和定时器
struct client_data{
    sockaddr_in address;
    int sockfd;
    char buf[BUFFRE_SIZE];
    tw_timer* timer;
};

// 定时器类
class tw_timer{
public:
    tw_timer(int rot, int ts)
    :next(NULL),prev(NULL),rotation(rot), time_slot(ts){} 

public:
    int rotation;       // 记录定时器在时间轮赚多少圈后生效
    int time_slot;      // 记录定时器数据时间轮上哪个槽
    void (*cb_func)(client_data*);      // 定时器回调函数
    client_data* user_data;         // 客户数据
    tw_timer* next;             // 下一个定时器
    tw_timer* prev;             // 前一个定时器
};

class time_wheel{
public:
    // 初始化每个槽的头节点
    time_wheel():cur_slot(0){
        for(int i=0;inext;
                delete tmp;
                tmp = slots[i];
            }
        }
    }

    // 根据定时器值timeout创建一个定时器，并插入到合适的槽中
    tw_timer* add_timer(int timeout){
        if(timeout<0){
            return NULL;
        }
        int ticks = 0;
        // 根据定时器超时值计算其在时间轮多少个抵达后被触发
        // 将滴答数存在变量ticks中
        if(timeout < SI){
            ticks = 1;
        }
        else{
            ticks = timeout/SI;
        }
        // 计算待插入的定时器在多少圈后被触发
        int rotation = ticks/N;
        // 计算待插入的定时器应该被插入到哪个槽中
        int ts = (cur_slot+ (ticks % N) ) %N;
        // 创建新的定时器，其在时间轮转动rotation后被触发，位于第ts个槽上
        tw_timer* timer = new tw_timer(rotation, ts);
        // 如果ts槽上无任何定时器，则将其设为头节点
        if(!slots[ts]){
            printf("add timer, rotation is %d, ts is %d, cur_slot is %dn", rotation, ts, cur_slot);
            slots[ts] = timer;
        }
        // 否则插入第ts槽中
        else{
            timer->next = slots[ts];
            slots[ts]->prev = timer;
            slots[ts] = timer;
        }
        return timer;
    }

    // 删除目标定时器
    void del_timer(tw_timer* timer){
        if(!timer){
            return;
        }
        int ts = timer->time_slot;
        // slots[ts]是目标定时器所在的头节点
        if(timer == slots[ts]){
            slots[ts] = slots[ts]->next;
            if(slots[ts]){
                slots[ts]->prev = NULL;
            }
            delete timer;
        }
        else{
            timer->prev->next = timer->next;
            if(timer->next){
                timer->next->prev = timer->prev;
            }
            delete timer;
        }
    }

    // SI时间到后，调用该函数，时间轮向前滚动一个槽的间隔
    void tick(){
        tw_timer* tmp = slots[cur_slot];
        printf("current slot is %dn", cur_slot);
        while(tmp){
            printf("tick the timer oncen");
            // 如果定时器的rotation值大于0，则它在这一轮不起作用
            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_slotn");
                    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;      // 更新时间轮的当前槽
    }
private:
    static const int N = 60;        // 时间轮上的槽数
    static const int SI = 1;        // 每1s转动一次，即槽间隔为1s
    tw_timer* slots[N];             // 时间轮的槽，每个元素指向一个定时器链表，链表无序
    int cur_slot;                   // 时间轮的当前槽
}

#endif // !TIME_WHEEL_TIMER

  思想：将所有定时器中超时时间最小的一个定时器的超时值作为心搏间隔。这样一旦心搏函数tick被调用，超时时间最小的定时器必然到期，从而处理该定时器。依次反复，实现精确定时。
  利用最小堆可实现上述思想，最小堆是一种完全二叉树，可以使用数组来组织其中的元素。

实例：时间堆

#ifndef MIN_HEAP
#define MIN_HEAP

#include 
#include 
#include 
using std::exception;

#define BUFFER_SIZE 64

class heap_timer;

// 绑定socket和定时器
struct client_data{
    sockaddr_in address;
    int sockfd;
    char buf[BUFFER_SIZE];
    heap_timer* timer;
};

// 定时器类
class heap_timer{
public:
    heap_timer(int delay){
        expire = time(NULL)+delay;
    }
public:
    time_t expire;          // 定时器生效绝对事件
    void (*cb_func)(client_data*);      // 回调函数
    client_data* user_data;         // 用户数据
};

// 时间堆类
class time_heap{
public:
    // 初始化一个大小为cap的空堆
    time_heap(int cap) throw(std::exception):capacity(cap), cur_size(0){
        array = new heap_timer*[capacity];
        if(!array){
            throw std::exception();
        }
        for(int i=0;i=0;i--){
                // 堆数组中第[(cur_size-1)/2]进行下虑操作
                percolate_down(i);
            }
        }
    }
    // 析构
    ~time_heap(){
        for(int i=0;i=capacity){
            resize();
        }
        // 堆大小+1， hole新建空穴位置
        int hole = cur_size++;
        int parent = 0;
        // 堆空穴到根节点的路径上所有节点执行上虑操作
        for(;hole>0;hole=parent){
            parent = (hole-1)/2;
            if(array[parent]->expire <= timer->expire){
                break;
            }
            array[hole] = array[parent];
        }
        array[hole] = timer;
    }

    // 删除目标定时器
    void del_timer(heap_timer* timer){
        if(!timer){
            return;
        }
        timer->cb_func = NULL;
    }

    // 获得堆顶部的定时器
    heap_timer* top() const{
        if(empty()){
            return NULL;
        }
        return array[0];
    }
    // 删除堆顶部的定时器
    void pop_timer(){
        if(empty()){
            return;
        }
        if(array[0]){
            delete array[0];
            // 将原来对顶元素替换为堆数组最后一个元素
            array[0] = array[--cur_size]；
            percolate_down(0);  // 对新的堆顶元素进行下虑操作
        }
    }
    // 心搏函数
    void ticl(){
        heap_timer* tmp = array[0];
        timer_t cur = time(NULL);       // 循环处理堆中到期的定时器
        while(!empty()){
            if(!tmp){
                break;
            }
            // 如果堆顶定时器没到期，则退出循环
            if(tmp->expire > cur){
                break;
            }
            // 开始执行
            if(array[0]->cb_func){
                array[0]->cb_func(array[0]->user_data);
            }
            // 删除对顶元素；
            pop_timer();
            tmp = arrya[0];
        }
    }
    bool empty() const {return cur_size == 0;}

private:
    // 下虑操作，保证堆数组中以第hole个节点作为根的子树都有最小堆性质
    void percolate_down(int hole){
        heap_timer* temp = array[hole];
        int child = 0;
        for(;((hole*2+1)<=(cur_size+1));hole = child){
            child = hole*2+1;
            if((child<(cur_size-1))&&(array[child+1]->expire < array[child]->expire)){
                ++child;
            }
            if(array[child]->expire < temp->expire){
                array[hole] = array[child];
            }
            else{
                break;
            }
        }
        array[hole] = temp;
    }
    // 将堆数组容量扩大1倍
    void resize() throw (std::exception){
        heap_timer** temp = new heap_timer*[2*capacity];
        for(int i=0;i<2*capacity;i++){
            temp[i] = NULL;
        }
        if(!temp){
            throw std::excetion();
        }
        capacity = 2*capacity;
        for(int i=0;i