Linux(程序设计):58---epoll复用技术实现统一处理信号事件源

Posted 2021-08-28 董哥的黑板报

一、统一信号处理事件源概述

• 信号是一种异步事件：信号处理函数和程序的主循环是两条不同的执行路线。显然，信号处理函数需要尽可能快地执行完毕，以确保该信号不被屏蔽（为了避免一些竞态条件，信号在处理期间，系统不会再次触发它）太久
• 一种典型的解决办法是：
  • 把信号的主要处理逻辑放到程序的主循环中
  • 当信号处理函数被触发时，它只是简单地通过主循环程序接收到信号，并把信号值传递给主循环
  • 主循环再根据接收到的信号值执行目标信号对应的逻辑代码
• 信号处理函数通常使用管道来将信号“传递”给主循环：
  • 信号处理函数往管道的写端写入信号值，主循环则从管道的读端读出该信号值
  • 主循环使用I/O复用系统调用来监听管道的读端文件描述符上的可读时间
• 如此一来，信号事件就能和其他I/O事件一样被处理，即统一事件源
• 很多优秀的I/O框架和后台服务器程序都统一处理信号和I/O事件，比如LiBEVENT I/O框架库和xinetd超级服务

二、编码实现

#include <stdio.h>
#include <stdlib.h>
#include <strings.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <libgen.h>
#include <signal.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/types.h>
#include <arpa/inet.h>

#define LISTEM_NUM 5
#define MAX_EVENT_NUM 1024

int setnonblocking(int fd);

void add_epoll_fd(int epoll_fd,int fd);

void sig_handler(int sigalno);

void add_sig(int sigalno);

static int pipe_fd[2];

int main(int argc,char* argv[])
{
    if(argc!=3){
        printf("usage:./%s [server ip] [server port]\\n",basename(argv[1]));
        exit(EXIT_FAILURE);
    }

    int ser_fd,server_port;
    const char* server_ip;

    //创建套接字
    if((ser_fd=socket(AF_INET,SOCK_STREAM,0))==-1){
        perror("socket");
        exit(EXIT_FAILURE);
    }

    //初始化服务端地址
    struct sockaddr_in server_address;

    server_ip=argv[1];
    server_port=atoi(argv[2]);
    bzero(&server_address,sizeof(server_address));
    server_address.sin_family=AF_INET;
    server_address.sin_port=htons(server_port);
    if(inet_pton(AF_INET,server_ip,&server_address.sin_addr.s_addr)==-1){
        perror("inet_pton");
        exit(EXIT_FAILURE);
    }

    //绑定服务端地址
    if(bind(ser_fd,(struct sockaddr*)&server_address,sizeof(server_address))==-1){
        perror("bind");
        exit(EXIT_FAILURE);
    }

    //开启监听
    if(listen(ser_fd,LISTEM_NUM)==-1){
        perror("bind");
        exit(EXIT_FAILURE);
    }

    int epoll_fd;
    //创建epoll事件表句柄
    if((epoll_fd=epoll_create(5))==-1){
        perror("epoll_create");
        exit(EXIT_FAILURE);
    }
    //将服务端套接字加入到事件表中
    add_epoll_fd(epoll_fd,ser_fd);

    //创建管道
    if(socketpair(PF_UNIX,SOCK_STREAM,0,pipe_fd)==-1){
        perror("socketpair");
        exit(EXIT_FAILURE);
    }
    /*sockpair函数创建的管道是全双工的，不区分读写端
        此处我们假设pipe_fd[1]为写端，非阻塞
        pipe_fd[0]为读端
    */
    setnonblocking(pipe_fd[1]);
    add_epoll_fd(epoll_fd,pipe_fd[0]);

    //为一些信号绑定信号处理函数
    add_sig(SIGHUP); //终端接口检测到一个连接断开，发送此信号
    add_sig(SIGCHLD);//子进程终止或停止时，子进程发送此信号
    add_sig(SIGTERM);//接收到kill命令
    add_sig(SIGINT); //用户按下中断键(Delete或Ctrl+C)

    int server_running=1;
    int epoll_wait_ret_value;

    struct epoll_event events[MAX_EVENT_NUM];
    
    while(server_running)
    {
        bzero(events,sizeof(events));
        epoll_wait_ret_value=epoll_wait(epoll_fd,events,MAX_EVENT_NUM,-1);

        //epoll_wait函数出错
        if((epoll_wait_ret_value==-1)&&(errno!=EINTR)){
            close(ser_fd);
            perror("epoll_wait");
            exit(EXIT_FAILURE);
        }

        //遍历就绪的事件
        for(int i=0;i<epoll_wait_ret_value;++i)
        {
            int sock_fd=events[i].data.fd;//获取文件描述符
            //如果是服务端套接字,接收客户端的连接
            if(sock_fd==ser_fd){
                int client_fd;
                char client_address_ip[24];
                struct sockaddr_in client_address;
                socklen_t address_len=sizeof(client_address);
                bzero(&client_address,sizeof(client_address));

                if((client_fd=accept(ser_fd,(struct sockaddr*)&client_address,&address_len))==-1){
                    perror("accept");
                    continue;
                }

                //将新的客户端套接字放入到事件集中
                add_epoll_fd(epoll_fd,client_fd);
                //打印客户端地址信息
                inet_ntop(AF_INET,&client_address.sin_addr.s_addr,client_address_ip,sizeof(client_address_ip));
                printf("get a new client,ip:%s,port:%d\\n",client_address_ip,ntohs(client_address.sin_port));
            }
            //如果是管道的一端有数据可读，那么处理信号
            else if((sock_fd==pipe_fd[0])&&(events[i].events&EPOLLIN)){
                char signals[1024];
                int sig;
                int recv_ret_value;
                recv_ret_value=recv(pipe_fd[0],signals,sizeof(signals),0);
                if(recv_ret_value<=0)
                    continue;
                else{
                    //每个信号值占1字节，所以按字节来逐个接收信号
                    for(int i=0;i<recv_ret_value;++i){
                        printf("server:I caugh the signal %d\\n", signals[i]);
                        switch (signals[i])
                        {
                        case SIGCHLD:
                        case SIGHUP:
                            continue;
                        //接收到下面这两个信号，终止程序
                        case SIGTERM: //kill
                        case SIGINT:  //ctrl +c
                            server_running=0;
                        }
                    }
                }
            }
            //如果是客户端
            else{
                
            }
        }
    }

    printf("service is down\\n");
    close(ser_fd);
    close(pipe_fd[1]);
    close(pipe_fd[0]);
    exit(EXIT_SUCCESS);
}

