使用epoll写一个reactor框架实现百万fd连接

这里实现一个单线程的reactor框架，没有任何业务处理。

这里先介绍三个重要的结构体

struct ntyevent
{
    int fd;
    int events;
    void *arg;
    int (*callback)(int fd,int events,void *arg);
    long last_active;
    int status;
    char buffer[MAXLEN];
    int length;

};

struct ntyreactor
{
    int epfd;
    int blkcnt;
    struct eventblock *evblk;
};


struct eventblock
{
    struct eventblock *next;
    struct ntyevent *events;
};

ntyevent用来存储连接的fd和设置该连接的监听事件和回调函数。eventblock则是用来存储ntyevent的块，我们做成一个链表的形式，当这个块存储满时创建一个新块，这样就可以容纳足够多的ntyevent。ntyreactor则是存储epoll的句柄和记录当前块的数量并指向它。

内存模型是这样的

我们知道，fd是顺序往上增加的，当fd大于块的大小时，就需要创建一个新块存储，想对fd进行操作时，就得先找到该fd所对应的ntyevent。

struct ntyevent* ntyreactor_idx(struct ntyreactor *reactor,int sockfd)
{
    int blkidx = sockfd / MAX_EPOLL_SIZE;

    while(blkidx >= reactor->blkcnt)//如果fd大于现有的块，则需要创建新块存储
    {
        ntyreactor_alloc(reactor);
    }

    int i = 0;
    struct eventblock *blk = reactor->evblk;
    while(i++ < blkidx && blk != NULL)//找到fd对应的块
    {
        blk = blk->next;
    }

    return &blk->events[sockfd % MAX_EPOLL_SIZE];


}

而每个端口只能监听65535个io，所以我这里监听了100个端口，下面上完整代码

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



#define MAXLEN 4096
#define MAX_EPOLL_SIZE 1024
#define SERVER_PORT 8888
#define PORT_COUNT 100

typedef int ALLBACK(int,int,void*);

struct ntyevent
{
    int fd;
    int events;
    void *arg;
    int (*callback)(int fd,int events,void *arg);
    long last_active;
    int status;
    char buffer[MAXLEN];
    int length;

};

struct ntyreactor
{
    int epfd;
    int blkcnt;
    struct eventblock *evblk;
};


struct eventblock
{
    struct eventblock *next;
    struct ntyevent *events;
};

int recv_cb(int fd,int events,void *arg);
int send_cb(int fd,int events,void *arg);
struct ntyevent* ntyreactor_idx(struct ntyreactor *reactor,int sockfd);


void nty_event_set(struct ntyevent *ev,int fd,ALLBACK callback,void *arg)
{
    ev->fd = fd;
    ev->callback = callback;
    ev->events = 0;
    ev->arg = arg;
    ev->last_active = time(NULL);
    return ;
}

int nty_event_del(int epfd,struct ntyevent *ev)
{
    struct epoll_event ep_ev = {0,{0}};

    if(ev->status != 1)
    {
        return -1;
    }

    ep_ev.data.ptr = ev;
    ev->status = 0;
    epoll_ctl(epfd,EPOLL_CTL_DEL,ev->fd,&ep_ev);
    return 0;
}

int nty_event_add(int epfd,int events,struct ntyevent *ev)
{
    struct epoll_event ep_ev = {0, {0}};
    ep_ev.data.ptr = ev;
    ep_ev.events = ev->events = events;

    int op;
    if(ev->status == 1)
    {
        op = EPOLL_CTL_MOD;
    }else{
        op = EPOLL_CTL_ADD;
        ev->status = 1;
    }

    if (epoll_ctl(epfd,op,ev->fd,&ep_ev) < 0)
    {
        printf("event add fail fd=%d,events[%d]n",ev->fd,events);
        return -1;
    }

    return 0;
}

