lt，recvbuff中有数据就一直触发；
et，recvbuff中收到数据，只触发一次。如果recvbuff中数据没有读完，不会再次触发，当recvbuff中收到新的数据时，再次触发。也就是收到一个包，只触发一次。比如客户端发送32byte的包，服务器只recv了10byte的数据，那么epoll不会再次触发，等到下一次客户端再发送32byte的数据，epoll会再一次触发。对于et，对于recv，最好是一个循环的读，直到读完，返回-1。
所以lt适合用于大包，数据没读完会一直触发；et适合小包，只触发一次，需要应用程序循环把数据读完。
et模式下，sendbuff从不可发送到可发送，只触发一次。
send的情况，如果sendbuff一直为空，如果用et，epoll会一直触发吗？
测试过，只会触发一次。
哪些场景使用水平触发？
- 小数据，使用边沿触发
- 数据块，数据量比较大使用水平触发。防止一次性接收不完。
listenfd用水平触发，如果多个client同时连接进来，listenfd里面积攒多个连接的话，accept一次只处理一个连接，防止漏掉连接，选择水平触发。
水平触发和边沿触发分界点，recv的buffer_length如果一次能接收完recv buffer中的数据，就是小数据，一次接收不完就是大数据
reactor怎么存储100万个event
使用reactor实现百万并发连接的服务器，需要考虑event怎么保存，怎样分配内存, 存储百万级别的event。
数据结构设计
利用fd是递增的特性，可以设计成下面的结构。这样做可扩展性非常好，reactor存储的event数量不受限制。
typedef struct _eventblock {
struct _eventblock *next;
nevent *events; // 每一个block 1024个event
} eventblock;
typedef struct _nreactor {
int epfd;
int blkcnt;
eventblock *evblk;
} nreactor;
代码实现
通过fd，可以计算得出相应的eventblock位置，以及event在该eventblock中的相应位置。之前代码中所有使用reactor->events的都需要做相应的修改。
nevent *nreactor_idx(nreactor *reactor, int sockfd) {
int blkidx = sockfd / max_epoll_events;
while (blkidx >= reactor->blkcnt) {
nreactor_alloc(reactor);
}
int i = 0;
eventblock *blk = reactor->evblk;
while (i < blkidx && blk != null) {
blk = blk->next;
}
return &blk->events[sockfd % max_epoll_events];
}
reactor初始化时，也需要相应的申请eventblock内存，以及events
int nreactor_alloc(nreactor *reactor) {
if (reactor == null) return -1;
if (reactor->evblk == null) return -1;
eventblock *blk = reactor->evblk;
while (blk->next != null) {
blk = blk->next;
}
nevent *evs = (nevent *)malloc(max_epoll_events * sizeof(nevent));
if (evs == null) {
perror("nreactor_alloc malloc events failed");
return -2;
}
memset(evs, 0, max_epoll_events * sizeof(nevent));
eventblock *block = (eventblock *)malloc(sizeof(eventblock));
if (block == null) {
perror("nreactor_alloc malloc block failed");
return -2;
}
memset(block, 0, sizeof(eventblock));
block->events = evs;
block->next = null;
blk->next = block;
reactor->blkcnt ;
return 0;
}
完整代码
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define buffer_length 1024
#define max_epoll_events 1024
#define server_port 9105
#define port_count 100
typedef int (*ncallback)(int fd, void *arg);
typedef struct _nevent {
int fd;
int events;
void *arg;
ncallback callback;
int status; // whether fd is in epoll now.
char buffer[buffer_length];
int length;
} nevent;
typedef struct _eventblock {
struct _eventblock *next;
nevent *events; // 每一个block 1024个event
} eventblock;
typedef struct _nreactor {
int epfd;
int blkcnt;
eventblock *evblk;
} nreactor;
int recv_cb(int client_fd, void *arg);
int send_cb(int client_fd, void *arg);
int accept_cb(int listen_fd, void *arg);
nevent *nreactor_idx(nreactor *reactor, int sockfd);
int init_sock(unsigned short port) {
int listen_fd = socket(af_inet, sock_stream, 0);
fcntl(listen_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(listen_fd, (struct sockaddr *)&server_addr, sizeof(server_addr));
if (listen(listen_fd, 20) < 0) {
perror("listen");
}
return listen_fd;
}
int nreactor_alloc(nreactor *reactor) {
if (reactor == null) return -1;
if (reactor->evblk == null) return -1;
eventblock *blk = reactor->evblk;
while (blk->next != null) {
blk = blk->next;
}
nevent *evs = (nevent *)malloc(max_epoll_events * sizeof(nevent));
if (evs == null) {
perror("nreactor_alloc malloc events failed");
return -2;
}
memset(evs, 0, max_epoll_events * sizeof(nevent));
eventblock *block = (eventblock *)malloc(sizeof(eventblock));
if (block == null) {
perror("nreactor_alloc malloc block failed");
return -2;
}
memset(block, 0, sizeof(eventblock));
block->events = evs;
block->next = null;
blk->next = block;
reactor->blkcnt ;
return 0;
}
int nreactor_init(nreactor *reactor) {
if (reactor == null) return -1;
memset(reactor, 0, sizeof(nreactor));
reactor->epfd = epoll_create(1);
if (reactor->epfd < 0) {
perror("epoll_create");
return -2;
}
nevent *evs = (nevent *)malloc(max_epoll_events * sizeof(nevent));
if (evs == null) {
perror("nreactor_init malloc events failed");
return -2;
}
memset(evs, 0, max_epoll_events * sizeof(nevent));
eventblock *block = (eventblock *)malloc(sizeof(eventblock));
if (block == null) {
perror("nreactor_init malloc block failed");
return -2;
}
memset(block, 0, sizeof(eventblock));
block->events = evs;
block->next = null;
reactor->evblk = block;
reactor->blkcnt = 1;
return 0;
}
int nreactor_destroy(nreactor *reactor) {
close(reactor->epfd);
eventblock *blk = reactor->evblk;
eventblock *blk_next = null;
while (blk != null) {
blk_next = blk->next;
free(blk->events);
free(blk);
blk = blk_next;
}
return 0;
}
void nreactor_event_set(nevent *ev, int fd, ncallback callback, void *arg) {
ev->fd = fd;
ev->callback = callback;
ev->arg = arg;
ev->events = 0;
}
int nreactor_event_add(int epfd, nevent *ev, int events) {
struct epoll_event ep_ev = {
0, {
0}};
ep_ev.events = ev->events = events;
ep_ev.data.ptr = ev;
int op;
if (ev->status == 1) {
op = epoll_ctl_mod;
} else {
ev->status = 1;
op = epoll_ctl_add;
}
if (epoll_ctl(epfd, op, ev->fd, &ep_ev) < 0) {
perror("epoll_ctl");
return -1;
}
return 0;
}
int nreactor_event_del(int epfd, nevent *ev) {
struct epoll_event ep_ev = {
0, {
0}};
if (ev->status != 1) {
return -1;
}
ev->status = 0;
epoll_ctl(epfd, epoll_ctl_del, ev->fd, null);
return 0;
}
int recv_cb(int client_fd, void *arg) {
nreactor *reactor = (nreactor *)arg;
if (reactor == null) return -1;
nevent *ev = nreactor_idx(reactor, client_fd);
int len = recv(client_fd, ev->buffer, buffer_length, 0);
nreactor_event_del(reactor->epfd, ev);
if (len > 0) {
ev->length = len;
ev->buffer[len] = '\0';
printf("client_fd[%d]:%s\n", client_fd, ev->buffer);
nreactor_event_set(ev, client_fd, send_cb, reactor);
nreactor_event_add(reactor->epfd, ev, epollout);
} else if (len == 0) {
close(client_fd);
// printf("[client_fd=%d] pos[%ld], closed\n", client_fd, ev - reactor->events);
} else {
close(client_fd);
perror("recv");
}
return len;
}
int send_cb(int client_fd, void *arg) {
nreactor *reactor = (nreactor *)arg;
if (reactor == null) return -1;
nevent *ev = nreactor_idx(reactor, client_fd);
int len = send(client_fd, ev->buffer, ev->length, 0);
nreactor_event_del(reactor->epfd, ev);
if (len > 0) {
printf("send[client_fd=%d], [%d]%s\n", client_fd, len, ev->buffer);
nreactor_event_set(ev, client_fd, recv_cb, reactor);
nreactor_event_add(reactor->epfd, ev, epollin);
} else {
close(ev->fd);
perror("send");
}
return len;
}
int accept_cb(int listen_fd, void *arg) {
nreactor *reactor = (nreactor *)arg;
if (reactor == null) return -1;
struct sockaddr_in client_addr;
socklen_t len = sizeof(client_addr);
int client_fd = accept(listen_fd, (struct sockaddr *)&client_addr, &len);
if (client_fd < 0) {
perror("accept");
return -1;
}
int flag = fcntl(client_fd, f_setfl, o_nonblock);
if (flag < 0) {
perror("fcntl");
return -2;
}
nevent *ev = nreactor_idx(reactor, client_fd);
nreactor_event_set(ev, client_fd, recv_cb, reactor);
nreactor_event_add(reactor->epfd, ev, epollin);
printf("new connect [%s:%d], client_fd[%d]\n",
inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), client_fd);
}
nevent *nreactor_idx(nreactor *reactor, int sockfd) {
int blkidx = sockfd / max_epoll_events;
while (blkidx >= reactor->blkcnt) {
nreactor_alloc(reactor);
}
int i = 0;
eventblock *blk = reactor->evblk;
while (i < blkidx && blk != null) {
blk = blk->next;
}
return &blk->events[sockfd % max_epoll_events];
}
int nreactor_addlistener(nreactor *reactor, int listen_fd, ncallback accept_cb) {
if (reactor == null || reactor->evblk == null) {
return -1;
}
nevent *event = nreactor_idx(reactor, listen_fd);
nreactor_event_set(event, listen_fd, accept_cb, reactor);
nreactor_event_add(reactor->epfd, event, epollin);
return 0;
}
int nreactor_run(nreactor *reactor) {
if (reactor == null) return -1;
if (reactor->evblk == null) return -1;
struct epoll_event events[max_epoll_events];
while (1) {
int nready = epoll_wait(reactor->epfd, events, max_epoll_events, 1000);
if (nready < 0) {
perror("epoll_wait");
continue;
}
int i;
for (i = 0; i < nready; i ) {
nevent *ev = (nevent *)events[i].data.ptr;
if ((events[i].events & epollin) && (ev->events & epollin)) {
ev->callback(ev->fd, ev->arg);
}
if ((events[i].events & epollout) && (ev->events & epollout)) {
ev->callback(ev->fd, ev->arg);
}
}
}
}
#if 1
int main(int argc, char *argv[]) {
unsigned short port = server_port;
if (argc == 2) {
port = atoi(argv[1]);
}
nreactor *reactor = (nreactor *)malloc(sizeof(nreactor));
nreactor_init(reactor);
int i = 0;
int listen_fds[port_count] = {
0};
for (i = 0; i < port_count; i ) {
listen_fds[i] = init_sock(port i);
nreactor_addlistener(reactor, listen_fds[i], accept_cb);
}
nreactor_run(reactor);
nreactor_destroy(reactor);
for (i = 0; i < port_count; i ) {
close(listen_fds[i]);
}
free(reactor);
return 0;
}
#endif
服务器端修改配置
修改ulimit
# /etc/security/limits.conf
* hard nofile 1048576
* soft nofile 1048576
修改file-max
# vim /etc/sysctl.conf
# cat /proc/sys/fs/file-max
fs.file-max = 1048576
修改nf_conntrack_max
# /etc/sysctl.conf
# cat /proc/sys/net/netfilter/nf_conntrack_max
# sysctl -p
net.netfilter.nf_conntrack_max = 1048576
修改tcp_wmem， tcp_rmem
如果服务器的内存不够，为了测试，修改一下wmem和rmem，即将sendbuff、recvbuff改小，单位是byte。测试后，记得还原回去。
# vim /etc/sysctl.conf
# cat /proc/sys/net/ipv4/tcp_wmem
# sysctl -p
net.ipv4.tcp_rmem = 512 512 1024
net.ipv4.tcp_wmem = 512 512 1024
修改tcp_mem
tcp_mem是tcp协议栈的大小，单位是页，一页4k
# vi /etc/sysctl.conf
# sysctl -p
# cat /proc/sys/net/ipv4/tcp_mem
net.ipv4.tcp_mem = 757596 1010128 1515192
测试结果
最后只到61w多，四台机器，每个都是4核16g。
跑到50w左右的时候，连接处理的速度就比较慢了。
server：
client：
- 大量客户端连到服务器，要使连接的处理速度更快，可以把100个listenfd放在不同的线程。
可以通过一个线程一个reactor来实现。每个线程一个reactor，对应一个listenfd。
- 如果服务器只监听一个端口，怎么做？开进程
一个master进程，多个worker进程。通过加accept锁来决定由那个worker进行处理该连接。
- 可以把listenfd和clientfd放到不同的线程里面。
可以看下libevent/redis的reactor, 将listenfd和clientfd，使用不同的线程进行处理。
开一个worker线程，main线程处理accept，worker线程处理clientfd。main线程可以使用poll。
单线程，libevent/redis
多线程，memcached，每个线程一个epoll_wait
多进程，nginx
我们实现的reactor是个单线程的，100个listenfd都在一个线程里，如果想要性能更高，可以把100个listenfd放在不同的线程里面。
3个思考问题：
1）reactor怎么使用多线程，比如listen 10个端口，怎么做到每个端口(listenfd)一个线程？
每个线程一个reactor，对应一个listenfd
2）服务器一般只监听一个port，比如8888，怎么解决？可以通过开进程解决。
怎么做到多个进程listen一个端口？
nginx的ag真人游戏的解决方案
3）如何做到listenfd和clientfd在不同的线程？
一个main线程，一个worker线程。main线程负责listen，处理连接；worker线程负责处理clientfd。
listen的backlog参数，linux系统上指的是accpt队列长度；unix系统上是sync accept队列长度。