Linux-多路转接-epoll

接口认识

epoll_create

int epoll_create(int size);
1

🚀创建一个epoll模型，在Linux2.8.6之后size参数是被忽略的，通常传递一个大于0的整数。
🚀返回值是一个文件描述符，所以在使用完毕后要close掉，防止资源泄漏。

epoll_ctl

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
1

🚀第一个参数为创建的epoll模型对应的fd。
🚀第二个参数为在epoll模型中对让内核关心的文件描述符某些事件进行增删改。

EPOLL_CTL_ADD:在epoll模型中添加让内核关心对某个fd的某些事件。
EPOLL_CTL_MOD:修改epoll模型中某个fd所关心的事件。
EPOLL_CTL_MOD:删除epoll模型中某个fd，不用让内核关心此fd了。

🚀第三个参数为对哪个文件描述符fd进行操作。
🚀第四个参数是一个结构体类型，其内部结构如下：

epoll_event结构体中包含两个字段：uint32_t表示要让内核关心的哪些事件，epoll_data_t是一个联合体，通常设置fd字段即可，ptr指针可以用来存放用户自定义数据的地址，在epoll_wait返回时，会将我们当初设置的字段原样返回。
events字段：

epoll_wait

int epoll_wait(int epfd, struct epoll_event *events,
                      int maxevents, int timeout);
1
2

🚀第一个参数为创建的epoll模型对应的fd。
🚀第二个参数需要用户传递一个数组进去，返回时其内容会被填充，表示内核告诉用户哪些fd的哪些事件已经就绪了。
🚀第三个参数为用户传递数组的大小。
🚀第四个参数为等待时间：
大于0：表示经过timeout时间的等待没有事件就绪，就返回。
等于0：表示没有事件就绪就直接返回。返回值设置为0。
-1：表示阻塞等待模式

🚀返回值：大于0：表示用户传入的数组中多少个元素被设置。
等于0：表明timeout。
小于0：表明epoll_wait出错。

epoll工作原理

在内核中创建的数据结构

🚀在用户通过epoll_create创建一个epoll模型时，在内核中其实创建了以下的数据结构：

🚀一颗红黑树：

红黑树结点中存放的内容包括：fd，用户让内核关心这个fd的哪些事件，这个fd的哪些事件已经就绪了(revents)等。除此之外还会包括一些连接字段。红黑树是一颗近似平衡的搜索二叉树，所以用户通过epoll_ctl对某个fd进行插入，修改，删除，就是对红黑树结点做插入，修改，删除，效率是很高的。

🚀一个就绪队列：

就绪队列中的结点就表示，让内核关心的哪些fd上的哪些事件已经就绪了，用户调用epoll_wait时，内核就是将就绪队列中的信息拷贝给用户，另外epoll_wait可以以O（1）的事件复杂度来判断有没有事件就绪，因为如果就绪队列为空，表示没有事件就绪，如果事件不为空表示有事件就绪。

🚀struct file中的回调指针被设置

以读事件为例，内核怎么知道读事件就绪了呢？这我们就要弄清楚数据是从哪里来的，首先网络通信中接收到的数据一定是网卡先收到的，当网卡收到数据时可能会触发某种中断机制，通知操作系统，随后数据通过网络协议栈的层层解析最终被放到struct file中的缓冲区中，对于用户而言，数据被放到了struct file缓冲区中才代表实践就绪了，那么既然是操作系统将数据拷贝到的struct file缓冲区中的，在epoll的工作原理下，OS将数据拷贝到struct file缓冲区中时，会触发回调机制，将红黑树的结点链接到就绪队列中（内核中的数据结构的某个结点可以数据多个数据结构）。
写事件与读事件类似。

epoll模型的一个完整工作流程

epoll工作模式

LT-水平触发

🚀LT模式是epoll的默认工作方式，当内核缓冲区的数据越过设定的水位线后，epoll_wait就会返回，通知用户有事件已经就绪了，如果用户置之不理不去处理这个事件，那么epoll会一直通知用户这个事件已经就绪了。

ET-边缘触发