int accept_cb(int fd,int events,void *arg)
{
    struct ntyreactor *reactor = (struct ntyreactor*)arg;
    if (reactor == NULL )return -1;
    struct sockaddr_in client;
    socklen_t len = sizeof(client);
    int clientfd;
    
    if((clientfd = accept(fd,(struct sockaddr*)&client,&len)) == -1)
    {
        //if(errno != EAGAIN && errno != EINTR)
        printf("accept err:%sn",strerror(errno));
        return -1;
    }

    int flag = 0;

    if((flag = fcntl(clientfd,F_SETFL,O_NONBLOCK)) < 0)
    {
        printf("%s:fcntl nonblocking fail,%dn",__func__,clientfd);
        return -1;
    }
    struct ntyevent *event = ntyreactor_idx(reactor,clientfd);

    nty_event_set(event,clientfd,recv_cb,reactor);
    nty_event_add(reactor->epfd,EPOLLIN,event);

    printf("new connect [%s:%d],pos[%d]n",inet_ntoa(client.sin_addr),ntohs(client.sin_port),clientfd);
    return 0;

}

int recv_cb(int fd,int events,void *arg)
{
    struct ntyreactor *reactor = (struct ntyreactor*) arg;
    struct ntyevent *ev = ntyreactor_idx(reactor,fd);

    int len = recv(fd,ev->buffer,MAXLEN,0);
    nty_event_del(reactor->epfd,ev);

    if(len > 0)
    {
        ev->length = len;
        ev->buffer[len] = '';

        printf("C[%d]:%sn",fd,ev->buffer);
        nty_event_set(ev,fd,send_cb,reactor);
        nty_event_add(reactor->epfd,EPOLLOUT,ev);
    }
    else if(len == 0)
    {
        printf("close fd:%dn",ev->fd);
        close(ev->fd);
    }
    else
    {
        close(ev->fd);
        printf("recv[fd:%d] error[%d]:%sn",fd,errno,strerror(errno));
    }
    return len;
}

int send_cb(int fd,int events,void *arg)
{
    struct ntyreactor *reactor = (struct ntyreactor*) arg;
    struct ntyevent *ev = ntyreactor_idx(reactor,fd);
    int len = send(fd,ev->buffer,ev->length,0);
    if(len > 0)
    {
        printf("S[%d]:%sn",fd,ev->buffer);
        nty_event_del(reactor->epfd,ev);
        nty_event_set(ev,fd,recv_cb,arg);
        nty_event_add(reactor->epfd,EPOLLIN,ev);

    }else{
        close(ev->fd);
        nty_event_del(reactor->epfd,ev);
        printf("send fd[%d] error:%sn",fd,strerror(errno));
    }
    return len;

}

int ntyreactor_init(struct ntyreactor *reactor)
{
    if(reactor == NULL) return -1;
    reactor->epfd = epoll_create(1);
    if(reactor->epfd <= 0)
    {
        printf("create epfd in %s err %sn",__func__,strerror(errno));
        return -2;
    }

    struct ntyevent* evs = (struct ntyevent*)calloc(MAX_EPOLL_SIZE,sizeof(struct ntyevent));
    if(evs == NULL)
    {
        printf("calloc ntyevent failn");
        return -2;
    }

    struct eventblock *block = (struct eventblock*)calloc(1,sizeof(struct eventblock));
    if(block == NULL)
    {
        printf("calloc eventblock failn");
        return -2;
    }

    block->events = evs;
    block->next = NULL;

    reactor->blkcnt = 1;
    reactor->evblk = block;
    return 0;

}

int ntyreactor_run(struct ntyreactor *reactor)
{
    if(reactor == NULL) return -1;
    if(reactor->evblk == NULL ) return -1;
    if(reactor->epfd < 0) return -1;

    struct epoll_event events[MAX_EPOLL_SIZE + 1];
    int check = 0;

    while(1)
    {

        int nready = epoll_wait(reactor->epfd,events,MAX_EPOLL_SIZE,1000);
        if(nready < 0)
        {
            printf("epoll_wait error,exitn");
            continue;
        }

        for(int i = 0;i < nready;i++)
        {
            struct ntyevent *ev = (struct ntyevent*)events[i].data.ptr;

            if((events[i].events & EPOLLIN) && (ev->events & EPOLLIN))
            {
                ev->callback(ev->fd,events[i].events,ev->arg);
            }
            if((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT))
            {
                ev->callback(ev->fd,events[i].events,ev->arg);
            }
        }
    }

}

