介紹reactor的四種模型
前言
本文將由淺入深的介紹reactor，深入淺出的封裝epoll，一步步變成reactor模型，并在文末介紹reactor的四種模型。
reactor是什么？
reactor是一種高并發(fā)服務(wù)器模型，是一種框架，一個(gè)概念，所以reactor沒有一個(gè)固定的代碼，可以有很多變種，后續(xù)會介紹到。
reactor中的IO使用的是select，poll，epoll這些IO多路復(fù)用，使用IO多路復(fù)用系統(tǒng)不必創(chuàng)建維護(hù)大量線程，只使用一個(gè)線程、一個(gè)選擇器就可同時(shí)處理成千上萬連接，大大減少了系統(tǒng)開銷。
reactor中文譯為反應(yīng)堆，將epoll中的IO變成事件驅(qū)動，比如讀事件，寫事件。來了個(gè)讀事件，立馬進(jìn)行反應(yīng)，執(zhí)行提前注冊好的事件回調(diào)函數(shù)。
回想一下普通函數(shù)調(diào)用的機(jī)制：程序調(diào)用某函數(shù)，函數(shù)執(zhí)行，程序等待，函數(shù)將結(jié)果和控制權(quán)返回給程序，程序繼續(xù)處理。reactor反應(yīng)堆，是一種事件驅(qū)動機(jī)制，和普通函數(shù)調(diào)用的不同之處在于：應(yīng)用程序不是主動的調(diào)用某個(gè) API 完成處理，而是恰恰相反，reactor逆置了事件處理流程，應(yīng)用程序需要提供相應(yīng)的接口并注冊到 reactor上，如果相應(yīng)的事件發(fā)生，reactor將主動調(diào)用應(yīng)用程序注冊的接口，這些接口又稱為“回調(diào)函數(shù)”。
說白了，reactor就是對epoll進(jìn)行封裝，進(jìn)行網(wǎng)絡(luò)IO與業(yè)務(wù)的解耦，將epoll管理IO變成管理事件，整個(gè)程序由事件進(jìn)行驅(qū)動執(zhí)行。就像下圖一樣，有就緒事件返回，reactor：由事件驅(qū)動執(zhí)行對應(yīng)的回調(diào)函數(shù)；epoll：需要自己判斷。
reactor模型三個(gè)重要組件與流程分析
reactor是處理并發(fā) I/O 比較常見的一種模式，用于同步 I/O，中心思想是將所有要處理的 I/O 事件注冊到一個(gè)中心 I/O 多路復(fù)用器（epoll）上，同時(shí)主線程/進(jìn)程阻塞在多路復(fù)用器上；一旦有 I/O 事件到來或是準(zhǔn)備就緒(文件描述符或 socket 可讀、寫)，多路復(fù)用器返回并將事先注冊的相應(yīng) I/O 事件分發(fā)到對應(yīng)的處理器中。
組件
reactor模型有三個(gè)重要的組件
多路復(fù)用器：由操作系統(tǒng)提供，在 linux 上一般是 select, poll, epoll 等系統(tǒng)調(diào)用。
事件分發(fā)器：將多路復(fù)用器中返回的就緒事件分到對應(yīng)的處理函數(shù)中。
事件處理器：負(fù)責(zé)處理特定事件的處理函數(shù)。
流程
具體流程：
- 注冊相應(yīng)的事件處理器（剛開始listenfd注冊都就緒事件）
- 多路復(fù)用器等待事件
- 事件到來，激活事件分發(fā)器，分發(fā)器調(diào)用事件到對應(yīng)的處理器
- 事件處理器處理事件，然后注冊新的事件（比如讀事件，完成讀操作后，根據(jù)業(yè)務(wù)處理數(shù)據(jù)，注冊寫事件，寫事件根據(jù)業(yè)務(wù)響應(yīng)請求；比如listen讀事件，肯定要給新的連接注冊讀事件）
將epoll封裝成reactor事件驅(qū)動
封裝每一個(gè)連接sockfd變成ntyevent
我們知道一個(gè)連接對應(yīng)一個(gè)文件描述符fd，對于這個(gè)連接（fd）來說，它有自己的事件（讀，寫）。我們將fd都設(shè)置成非阻塞的，所以這里我們需要添加兩個(gè)buffer，至于大小就是看業(yè)務(wù)需求了。
struct ntyevent {
int fd;//socket fd
int events;//事件
char sbuffer[BUFFER_LENGTH];//寫緩沖buffer
int slength;
char rbuffer[BUFFER_LENGTH];//讀緩沖buffer
int rlength;
// typedef int (*NtyCallBack)(int, int, void *);
NtyCallBack callback;//回調(diào)函數(shù)
void *arg;
int status;//1MOD 0 null
};
int fd;//socket fd
int events;//事件
char sbuffer[BUFFER_LENGTH];//寫緩沖buffer
int slength;
char rbuffer[BUFFER_LENGTH];//讀緩沖buffer
int rlength;
// typedef int (*NtyCallBack)(int, int, void *);
NtyCallBack callback;//回調(diào)函數(shù)
void *arg;
int status;//1MOD 0 null
};
封裝epfd和ntyevent變成ntyreactor
我們知道socket fd已經(jīng)被封裝成了ntyevent，那么有多少個(gè)ntyevent呢？這里demo初始化reactor的時(shí)候其實(shí)是將*events指向了一個(gè)1024的ntyevent數(shù)組(按照道理來說客戶端連接可以一直連，不止1024個(gè)客戶端，后續(xù)文章有解決方案，這里從簡)。epfd肯定要封裝進(jìn)行，不用多說。
struct ntyreactor {
int epfd;
struct ntyevent *events;
//struct ntyevent events[1024];
};
int epfd;
struct ntyevent *events;
//struct ntyevent events[1024];
};
封裝讀、寫、接收連接等事件對應(yīng)的操作變成callback
前面已經(jīng)說了，把事件寫成回調(diào)函數(shù)，這里的參數(shù)fd肯定要知道自己的哪個(gè)連接，events是什么事件的意思，arg傳的是ntyreactor （考慮到后續(xù)多線程多進(jìn)程，如果將ntyreactor設(shè)為全局感覺不太好 ）
typedef int (*NtyCallBack)(int, int, void *);
int recv_cb(int fd, int events, void *arg);
int send_cb(int fd, int events, void *arg);
int accept_cb(int fd, int events, void *arg);
int recv_cb(int fd, int events, void *arg);
int send_cb(int fd, int events, void *arg);
int accept_cb(int fd, int events, void *arg);
給每個(gè)客戶端的ntyevent設(shè)置屬性
具兩個(gè)例子，我們知道第一個(gè)socket一定是listenfd，用來監(jiān)聽用的，那么首先肯定是設(shè)置ntyevent的各項(xiàng)屬性。 本來是讀事件，讀完后要改成寫事件，那么必然要把原來的讀回調(diào)函數(shù)設(shè)置成寫事件回調(diào)。
void nty_event_set(struct ntyevent *ev, int fd, NtyCallBack callback, void *arg) {
ev->fd = fd;
ev->callback = callback;
ev->events = 0;
ev->arg = arg;
}
ev->fd = fd;
ev->callback = callback;
ev->events = 0;
ev->arg = arg;
}
將ntyevent加入到epoll中由內(nèi)核監(jiān)聽
int nty_event_add(int epfd, int events, struct ntyevent *ntyev) {
struct epoll_event ev = {0, {0}};
ev.data.ptr = ntyev;
ev.events = ntyev->events = events;
int op;
if (ntyev->status == 1) {
op = EPOLL_CTL_MOD;
}
else {
op = EPOLL_CTL_ADD;
ntyev->status = 1;
}
if (epoll_ctl(epfd, op, ntyev->fd, &ev) < 0) {
printf("event add failed [fd=%d], events[%d],err:%s,err:%dn", ntyev->fd, events, strerror(errno), errno);
return -1;
}
return 0;
}
struct epoll_event ev = {0, {0}};
ev.data.ptr = ntyev;
ev.events = ntyev->events = events;
int op;
if (ntyev->status == 1) {
op = EPOLL_CTL_MOD;
}
else {
op = EPOLL_CTL_ADD;
ntyev->status = 1;
}
if (epoll_ctl(epfd, op, ntyev->fd, &ev) < 0) {
printf("event add failed [fd=%d], events[%d],err:%s,err:%dn", ntyev->fd, events, strerror(errno), errno);
return -1;
}
return 0;
}
將ntyevent從epoll中去除
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);
//epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, NULL);
return 0;
}
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);
//epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, NULL);
return 0;
}
讀事件回調(diào)函數(shù)
這里就是被觸發(fā)的回調(diào)函數(shù)，具體代碼要與業(yè)務(wù)結(jié)合，這里的參考意義不大（這里就是讀一次，改成寫事件）
int recv_cb(int fd, int events, void *arg) {
struct ntyreactor *reactor = (struct ntyreactor *) arg;
struct ntyevent *ntyev = &reactor->events[fd];
int len = recv(fd, ntyev->buffer, BUFFER_LENGTH, 0);
nty_event_del(reactor->epfd, ntyev);
if (len > 0) {
ntyev->length = len;
ntyev->buffer[len] = '?';
printf("C[%d]:%sn", fd, ntyev->buffer);
nty_event_set(ntyev, fd, send_cb, reactor);
nty_event_add(reactor->epfd, EPOLLOUT, ntyev);
}
else if (len == 0) {
close(ntyev->fd);
printf("[fd=%d] pos[%ld], closedn", fd, ntyev - reactor->events);
}
else {
close(ntyev->fd);
printf("recv[fd=%d] error[%d]:%sn", fd, errno, strerror(errno));
}
return len;
}
struct ntyreactor *reactor = (struct ntyreactor *) arg;
struct ntyevent *ntyev = &reactor->events[fd];
int len = recv(fd, ntyev->buffer, BUFFER_LENGTH, 0);
nty_event_del(reactor->epfd, ntyev);
if (len > 0) {
ntyev->length = len;
ntyev->buffer[len] = '?';
printf("C[%d]:%sn", fd, ntyev->buffer);
nty_event_set(ntyev, fd, send_cb, reactor);
nty_event_add(reactor->epfd, EPOLLOUT, ntyev);
}
else if (len == 0) {
close(ntyev->fd);
printf("[fd=%d] pos[%ld], closedn", fd, ntyev - reactor->events);
}
else {
close(ntyev->fd);
printf("recv[fd=%d] error[%d]:%sn", fd, errno, strerror(errno));
}
return len;
}
寫事件回調(diào)函數(shù)
這里就是被觸發(fā)的回調(diào)函數(shù)，具體代碼要與業(yè)務(wù)結(jié)合，這里的參考意義不大（將讀事件讀的數(shù)據(jù)寫回，再改成讀事件，相當(dāng)于echo）
int send_cb(int fd, int events, void *arg) {
struct ntyreactor *reactor = (struct ntyreactor *) arg;
struct ntyevent *ntyev = &reactor->events[fd];
int len = send(fd, ntyev->buffer, ntyev->length, 0);
if (len > 0) {
printf("send[fd=%d], [%d]%sn", fd, len, ntyev->buffer);
nty_event_del(reactor->epfd, ntyev);
nty_event_set(ntyev, fd, recv_cb, reactor);
nty_event_add(reactor->epfd, EPOLLIN, ntyev);
}
else {
close(ntyev->fd);
nty_event_del(reactor->epfd, ntyev);
printf("send[fd=%d] error %sn", fd, strerror(errno));
}
return len;
}
struct ntyreactor *reactor = (struct ntyreactor *) arg;
struct ntyevent *ntyev = &reactor->events[fd];
int len = send(fd, ntyev->buffer, ntyev->length, 0);
if (len > 0) {
printf("send[fd=%d], [%d]%sn", fd, len, ntyev->buffer);
nty_event_del(reactor->epfd, ntyev);
nty_event_set(ntyev, fd, recv_cb, reactor);
nty_event_add(reactor->epfd, EPOLLIN, ntyev);
}
else {
close(ntyev->fd);
nty_event_del(reactor->epfd, ntyev);
printf("send[fd=%d] error %sn", fd, strerror(errno));
}
return len;
}
接受新連接事件回調(diào)函數(shù)
本質(zhì)上就是accept，然后將其加入到epoll監(jiān)聽
int accept_cb(int fd, int events, void *arg) {
struct ntyreactor *reactor = (struct ntyreactor *) arg;
if (reactor == NULL) return -1;
struct sockaddr_in client_addr;
socklen_t len = sizeof(client_addr);
int clientfd;
if ((clientfd = accept(fd, (struct sockaddr *) &client_addr, &len)) == -1) {
printf("accept: %sn", strerror(errno));
return -1;
}
printf("client fd = %dn", clientfd);
if ((fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) {
printf("%s: fcntl nonblocking failed, %dn", __func__, MAX_EPOLL_EVENTS);
return -1;
}
nty_event_set(&reactor->events[clientfd], clientfd, recv_cb, reactor);
nty_event_add(reactor->epfd, EPOLLIN, &reactor->events[clientfd]);
printf("new connect [%s:%d][time:%ld], pos[%d]n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port),
reactor->events[clientfd].last_active, clientfd);
return 0;
}
struct ntyreactor *reactor = (struct ntyreactor *) arg;
if (reactor == NULL) return -1;
struct sockaddr_in client_addr;
socklen_t len = sizeof(client_addr);
int clientfd;
if ((clientfd = accept(fd, (struct sockaddr *) &client_addr, &len)) == -1) {
printf("accept: %sn", strerror(errno));
return -1;
}
printf("client fd = %dn", clientfd);
if ((fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) {
printf("%s: fcntl nonblocking failed, %dn", __func__, MAX_EPOLL_EVENTS);
return -1;
}
nty_event_set(&reactor->events[clientfd], clientfd, recv_cb, reactor);
nty_event_add(reactor->epfd, EPOLLIN, &reactor->events[clientfd]);
printf("new connect [%s:%d][time:%ld], pos[%d]n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port),
reactor->events[clientfd].last_active, clientfd);
return 0;
}
reactor運(yùn)行
就是將原來的epoll_wait從main函數(shù)中封裝到ntyreactor_run函數(shù)中
int ntyreactor_run(struct ntyreactor *reactor) {
if (reactor == NULL) return -1;
if (reactor->epfd < 0) return -1;
if (reactor->events == NULL) return -1;
struct epoll_event events[MAX_EPOLL_EVENTS];
int checkpos = 0, i;
while (1) {
int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);
if (nready < 0) {
printf("epoll_wait error, exitn");
continue;
}
for (i = 0; i < nready; i++) {
struct ntyevent *ev = (struct ntyevent *) events[i].data.ptr;
ev->callback(ev->fd, events[i].events, ev->arg);
}
}
}
if (reactor == NULL) return -1;
if (reactor->epfd < 0) return -1;
if (reactor->events == NULL) return -1;
struct epoll_event events[MAX_EPOLL_EVENTS];
int checkpos = 0, i;
while (1) {
int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);
if (nready < 0) {
printf("epoll_wait error, exitn");
continue;
}
for (i = 0; i < nready; i++) {
struct ntyevent *ev = (struct ntyevent *) events[i].data.ptr;
ev->callback(ev->fd, events[i].events, ev->arg);
}
}
}
reactor簡單版代碼與測試
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define BUFFER_LENGTH 4096
#define MAX_EPOLL_EVENTS 1024
#define SERVER_PORT 8082
typedef int (*NtyCallBack)(int, int, void *);
struct ntyevent {
int fd;
int events;
void *arg;
NtyCallBack callback;
int status;//1MOD 0 null
char buffer[BUFFER_LENGTH];
int length;
long last_active;
};
struct ntyreactor {
int epfd;
struct ntyevent *events;
};
int recv_cb(int fd, int events, void *arg);
int send_cb(int fd, int events, void *arg);
int accept_cb(int fd, int events, void *arg);
void nty_event_set(struct ntyevent *ev, int fd, NtyCallBack callback, void *arg) {
ev->fd = fd;
ev->callback = callback;
ev->events = 0;
ev->arg = arg;
ev->last_active = time(NULL);
}
int nty_event_add(int epfd, int events, struct ntyevent *ntyev) {
struct epoll_event ev = {0, {0}};
ev.data.ptr = ntyev;
ev.events = ntyev->events = events;
int op;
if (ntyev->status == 1) {
op = EPOLL_CTL_MOD;
}
else {
op = EPOLL_CTL_ADD;
ntyev->status = 1;
}
if (epoll_ctl(epfd, op, ntyev->fd, &ev) < 0) {
printf("event add failed [fd=%d], events[%d],err:%s,err:%dn", ntyev->fd, events, strerror(errno), errno);
return -1;
}
return 0;
}
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);
//epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, NULL);
return 0;
}
int recv_cb(int fd, int events, void *arg) {
struct ntyreactor *reactor = (struct ntyreactor *) arg;
struct ntyevent *ntyev = &reactor->events[fd];
int len = recv(fd, ntyev->buffer, BUFFER_LENGTH, 0);
nty_event_del(reactor->epfd, ntyev);
if (len > 0) {
ntyev->length = len;
ntyev->buffer[len] = '?';
printf("C[%d]:%sn", fd, ntyev->buffer);
nty_event_set(ntyev, fd, send_cb, reactor);
nty_event_add(reactor->epfd, EPOLLOUT, ntyev);
}
else if (len == 0) {
close(ntyev->fd);
printf("[fd=%d] pos[%ld], closedn", fd, ntyev - reactor->events);
}
else {
close(ntyev->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 *ntyev = &reactor->events[fd];
int len = send(fd, ntyev->buffer, ntyev->length, 0);
if (len > 0) {
printf("send[fd=%d], [%d]%sn", fd, len, ntyev->buffer);
nty_event_del(reactor->epfd, ntyev);
nty_event_set(ntyev, fd, recv_cb, reactor);
nty_event_add(reactor->epfd, EPOLLIN, ntyev);
}
else {
close(ntyev->fd);
nty_event_del(reactor->epfd, ntyev);
printf("send[fd=%d] error %sn", fd, strerror(errno));
}
return len;
}
int accept_cb(int fd, int events, void *arg) {
struct ntyreactor *reactor = (struct ntyreactor *) arg;
if (reactor == NULL) return -1;
struct sockaddr_in client_addr;
socklen_t len = sizeof(client_addr);
int clientfd;
if ((clientfd = accept(fd, (struct sockaddr *) &client_addr, &len)) == -1) {
printf("accept: %sn", strerror(errno));
return -1;
}
printf("client fd = %dn", clientfd);
if ((fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) {
printf("%s: fcntl nonblocking failed, %dn", __func__, MAX_EPOLL_EVENTS);
return -1;
}
nty_event_set(&reactor->events[clientfd], clientfd, recv_cb, reactor);
nty_event_add(reactor->epfd, EPOLLIN, &reactor->events[clientfd]);
printf("new connect [%s:%d][time:%ld], pos[%d]n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port),
reactor->events[clientfd].last_active, clientfd);
return 0;
}
int init_sock(short port) {
int fd = socket(AF_INET, SOCK_STREAM, 0);
struct sockaddr_in server_addr;
memset(&server_addr, 0, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
server_addr.sin_port = htons(port);
bind(fd, (struct sockaddr *) &server_addr, sizeof(server_addr));
if (listen(fd, 20) < 0) {
printf("listen failed : %sn", strerror(errno));
}
return fd;
}
int ntyreactor_init(struct ntyreactor *reactor) {
if (reactor == NULL) return -1;
memset(reactor, 0, sizeof(struct ntyreactor));
reactor->epfd = epoll_create(1);
if (reactor->epfd <= 0) {
printf("create epfd in %s err %sn", __func__, strerror(errno));
return -2;
}
reactor->events = (struct ntyevent *) malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
memset(reactor->events, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
if (reactor->events == NULL) {
printf("create epll events in %s err %sn", __func__, strerror(errno));
close(reactor->epfd);
return -3;
}
return 0;
}
int ntyreactor_destory(struct ntyreactor *reactor) {
close(reactor->epfd);
free(reactor->events);
}
int ntyreactor_addlistener(struct ntyreactor *reactor, int sockfd, NtyCallBack acceptor) {
if (reactor == NULL) return -1;
if (reactor->events == NULL) return -1;
nty_event_set(&reactor->events[sockfd], sockfd, acceptor, reactor);
nty_event_add(reactor->epfd, EPOLLIN, &reactor->events[sockfd]);
return 0;
}
_Noreturn int ntyreactor_run(struct ntyreactor *reactor) {
if (reactor == NULL) return -1;
if (reactor->epfd < 0) return -1;
if (reactor->events == NULL) return -1;
struct epoll_event events[MAX_EPOLL_EVENTS];
int checkpos = 0, i;
while (1) {
//心跳包 60s 超時(shí)則斷開連接
long now = time(NULL);
for (i = 0; i < 100; i++, checkpos++) {
if (checkpos == MAX_EPOLL_EVENTS) {
checkpos = 0;
}
if (reactor->events[checkpos].status != 1 || checkpos == 3) {
continue;
}
long duration = now - reactor->events[checkpos].last_active;
if (duration >= 60) {
close(reactor->events[checkpos].fd);
printf("[fd=%d] timeoutn", reactor->events[checkpos].fd);
nty_event_del(reactor->epfd, &reactor->events[checkpos]);
}
}
int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);
if (nready < 0) {
printf("epoll_wait error, exitn");
continue;
}
for (i = 0; i < nready; i++) {
struct ntyevent *ev = (struct ntyevent *) events[i].data.ptr;
ev->callback(ev->fd, events[i].events, ev->arg);
}
}
}
int main(int argc, char *argv[]) {
int sockfd = init_sock(SERVER_PORT);
struct ntyreactor *reactor = (struct ntyreactor *) malloc(sizeof(struct ntyreactor));
if (ntyreactor_init(reactor) != 0) {
return -1;
}
ntyreactor_addlistener(reactor, sockfd, accept_cb);
ntyreactor_run(reactor);
ntyreactor_destory(reactor);
close(sockfd);
return 0;
}
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define BUFFER_LENGTH 4096
#define MAX_EPOLL_EVENTS 1024
#define SERVER_PORT 8082
typedef int (*NtyCallBack)(int, int, void *);
struct ntyevent {
int fd;
int events;
void *arg;
NtyCallBack callback;
int status;//1MOD 0 null
char buffer[BUFFER_LENGTH];
int length;
long last_active;
};
struct ntyreactor {
int epfd;
struct ntyevent *events;
};
int recv_cb(int fd, int events, void *arg);
int send_cb(int fd, int events, void *arg);
int accept_cb(int fd, int events, void *arg);
void nty_event_set(struct ntyevent *ev, int fd, NtyCallBack callback, void *arg) {
ev->fd = fd;
ev->callback = callback;
ev->events = 0;
ev->arg = arg;
ev->last_active = time(NULL);
}
int nty_event_add(int epfd, int events, struct ntyevent *ntyev) {
struct epoll_event ev = {0, {0}};
ev.data.ptr = ntyev;
ev.events = ntyev->events = events;
int op;
if (ntyev->status == 1) {
op = EPOLL_CTL_MOD;
}
else {
op = EPOLL_CTL_ADD;
ntyev->status = 1;
}
if (epoll_ctl(epfd, op, ntyev->fd, &ev) < 0) {
printf("event add failed [fd=%d], events[%d],err:%s,err:%dn", ntyev->fd, events, strerror(errno), errno);
return -1;
}
return 0;
}
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);
//epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, NULL);
return 0;
}
int recv_cb(int fd, int events, void *arg) {
struct ntyreactor *reactor = (struct ntyreactor *) arg;
struct ntyevent *ntyev = &reactor->events[fd];
int len = recv(fd, ntyev->buffer, BUFFER_LENGTH, 0);
nty_event_del(reactor->epfd, ntyev);
if (len > 0) {
ntyev->length = len;
ntyev->buffer[len] = '?';
printf("C[%d]:%sn", fd, ntyev->buffer);
nty_event_set(ntyev, fd, send_cb, reactor);
nty_event_add(reactor->epfd, EPOLLOUT, ntyev);
}
else if (len == 0) {
close(ntyev->fd);
printf("[fd=%d] pos[%ld], closedn", fd, ntyev - reactor->events);
}
else {
close(ntyev->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 *ntyev = &reactor->events[fd];
int len = send(fd, ntyev->buffer, ntyev->length, 0);
if (len > 0) {
printf("send[fd=%d], [%d]%sn", fd, len, ntyev->buffer);
nty_event_del(reactor->epfd, ntyev);
nty_event_set(ntyev, fd, recv_cb, reactor);
nty_event_add(reactor->epfd, EPOLLIN, ntyev);
}
else {
close(ntyev->fd);
nty_event_del(reactor->epfd, ntyev);
printf("send[fd=%d] error %sn", fd, strerror(errno));
}
return len;
}
int accept_cb(int fd, int events, void *arg) {
struct ntyreactor *reactor = (struct ntyreactor *) arg;
if (reactor == NULL) return -1;
struct sockaddr_in client_addr;
socklen_t len = sizeof(client_addr);
int clientfd;
if ((clientfd = accept(fd, (struct sockaddr *) &client_addr, &len)) == -1) {
printf("accept: %sn", strerror(errno));
return -1;
}
printf("client fd = %dn", clientfd);
if ((fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) {
printf("%s: fcntl nonblocking failed, %dn", __func__, MAX_EPOLL_EVENTS);
return -1;
}
nty_event_set(&reactor->events[clientfd], clientfd, recv_cb, reactor);
nty_event_add(reactor->epfd, EPOLLIN, &reactor->events[clientfd]);
printf("new connect [%s:%d][time:%ld], pos[%d]n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port),
reactor->events[clientfd].last_active, clientfd);
return 0;
}
int init_sock(short port) {
int fd = socket(AF_INET, SOCK_STREAM, 0);
struct sockaddr_in server_addr;
memset(&server_addr, 0, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
server_addr.sin_port = htons(port);
bind(fd, (struct sockaddr *) &server_addr, sizeof(server_addr));
if (listen(fd, 20) < 0) {
printf("listen failed : %sn", strerror(errno));
}
return fd;
}
int ntyreactor_init(struct ntyreactor *reactor) {
if (reactor == NULL) return -1;
memset(reactor, 0, sizeof(struct ntyreactor));
reactor->epfd = epoll_create(1);
if (reactor->epfd <= 0) {
printf("create epfd in %s err %sn", __func__, strerror(errno));
return -2;
}
reactor->events = (struct ntyevent *) malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
memset(reactor->events, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
if (reactor->events == NULL) {
printf("create epll events in %s err %sn", __func__, strerror(errno));
close(reactor->epfd);
return -3;
}
return 0;
}
int ntyreactor_destory(struct ntyreactor *reactor) {
close(reactor->epfd);
free(reactor->events);
}
int ntyreactor_addlistener(struct ntyreactor *reactor, int sockfd, NtyCallBack acceptor) {
if (reactor == NULL) return -1;
if (reactor->events == NULL) return -1;
nty_event_set(&reactor->events[sockfd], sockfd, acceptor, reactor);
nty_event_add(reactor->epfd, EPOLLIN, &reactor->events[sockfd]);
return 0;
}
_Noreturn int ntyreactor_run(struct ntyreactor *reactor) {
if (reactor == NULL) return -1;
if (reactor->epfd < 0) return -1;
if (reactor->events == NULL) return -1;
struct epoll_event events[MAX_EPOLL_EVENTS];
int checkpos = 0, i;
while (1) {
//心跳包 60s 超時(shí)則斷開連接
long now = time(NULL);
for (i = 0; i < 100; i++, checkpos++) {
if (checkpos == MAX_EPOLL_EVENTS) {
checkpos = 0;
}
if (reactor->events[checkpos].status != 1 || checkpos == 3) {
continue;
}
long duration = now - reactor->events[checkpos].last_active;
if (duration >= 60) {
close(reactor->events[checkpos].fd);
printf("[fd=%d] timeoutn", reactor->events[checkpos].fd);
nty_event_del(reactor->epfd, &reactor->events[checkpos]);
}
}
int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);
if (nready < 0) {
printf("epoll_wait error, exitn");
continue;
}
for (i = 0; i < nready; i++) {
struct ntyevent *ev = (struct ntyevent *) events[i].data.ptr;
ev->callback(ev->fd, events[i].events, ev->arg);
}
}
}
int main(int argc, char *argv[]) {
int sockfd = init_sock(SERVER_PORT);
struct ntyreactor *reactor = (struct ntyreactor *) malloc(sizeof(struct ntyreactor));
if (ntyreactor_init(reactor) != 0) {
return -1;
}
ntyreactor_addlistener(reactor, sockfd, accept_cb);
ntyreactor_run(reactor);
ntyreactor_destory(reactor);
close(sockfd);
return 0;
}
reactor優(yōu)點(diǎn)
reactor模式是編寫高性能網(wǎng)絡(luò)服務(wù)器的必備技術(shù)之一，它具有如下優(yōu)點(diǎn)：
- 響應(yīng)快，不必為單個(gè)同步時(shí)間所阻塞，雖然 reactor本身依然是同步的
- 編程相對簡單，可以最大程度的避免復(fù)雜的多線程及同步問題，并且避免了多線程/進(jìn)程的切換開銷
- 可擴(kuò)展性，可以方便的通過增加 reactor實(shí)例個(gè)數(shù)來充分利用 CPU 資源
- 可復(fù)用性，reactor 框架本身與具體事件處理邏輯無關(guān)，具有很高的復(fù)用性
reactor模型開發(fā)效率上比起直接使用 IO 復(fù)用要高，它通常是單線程的，設(shè)計(jì)目標(biāo)是希望單線程使用一顆 CPU 的全部資源，但也有附帶優(yōu)點(diǎn)，即每個(gè)事件處理中很多時(shí)候可以不考慮共享資源的互斥訪問?？墒侨秉c(diǎn)也是明顯的，現(xiàn)在的硬件發(fā)展，已經(jīng)不再遵循摩爾定律，CPU 的頻率受制于材料的限制不再有大的提升，而改為是從核數(shù)的增加上提升能力，當(dāng)程序需要使用多核資源時(shí)，reactor模型就會悲劇。
如果程序業(yè)務(wù)很簡單，例如只是簡單的訪問一些提供了并發(fā)訪問的服務(wù)，就可以直接開啟多個(gè)反應(yīng)堆，每個(gè)反應(yīng)堆對應(yīng)一顆 CPU 核心，這些反應(yīng)堆上跑的請求互不相關(guān)，這是完全可以利用多核的。例如 Nginx 這樣的 http 靜態(tài)服務(wù)器。
reactor多種模型
單reactor + 單線程模型
單reactor單線程模型，指的是所有的 IO 操作（讀，寫，建立連接）都在同一個(gè)線程上面完成
缺點(diǎn)：
- 由于只有一個(gè)線程，因此事件是順序處理的，一個(gè)線程同時(shí)只能做一件事情，事件的優(yōu)先級得不到保證
- 不能充分利用多核CPU
單reactor + 線程池(Thread Pool)模型
相比于單reactor單線程模型，此模型中收到請求后，不在reactor線程計(jì)算，而是使用線程池來計(jì)算，這會充分的利用多核CPU。采用此模式時(shí)有可能存在多個(gè)線程同時(shí)計(jì)算同一個(gè)連接上的多個(gè)請求，算出的結(jié)果的次序是不確定的， 所以需要網(wǎng)絡(luò)框架在設(shè)計(jì)協(xié)議時(shí)帶一個(gè)id標(biāo)示，以便以便讓客戶端區(qū)分response對應(yīng)的是哪個(gè)request。
多reactor + 多線程模型
此模式的特點(diǎn)是每個(gè)線程一個(gè)循環(huán)， 有一個(gè)main reactor負(fù)責(zé)accept連接, 然后把該連接掛在某個(gè)sub reactor中，這樣該連接的所有操作都在那個(gè)sub reactor所處的線程中完成。多個(gè)連接可能被分配到多個(gè)線程中，充分利用CPU。在應(yīng)用場景中，reactor的個(gè)數(shù)可以采用 固定的個(gè)數(shù)，比如跟CPU數(shù)目一致。此模型與單reactor多線程模型相比，減少了進(jìn)出thread pool兩次上下文切換，小規(guī)模的計(jì)算可以在當(dāng)前IO線程完成并且返回結(jié)果，降低響應(yīng)的延遲。并可以有效防止當(dāng)IO壓力過大時(shí)一個(gè)reactor處理能力飽和問題。
多reactor + 線程池(Thread Pool)模型
此模型是上面兩個(gè)的混合體，它既使用多個(gè) reactors 來處理 IO，又使用線程池來處理計(jì)算。此模式適適合既有突發(fā)IO(利用Multiple Reactor分擔(dān))，又有突發(fā)計(jì)算的應(yīng)用（利用線程池把一個(gè)連接上的計(jì)算任務(wù)分配給多個(gè)線程）。
注意點(diǎn)
注意：前面介紹的四種reactor 模式在具體實(shí)現(xiàn)時(shí)為了簡應(yīng)該遵循的原則是：每個(gè)文件描述符只由一個(gè)線程操作。這樣可以輕輕松松解決消息收發(fā)的順序性問題，也避免了關(guān)閉文件描述符的各種race condition。一個(gè)線程可以操作多個(gè)文件描述符，但是一個(gè)線程不能操作別的線程擁有的文件描述符。這一點(diǎn)不難做到。epoll也遵循了相同的原則。Linux文檔中并沒有說明，當(dāng)一個(gè)線程證阻塞在epoll_wait時(shí)，另一個(gè)線程往epoll fd添加一個(gè)新的監(jiān)控fd會發(fā)生什么。新fd上的事件會不會在此次epoll_wait調(diào)用中返回？為了穩(wěn)妥起見，我們應(yīng)該吧對同一個(gè) epoll fd的操作(添加、刪除、修改等等)都放到同一個(gè)線程中執(zhí)行。
reactor完善版代碼
由于fd的數(shù)量未知，這里設(shè)計(jì)ntyreactor 里面包含 eventblock ，eventblock 包含1024個(gè)fd。每個(gè)fd通過 fd/1024定位到在第幾個(gè)eventblock，通過fd%1024定位到在eventblock第幾個(gè)位置。
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define BUFFER_LENGTH 4096
#define MAX_EPOLL_EVENTS 1024
#define SERVER_PORT 8081
#define PORT_COUNT 100
typedef int (*NCALLBACK)(int, int, void *);
struct ntyevent {
int fd;
int events;
void *arg;
NCALLBACK callback;
int status;
char buffer[BUFFER_LENGTH];
int length;
};
struct eventblock {
struct eventblock *next;
struct ntyevent *events;
};
struct ntyreactor {
int epfd;
int blkcnt;
struct eventblock *evblk;
};
int recv_cb(int fd, int events, void *arg);
int send_cb(int fd, int events, void *arg);
struct ntyevent *ntyreactor_find_event_idx(struct ntyreactor *reactor, int sockfd);
void nty_event_set(struct ntyevent *ev, int fd, NCALLBACK *callback, void *arg) {
ev->fd = fd;
ev->callback = callback;
ev->events = 0;
ev->arg = arg;
}
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 failed [fd=%d], events[%d]n", ev->fd, events);
return -1;
}
return 0;
}
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 recv_cb(int fd, int events, void *arg) {
struct ntyreactor *reactor = (struct ntyreactor *) arg;
struct ntyevent *ev = ntyreactor_find_event_idx(reactor, fd);
int len = recv(fd, ev->buffer, BUFFER_LENGTH, 0); //
nty_event_del(reactor->epfd, ev);
if (len > 0) {
ev->length = len;
ev->buffer[len] = '?';
// printf("recv[%d]:%sn", fd, ev->buffer);
printf("recv fd=[%dn", fd);
nty_event_set(ev, fd, send_cb, reactor);
nty_event_add(reactor->epfd, EPOLLOUT, ev);
}
else if (len == 0) {
close(ev->fd);
//printf("[fd=%d] pos[%ld], closedn", fd, ev-reactor->events);
}
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_find_event_idx(reactor, fd);
int len = send(fd, ev->buffer, ev->length, 0);
if (len > 0) {
// printf("send[fd=%d], [%d]%sn", fd, len, ev->buffer);
printf("send fd=[%dn]", fd);
nty_event_del(reactor->epfd, ev);
nty_event_set(ev, fd, recv_cb, reactor);
nty_event_add(reactor->epfd, EPOLLIN, ev);
}
else {
nty_event_del(reactor->epfd, ev);
close(ev->fd);
printf("send[fd=%d] error %sn", fd, strerror(errno));
}
return len;
}
int accept_cb(int fd, int events, void *arg) {//非阻塞
struct ntyreactor *reactor = (struct ntyreactor *) arg;
if (reactor == NULL) return -1;
struct sockaddr_in client_addr;
socklen_t len = sizeof(client_addr);
int clientfd;
if ((clientfd = accept(fd, (struct sockaddr *) &client_addr, &len)) == -1) {
printf("accept: %sn", strerror(errno));
return -1;
}
if ((fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) {
printf("%s: fcntl nonblocking failed, %dn", __func__, MAX_EPOLL_EVENTS);
return -1;
}
struct ntyevent *event = ntyreactor_find_event_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_addr.sin_addr), ntohs(client_addr.sin_port), clientfd);
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_family = AF_INET;
server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
server_addr.sin_port = htons(port);
bind(fd, (struct sockaddr *) &server_addr, sizeof(server_addr));
if (listen(fd, 20) < 0) {
printf("listen failed : %sn", strerror(errno));
}
return fd;
}
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 *) malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
if (evs == NULL) {
printf("ntyreactor_alloc ntyevents failedn");
return -2;
}
memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
struct eventblock *block = (struct eventblock *) malloc(sizeof(struct eventblock));
if (block == NULL) {
printf("ntyreactor_alloc eventblock failedn");
return -2;
}
memset(block, 0, sizeof(struct eventblock));
block->events = evs;
block->next = NULL;
blk->next = block;
reactor->blkcnt++; //
return 0;
}
struct ntyevent *ntyreactor_find_event_idx(struct ntyreactor *reactor, int sockfd) {
int blkidx = sockfd / MAX_EPOLL_EVENTS;
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_EVENTS];
}
int ntyreactor_init(struct ntyreactor *reactor) {
if (reactor == NULL) return -1;
memset(reactor, 0, sizeof(struct ntyreactor));
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 *) malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
if (evs == NULL) {
printf("ntyreactor_alloc ntyevents failedn");
return -2;
}
memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
struct eventblock *block = (struct eventblock *) malloc(sizeof(struct eventblock));
if (block == NULL) {
printf("ntyreactor_alloc eventblock failedn");
return -2;
}
memset(block, 0, sizeof(struct eventblock));
block->events = evs;
block->next = NULL;
reactor->evblk = block;
reactor->blkcnt = 1;
return 0;
}
int ntyreactor_destory(struct ntyreactor *reactor) {
close(reactor->epfd);
//free(reactor->events);
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 ntyreactor_addlistener(struct ntyreactor *reactor, int sockfd, NCALLBACK *acceptor) {
if (reactor == NULL) return -1;
if (reactor->evblk == NULL) return -1;
struct ntyevent *event = ntyreactor_find_event_idx(reactor, sockfd);
nty_event_set(event, sockfd, acceptor, reactor);
nty_event_add(reactor->epfd, EPOLLIN, event);
return 0;
}
_Noreturn int ntyreactor_run(struct ntyreactor *reactor) {
if (reactor == NULL) return -1;
if (reactor->epfd < 0) return -1;
if (reactor->evblk == NULL) return -1;
struct epoll_event events[MAX_EPOLL_EVENTS + 1];
int i;
while (1) {
int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);
if (nready < 0) {
printf("epoll_wait error, exitn");
continue;
}
for (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 main(int argc, char *argv[]) {
unsigned short port = SERVER_PORT; // listen 8081
if (argc == 2) {
port = atoi(argv[1]);
}
struct ntyreactor *reactor = (struct ntyreactor *) malloc(sizeof(struct ntyreactor));
ntyreactor_init(reactor);
int i = 0;
int sockfds[PORT_COUNT] = {0};
for (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 (i = 0; i < PORT_COUNT; i++) {
close(sockfds[i]);
}
free(reactor);
return 0;
},>
#include
#include
#include
#include
#include
#include
#include
#include
#define BUFFER_LENGTH 4096
#define MAX_EPOLL_EVENTS 1024
#define SERVER_PORT 8081
#define PORT_COUNT 100
typedef int (*NCALLBACK)(int, int, void *);
struct ntyevent {
int fd;
int events;
void *arg;
NCALLBACK callback;
int status;
char buffer[BUFFER_LENGTH];
int length;
};
struct eventblock {
struct eventblock *next;
struct ntyevent *events;
};
struct ntyreactor {
int epfd;
int blkcnt;
struct eventblock *evblk;
};
int recv_cb(int fd, int events, void *arg);
int send_cb(int fd, int events, void *arg);
struct ntyevent *ntyreactor_find_event_idx(struct ntyreactor *reactor, int sockfd);
void nty_event_set(struct ntyevent *ev, int fd, NCALLBACK *callback, void *arg) {
ev->fd = fd;
ev->callback = callback;
ev->events = 0;
ev->arg = arg;
}
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 failed [fd=%d], events[%d]n", ev->fd, events);
return -1;
}
return 0;
}
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 recv_cb(int fd, int events, void *arg) {
struct ntyreactor *reactor = (struct ntyreactor *) arg;
struct ntyevent *ev = ntyreactor_find_event_idx(reactor, fd);
int len = recv(fd, ev->buffer, BUFFER_LENGTH, 0); //
nty_event_del(reactor->epfd, ev);
if (len > 0) {
ev->length = len;
ev->buffer[len] = '?';
// printf("recv[%d]:%sn", fd, ev->buffer);
printf("recv fd=[%dn", fd);
nty_event_set(ev, fd, send_cb, reactor);
nty_event_add(reactor->epfd, EPOLLOUT, ev);
}
else if (len == 0) {
close(ev->fd);
//printf("[fd=%d] pos[%ld], closedn", fd, ev-reactor->events);
}
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_find_event_idx(reactor, fd);
int len = send(fd, ev->buffer, ev->length, 0);
if (len > 0) {
// printf("send[fd=%d], [%d]%sn", fd, len, ev->buffer);
printf("send fd=[%dn]", fd);
nty_event_del(reactor->epfd, ev);
nty_event_set(ev, fd, recv_cb, reactor);
nty_event_add(reactor->epfd, EPOLLIN, ev);
}
else {
nty_event_del(reactor->epfd, ev);
close(ev->fd);
printf("send[fd=%d] error %sn", fd, strerror(errno));
}
return len;
}
int accept_cb(int fd, int events, void *arg) {//非阻塞
struct ntyreactor *reactor = (struct ntyreactor *) arg;
if (reactor == NULL) return -1;
struct sockaddr_in client_addr;
socklen_t len = sizeof(client_addr);
int clientfd;
if ((clientfd = accept(fd, (struct sockaddr *) &client_addr, &len)) == -1) {
printf("accept: %sn", strerror(errno));
return -1;
}
if ((fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) {
printf("%s: fcntl nonblocking failed, %dn", __func__, MAX_EPOLL_EVENTS);
return -1;
}
struct ntyevent *event = ntyreactor_find_event_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_addr.sin_addr), ntohs(client_addr.sin_port), clientfd);
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_family = AF_INET;
server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
server_addr.sin_port = htons(port);
bind(fd, (struct sockaddr *) &server_addr, sizeof(server_addr));
if (listen(fd, 20) < 0) {
printf("listen failed : %sn", strerror(errno));
}
return fd;
}
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 *) malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
if (evs == NULL) {
printf("ntyreactor_alloc ntyevents failedn");
return -2;
}
memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
struct eventblock *block = (struct eventblock *) malloc(sizeof(struct eventblock));
if (block == NULL) {
printf("ntyreactor_alloc eventblock failedn");
return -2;
}
memset(block, 0, sizeof(struct eventblock));
block->events = evs;
block->next = NULL;
blk->next = block;
reactor->blkcnt++; //
return 0;
}
struct ntyevent *ntyreactor_find_event_idx(struct ntyreactor *reactor, int sockfd) {
int blkidx = sockfd / MAX_EPOLL_EVENTS;
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_EVENTS];
}
int ntyreactor_init(struct ntyreactor *reactor) {
if (reactor == NULL) return -1;
memset(reactor, 0, sizeof(struct ntyreactor));
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 *) malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
if (evs == NULL) {
printf("ntyreactor_alloc ntyevents failedn");
return -2;
}
memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
struct eventblock *block = (struct eventblock *) malloc(sizeof(struct eventblock));
if (block == NULL) {
printf("ntyreactor_alloc eventblock failedn");
return -2;
}
memset(block, 0, sizeof(struct eventblock));
block->events = evs;
block->next = NULL;
reactor->evblk = block;
reactor->blkcnt = 1;
return 0;
}
int ntyreactor_destory(struct ntyreactor *reactor) {
close(reactor->epfd);
//free(reactor->events);
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 ntyreactor_addlistener(struct ntyreactor *reactor, int sockfd, NCALLBACK *acceptor) {
if (reactor == NULL) return -1;
if (reactor->evblk == NULL) return -1;
struct ntyevent *event = ntyreactor_find_event_idx(reactor, sockfd);
nty_event_set(event, sockfd, acceptor, reactor);
nty_event_add(reactor->epfd, EPOLLIN, event);
return 0;
}
_Noreturn int ntyreactor_run(struct ntyreactor *reactor) {
if (reactor == NULL) return -1;
if (reactor->epfd < 0) return -1;
if (reactor->evblk == NULL) return -1;
struct epoll_event events[MAX_EPOLL_EVENTS + 1];
int i;
while (1) {
int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);
if (nready < 0) {
printf("epoll_wait error, exitn");
continue;
}
for (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 main(int argc, char *argv[]) {
unsigned short port = SERVER_PORT; // listen 8081
if (argc == 2) {
port = atoi(argv[1]);
}
struct ntyreactor *reactor = (struct ntyreactor *) malloc(sizeof(struct ntyreactor));
ntyreactor_init(reactor);
int i = 0;
int sockfds[PORT_COUNT] = {0};
for (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 (i = 0; i < PORT_COUNT; i++) {
close(sockfds[i]);
}
free(reactor);
return 0;
},>
聲明：本文內(nèi)容及配圖由入駐作者撰寫或者入駐合作網(wǎng)站授權(quán)轉(zhuǎn)載。文章觀點(diǎn)僅代表作者本人，不代表電子發(fā)燒友網(wǎng)立場。文章及其配圖僅供工程師學(xué)習(xí)之用，如有內(nèi)容侵權(quán)或者其他違規(guī)問題，請聯(lián)系本站處理。
舉報(bào)投訴
-
封裝
+關(guān)注
關(guān)注
125文章
7593瀏覽量
142145 -
服務(wù)器
+關(guān)注
關(guān)注
12文章
8701瀏覽量
84551 -
模型
+關(guān)注
關(guān)注
1文章
3032瀏覽量
48361 -
代碼
+關(guān)注
關(guān)注
30文章
4671瀏覽量
67767
發(fā)布評論請先 登錄
相關(guān)推薦
四種模擬輸入信號的保護(hù)電路實(shí)現(xiàn)方法
本文介紹了四種模擬輸入信號的保護(hù)電路的實(shí)現(xiàn)方法。
發(fā)表于 03-28 09:55
?1038次閱讀
FPGA 設(shè)計(jì)的四種常用思想與技巧
FPGA 設(shè)計(jì)的四種常用思想與技巧FPGA設(shè)計(jì)的四種常用思想與技巧 討論的四種常用FPGA/CPLD設(shè)計(jì)思想與技巧：乒乓操作、串并轉(zhuǎn)換、流水線操作、數(shù)據(jù)接口同步化，都是FPGA/CPLD 邏輯設(shè)計(jì)
發(fā)表于 08-11 10:30
四種不同供電模式的LED拓?fù)?b class='flag-5'>介紹
本文中，小編將為大家介紹四種在LED供電當(dāng)中經(jīng)常使用的四種拓?fù)浣Y(jié)構(gòu)。感興趣的朋友快來看一看吧。 首先需要從了解轉(zhuǎn)換器的最小及最大輸出電壓入手。這只是將所有LED正向壓降與傳感電阻器電壓相加的總數(shù)
發(fā)表于 10-10 15:07
介紹UPS電源的四種工作方式
UPS電源是較為常見的應(yīng)急電源系統(tǒng)，其在市電正常與市電異常的情況下，工作方式也有所不同，以下介紹UPS電源的四種工作方式：正常運(yùn)行、電池工作、旁路運(yùn)行和旁路維護(hù)。1、正常運(yùn)行方式 UPS電源系統(tǒng)
發(fā)表于 11-16 06:19
介紹AUTOSAR支持的四種功能安全機(jī)制
1、AUTOSAR的四種功能安全機(jī)制雖然AUTOSAR不是一個(gè)完整的安全解決方案，但它提供了一些安全機(jī)制用于支持安全關(guān)鍵系統(tǒng)的開發(fā)。本文用于介紹AUTOSAR支持的四種功能安全機(jī)制：內(nèi)存分區(qū)
發(fā)表于 06-10 17:33
四種典型瞬態(tài)介紹
，我將介紹應(yīng)該注意的幾種典型瞬態(tài)，以及TI如何幫助滿足瞬態(tài)保護(hù)需求。 瀏覽此文章，并查看參考設(shè)計(jì)：《汽車瞬態(tài)和過流保護(hù)濾波器參考設(shè)計(jì)》 典型瞬態(tài)在四種常見場景中可能會發(fā)生瞬變。圖1所示為第一種場景
發(fā)表于 11-07 08:02
云計(jì)算的三種服務(wù)模式和四種部署模型
云計(jì)算基于3種特殊的云計(jì)算服務(wù)模式，具體架構(gòu)包括：基礎(chǔ)設(shè)施即服務(wù)、軟件即服務(wù)、平臺即服務(wù)。四種部署模型：公有云、私有云、社區(qū)云、混合云。
發(fā)表于 01-31 15:10
?3049次閱讀
無線充電技術(shù)的四種方式及其原理和應(yīng)用介紹
本文介紹了無線充電技術(shù)的應(yīng)用范圍及其電磁感應(yīng)方式等四種充電方式的詳細(xì)介紹。
發(fā)表于 10-12 16:16
?27次下載
jquery四種選擇器介紹
本文給大家匯總介紹了jQuery的四種選擇器的使用方法以及示例，非常的簡單實(shí)用，希望對大家學(xué)習(xí)jquery能夠有所幫助。
發(fā)表于 12-01 16:40
?2992次閱讀
四種溫度傳感器的數(shù)據(jù)介紹
本文檔的主要內(nèi)容詳細(xì)介紹的是四種溫度傳感器的數(shù)據(jù)介紹包括了： 球形傳感器，卡箍式傳感器，地面?zhèn)鞲衅?，侵入式傳感?/div>
發(fā)表于 02-28 17:09
?9次下載
四種常見的圖像濾波算法介紹
作者丨一支程序媛@知乎 來源丨h(huán)ttps://zhuanlan.zhihu.com/p/279602383 編輯丨極市平臺 導(dǎo)讀 圖像濾波是一種非常重要的圖像處理技術(shù)，本文詳細(xì)介紹了四種常見的圖像
介紹MCUboot支持的四種升級模式(2)
介紹MCUboot支持的四種升級模式，分別是Overwrite、Swap、Direct XIP和加載到RAM中執(zhí)行。由于FSP不支持第四種——加載到RAM中執(zhí)行，因?yàn)槲覀冎攸c(diǎn)介紹前三
- 設(shè)計(jì)技術(shù)
- 可編程邏輯
- 電源/新能源
- MEMS/傳感技術(shù)
- 測量儀表
- 嵌入式技術(shù)
- 制造/封裝
- 模擬技術(shù)
- RF/無線
- 接口/總線/驅(qū)動
- 處理器/DSP
- EDA/IC設(shè)計(jì)
- 存儲技術(shù)
- 光電顯示
- EMC/EMI設(shè)計(jì)
- 連接器
- 行業(yè)應(yīng)用
- LEDs
- 汽車電子
- 音視頻及家電
- 通信網(wǎng)絡(luò)
- 醫(yī)療電子
- 人工智能
- 虛擬現(xiàn)實(shí)
- 可穿戴設(shè)備
- 機(jī)器人
- 安全設(shè)備/系統(tǒng)
- 軍用/航空電子
- 移動通信
- 工業(yè)控制
- 便攜設(shè)備
- 觸控感測
- 物聯(lián)網(wǎng)
- 智能電網(wǎng)
- 區(qū)塊鏈
- 新科技
- 特色內(nèi)容
- 專欄推薦
- 學(xué)院
- 設(shè)計(jì)資源
- 設(shè)計(jì)技術(shù)
- 電子百科
- 電子視頻
- 元器件知識
- 工具箱
- VIP會員
- 最新技術(shù)文章
- 供應(yīng)鏈服務(wù)
- 硬件開發(fā)
- 華秋電路
- 華秋商城
- 華秋智造
- nextPCB
- BOM配單
- 媒體服務(wù)
- 網(wǎng)站廣告
- 在線研討會
- 活動策劃
- 新聞發(fā)布
- 新品發(fā)布
- 小測驗(yàn)
- 設(shè)計(jì)大賽
- 華秋
- 關(guān)于我們
- 投資關(guān)系
- 新聞動態(tài)
- 加入我們
- 聯(lián)系我們
- 舉報(bào)投訴
- 社交網(wǎng)絡(luò)
- 微博
- 移動端
- 發(fā)燒友APP
- 硬聲APP
- WAP
- 聯(lián)系我們
- 廣告合作
- 王婉珠：wangwanzhu@elecfans.com
- 內(nèi)容合作
- 黃晶晶：huangjingjing@elecfans.com
- 內(nèi)容合作（海外）
- 張迎輝：mikezhang@elecfans.com
- 供應(yīng)鏈服務(wù) PCB/IC/PCBA
- 江良華：lanhu@huaqiu.com
- 投資合作
- 曾海銀：zenghaiyin@huaqiu.com
- 社區(qū)合作
- 劉勇：liuyong@huaqiu.com
-
關(guān)注我們的微信
-
下載發(fā)燒友APP
-
電子發(fā)燒友觀察
版權(quán)所有 ? 湖南華秋數(shù)字科技有限公司
長沙市望城經(jīng)濟(jì)技術(shù)開發(fā)區(qū)航空路6號手機(jī)智能終端產(chǎn)業(yè)園2號廠房3層（0731-88081133）
電子發(fā)燒友 （電路圖） 湘公網(wǎng)安備43011202000918
工商網(wǎng)監(jiān)
湘ICP備2023018690號-1
評論