350 行 C 代碼實現 epoll reactor 模型

有了 epoll 爲什麼還需要 reactor？

響應快，不必爲單個同步事件所阻塞，雖然 Reactor 本身依然是同步的；
編程相對簡單，可以最大程度的避免複雜的多線程及同步問題，並且避免了多線程 / 進程的切換開銷；
可擴展性，可以方便的通過增加 Reactor 實例個數來充分利用 CPU 資源；
可複用性，reactor 框架本身與具體事件處理邏輯無關，具有很高的複用性；

reactor 的思想：將對 IO 的處理轉化爲對事件的處理。

reactor 包括 5 個部分

reactor 管理器：註冊和刪除事件，運行事件循環，分發事件到已註冊的回調函數上
事件解複用器：epoll 等各個平臺的 IO 多路複用 API 的封裝
fd
事件處理器接口
具體的事件處理器：由用戶實現事件處理器接口

reactor 封裝

reactor_init
reactor_run
reactor_free
accept_cb
recv_cb
send_cb

accept_cb 裏可以實現 IP 限制，負載均衡

reactor 注意事項

1 個 fd 對應一個 user_data，epoll 管理所有的 IO
ET 模式收發數據要循環，一次 recv BUF_LEN 數據，直到返回 - 1 或 recv_len 等於 BUF_LEN。send 返回長度如果小於指定大小，則需要註冊寫事件
爲了避免讀寫數據時阻塞，socket 需要設置成非阻塞的
ocket 是否阻塞影響的 API：send,recv,connect,accpet

reactor 模型

單 reactor

多 reactor（one loop one thread）（可以有一個 reactor 專門監聽連接）

多進程 reactor：多個子進程監聽同一個端口，通過共享內存決定哪個進程來 accept，nginx 在用戶態解決驚羣問題

thread per request

單線程模型

多線程模型（單 reactor）

多線程模型（多 reactor）

單線程 reactor 實現

一個數據塊有 1024 個 fd，數據塊之間用鏈表連接。每次 fd 超過上限時分配 1 個數據塊然後添加到鏈表尾部。

代碼實現：

#include <sys/epoll.h>
#include <sys/socket.h>
#include <stdio.h>
#include <netinet/in.h>     // sockaddr_in
#include <netinet/ip.h>
#include <arpa/inet.h>      // inet_addr
#include <errno.h>
#include <string.h>
#include <stdlib.h>         // atoi
#include <unistd.h>
#include <fcntl.h>          // fcntl
 
#define MAX_EPOLL_EVENT 1024
#define BUF_SIZE        1024
 
#define NO_EVENT        0
#define ACCEPT_EVENT    1
#define READ_EVENT      2
#define WRITE_EVENT     3
 
#define exitif(s, err_str) do { \
    if(s) { \
        printf("%s: %s(code:%d)\n", err_str, strerror(errno), errno); \
    } \
} while(0);
 
typedef int (*EVENT_CALLBACK)(int fd, int events, void* args);
 
typedef struct _user_data {
    int             fd;
    uint32_t        events;
    void*           args;
 
    EVENT_CALLBACK  accept_cb;
    EVENT_CALLBACK  read_cb;
    EVENT_CALLBACK  write_cb;
 
    char            send_buf[BUF_SIZE];
    char            recv_buf[BUF_SIZE];
    int             send_buf_len;
    int             recv_buf_len;
} user_data;
 
// 1個用戶數據塊有1024個用戶數據
typedef struct _user_data_block {
    struct _user_data_block*    next;
    user_data*                  user_data_array;
} user_data_block;
 
typedef struct _reactor {
    short               listenfd;
    int                 epfd;
    user_data_block*    head;   // 用戶數據塊鏈表，head指向第1個數據快
} reactor;
 
 
void reactor_set_event(/* reactor* r, */ int fd, int event_type, void* args);
void reactor_del_event(int fd, int event_type, void* args);
 
int set_fd_nonblock(int fd) {
    return fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK);
}
 
int init_server(short port) {
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    exitif(-1 == fd, "socket");
 
    int ret = set_fd_nonblock(fd);
    exitif(-1 == ret, "set_fd_nonblock");
 
    struct sockaddr_in local_addr;
    int addr_len = sizeof(struct sockaddr_in);
    memset(&local_addr, 0, addr_len);
    local_addr.sin_family = AF_INET;
    local_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    local_addr.sin_port = htons(port);
    ret = bind(fd, (struct sockaddr *)&local_addr, addr_len);
    exitif(-1 == ret, "bind");
 
    ret = listen(fd, 5);
    exitif(-1 == ret, "listen");
 
    return fd;
}
 