🚀ET工作模式下，epoll检测到某个套接字上的事件已经就绪了，但是用户并没有处理，那么下次调用epoll_wait的时候就不会返回，也就意味这只会通知一次，只有当套接字中有新的数据到来时才会再次通知用户一次。
🚀对于ET工作模式，假设某个客户端就给服务器发送了一次数据，那么对于服务器而言就必须要将数据全部读取上来，因为epoll不会通我第二次，并且如果采用阻塞式的读取，如果某次恰好将数据读取完，那么下次再读取时就会阻塞住，这时不合理的，所以ET模式一定要搭配非阻塞IO使用，ET这种工作模式倒逼用户一次就将数据读取完。

两种方式的对比

🚀由于ET模式倒逼用户一次将数据读取完，那么就能有效的减少epoll触发的次数，epoll可以利用重复通知某一进程的资源去通知其他进程，其IO效率是比LT要高的。但是，对于LT而言也可以采用非阻塞IO的方式一次将数据读取完，这种情况下LT和ET的IO效率是相等的。所以ET的IO效率是>=LT的iIO效率的。

🚀ET模式适合于高IO的场景。
🚀LT模式适合于高响应的模式，对于LT模式，可以每次读取一部分，对这一部分数据做处理尽快的响应客户端请求，所以LT适用于高响应的场景。

epoll的使用场景

🚀epoll适用于存在大量的客户端连接，但只有少部分的连接处于活跃状态的的场景下，例如各种互联网APP的入口服务器，就很适合使用epoll。但是，如果只有很少的连接，就不太适合使用epoll。

对于poll的改进

🚀在用户层面不再用维护pollfd这样类型的数组了，在epoll中由内核进行管理。
🚀不用将用户维护关于fd的某些事件的的信息，每次调用时都拷贝给内核。
🚀在epoll_wait返回时，不用对用户维护的数组做遍历来检测哪些fd上的哪些事件已经就绪，因为返回的都是已经就绪的事件。

惊群效应

什么是惊群效应

🚀在多线程或者多进程的场景下，多个线程或者多个进程共同监听同一个端口，当一个新的连接到来时，操作系统不知道通知哪个进程/线程去获取这个连接，所以就直接都通知，导致多个进程/线程只有一个能够成功的获取这个连接，而对于其他没获取连接成功的进程/线程，本有阻塞状态->运行状态，这种进程/线程的切换是浪费资源的，在一个多用户的服务上，如果存在这种惊群效应，那么对CPU是有一定消耗的。

如何解决惊群效应

🚀在Nginx中，是采用加锁的方式来处理惊群效应的，在连接到来时，多个进程谁先抢到锁，谁去获取这个连接。
🚀Nginx中锁的设计：在多个进程之间共享一块内存，在内存中设置一个变量-share_mtx,其值只有两种状态，0和1，当某个进程想去获取连接前，先去判断share_mtx，如果这个值为0，将其改为1，然后去获取连接。如果有的进程判断时，这个值本身就是1，那么去休眠一定时间，并将这个值改为0。这样设计肯定是存在漏洞的，势必会存在某几个进程同时判断share_mtx的值都是0，那么这几个进程都会去获取连接，所以对这个过程要做处理：1.可以采用CAS将这个判断+赋值的过程整合成一个原子操作2.使用mutex互斥锁3.使用spinlock自旋锁。

原子操作/mutex/spinlock如何选择

🚀临界区内代码量较小，且CPU体系结构的指令集中有指令能够支持，则选择原子操作。
🚀临界区内代码量很大，使用mutex互斥锁。
🚀介于两者之间的使用spinlock。

简单reactor模式的epoll服务器的编写

