epoll的陷阱实践
source link: https://www.daqianduan.com/17674.html
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
在使用epoll的时候，我们上篇文章 epoll的陷阱 大体介绍了epoll中会有哪些问题。这篇文章我们就针对必须要了解，也是绕不过去的陷阱进行实验，看看现象是什么，并且如何编写才能达到我们想要的效果。
https://idea.popcount.org/2017-02-20-epoll-is-fundamentally-broken-12/
https://juejin.cn/post/6844904122018168845系统环境
Linux version 4.19.0-6-amd64 ([email protected]) (gcc version 8.3.0 (Debian 8.3.0-6)) #1 SMP Debian 4.19.67-2+deb10u1 (2019-09-20)
监听套接字的惊群现象
问题描述
创建一个epoll实例，多个线程调用epoll_wait监听，一个客户端请求连接，所有调用epoll_wait的线程都会返回。因为我们的listen socket是非阻塞的，所以只有一个线程的accept返回成功，其他的线程返回EAGAIN。
性能影响
https://www.ichenfu.com/2017/05/03/proxy-epoll-thundering-herd/
我们知道系统中影响性能最大的就是上下文切换，IO操作和轮询。accept的惊群现象至少满足了两种：
-
频繁的唤起线程，上下文切换
-
频繁的操作socket，IO操作
重现问题
#ifndef EPOLL1_H
#define EPOLL1_H
#include <pthread.h>
class epoll1;
struct THREADDATA
{
epoll1* pepoll;
pthread_t threadid;
int num;
THREADDATA()
{
pepoll = NULL;
threadid = 0;
num = 0;
}
};
class epoll1
{
public:
epoll1();
~epoll1();
void start();
void setnonblocking(int fd);
static void* start_routine(void * parg);
int m_epollfd;
int m_listenfd;
THREADDATA m_threaddata[5];
};
#endif // EPOLL1_H
#include "epoll1.h"
#include <iostream>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <errno.h>
#define MAX_EVENTS 100
#define MAX_BUF 1024
using namespace std;
epoll1::epoll1()
{
m_epollfd = -1;
m_listenfd = -1;
for(auto& iter : m_threaddata)
{
iter.pepoll = this;
}
}
epoll1::~epoll1()
{
close(m_listenfd);
close(m_epollfd);
}
void epoll1::start()
{
struct sockaddr_in servaddr;
struct epoll_event ev;
bzero(&servaddr, sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(9000);
if((m_listenfd = socket(AF_INET, SOCK_STREAM|SOCK_NONBLOCK, 0)) == -1)
{
cout << "create listen socket err" << endl;
exit(EXIT_FAILURE);
}
if(bind(m_listenfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr)) == -1)
{
cout << "bind listen socket err" << endl;
cout << strerror(errno) << endl;
exit(EXIT_FAILURE);
}
if(listen(m_listenfd, 5) == -1)
{
cout << "listen listen socket err" << endl;
exit(EXIT_FAILURE);
}
m_epollfd = epoll_create1(0);
ev.events = EPOLLIN;
ev.data.fd = m_listenfd;
if(epoll_ctl(m_epollfd, EPOLL_CTL_ADD, m_listenfd, &ev) < 0)
{
cout << "add listen socket to epoll err" << endl;
exit(EXIT_FAILURE);
}
for(int i = 0; i < 5; i++)
{
m_threaddata[i].num = i;
}
pthread_create(&(m_threaddata[0].threadid), NULL, start_routine, (void*)(&m_threaddata[0]));
pthread_create(&(m_threaddata[1].threadid), NULL, start_routine, (void*)(&m_threaddata[1]));
pthread_create(&(m_threaddata[2].threadid), NULL, start_routine, (void*)(&m_threaddata[2]));
pthread_create(&(m_threaddata[3].threadid), NULL, start_routine, (void*)(&m_threaddata[3]));
pthread_create(&(m_threaddata[4].threadid), NULL, start_routine, (void*)(&m_threaddata[4]));
for(auto& iter : m_threaddata)
{
pthread_join(iter.threadid, NULL);
}
}
void epoll1::setnonblocking(int fd)
{
fcntl(fd, F_SETFL, (fcntl(fd, F_GETFL)|O_NONBLOCK));
}
void* epoll1::start_routine(void *parg)
{
epoll1 * pepoll = ((THREADDATA*)parg)->pepoll;
struct epoll_event events[MAX_EVENTS];
struct sockaddr_in cliaddr;
char buf[MAX_BUF] = {0};
struct epoll_event ev;
socklen_t len = sizeof(struct sockaddr_in);
while(true)
{
int nfds = epoll_wait(pepoll->m_epollfd, events, MAX_EVENTS, -1);
if(nfds == -1)
{
break;
}
cout << "thread id: " << ((THREADDATA*)parg)->num << " wake up" << endl;
for(int i = 0; i < nfds; i++)
{
if(events[i].data.fd == pepoll->m_listenfd)
{
int connfd = accept(pepoll->m_listenfd, (struct sockaddr *)&cliaddr, &len);
if(connfd == -1)
{
if(errno == EAGAIN)
{
cout << "thread id: " << ((THREADDATA*)parg)->num << " err eagain" << endl;
continue;
}
else
{
cout << "thread id: " << ((THREADDATA*)parg)->num << " err return" << endl;
break;
}
}
pepoll->setnonblocking(connfd);
cout << "thread id: " << ((THREADDATA*)parg)->num << " accept" << endl;
ev.events = EPOLLIN;
ev.data.fd = connfd;
if( epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_ADD, connfd, &ev) < 0 )
{
break;
}
}
else
{
int npro = read(events[i].data.fd, buf, sizeof(buf));
if(npro == -1)
{
close(events[i].data.fd );
epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_DEL, events[i].data.fd, &ev);
}
else
{
cout << buf << endl;
npro = write(events[i].data.fd, buf, npro);
if(npro == -1)
{
close(events[i].data.fd );
epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_DEL, events[i].data.fd, &ev);
}
}
}
}
}
return NULL;
}
当有一个客户端请求连接时，输出日志如下
thread id: thread id: thread id: 4 wake up0 wake up3 wake up thread id: thread id: 1 wake up thread id: 0 accept 2 wake up thread id: 3 err eagain thread id: thread id: 4 err eagain 2 err eagain thread id: 1 err eagain
因为是多线程，终端输出的时候不能保证顺序，所以出现了乱序，不过从日志中也能看出，5个空闲的线程都唤醒了，但是呢，只有0序号的线程成功接收了请求，其他的线程都报了错。
解决方案一
EPOLLEXCLUSIVE (since Linux 4.5)
https://man7.org/linux/man-pages/man2/epoll_ctl.2.html
https://github.com/torvalds/linux/commit/df0108c5da561c66c333bb46bfe3c1fc65905898
为epoll文件描述符设置独占唤醒模式也会被附加到目标文件描述符fd上。当需要唤醒事件的时候，如果有多个epoll文件描述符被附加到同一个目标文件，使用的是EPOLLEXCLUSIVE标识，一个或者更多epoll文件描述符会通过epoll_wait收到事件。默认情况下（没有设置EPOLLEXCLUSIVE），所有的epoll文件描述符都会收到事件。EPOLLEXCLUSIVE在这种情形下，对于避免惊群问题非常有效。
如果同一个文件描述符是在多个epoll实例中，一些有EPOLLEXCLUSIVE标识，一些没有，事件会唤醒所有没有定义EPOLLEXCLUSIVE标识的实例，至少唤醒一个定义了EPOLLEXCLUSIVE标识的epoll实例。
如下的一些值可以和EPOLLEXCLUSIVE同时使用：EPOLLIN EPOLLOUT EPOLLWAKEUP EPOLLET。EPOLLHUP和EPOLLERR也可以被定义，但是没必要：正常情况下，这些事件只会在发生的时候才会通知，不管有没有定义在事件中。如果设定了其他的值，会触发EINVAL的错误。
EPOLLEXCLUSIVE可能会用在EPOLL_CTL_ADD的操作中；如果想在EPOLL_CTL_MOD使用，会报错。如果EPOLLEXCLUSIVE已经经过epoll_ctl()设置了，那么在后续EPOLL_CTL_MOD中操作同一个epfd和fd对也会报错。调用epoll_ctl()为事件设定EPOLLEXCLUSIVE，然后设定这个目标文件描述符为epoll的实例也会报错。所有这些操作的错误代码都是EINVAL。
EPOLLEXCLUSIVE在调用epoll_ctl时为传入事件设置的一个输入标识；永远也不会通过epoll_wait返回给用户。
我们从官方的文档可以看出，这个标识是用来避免，一个事件到达时，唤醒所有epoll_wait的问题。但是官方也只说，会唤起一个或者多个，也就是这个标识是减少了惊群问题的触发概率，但是没有完全避免。
那我们为监听socket设置EPOLLEXCLUSIVE标识进行测试
m_epollfd = epoll_create1(0);
ev.events = EPOLLIN|EPOLLEXCLUSIVE;
ev.data.fd = m_listenfd;
if(epoll_ctl(m_epollfd, EPOLL_CTL_ADD, m_listenfd, &ev) < 0)
{
cout << "add listen socket to epoll err" << endl;
exit(EXIT_FAILURE);
}
在把监听socket添加到epoll中的时候，增加EPOLLEXCLUSIVE标识，测试结果如下
thread id: thread id: thread id: 4 wake up1 wake up0 wake up thread id: thread id: 0 err eagain thread id: 3 wake up 4 accept thread id: 3 err eagain thread id: 2 wake up thread id: 2 err eagain thread id: 1 err eagain
非常奇怪，按照官方文档和github上的这个flag的patch来讲，应该是只有一个线程唤醒，或者会有多个唤醒，但是不管如何测试，结果都是5个线程全部唤醒。针对这个问题，我也没找到对应的解释和解决方法，也没有查找到在LT模式下EPOLLEXCLUSIVE的用法。目前得到的结论就是
-
这个标识肯定是有作用的。因为源码是公开的，并且官方文档中也写明了。
-
导致所有线程都被唤醒的原因可能是因为LT模式。LT模式的特点是，如果epoll中监听的socket有事件，那么不管什么时候调用，epoll_wait都会返回。我们设置了EPOLLEXCLUSIVE标识，可以保证所有的epoll_wait同时调用的时候只有一个返回，但是这个线程返回之后，在调用accept把监听事件从epoll中取出来之前，别的线程的epoll_wait也在阻塞监听，就相当于这个时候又有epoll_wait调用，基于LT的特点，所以又返回了。
最后的结果就是，LT模式下的EPOLLEXCLUSIVE并不能解决监听套接字的惊群问题
进一步验证
如果说EPOLLEXCLUSIVE对于监听套接字的惊群问题没有任何作用，根据官方所说，惊群的现象是所有等待线程都会被唤醒，那我们把线程增加到50个，再看看惊群的现象。50个线程去掉EPOLLEXCLUSIVE标识，测试结果如下
thread id: 49 wake up thread id: 47 wake up thread id: 42 wake up thread id: 48 wake up thread id: 42 err eagain thread id: thread id: 49 accept thread id: thread id: 4446 wake up wake up thread id: 46 err eagain thread id: thread id: 47 err eagain 4845 wake up err eagainthread id: 45 err eagain thread id: 43 wake up thread id: 44 err eagain thread id: 43 err eagain
奇怪的问题又来了，并没有唤起所有线程，而是唤起了部分线程。根据官方文档的描述和理解（ https://man7.org/linux/man-pages/man7/epoll.7.html ），导致这个的原因有可能是，epoll在遍历的时候，一个个的唤起线程，但是因为每个线程互不影响，导致某一个系统时钟切片，有一个线程运行到了accept并且成功了，这个时候把事件从epoll中读取出来，所以切换回epoll继续遍历唤醒的时候，发现没有事件了，也就不会唤起其他线程了。
具体的证明还需要后续查看epoll的源码。
解决方案二
如果新的标识不能解决我们的问题，那么我们就使用另一个
EPOLLONESHOT (since Linux 2.6.2)
在相关的文件描述符上请求只有一次的通知。也就是，当一个事件在epoll_wait等待的文件描述符上触发时，这个文件描述符就会从通知列表中禁用，不会再通知到其他的epoll实例。用户必须调用epoll_ctl()和EPOLL_CTL_MOD把这个文件描述符用新的事件标识重新启用。
这个标识是event.events的输入标识，在调用epoll_ctl()的时候使用；不可能在epoll_wait返回。
使用EPOLLONESHOT的话，我们只需要改两个地方：一个是第一次把监听套接字添加到epoll；另一个是在epoll_wait返回的时候，需要通过EPOLL_CTL_MOD再次重置。
m_epollfd = epoll_create1(0);
ev.events = EPOLLIN|EPOLLONESHOT;
ev.data.fd = m_listenfd;
if(epoll_ctl(m_epollfd, EPOLL_CTL_ADD, m_listenfd, &ev) < 0)
{
cout << "add listen socket to epoll err" << endl;
exit(EXIT_FAILURE);
}
while(true)
{
int nfds = epoll_wait(pepoll->m_epollfd, events, MAX_EVENTS, -1);
if(nfds == -1)
{
break;
}
cout << "thread id: " << ((THREADDATA*)parg)->num << " wake up " << nfds << endl;
for(int i = 0; i < nfds; i++)
{
if(events[i].data.fd == pepoll->m_listenfd)
{
int connfd = accept(pepoll->m_listenfd, (struct sockaddr *)&cliaddr, &len);
if(connfd == -1)
{
if(errno == EAGAIN)
{
cout << "thread id: " << ((THREADDATA*)parg)->num << " err eagain" << endl;
continue;
}
else
{
cout << "thread id: " << ((THREADDATA*)parg)->num << " err return" << endl;
break;
}
}
pepoll->setnonblocking(connfd);
cout << "thread id: " << ((THREADDATA*)parg)->num << " accept" << endl;
ev.events = EPOLLIN;
ev.data.fd = connfd;
if(epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_ADD, connfd, &ev) < 0)
{
break;
}
//在这里重新把监听socket添加到epoll
ev.events = EPOLLIN|EPOLLONESHOT;
ev.data.fd = pepoll->m_listenfd;
epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_MOD, pepoll->m_listenfd, &ev);
}
else
{
...
}
这种方法是否可行呢，我们测试一下
thread id: 49 wake up 1 thread id: 49 accept thread id: 49 wake up 1 thread id: 49 accept thread id: 49 wake up 1 thread id: 49 accept thread id: 49 wake up 1 thread id: 49 accept
从输出的日志可以看出，同时来了4个链接，每一个链接都是单独唤醒一个线程，并且没有惊群的问题。算是解决了。
注意
上面代码可能会有一个问题，就是epoll_wait返回的时候，有多个监听的事件，那么应该在所有监听事件都处理完成后再激活监听socket。不然有可能在处理完第一个监听事件的时候，激活了套接字，在处理第二个监听事件之前，另一个线程的epoll_wait也返回了，就会出现多个线程同时accept同一个链接，浪费一次调用。
虽然多次测试发现都是只有一个监听事件返回，建议还是修改一下，做一个标识，如果有监听事件，那么在这些事件都处理完成后，再次把监听socket激活。但是这样做呢，又有另一个问题，就是如果这次返回的事件中，有一个处理起来非常慢，就会导致监听迟迟无法响应。
进一步完善的话，就是，先处理epoll_wait返回的所有监听事件，如果有，把监听套接字重新激活，然后再处理剩下的事件。
可能的影响
从日志中可以看出，这种方式不能很好的利用多线程/多进程/CPU多内核的优势。因为当有一个连接请求进来时，epoll_wait就会返回，并且把监听事件禁止掉，其他进来的连接只能等当前的处理完再次激活监听事件后才能处理，也就是强制变成了串行的。我们理想的模型是不管进来几个连接，都会对每个连接唤起一个空闲的线程进行处理；不管是一次处理一个请求还是多个请求，都不会出现一个请求多个线程唤起最后只有一个成功（惊群问题）或者多个线程同时处理一个请求（比如接收数据的乱序问题，下面会介绍）。很显然epoll并不能很容易的满足我们的要求。
解决方案三
上面的解决方案是一个epoll实例，然后多个线程同时监听，也可以写成每个线程/进程创建一个自己的epoll实例，监听同一个端口。
SO_REUSEPORT (since Linux 3.9)
允许多个AF_INET或者AF_INET6套接字绑定到一个制定的套接字地址上。必须为每一个套接字设置这个属性，包括第一个套接字，在调用bind前设置这个属性。为了防止端口劫持，所有的绑定到同一个地址的进程必须有着同一个有效的UID。这个选项对与TCP和UDP都有效。
对于TCP套接字，这个选项允许accept分布式的加载在多线程服务器，对于每一个线程用不同的监听套接字，用于改善性能。对比传统的技术这种方法提供了改进的负载分配，比如使用单独的一个accept线程，或者有多个线程对于同一个socket竞争的accept。
对于UDP套接字，这个选项对比传统技术多个进程竞争的获取同一个套接字上的数据包可以在多进程（或者多线程）提供更好的负载对于将要到达的数据包。
这个方案是针对每一个线程/进程，都开启一个独立的监听套接字，每一个线程/进程都创建一个单独的epoll实例。这样的话，每个线程/进程内部，相当于是单线程的，每个线程/进程内部的监听套接字是唯一的，不会有其他的线程/进程在监听当前线程的套接字，所以就可以避免多个线程唤起了。
Address already in use
当我们启用SO_REUSEPORT的时候，开启多个线程监听，结果提示Address already in use。这是因为SO_REUSEPORT是解决多个进程绑定同一个端口的问题，而我们的是多线程，属于同一个线程绑定同一个端口，那么还需要增加一个参数SO_REUSEADDR
SO_REUSEADDR
在调用bind前使用的指定地址可以在本地重复利用。对于AF_INET套接字，意思就是一个套接字可以绑定，就算已经有一个激活的监听套接字已经绑定到这个地址上了。如果一个监听套接字已经通过INADDR_ANY标识绑定到了一个指定的端口上，那么就不能把这个端口绑定到本地其他地址上了。参数是整数的布尔类型的标识。
SO_REUSEADDR和SO_REUSEPORT的作用分别是指定地址和端口是否允许多次绑定。
代码
#include "epoll1.h"
#include <iostream>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <errno.h>
#include <arpa/inet.h>
#define MAX_EVENTS 100
#define MAX_BUF 1024
using namespace std;
epoll1::epoll1()
{
for(auto& iter : m_threaddata)
{
iter.pepoll = this;
}
}
epoll1::~epoll1()
{
}
void epoll1::start()
{
int i = 0;
for(auto& iter : m_threaddata)
{
iter.num = i;
pthread_create(&(iter.threadid), NULL, start_routine, (void*)(&iter));
i++;
}
for(auto& iter : m_threaddata)
{
pthread_join(iter.threadid, NULL);
}
}
void epoll1::setnonblocking(int fd)
{
fcntl(fd, F_SETFL, (fcntl(fd, F_GETFL)|O_NONBLOCK));
}
void* epoll1::start_routine(void *parg)
{
int epollfd = epoll_create1(0);
bool bcontinue = true;
//cout << "enter thread " << ((THREADDATA1*)parg)->num << endl;
epoll1 * pepoll = ((THREADDATA1*)parg)->pepoll;
struct epoll_event events[MAX_EVENTS];
struct sockaddr_in cliaddr;
char buf[MAX_BUF] = {0};
struct epoll_event ev;
socklen_t len = sizeof(struct sockaddr_in);
int listenfd = 0;
struct sockaddr_in servaddr;
bzero(&servaddr, sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(9000);
if((listenfd = socket(AF_INET, SOCK_STREAM|SOCK_NONBLOCK, 0)) == -1)
{
cout << "create listen socket err" << endl;
bcontinue = false;
}
int val = 1;
if(bcontinue && setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof (val)) == -1)
{
cout << "set listen socket reuseport err" << endl;
bcontinue = false;
}
if(bcontinue && setsockopt(listenfd, SOL_SOCKET, SO_REUSEPORT, &val, sizeof (val)) == -1)
{
cout << "set listen socket reuseport err" << endl;
bcontinue = false;
}
if(bcontinue && bind(listenfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr)) == -1)
{
cout << "bind listen socket err" << endl;
cout << strerror(errno) << endl;
bcontinue = false;
}
if(bcontinue && listen(listenfd, 5) == -1)
{
cout << "listen listen socket err" << endl;
bcontinue = false;
}
ev.events = EPOLLIN;
ev.data.fd = listenfd;
if(bcontinue && epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &ev) < 0)
{
cout << "add listen socket to epoll err" << endl;
bcontinue = false;
}
while(bcontinue)
{
int nfds = epoll_wait(epollfd, events, MAX_EVENTS, -1);
if(nfds == -1)
{
break;
}
cout << "thread id: " << ((THREADDATA1*)parg)->num << " wake up " << nfds << endl;
for(int i = 0; i < nfds; i++)
{
if(events[i].data.fd == listenfd)
{
int connfd = accept(listenfd, (struct sockaddr *)&cliaddr, &len);
if(connfd == -1)
{
if(errno == EAGAIN)
{
cout << "thread id: " << ((THREADDATA1*)parg)->num << " err eagain" << endl;
continue;
}
else
{
cout << "thread id: " << ((THREADDATA1*)parg)->num << " err return" << endl;
break;
}
}
pepoll->setnonblocking(connfd);
cout << "thread id: " << ((THREADDATA1*)parg)->num << " accept" << endl;
ev.events = EPOLLIN;
ev.data.fd = connfd;
if(epoll_ctl(epollfd, EPOLL_CTL_ADD, connfd, &ev) < 0)
{
break;
}
}
else
{
int npro = read(events[i].data.fd, buf, sizeof(buf));
if(npro == -1)
{
close(events[i].data.fd );
epoll_ctl(epollfd, EPOLL_CTL_DEL, events[i].data.fd, &ev);
}
else
{
cout << buf << endl;
npro = write(events[i].data.fd, buf, npro);
if(npro == -1)
{
close(events[i].data.fd );
epoll_ctl(epollfd, EPOLL_CTL_DEL, events[i].data.fd, &ev);
}
}
}
}
}
cout << "quit thread " << ((THREADDATA1*)parg)->num << endl;
close(listenfd);
close(epollfd);
return NULL;
}
我们把创建监听socket放到了线程内，并且在bind之前，先调用setsockopt设置了SO_REUSEADDR，再设置了SO_REUSEPORT。记住，这个顺序不能颠倒，不然还是会提示Address already in use。我们同时连接4个请求，输出日志如下
thread id: 13 wake up 1 thread id: 13 accept thread id: 0 wake up 1 thread id: 0 accept thread id: 26 wake up 1 thread id: 15 wake up 1 thread id: 26 accept thread id: 15 accept
从日志可以看出，解决了监听socket的惊群问题。方法三比方法二的优势是，多个监听套接字，如果其中一个监听套接字出现问题，不影响程序运行。这里需要注意的是需要为每一个线程创建一个epoll实例，在这个线程中创建监听套接字，并且绑定，在当前线程监听当前的epoll实例。
为什么不能创建一个epoll实例，然后多个线程使用多个监听套接字呢？是因为，相当于前面的一个epoll实例，一个监听套接字，多个线程监听的情形，只不过多添加了几个监听套接字。epoll_wait返回的时候，并不会因为你是在A线程通过epoll_ctl添加的监听套接字，就返回给A线程。还会出现惊群问题，相当于我们什么都没做，仅仅多绑定了几个监听套接字，变得更乱。所以必须每个线程一个epoll实例，一个监听套接字。
可能的影响
这种方式是可以了，不过因为创建了多个epoll实例，一个连接进来时，需要把它绑定到对应的epoll实例上，这里必须要把这个关系对应好。如果把一个套接字绑定到多个epoll实例上，当有事件来临时，与上面的问题一样，可能唤起多个线程处理。
解决方案四
与方案三类似，只不过是创建进程。这种方案适合每个连接关联比较弱的情形，不然的话，就需要进程间的通信，实现起来更复杂。目前我没有使用这种方式，就没有做更详细的测试。
解决方案五
使用ET模式。边缘触发模式，在状态触发的时候通知。比如当前监听套接字没有事件，这时来了一个连接，那么epoll就会通知一次，只有一个epoll_wait返回。不管我们没有处理，对于当前连接的事件，epoll都不会再次通知。如果包含没有处理的连接，那么就算后续再次来新的连接，epoll也不会通知。
使用ET模式也有需要处理的问题：
无用的唤醒
虽然ET模式基本上可以避免惊群问题了，但是还是会有这种情况：
-
线程A处理监听套接字
-
处理完最后一个连接，监听套接字的事件列表为空
-
因为A不知道已经是最后一个连接了，所以需要再次调用accept
-
这时来到一个新连接，因为原来的监听套接字的事件列表是空的，所以epoll通知事件
-
因为线程A的epoll_wait返回了，并且没有再次调用，所以epoll唤起线程B处理
-
这就出现了线程A和线程B同时accept同一个连接的情况，其中一个会返回EAGAIN
这种情形实际上比较少见，并且影响不大，只是多唤起了一个线程。
饥饿
-
有事件到达，唤起线程A
-
线程A的处理速度比后续事件到达的速度慢
-
线程A不断的处理事件
-
其他线程没有事情做
这个也可以叫做“撑死”，就是一种事件触发，比如监听套接字有新的连接到达，那么唤起线程A处理。但是由于并发太大，监听套接字的事件列表一直有内容，线程A需要不断的处理事件，epoll也不会再次通知其他线程。
这个问题处理起来就比较麻烦了，因为epoll不会再次通知其他线程处理，只能这个线程接收，然后分发。也不能说，这个线程设定一个处理上限，到达了就不处理了，因为其他epoll_wait不会返回，就导致有事件，但是没有线程知道的问题。
还有一种方案就是记录一下当前事件的状态，比如正在接收监听套接字的事件，那么唤起一个线程一起接收。
不管如何操作，都不是我们的初衷，需要我们花费大量心思在各种细节上来实现负载均衡的处理网络消息。
总结
根据上面的内容，建议还是使用方案三：多个线程，多个epoll实例，多个监听套接字。这种实现即简单，也能负载均衡。
接收数据乱序
问题描述
-
套接字A有数据到达
-
唤起线程T1处理
-
在T1接收数据过程中，套接字A有新数据到达或者T1第一次接收没有把数据读完
-
唤起线程T2处理
这个问题非常严重，并且难以解决。如果是accept，那么就算多个线程调用，顶多是返回EAGAIN的错误，或者各自接收到一个连接请求，因为accept系统内部处理了，不会出现同时返回成功同一个连接请求。但是多个线程调用recv本来就是合理的，系统也不会做额外的限制。如果我们在这个地方没有做处理（比如通过锁把数据按照接收的顺序放入同一个缓冲区，或者强制限制只能一个线程接收数据等），那么就会出现乱序。对方发送的是ABCDE，我们接收的是ACDBE，这肯定是不行的。
问题重现
#include "epoll2.h"
#include <iostream>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <errno.h>
#define MAX_EVENTS 100
#define MAX_BUF 1
using namespace std;
epoll2::epoll2()
{
m_epollfd = -1;
m_listenfd = -1;
for(auto& iter : m_threaddata)
{
iter.pepoll = this;
}
}
epoll2::~epoll2()
{
close(m_listenfd);
close(m_epollfd);
}
void epoll2::start()
{
struct sockaddr_in servaddr;
struct epoll_event ev;
bzero(&servaddr, sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(9000);
if((m_listenfd = socket(AF_INET, SOCK_STREAM|SOCK_NONBLOCK, 0)) == -1)
{
cout << "create listen socket err" << endl;
exit(EXIT_FAILURE);
}
int val = 1;
if(setsockopt(m_listenfd, SOL_SOCKET, SO_REUSEPORT, &val, sizeof (val)) == -1)
{
cout << "set listen socket reuseport err" << endl;
}
if(bind(m_listenfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr)) == -1)
{
cout << "bind listen socket err" << endl;
cout << strerror(errno) << endl;
exit(EXIT_FAILURE);
}
if(listen(m_listenfd, 5) == -1)
{
cout << "listen listen socket err" << endl;
exit(EXIT_FAILURE);
}
m_epollfd = epoll_create1(0);
ev.events = EPOLLIN|EPOLLONESHOT;
ev.data.fd = m_listenfd;
if(epoll_ctl(m_epollfd, EPOLL_CTL_ADD, m_listenfd, &ev) < 0)
{
cout << "add listen socket to epoll err" << endl;
exit(EXIT_FAILURE);
}
int i = 0;
for(auto& iter : m_threaddata)
{
iter.num = i;
pthread_create(&(iter.threadid), NULL, start_routine, (void*)(&iter));
i++;
}
for(auto& iter : m_threaddata)
{
pthread_join(iter.threadid, NULL);
}
}
void epoll2::setnonblocking(int fd)
{
fcntl(fd, F_SETFL, (fcntl(fd, F_GETFL)|O_NONBLOCK));
}
void* epoll2::start_routine(void *parg)
{
epoll2 * pepoll = ((THREADDATA2*)parg)->pepoll;
struct epoll_event events[MAX_EVENTS];
struct sockaddr_in cliaddr;
char buf[MAX_BUF] = {0};
struct epoll_event ev;
socklen_t len = sizeof(struct sockaddr_in);
while(true)
{
int nfds = epoll_wait(pepoll->m_epollfd, events, MAX_EVENTS, -1);
if(nfds == -1)
{
break;
}
for(int i = 0; i < nfds; i++)
{
if(events[i].data.fd == pepoll->m_listenfd)
{
cout << "enter thread " << ((THREADDATA2*)parg)->num << " listen" << endl;
int connfd = accept(pepoll->m_listenfd, (struct sockaddr *)&cliaddr, &len);
if(connfd == -1)
{
if(errno == EAGAIN)
{
cout << "thread id: " << ((THREADDATA2*)parg)->num << " err eagain" << endl;
continue;
}
else
{
cout << "thread id: " << ((THREADDATA2*)parg)->num << " err return" << endl;
break;
}
}
pepoll->setnonblocking(connfd);
cout << "thread id: " << ((THREADDATA2*)parg)->num << " accept" << endl;
ev.events = EPOLLIN;
ev.data.fd = connfd;
if(epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_ADD, connfd, &ev) < 0)
{
break;
}
ev.events = EPOLLIN|EPOLLONESHOT;
ev.data.fd = pepoll->m_listenfd;
epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_MOD, pepoll->m_listenfd, &ev);
}
else
{
cout << "thread id: " << ((THREADDATA2*)parg)->num << " wake up " << nfds << endl;
int npro = read(events[i].data.fd, buf, sizeof(buf));
if(npro == -1)
{
close(events[i].data.fd );
epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_DEL, events[i].data.fd, &ev);
}
else
{
cout << buf << endl;
npro = write(events[i].data.fd, buf, npro);
if(npro == -1)
{
close(events[i].data.fd );
epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_DEL, events[i].data.fd, &ev);
}
}
}
}
}
cout << "quit thread " << ((THREADDATA2*)parg)->num << endl;
return NULL;
}
我们接收到一个连接，然后设置缓冲区为1，发送一个“hello world”。输出如下
enter thread 49 listen thread id: 49 accept thread id: 49 wake up 1 h thread id: 47 wake up 1 thread id: thread id: 49 wake up 1 e 48 wake up 1 thread id: l thread id: 46 wake up 149 wake up 1 thread id: 44 wake up 1 thread id: o thread id: 48 wake up 1 thread id: lthread id: 43 wake up 1 thread id: 37 wake up 1 thread id: 35 wake up 1 45s wake up thread id: 41 wake up 1 thread id: 48 wake up 1 thread id: 1 rthread id: 39 wake up 1 vthread id: e thread id: 31 wake up 1 thread id: 35 wake up 1 thread id: 24 wake up 1 thread id: thread id: thread id: 31 wake up 1 thread id: 23thread id: 21 wake up 1 thread id: 1922 wake up wake up 1 wake up 1 thread id: 36 wake up 1 126 thread id: 32 wake up 1 wake up 1 thread id: 34 wake up 1 thread id: 40 wake up 1 3833 wake up 1 thread id: r42 wake up �1 e wake up 1 44 wake up 1
我们看到多个线程被唤起，分别接收了部分数据。这种情况下我们处理起来很麻烦。
理论上接收数据与接收监听是同样的事件，对于epoll都是EPOLLIN。所以解决乱序的问题与解决监听套接字惊群的问题思路基本一样。
解决方案一 EPOLLEXCLUSIVE
我们对于新的连接设置EPOLLEXCLUSIVE标识，修改代码如下
ev.events = EPOLLIN|EPOLLEXCLUSIVE;
ev.data.fd = connfd;
if(epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_ADD, connfd, &ev) < 0)
{
break;
}
不出我们所料，这个标识并不能解决我们的问题，输出结果如下
he el r � ls r v e o
解决方案二 EPOLLONESHOT
//接收完连接，增加到epoll中的时候设置EPOLLONESHOT ev.events = EPOLLIN|EPOLLONESHOT; ev.data.fd = connfd; epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_ADD, connfd, &ev);
//接收完数据，把事件重新激活 ev.events = EPOLLIN|EPOLLONESHOT; ev.data.fd = events[i].data.fd; epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_MOD, events[i].data.fd, &ev);
发送两条消息，看一下输出日志
thread id: 49 wake up 1 hello server � thread id: 49 wake up 1 hello server �
可以解决这个问题。收发数据本来也不需要多线程同时操作，需要额外的锁，效率可能并不会太高。所以这种方案是可行的。
解决方案三
多个线程，多个epoll实例，多个监听套接字。当一个连接进来，会被一个监听套接字捕获，然后就会加入到当前线程的epoll实例中监听，不会有额外的线程监听当前的epoll实例，相当于单线程，所以是可以解决这个问题的。
代码与上面的基本一致，只不过改了接收的部内容。
#include "epoll1.h"
#include <iostream>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <errno.h>
#include <arpa/inet.h>
#define MAX_EVENTS 100
#define MAX_BUF 1024
using namespace std;
epoll1::epoll1()
{
for(auto& iter : m_threaddata)
{
iter.pepoll = this;
}
}
epoll1::~epoll1()
{
}
void epoll1::start()
{
int i = 0;
for(auto& iter : m_threaddata)
{
iter.num = i;
pthread_create(&(iter.threadid), NULL, start_routine, (void*)(&iter));
i++;
}
for(auto& iter : m_threaddata)
{
pthread_join(iter.threadid, NULL);
}
}
void epoll1::setnonblocking(int fd)
{
fcntl(fd, F_SETFL, (fcntl(fd, F_GETFL)|O_NONBLOCK));
}
void* epoll1::start_routine(void *parg)
{
int epollfd = epoll_create1(0);
bool bcontinue = true;
//cout << "enter thread " << ((THREADDATA1*)parg)->num << endl;
epoll1 * pepoll = ((THREADDATA1*)parg)->pepoll;
struct epoll_event events[MAX_EVENTS];
struct sockaddr_in cliaddr;
char buf[MAX_BUF] = {0};
struct epoll_event ev;
socklen_t len = sizeof(struct sockaddr_in);
int listenfd = 0;
struct sockaddr_in servaddr;
bzero(&servaddr, sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(9000);
if((listenfd = socket(AF_INET, SOCK_STREAM|SOCK_NONBLOCK, 0)) == -1)
{
cout << "create listen socket err" << endl;
bcontinue = false;
}
int val = 1;
if(bcontinue && setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof (val)) == -1)
{
cout << "set listen socket reuseport err" << endl;
bcontinue = false;
}
if(bcontinue && setsockopt(listenfd, SOL_SOCKET, SO_REUSEPORT, &val, sizeof (val)) == -1)
{
cout << "set listen socket reuseport err" << endl;
bcontinue = false;
}
if(bcontinue && bind(listenfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr)) == -1)
{
cout << "bind listen socket err" << endl;
cout << strerror(errno) << endl;
bcontinue = false;
}
if(bcontinue && listen(listenfd, 5) == -1)
{
cout << "listen listen socket err" << endl;
bcontinue = false;
}
ev.events = EPOLLIN;
ev.data.fd = listenfd;
if(bcontinue && epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &ev) < 0)
{
cout << "add listen socket to epoll err" << endl;
bcontinue = false;
}
while(bcontinue)
{
int nfds = epoll_wait(epollfd, events, MAX_EVENTS, -1);
if(nfds == -1)
{
break;
}
for(int i = 0; i < nfds; i++)
{
if(events[i].data.fd == listenfd)
{
int connfd = accept(listenfd, (struct sockaddr *)&cliaddr, &len);
if(connfd == -1)
{
if(errno == EAGAIN)
{
cout << "thread id: " << ((THREADDATA1*)parg)->num << " err eagain" << endl;
continue;
}
else
{
cout << "thread id: " << ((THREADDATA1*)parg)->num << " err return" << endl;
break;
}
}
pepoll->setnonblocking(connfd);
cout << "thread id: " << ((THREADDATA1*)parg)->num << " accept" << endl;
ev.events = EPOLLIN;
ev.data.fd = connfd;
epoll_ctl(epollfd, EPOLL_CTL_ADD, connfd, &ev);
}
else
{
cout << "thread id: " << ((THREADDATA1*)parg)->num << " wake up " << nfds << endl;
while (read(events[i].data.fd, buf, sizeof(buf)) != -1)
{
cout << buf;
}
cout << endl;
if(errno != EAGAIN)
{
close(events[i].data.fd );
epoll_ctl(pepoll->m_epollfd, EPOLL_CTL_DEL, events[i].data.fd, &ev);
}
}
}
}
cout << "quit thread " << ((THREADDATA1*)parg)->num << endl;
close(listenfd);
close(epollfd);
return NULL;
}
输出日志如下
thread id: 34 accept thread id: 34 wake up 1 hello server thread id: 34 wake up 1 hello server
同样很好的解决了我们的问题
其他解决方案
与上面监听套接字的四和五一样，也有着同样的影响，我们不使用这种模式，没做验证。
结论
-
EPOLLEXCLUSIVE在LT模式下并不能解决我们的问题，放弃
-
EPOLLONESHOT可以解决问题，但是在并发比较大的时候，不能很好的利用系统资源
-
多线程，多epoll实例。目前来看是最适合我们的方案
-
多进程，多epoll实例。更适合网站类型的，比如nginx，每个连接关联比较弱。
-
ET模式，有着无法避免的缺陷，放弃
所以我们最终选择多线程，多epoll实例的方案。
#感谢您访问本站# #本文转载自互联网，若侵权，请联系删除，谢谢！657271#qq.com#
Recommend
About Joyk
Aggregate valuable and interesting links.
Joyk means Joy of geeK