// 獲取第fd / MAX_EPOLL_EVENT個數據塊的第fd % MAX_EPOLL_EVENT個user_data
user_data* reactor_user_data(reactor* r, int fd) {
    if (!r || !r->head) {
        return NULL;
    }
    user_data_block* block = r->head;
    user_data_block* prev_block = NULL;
    int index = fd / MAX_EPOLL_EVENT;
    while (index >= 0) {
        if (!block) {
            block = calloc(sizeof(user_data_block), 1);
            block->next = NULL;
            block->user_data_array = calloc(sizeof(user_data), MAX_EPOLL_EVENT);
            if (NULL == block->user_data_array) {
                free(block);
                block = NULL;
                close(r->epfd);
                exitif(1, "NULL == block->user_data_array");
            }
 
            if (prev_block) {
                prev_block->next = block;
            }
        }
        prev_block = block;
        block = block->next;
        --index;
    }
 
    return &prev_block->user_data_array[fd % MAX_EPOLL_EVENT];
}
 
// 沒有returngcc爲什麼沒有報錯？
reactor* reactor_create() {
    reactor* r = calloc(sizeof(reactor), 1);
 
    r->epfd = epoll_create(1);
    r->head = calloc(sizeof(user_data_block), 1);
    if (NULL == r->head) {
        close(r->epfd);
        exitif(1, "NULL == r->head");
    }
    r->head->next = NULL;
    r->head->user_data_array = calloc(sizeof(user_data), MAX_EPOLL_EVENT);
    if (NULL == r->head->user_data_array) {
        free(r->head);
        r->head = NULL;
        close(r->epfd);
        exitif(1, "NULL == r->head->user_data_array");
    }
 
    return r;
}
 
int reactor_accept_cb(int fd, int events, void* args) {
    printf("reactor_accept_cb\n");
 
    struct sockaddr_in peer_addr;
    memset(&peer_addr, 0, sizeof(struct sockaddr_in));
    socklen_t peer_addr_len = sizeof(peer_addr);
    // ADDR_LEN要設爲sockaddr_in的實際大小，否則會出現端口爲0的情況
    int clientfd = accept(fd, (struct sockaddr *)&peer_addr, (socklen_t *)&peer_addr_len);
    exitif(-1 == clientfd, "accept");
 
    int ret = set_fd_nonblock(clientfd);
    exitif(-1 == ret, "set_fd_nonblock");
 
    printf("new connection from fd: %d, address:%s:%d\n\n", clientfd, inet_ntoa(peer_addr.sin_addr), ntohs(peer_addr.sin_port));
    
    reactor_set_event(clientfd, READ_EVENT, args);
 
    return 0;
}
 
int reactor_read_cb(int fd, int events, void* args) {
    printf("reactor_read_cb\n");
    
    reactor* r = args;
    user_data* ud = reactor_user_data(r, fd);
 
    // TODO: socket是非阻塞的，需要循環接收
    char *recv_buf = ud->recv_buf;
    int n = recv(fd, recv_buf, BUF_SIZE, 0);
    if (n < 0)
    {
        printf("recv errno: %s\n", strerror(errno));
        return -1;
    }
    else if (n == 0)
    {
        printf("client closed, fd: %d\n", fd);
        reactor_del_event(fd, 0, args);
        close(fd);
        // free(ud);
    }
    else
    {
        recv_buf[n] = '\0';
        printf("recv from fd %d, msg: %s", fd, recv_buf);
 
        char *send_buf = ud->send_buf;
        ud->recv_buf_len = n;
        ud->send_buf_len = n;
        memcpy(send_buf, recv_buf, n);
        reactor_set_event(fd, WRITE_EVENT, args);
    }
 
    return 0;
}
 
int reactor_write_cb(int fd, int events, void* args) {
    printf("reactor_write_cb\n");
 
    reactor* r = args;
    user_data* ud = reactor_user_data(r, fd);
 
    int send_len = ud->send_buf_len;
    char* send_buf = ud->send_buf;
    send_buf[send_len] = '\0';
    int nsend = send(fd, send_buf, send_len, 0);
    if (nsend < 0) {
        printf("send errno: %s\n", strerror(errno));
        return -1;
    }
    else if (nsend == 0) {
        printf("send, client closed, fd: %d\n", fd);
        reactor_del_event(fd, 0, args);
        close(fd);
    }
    else {
        printf("send to fd %d, msg:%s\n", fd, send_buf);
        reactor_set_event(fd, READ_EVENT, args);
    }
 
    return 0;
}
 