#ifndef __M_SERVER_H__
#define __M_SERVER_H__ 
#include "Sock.hpp"
#include "err.hpp"
#include "log.hpp"
#include 
#include 
#include "epoll.hpp"
#include "Protocol.hpp"
#include "util.hpp"
#include 
#include 
#include 
const int N = 1024;
const int PORT = 8888;
class Connection;
using func_t = std::function<void (Connection*)>;
using Func_t = std::function<void (const Request&,Response&)>;
class Connection {
    private:
        int _fd;
        std::string _inbuffer;
        std::string _outbuffer;
        std::string _ip;
        uint16_t _port;
        uint32_t _events = 0;
        func_t _recver;
        func_t _sender;
        func_t _excepter;
    public:
        Connection(int fd,const std::string& ip,uint16_t port) :_fd(fd),_ip(ip),_port(port) {}
        int GetFd() {
            return _fd;
        }
        std::string Ip() {
            return _ip;
        }
        uint16_t Port() {
            return _port;
        }
        void SetEvents(uint32_t events) {
            _events = events;
        }
        void SetRecver(const func_t& recver) {
            _recver = recver;
        }
        void SetSender(const func_t& sender) {
            _sender = sender;
        }
        void SetExcepter(const func_t& excepter) {
            _excepter = excepter;
        }
        uint32_t GetEvents() {
            return _events;
        }
        func_t& GetRecver() {
            return _recver;
        }
        func_t& GetSender() {
            return _sender;
        }
        func_t& GetExcepter() {
            return _excepter;
        }
        string& GetInBuffer() {
            return _inbuffer;
        }
        string& GetOutBuffer() {
            return _outbuffer;
        }
};
class EpollServer {
    private:
        std::string _server_ip;
        uint16_t _server_port;
        sock::Sock _listensock;
        Epoll _epoll;
        epollfd _events[N];
        std::unordered_map<int,Connection*> _connections;
        Func_t _func;
    public:
        using ptr = std::unique_ptr<EpollServer>;
        EpollServer(const Func_t& func,uint16_t port = PORT,const std::string& ip = "0.0.0.0")
            :_server_ip(ip),_server_port(port),_func(func) 
        {}
        ~EpollServer() {
            _listensock.Close();
            _epoll.Close();
        }
        void AddEvents(int fd,uint32_t events,std::string ip = "127.0.0.1",uint16_t port = 0) {
            //1.将套接字添加到epoll模型中
            if(false == _epoll.AddEvent(fd,events)) {
                std::cerr << "将套接字设置进epoll模型失败\n";
            }
            //2.将套接字添加到_connections中
            Connection* conn = new Connection(fd,ip,port);
            conn->SetEvents(events);
            //注册recv方法
            if(fd == _listensock.Fd()) {
                conn->SetRecver(std::bind(&EpollServer::Acceptor,this,std::placeholders::_1));
            } else {
                conn->SetRecver(std::bind(&EpollServer::Reader,this,std::placeholders::_1));
            }
            //注册send方法
            conn->SetSender(std::bind(&EpollServer::Writer,this,std::placeholders::_1));
            //注册异常方法
            conn->SetExcepter(std::bind(&EpollServer::Excepter,this,std::placeholders::_1));
            _connections.insert({fd,conn});
        }
        void Init() {
            _listensock.Socket();
            _listensock.Bind(_server_port);
            _listensock.Listen();
            //创建epoll模型
            if(_epoll.Create() < 0) {
                std::cerr << "创建epoll模型失败\n";
            }
            //将listen套接字设置为非阻塞模式
            Util::SetNonBlock(_listensock.Fd());
            AddEvents(_listensock.Fd(),READ | ET | EXCEPT);
        }
        void Acceptor(Connection* conn) {
            do {
                //处理新获取的连接
                std::string ip;
                uint16_t port;
                int err = 0;
                int n = _listensock.Accept(&ip,&port,&err);
                if(n > 0) {
                    //获取新的连接成功
                    
                    //将套接字设置设置为非阻塞
                    Util::SetNonBlock(n);
                    std::cout << "获取了一个新的连接: " << ip << ":" << port << std::endl; 
                    AddEvents(n,READ | ET | EXCEPT,ip,port);
                } else if(n < 0) {
                    if((err & EAGAIN) || (err & EWOULDBLOCK)) {
                        break;
                    } else if(err & EINTR){
                        continue;
                    } else {
                        std::cerr << "获取连接失败\n";
                        continue;
                    }
                } 
            }while(conn->GetEvents() & ET);
        }
        void HandlerBusiness(Connection* conn) {
            bool stop = false;
            while(!stop) {
                //1.获取inbuffer中完整的报文
                std::string read_str;
                int n = ReadPackge(conn->GetInBuffer(),&read_str);
                if(n <= 0) {
                    break;
                } //这一轮没有读取到完整的一个报文
                //2.去掉报头
                read_str = RemoveHeaders(read_str,n);
                //3.反序列化
                Request req;
                req.Deserialize(read_str);
                //4.业务处理
                Response res;
                _func(req,res);
                //5.序列化
                std::string out_str;
                res.Serialize(&out_str);
                //6.添加报头
                out_str = AddHeaders(out_str);
                conn->GetOutBuffer() += out_str;
                //7.给客户端响应
                conn->GetSender()(conn);
            }
        }
        bool ReaderHelp(Connection* conn) {
            bool safe = true;
            do {
                std::cout << "读取开始,fd: " << conn->GetFd() << std::endl;
                std::string ip = conn->Ip();
                uint16_t port = conn->Port();
                char buffer[4096] {0};
                ssize_t n = recv(conn->GetFd(),buffer,sizeof(buffer) - 1,0);
                if(n > 0) {
                    buffer[n] = 0;
                    conn->GetInBuffer() += buffer;
                    std::cout << ip << ":" << port << "inbuffer>>>" << conn->GetInBuffer(); 
                } else {
                    //读取失败
                    if(n < 0) {
                        if((errno & EAGAIN) || (errno & EWOULDBLOCK)) {
                            break;
                        } else if(errno & EINTR) {
                            continue;
                        } else {
                            //读出错，将inbuffer置空然后断开连接
                            conn->GetInBuffer() = "";
                            conn->GetExcepter()(conn);
                            safe = false;
                            break;
                        }
                    }else if(n == 0) { //客户端关闭
                        std::cout << "客户端关闭连接: " << ip << ":" << port << std::endl;
                        conn->GetExcepter()(conn);
                        safe = false;
                        break;
                    }
                }
            }while(conn->GetEvents() & ET);
            return safe;
        }
        void Reader(Connection* conn) {
            bool res = ReaderHelp(conn);
            if(false == res) return;
            //完成读取后进行业务处理
            HandlerBusiness(conn);
        }
        void Writer(Connection* conn) {
            bool safe = true;
            do {
                ssize_t n = send(conn->GetFd(),conn->GetOutBuffer().c_str(),conn->GetOutBuffer().size(),0);
                if(n > 0) {
                    conn->GetOutBuffer().erase(0,n);
                    //发送完全部数据直接break
                    if(conn->GetOutBuffer().empty()) {
                        break;
                    }
                } else {
                    if((errno & EAGAIN) || (errno & EWOULDBLOCK)) {
                        break;
                    } else if(errno & EINTR) {
                        continue;
                    } else {
                        conn->GetExcepter()(conn);
                        safe = false;
                    }
                }
            }while(conn->GetEvents() & ET);
            if(false == safe) {
                return;
            }
            if(conn->GetOutBuffer().empty()) {
                //数据都发送完了,让epoll不再关心此次写事件
                _epoll.ModifyEvent(conn->GetFd(),READ | ET | EXCEPT); 
            } else {
                _epoll.ModifyEvent(conn->GetFd(),WRITE | READ | ET | EXCEPT);
                //没有写完让将写事件添加到epoll中
            }
        }
        void Excepter(Connection* conn) {
            //从epoll模型中去除此fd
             _epoll.DelEvent(conn->GetFd());
            //从connections中去除此fd
            _connections.erase(conn->GetFd());
            //关闭fd
            close(conn->GetFd());
            //delete connection对象
            delete conn;
        }
        bool IsConnExist(int fd) {
            return _connections.find(fd) != _connections.end();
        }
        void LoopOnce(int timeout) {
            epollfd fds[N];
            int n = _epoll.Wait(fds,N,timeout);
            for(int i = 0; i < n; i++) {
                if(fds[i].events & EXCEPT) {
                    fds[i].events |= (READ | WRITE);
                }
                if((fds[i].events & READ) && IsConnExist(fds[i].fd)) {
                    std::cout << "有一个读事件就绪\n";
                    std::cout << "fd: " << fds[i].fd << std::endl;
                    _connections[fds[i].fd]->GetRecver()(_connections[fds[i].fd]);
                } 
                if((fds[i].events & WRITE) && IsConnExist(fds[i].fd))  {
                    //写事件就绪
                    _connections[fds[i].fd]->GetSender()(_connections[fds[i].fd]);    
                }
            }
        }
        void Start() {
            int timeout = -1;
            while(true) {
                LoopOnce(timeout);
            }
        }
};

#endif 
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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288