int setnonblocking(int fd)
{
    int old_options,new_options;
    //获取原先的描述符标志
    if((old_options=fcntl(fd,F_GETFL))==-1){
        perror("fcntl");
        exit(EXIT_FAILURE);
    }
    //设置非阻塞
    new_options=old_options|O_NONBLOCK;
    if(fcntl(fd,F_SETFL,new_options)==-1){
        perror("fcntl");
        exit(EXIT_FAILURE);
    }

    return old_options;
}

void add_epoll_fd(int epoll_fd,int fd)
{
    struct epoll_event new_event;
    bzero(&new_event,sizeof(new_event));
    
    new_event.events=EPOLLIN|EPOLLET;
    new_event.data.fd=fd;

    //将新事件加入到事件表中
    if(epoll_ctl(epoll_fd,EPOLL_CTL_ADD,fd,&new_event)==-1){
        perror("epoll_ctl");
        exit(EXIT_FAILURE);
    }

    //设置为非阻塞
    setnonblocking(fd);
}

void add_sig(int sigalno)
{
    struct sigaction act;
    bzero(&act,sizeof(act));
    act.sa_handler=sig_handler;//设置信号处理函数
    sigfillset(&act.sa_mask); //初始化信号屏蔽集
    act.sa_flags|=SA_RESTART; //由此信号中断的系统调用自动重启动

    //初始化信号处理函数
    if(sigaction(sigalno,&act,NULL)==-1){
        printf("capture signal,but to deal with failure\\n");
        return;
    }
}

void sig_handler(int sigalno)
{
    printf("capture signal,signal num is %d",sigalno);
    //保留原来的errno，在函数最后回复，以保证函数的可重入性
    int save_errno=errno;
    int msg=sigalno;
    //将信号值写入管代，通知主循环
    if(send(pipe_fd[1],(char*)&msg,1,0)<=0){
        printf("The message sent to the server failed\\n");
    }
    printf(",signal is send to server\\n");
    errno=save_errno;
}

代码解析

• 创建一个无名管道，管道[0]端用来读取数据，[1]端用来发送数据。读写端都设置为非阻塞
• 当信号处理函数执行时，在处理函数中向[1]端发送信号编号
• 主函数使用epoll轮询，其中包括轮询管道[0]，一旦有信息(信号编号)发来，处理信号

代码演示

• 使用客户端工具连接程序，打印客户端连接信息

• 使用kill命令给服务端程序发送一个编码为1的信号，可以看到服务端接收到这个信号

• 按下Ctrl+C触发SIGINT信号，程序终止(与预期一致)