// 這裏用來設置事件，回調函數
void reactor_set_event(int fd, int event_type, void* args) {
    struct epoll_event watch_event = {0};
    reactor* r = args;
    user_data* ud = reactor_user_data(r, fd);
 
    ud->fd = fd;
    ud->args = args;
    if (ACCEPT_EVENT == event_type) {
        watch_event.events = EPOLLIN;
        ud->accept_cb = reactor_accept_cb;
    }
    else if (READ_EVENT == event_type) {
        watch_event.events = EPOLLIN;
        ud->read_cb = reactor_read_cb;
    }
    else if (WRITE_EVENT == event_type) {
        watch_event.events = EPOLLOUT;
        ud->write_cb = reactor_write_cb;
    }
 
    watch_event.data.ptr = ud;
 
    if (NO_EVENT == ud->events) {
        int ret = epoll_ctl(r->epfd, EPOLL_CTL_ADD, fd, &watch_event);
        exitif(-1 == ret, "EPOLL_CTL_ADD");
    }
    else {
        int ret = epoll_ctl(r->epfd, EPOLL_CTL_MOD, fd, &watch_event);
        exitif(-1 == ret, "EPOLL_CTL_MOD");
    }
    ud->events = event_type;
}
 
void reactor_del_event(int fd, int event_type, void* args) {
    struct epoll_event watch_event = {0};
    reactor* r = args;
    // fix bug: client退出沒有重置events導致 EPOLL_CTL_MOD: No such file or directory(code:2)
    user_data* ud = reactor_user_data(r, fd);
    ud->events = NO_EVENT;
    int ret = epoll_ctl(r->epfd, EPOLL_CTL_DEL, fd, &watch_event);
    exitif(-1 == ret, "epoll_ctl");
}
 
void reactor_create_server(short port, void* args) {
    reactor* r = args;
    r->listenfd = init_server(port);
    reactor_set_event(r->listenfd, ACCEPT_EVENT, r);
}
 
void reactor_loop(void* args) {
    reactor* r = args;
    // 就緒事件列表
    struct epoll_event ready_events[MAX_EPOLL_EVENT] = {0};
    while (1) {
        int nready = epoll_wait(r->epfd, ready_events, MAX_EPOLL_EVENT, 3000);
        if (nready < 0) {
            if (EINTR == errno) {
                continue;
            }
            exitif(-1 == nready, "epoll_wait");
        }
        // 就緒事件對應於events數組的下標[0, nready - 1]
        int i;
        for (i = 0; i < nready; ++i) {
            user_data* ud = (user_data *)ready_events[i].data.ptr;
            int fd = ud->fd;
            
            if (ACCEPT_EVENT == ud->events) {
                ud->accept_cb(fd, 0, r);
            }
            else {
                // 多個cb，用多個if
                if (EPOLLIN & ready_events[i].events) {
                    ud->read_cb(fd, 0, r);
                }
                if (EPOLLOUT & ready_events[i].events) {
                    // TODO: ET模式需要循環發送
                    ud->write_cb(fd, 0, r);
                }
            }
        }
    }
}
 
void reactor_free(void* args) {
    reactor* r = args;
 
    user_data_block* cur;
    while (r->head) {
        cur = r->head;
        r->head = r->head->next;
 
        if (cur) {
            if (cur->user_data_array)
            {
                free(cur->user_data_array);
                cur->user_data_array = NULL;
            }
            free(cur);
            cur = NULL;
        }
    }
 
    close(r->epfd);
    close(r->listenfd);
}
 
int main(int argc, char *argv[]) {
    if (argc < 2) {
        printf("usage: %s port\n", argv[0]);
        return -1;
    }
    
    // 創建reactor
    reactor* r = reactor_create();
 
    // 創建server
    reactor_create_server(atoi(argv[1]), r);
 
    // 事件循環
    reactor_loop(r);
 
    // 銷燬reactor
    reactor_free(r);
 
    return 0;
}

代碼理解

user_data 每個 fd 對應 1 個 item，存儲 fd 相關事件，回調函數，讀寫緩衝區
user_data_block 鏈表節點，每個結點有 1 個數據塊，包含 1024 個 fd item
reactor 管理所有 fd，包含數據塊鏈表的頭節點
reactor_create 初始化 reactor 結構體
reactor_accept_cb 接受連接，設置讀事件
reactor_read_cb 接收數據，設置寫數據（回發）
reactor_write_cb 發送數據，設置讀事件
reactor_create_server 創建監聽 fd，並添加到 epoll
reactor_loop 事件循環，根據事件類型調用回調函數
reactor_free 釋放 reactor 內存，關閉 socket

運行截圖

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/DjYMlOCGWOOgqqIBcWiU3A

猜你喜歡