Reactor網路模型核心思想探秘
Reactor網路模型核心思想探秘
在網路程式設計系列文章中,我們實作了一個基於epoll的網路框架,並在此基礎上開發了一個簡單的HTTP服務,在那個系列文章中我們使用了讀、寫兩個buffer將網路IO和資料的讀寫進行了分離,它們之間的扭轉完全透過epoll事件通知,如果你認真研究過源碼,會發現,所有針對網路IO的操作都是由事件觸發的。這種基於事件觸發的網路模型通常我們叫做Reactor網路模型。
由於網路程式設計系列文章中程式碼實現相對比較複雜,所以不太好講清楚。所以,我決定單獨出幾篇文章對那個系列文章進行一些拓展,主要涉及網路程式設計思想和效能測試。
這篇文章我們透過實作一個簡單的網路框架,來說明Reactor網路模型實現的一般思路,其本質思想和x-net專案基本上是一樣的,只是在程式碼上做了非常大的精簡,理解起來會輕鬆很多。
首先,我們來看一段程式碼
#include <sys/socket.h>
#include <errno.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <sys/epoll.h>
int main() {
int sockfd = socket(AF_INET, SOCK_STREAM, 0);
struct sockaddr_in servaddr;
memset(&servaddr, 0, sizeof(struct sockaddr_in));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(2048);
if (-1 == bind(sockfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr))) {
perror("bind fail");
return -1;
}
listen(sockfd, 10);
printf("sock-fd:%d\n", sockfd);
int epfd = epoll_create(1);
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.fd = sockfd;
epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &ev);
struct epoll_event events[1024] = {0};
while(1) {
int nready = epoll_wait(epfd, events, 1024, -1);
int i = 0;
for (i = 0; i < nready; i++) {
int connfd = events[i].data.fd;
if (events[i].events & EPOLLIN && sockfd == connfd) {
struct sockaddr_in clientaddr;
socklen_t len = sizeof(clientaddr);
int clientfd = accept(sockfd, (struct sockaddr*)&clientaddr, &len);
ev.events = EPOLLIN | EPOLLET;
ev.data.fd = clientfd;
epoll_ctl(epfd, EPOLL_CTL_ADD, clientfd, &ev);
printf("clientfd: %d\n", clientfd);
} else if (events[i].events & EPOLLIN) {
char buffer[10] = {0};
int count = recv(connfd, buffer, 10, 0);
if (count == 0) {
printf("discounnect\n");
epoll_ctl(epfd, EPOLL_CTL_DEL, connfd, NULL);
close(i);
continue;
}
send(connfd, buffer, count, 0);
printf("clientfd: %d, count: %d, buffer: %s\n", connfd, count, buffer);
}
}
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
- 32.
- 33.
- 34.
- 35.
- 36.
- 37.
- 38.
- 39.
- 40.
- 41.
- 42.
- 43.
- 44.
- 45.
- 46.
- 47.
- 48.
- 49.
- 50.
- 51.
- 52.
- 53.
- 54.
- 55.
- 56.
- 57.
- 58.
- 59.
- 60.
- 61.
- 62.
- 63.
- 64.
- 65.
- 66.
- 67.
- 68.
- 69.
- 70.
- 71.
- 72.
- 73.
- 74.
- 75.
- 76.
- 77.
- 78.
- 79.
- 80.
- 81.
- 82.
- 83.
- 84.
- 85.
- 86.
- 87.
- 88.
- 89.
- 90.
- 91.
- 92.
- 93.
- 94.
- 95.
- 96.
熟悉epoll的人應該對上面的程式碼比較熟悉,這段程式碼的核心在下面的while主循環,如果是當前Server的Socket說明有新的連接進來,調用accept拿到客戶端的fd,將其放在epoll的events中,並註冊EPOLLIN事件,一般我們理解為可讀事件。
如果不是sockfd,表示是客戶端的fd可讀,我們將資料讀出來再原樣發送回去。
上面的程式碼存在的主要問題在於,套接字的accept和讀寫操作我們是直接寫在主循環裡了,這將會讓程式碼的邏輯變得難以琢磨。
對於一個套接字,最直接的操作就是讀寫。所以,最容易想到的就是將讀和寫分開離開。為了實現讀取和寫入分離我們封裝兩個回調函數,如下:
int recv_callback(int fd, char *buffer, int size);
int send_callback(int fd, char *buffer, int size);
- 1.
- 2.
你可以想一下,這兩個函數該怎麼寫?以下是根據原有的邏輯將讀取和寫入封裝在了recv_callback和send_callback兩個函數中,程式碼如下:
int recv_callback(int fd, char *buffer, int size) {
int count = recv(fd, buffer, size, 0);
send_callback(fd, buffer, count, 0);
return count;
}
int send_callback(int fd, char *buffer, int size) {
int count = send(fd, buffer, size, 0);
return count;
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
然後,在主循環中就可以這樣使用
int main() {
...
while(1) {
int nready = epoll_wait(epfd, events, 1024, -1);
int i = 0;
for (i = 0; i < nready; i++) {
int connfd = events[i].data.fd;
if (events[i].events & EPOLLIN && sockfd == connfd) {
...
} else if (events[i].events & EPOLLIN) {
char buffer[10] = {0};
int count = recv_callback(fd, buffer, 10);
if (count == 0) {
printf("disconnect\\n");
epoll_ctl(epfd, EPOLL_CTL_DEL, connfd, NULL);
clise(i);
continue;
}
}
}
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
- 32.
雖然我們將讀取和寫入拆成了兩個方法,但讀和寫並沒有分離開,我們在recv_callback中每次收到資料之後調用send_callback將資料原樣又發回去,在這裡我們希望recv_callback和send_callback各管各的互不干擾,例如像下面這樣
int recv_callback(int fd, char *buffer, int size) {
int count = recv(fd, buffer, size, 0);
return count;
}
int send_callback(int fd, char *buffer, int size) {
int count = send(fd, buffer, size, 0);
return count;
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
但這明顯也是有問題的,在recv_callback中讀完了之後,要如何傳送資料呢?這裡,我們可以想一下,圍繞著一個套接字有哪些部分呢?是不是可以設計出一個類似字典的結構,這個字典的key對應的就是套接字,而value對應的就是圍繞套接字相關的各個組件。
我們將recv_callback和send_callback放在了一個conn_channel結構體中,並且設計了兩個buffer,一個用來讀數據,另一個用來發數據,conn_channel便是這個字典對應的value,代碼如下:
#define BUF_LEN 1024
typedef int(*callback)(int fd);
struct conn_channel {
int fd;
callback recv_call;
callback send_call;
char wbuf[BUF_LEN];
int wlen;
char rbuf[BUF_LEN];
int rlen;
};
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
其中,fd表示的是目前客戶端套接字。然後我們定義一個陣列來表示套接字到套接字value的映射關係,程式碼如下:
struct conn_channel conn_map[1024] = {0};
- 1.
這樣,我們在主循環中,就可以像下面這樣,往conn_map中加入對應的套接字了,程式碼如下:
int main() {
...
while(1) {
int nready = epoll_wait(epfd, events, 1024, -1);
int i = 0;
for (i = 0; i < nready; i++) {
int connfd = events[i].data.fd;
if (events[i].events & EPOLLIN && sockfd == connfd) {
struct sockaddr_in clientaddr;
socklen_t len = sizeof(clientaddr);
int clientfd = accept(sockfd, (struct sockaddr*)&clientaddr, &len);
ev.events = EPOLLIN;
ev.data.fd = clientaddr;
epoll_ctl(epfd, EPOLL_CTL_ADD, clientfd, &ev);
conn_map[clientfd].fd = clientfd;
conn_map[clientfd].rlen = 0;
conn_map[clientfd].wlen = 0;
conn_map[clientfd].recv_call = recv_callback;
conn_map[clientfd].send_call = send_callback;
memset(conn_map[clientfd].rbuf, 0, BUF_LEN);
memset(conn_map[clientfd].wbuf, 0, BUF_LEN);
printf("clientfd:%d\\n", clientfd);
} else if (events[i].events & EPOLLIN) {
...
}
}
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
- 32.
- 33.
- 34.
- 35.
- 36.
- 37.
- 38.
- 39.
- 40.
- 41.
- 42.
- 43.
- 44.
在上面的程式碼中,每當accept出來一個客戶端的套接字,我們就將它放到conn_map中,設定好讀寫buffer和回呼函數。但如果你細心會發現,recv_callback、send_callback和conn_channel中的回呼函數簽章是不一樣的。所以,我們要調整這兩個函數的實現,調整之後程式碼如下:
int recv_callback(int fd) {
int count = recv(fd, conn_map[fd].rbuf + conn_map[fd].rlen, BUF_LEN - conn_map[fd].rlen, 0);
// do something
memcpy(conn_map[fd].wbuf, conn_map[fd].rbuf, conn_map[fd].rlen);
conn_map[fd].wlen = conn_map[fd].rlen;
conn_map[fd].rlen = 0;
return count;
}
int send_callback(int fd) {
int count = send(fd, conn_map[fd].wbuffer, conn_map[fd].wlen, 0);
return count;
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
因為有了conn_map,所以原來傳進來的buffer和size都不需要了,在conn_channel中已經有記錄了。所以只需要一個fd參數就可以了。我們在recv_callback中模擬了回覆訊息,強行將讀到的資料寫到了wbuffer中。這裡補充一下,conn_channel中的rbuffer是用來從套接字中讀取資料的,wbuffer表示的是將要傳送到套接字的資料。
你可以試著把上面的程式碼跑起來,然後你會發現,並沒有照我們的預期執行,send_callback中的send似乎沒有運作。這是因為我們只是將資料從rbuffer寫到了wbuffer中,而send_callback並沒有機會呼叫。你可以想想send_callback放在哪裡呼叫比較合適呢?
在上面的例子中,顯然放在主循環中執行比較合適,在epoll中,EPOLLOUT表示可寫事件,我們可以利用這個事件。在recv_callback執行完之後我們註冊一個EPOLLOUT事件,然後在主循環中我們去監聽EPOLLOUT事件。這樣,當recv_callback將rbuffer的資料複製到wbuffer中之後,send_callback透過EPOLLOUT事件就可以在主循環中得以執行。
為了實現上面的效果我們要修改兩個地方,一個是recv_callback中我們要註冊一下EPOLLOUT事件,程式碼如下:
int recv_callback(int fd) {
int count = recv(fd, conn_map[fd].rbuf + conn_map[fd].rlen, BUF_LEN - conn_map[fd].rlen, 0);
// do something
memcpy(conn_map[fd].wbuf, conn_map[fd].rbuf, conn_map[fd].rlen);
conn_map[fd].wlen = conn_map[fd].rlen;
conn_map[fd].rlen = 0;
struct epoll_event ev;
ev.events = EPOLLOUT;
ev.data.fd = fd;
epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);
return count;
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
我們在rbuf拷貝到wbuf之後,給當前fd註冊了EPOLLOUT事件,然後我們在主循環中要處理EPOLLOUT事件,代碼如下:
int main() {
...
while(1) {
int nready = epoll_wait(epfd, events, 1024, -1);
int i = 0;
for (i = 0; i < nready; i++) {
int connfd = events[i].data.fd;
if (events[i].events & EPOLLIN && sockfd == connfd) {
struct sockaddr_in clientaddr;
socklen_t len = sizeof(clientaddr);
int clientfd = accept(sockfd, (struct sockaddr*)&clientaddr, &len);
ev.events = EPOLLIN;
ev.data.fd = clientaddr;
epoll_ctl(epfd, EPOLL_CTL_ADD, clientfd, &ev);
conn_map[clientfd].fd = clientfd;
conn_map[clientfd].rlen = 0;
conn_map[clientfd].wlen = 0;
conn_map[clientfd].recv_call = recv_callback;
conn_map[clientfd].send_call = send_callback;
memset(conn_map[clientfd].rbuf, 0, BUF_LEN);
memset(conn_map[clientfd].wbuf, 0, BUF_LEN);
printf("clientfd:%d\\n", clientfd);
} else if (events[i].events & EPOLLIN) {
int count = conn_map[connfd].recv_call(connfd);
printf("recv-count:%d\\n", count);
} else if (events[i].events & EPOLLOUT) { // 处理EPOLLOUT事件
int count = conn_map[connfd].send_call(connfd);
printf("send-count:%d\\n", count);
}
}
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
- 32.
- 33.
- 34.
- 35.
- 36.
- 37.
- 38.
- 39.
- 40.
- 41.
- 42.
- 43.
- 44.
- 45.
- 46.
- 47.
- 48.
要注意的是,epfd是在main函數中定義的,而我們在recv_callback中有使用,所以我們可以暫時將epfd宣告成一個全域變量,放在外面。
上面的程式碼有一個問題,EPOLLOUT事件觸發之後你會發現再向當前fd發送數據,就沒響應了,這是因為epoll事件被我們修改了,為了解決這個問題我們可以在send_callback執行完之後再設置回去,如下:
int send_callback(int fd) {
int count = send(fd, conn_map[fd].wbuffer, conn_map[fd].wlen, 0);
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.fd = fd;
epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);
return count;
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
這樣,我們就將IO操作給屏蔽了,在主循環中我們只專注於事件,不同的事件呼叫不同的回呼函數。在對應的回呼函數中只做自己該做的,做完後註冊事件通知其它的回呼函數。
但是,上面的程式碼還不夠優雅,對於accept和讀取事件來講在epoll中都是EPOLLIN事件,這兩個是不是可以合併在一起處理呢?答案是可以的,首先,我們要將accept相關的邏輯給拆出來,拆解之後的程式碼如下:
int accept_callback(int fd) {
struct sockaddr_in clientaddr;
socklen_t len = sizeof(clientaddr);
int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len);
ev.events = EPOLLIN;
ev.data.fd = clientaddr;
epoll_ctl(epfd, EPOLL_CTL_ADD, clientfd, &ev);
conn_map[clientfd].fd = clientfd;
conn_map[clientfd].rlen = 0;
conn_map[clientfd].wlen = 0;
conn_map[clientfd].recv_call = recv_callback;
conn_map[clientfd].send_call = send_callback;
memset(conn_map[clientfd].rbuf, 0, BUF_LEN);
memset(conn_map[clientfd].wbuf, 0, BUF_LEN);
return clientfd;
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
我們發現,accept_callback和recv_callback以及send_callback的簽名是一樣的,這樣我們可以在conn_channel用一個union,將accept_callback也放到conn_channel中來。如下:
struct conn_channel {
int fd;
union {
callback accept_call;
callback recv_call;
} call_t;
callback send_call;
char wbuf[BUF_LEN];
int wlen;
char rbuf[BUF_LEN];
int rlen;
};
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
在主循環中,我們就可以先給sockfd註冊好accept回呼函數,然後我們只需要在主循環中保留兩個邏輯就可以了,程式碼如下:
int main() {
int sockfd = create_serv(9000);
if (sockfd == -1) {
perror("create-server-fail");
return -1;
}
make_nonblocking(sockfd);
epfd = epoll_create1(1);
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.fd = sockfd;
epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &ev);
struct epoll_event events[1024] = {0};
conn_map[sockfd].rlen = 0;
conn_map[sockfd].wlen = 0;
conn_map[sockfd].fd = sockfd;
conn_map[sockfd].call_t.accept_call = accept_callback;
conn_map[sodkfd].send_call = send_callback;
memset(conn_map[sockfd].rbuf, 0, BUF_LEN);
memset(conn_map[sockfd].wbuf, 0, BUF_LEN);
while(1) {
int nready = epoll_wait(epfd, events, 1024, -1);
int i = 0;
for (i = 0; i < nready; i++) {
int connfd = events[i].data.fd;
if (events[i].events & EPOLLIN) {
int count = conn_map[connfd].call_t.recv_call(connfd);
printf("recv-count:%d\\n", count);
} else if (events[i].events & EPOLLOUT) {
int count = conn_map[connfd].send_call(connfd);
printf("send-count:%d\\n", count);
}
}
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
- 32.
- 33.
- 34.
- 35.
- 36.
- 37.
- 38.
- 39.
- 40.
- 41.
- 42.
- 43.
- 44.
- 45.
- 46.
- 47.
- 48.
- 49.
- 50.
- 51.
你可以想一下,我們註冊的是call_t.accept_call,但在呼叫的時候確是call_t.recv_call,為什麼這樣可行?
我們在網頁程式設計系列文章中,單獨為accept抽象化了一個對象,你可以比較一下這兩種實作方式,看看它們有什麼不同?在系列文件中我們為什麼要單獨抽象化一個accepter物件呢?
可以看到,最後主循環中的邏輯,只有兩個分支,這兩個分支代表了兩種事件,這種透過事件驅動的網路模型就是Reactor網路模型。本文為了容易理解,程式碼進行了精簡。在實際的工程中我們還要考慮許多情況。例如,上面的程式碼只支援epoll,我們是不是可以將事件驅動相關的程式碼抽象化成單獨的元件,讓其可以支援其它的事件模型。
本文雖然程式碼簡單,但Reactor網路模型的實作基本上都逃脫不了這個套路,只是在此基礎上可能會將各個部分進行單獨的封裝,例如我們在網路程式設計系列文章中就將channel和map進行了抽象,讓它能適配各種場景。
總結
reactor網路模型是網路程式設計中非常重要的一種程式設計思想,本文透過一個簡短的範例試圖講明白reactor網路程式設計模型的核心思想。當然,本文的實作還不是很完善,例如在呼叫回呼函數的時候還是傳入了fd,我們是否可以不需要這個參數,徹底和IO進行分離呢?