int ntyreactor_alloc(struct ntyreactor *reactor)
{
    if (reactor == NULL) return -1;
    if (reactor->evblk == NULL) return -1;

    struct eventblock *blk = reactor->evblk;
    while(blk->next != NULL)
    {
        blk = blk->next;
    }

    struct ntyevent *evs = (struct ntyevent*)calloc(MAX_EPOLL_SIZE,sizeof(struct ntyevent));
    if (evs == NULL)
    {
        printf("alloc ntyevents failn");
        return -2;
    }

    struct eventblock *block = (struct eventblock*)calloc(1,sizeof(struct eventblock));
    if (block == NULL)
    {
        printf("alloc eventblock failn");
        return -2;
    }

    block->events = evs;
    block->next = NULL;

    blk->next = block;
    reactor->blkcnt++;
    return 0;
}

struct ntyevent* ntyreactor_idx(struct ntyreactor *reactor,int sockfd)
{
    int blkidx = sockfd / MAX_EPOLL_SIZE;

    while(blkidx >= reactor->blkcnt)
    {
        ntyreactor_alloc(reactor);
    }

    int i = 0;
    struct eventblock *blk = reactor->evblk;
    while(i++ < blkidx && blk != NULL)
    {
        blk = blk->next;
    }

    return &blk->events[sockfd % MAX_EPOLL_SIZE];


}

int ntyreactor_addlistener(struct ntyreactor *reactor,int sockfd,ALLBACK *accpetor)
{
    if (reactor == NULL) return -1;
    if (reactor->evblk == NULL) return -1;
    
    struct ntyevent *event = ntyreactor_idx(reactor,sockfd);

    nty_event_set(event,sockfd,accpetor,reactor);
    nty_event_add(reactor->epfd,EPOLLIN,event);

    return 0;

    
}

int init_sock(short port)
{
    int fd = socket(AF_INET,SOCK_STREAM,0);
    fcntl(fd,F_SETFL,O_NONBLOCK);//

    struct sockaddr_in server_addr;
    memset(&server_addr,0,sizeof(server_addr));
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(port);

    bind(fd,(struct sockaddr*)&server_addr,sizeof(server_addr));

    if(listen(fd,20) < 0)
    {
        printf("listen fail:%sn",strerror(errno));
        return -1;
    }

    return fd;


}

int ntyreactor_destory(struct ntyreactor* reactor)
{
    close(reactor->epfd);

    struct eventblock* blk = reactor->evblk;
    struct eventblock* blk_next = NULL;

    while(blk != NULL)
    {
        blk_next = blk->next;
        free(blk->events);
        free(blk);

        blk = blk_next;
    }
    return 0;

}


int main(int argc,char* argv[])
{
    unsigned short port = SERVER_PORT;
    if(argc == 2)
    {
        port = atoi(argv[1]);
    }
    struct ntyreactor *reactor = (struct ntyreactor*)calloc(1,sizeof(struct ntyreactor));
    ntyreactor_init(reactor);

    int sockfds[PORT_COUNT] = {0};
    for(int i = 0;i < PORT_COUNT;i++)
    {
        sockfds[i] = init_sock(port + i);
        ntyreactor_addlistener(reactor,sockfds[i],accept_cb);

    }

    ntyreactor_run(reactor);

    ntyreactor_destory(reactor);

    for(int i = 0;i < PORT_COUNT;i++)
    {
        close(sockfds[i]);
    }

    free(reactor);

    return 0;



}

编译运行后，用客户端压测的效果如下：

 由图中可看出已经压到了10W个fd，内存还剩余很多，这里我每个块设置大小为1024，共创建了98个块，证明块的创建和fd的寻址是没问题的，实现百万的fd并发就是资源的问题了，这里就不继续压了

当然这只是一个框架，只实现了普通的收发功能，真正的业务是很复杂的，这就需要在此基础之上去扩展，如用线程池去执行事件的回调，处理业务，单独创建一个reactor来处理accpet等等，这里就不细说了