【在Linux世界中追寻巨大的One Piece】多路转接epoll(续)

目录
1 -> epoll的工作方式
1.1 -> 水平触发(Level Triggered)工作模式
1.2 -> 边沿触发(Edge Triggered)工作模式
2 -> 对比LT与ET
3 -> 明确ET模式和非壅闭文件形貌符
4 -> epoll的利用场景
5 -> epoll示例
5.1 -> epoll服务器(LT模式)
5.2 -> epoll服务器(ET模式)

---

1 -> epoll的工作方式

epoll有2中工作方式：

• 水平触发(LT)
  
• 边沿触发(ET)
  

加入有如许一个例子：

• 已经把一个tcp_socket添加到epoll形貌符。
  
• 这个时间socket的另一端被写入了2KB的数据。
  
• 调用epoll_wait，而且它会返回。说明它已经准备好读取操纵。
  
• 然后调用read，只读取了1KB的数据。
  
• 继续调用epoll_wait……
  

1.1 -> 水平触发(Level Triggered)工作模式

epoll默认状态下就是LT工作模式。 

• 当epoll检测到socket上事件就绪的时间，可以不立刻举行处理。或者只处理一部分。
  
• 如上面的例子，由于只读了1K的数据，缓冲区中还剩1K的数据，在第二次调用epoll_wait时，epoll_wait仍旧会立刻返回并关照socket读事件就绪。
  
• 直到缓冲区上所有的数据都被处理完，epoll_wait才不会立刻返回。
  
• 支持壅闭读写和非壅闭读写。
  

1.2 -> 边沿触发(Edge Triggered)工作模式

如果我们在第一步将socket添加到epoll形貌符的时间利用了EPOLLET标记，epoll进入ET工作模式。

• 当epoll检测到socket上事件就绪时，必须立刻处理。
  
• 如上面的例子，虽然只读了1K的数据，缓冲区还剩1K的数据，在第二次调用epoll_wait的时间，epoll_wait不会再返回了。
  
• 也就是说，ET模式下，文件形貌符上的事件就绪后，只有一次处理机会。
  
• ET的性能比LT的性能更高(epoll_wait返回的次数少了许多)。Nginx默认采用ET模式利用epoll。
  
• 只支持非壅闭的读写。
  

select和poll着实也是工作在LT模式下。epoll既可以支持LT，也可以支持ET。

2 -> 对比LT与ET

LT是epoll的默认行为。
利用ET能够减少epoll触发的次数。但是代价就是强逼着步伐员一次相应就绪过程中就把所有的数据都处理完。
相称于一个文件形貌符就绪后，不会反复被提示就绪，看起来就比LT更高效一些。但是在LT环境下如果也能做到每次就绪的文件形貌符都立刻处理，不让这个就绪被反复提示的话，着实性能也是一样的。
另一方面，ET的代码复杂水平更高了。


3 -> 明确ET模式和非壅闭文件形貌符

利用ET模式的epoll，需要将文件形貌设置为非壅闭。这个不是接口上的要求，而是“工程实践”上的要求。
假设一个场景：服务器接收到一个10k的请求，会向客户端返回一个应答数据。如果客户端收不到应答，不会发送第二个10K的请求。

如果服务端写的代码是壅闭式的read，而且一次只read1K数据的话(read不能包管一次就把所有的数据都读出来，参考man手册的说明，大概被信号打断)，剩下的9K数据就会待在缓冲区中。

此时由于epoll是ET模式，并不会认为文件形貌符读就绪。epoll_wait就不会再次返回。剩下的9K数据会不停在缓冲区中。直到下一次客户端再给服务器写数据。
但是问题来了。

• 服务器只读到1K个数据，要10K读完才会给客户端返回相应数据。
  
• 客户端要读到服务器的相应，才会发送下一个请求。
  
• 客户端发送了下一个请求，epoll_wait才会返回，才气去读缓冲区中剩余的数据。
  

以是，为了办理上述问题(壅闭read不肯定能一下把完整的请求读完)，于是就可以利用非壅闭轮训的方式来读缓冲区，包管肯定能把完整的请求都读出来。
而如果是LT没这个问题。只要缓冲区中的数据没读完，就能够让epoll_wait返回文件形貌符读就绪。

4 -> epoll的利用场景

epoll的高性能，是有肯定的特定场景的。如果场景选择的不适宜，epoll的性能大概适得其反。

• 对于多连接，且多连接中只有一部分连接比力活跃时，比力适合利用epoll。
  

比方，典型的一个需要处理上万个客户端的服务器，比方各种互联网APP的入口服务器，如许的服务器就很适合epoll。
如果只是系统内部，服务器和服务器之间举行通讯，只有少数的几个连接，这种环境下用epoll就并不适合。具体要根据需求和场景特点来决定利用哪种IO模子。


5 -> epoll示例

5.1 -> epoll服务器(LT模式)

tcp_epoll_server.hpp

1. ///
2. // 封装一个 Epoll 服务器, 只考虑读就绪的情况
3. ///
5. #pragma once
6. #include <vector>
7. #include <functional>
8. #include <iostream>
9. #include <sys/epoll.h>
10. #include "tcp_socket.hpp"
12. typedef std::function<void(const std::string&, std::string*resp)> Handler;
14. class Epoll
15. {
16. public:
17.         Epoll()
18.         {
19.                 epoll_fd_ = epoll_create(10);
20.         }
22.         ~Epoll()
23.         {
24.                 close(epoll_fd_);
25.         }
27.         bool Add(const TcpSocket& sock) const
28.         {
29.                 int fd = sock.GetFd();
30.                 printf("[Epoll Add] fd = %d\n", fd);
32.                 epoll_event ev;
33.                 ev.data.fd = fd;
34.                 ev.events = EPOLLIN;
36.                 int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev);
37.                 if (ret < 0)
38.                 {
39.                         perror("epoll_ctl ADD");
40.                         return false;
41.                 }
43.                 return true;
44.         }
46.         bool Del(const TcpSocket& sock) const
47.         {
48.                 int fd = sock.GetFd();
50.                 printf("[Epoll Del] fd = %d\n", fd);
52.                 int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, NULL);
53.                 if (ret < 0)
54.                 {
55.                         perror("epoll_ctl DEL");
56.                         return false;
57.                 }
59.                 return true;
60.         }
62.         bool Wait(std::vector<TcpSocket>* output) const
63.         {
64.                 output->clear();
65.                 epoll_event events[1000];
67.                 int nfds = epoll_wait(epoll_fd_, events, sizeof(events) / sizeof(events[0]), -1);
68.                 if (nfds < 0)
69.                 {
70.                         perror("epoll_wait");
71.                         return false;
72.                 }
74.                 // [注意!] 此处必须是循环到 nfds, 不能多循环
75.                 for (int i = 0; i < nfds; ++i)
76.                 {
77.                         TcpSocket sock(events[i].data.fd);
78.                         output->push_back(sock);
79.                 }
81.                 return true;
82.         }
84. private:
85.         int epoll_fd_;
86. };
88. class TcpEpollServer
89. {
90. public:
91.         TcpEpollServer(const std::string& ip, uint16_t port) : ip_(ip),
92.                 port_(port)
93.         {
94.         }
96.         bool Start(Handler handler)
97.         {
98.                 // 1. 创建 socket
99.                 TcpSocket listen_sock;
100.                 CHECK_RET(listen_sock.Socket());
102.                 // 2. 绑定
103.                 CHECK_RET(listen_sock.Bind(ip_, port_));
105.                 // 3. 监听
106.                 CHECK_RET(listen_sock.Listen(5));
108.                 // 4. 创建 Epoll 对象, 并将 listen_sock 加入进去
109.                 Epoll epoll;
110.                 epoll.Add(listen_sock);
112.                 // 5. 进入事件循环
113.                 for (;;)
114.                 {
115.                         // 6. 进行 epoll_wait
116.                         std::vector<TcpSocket> output;
117.                         if (!epoll.Wait(&output))
118.                         {
119.                                 continue;
120.                         }
122.                         // 7. 根据就绪的文件描述符的种类决定如何处理
123.                         for (size_t i = 0; i < output.size(); ++i)
124.                         {
125.                                 if (output[i].GetFd() == listen_sock.GetFd())
126.                                 {
127.                                         // 如果是 listen_sock, 就调用 accept
128.                                         TcpSocket new_sock;
129.                                         listen_sock.Accept(&new_sock);
130.                                         epoll.Add(new_sock);
131.                                 }
132.                                 else
133.                                 {
134.                                         // 如果是 new_sock, 就进行一次读写
135.                                         std::string req, resp;
136.                                         bool ret = output[i].Recv(&req);
137.                                         if (!ret)
138.                                         {
139.                                                 // [注意!!] 需要把不用的 socket 关闭
140.                                                 // 先后顺序别搞反. 不过在 epoll 删除的时候其实就已经关闭 socket 了
141.                                                 epoll.Del(output[i]);
142.                                                 output[i].Close();
143.                                                 continue;
144.                                         }
146.                                         handler(req, &resp);
147.                                         output[i].Send(resp);
148.                                 } // end for
149.                         } // end for (;;)
150.                 }
152.                 return true;
153.         }
155. private:
156.         std::string ip_;
157.         uint16_t port_;
158. };

复制代码

dict_server.cc只需要将server对象的类型改成TcpEpollServer即可。

5.2 -> epoll服务器(ET模式)

基于LT版本稍加修改即可。

• 修改tcp_socket.hpp，新增非壅闭读和非壅闭写接口。
  
• 对于accept返回的new_sock加上EPOLLET如许的选项。
  

注意：
此代码暂时未考虑listen_sock ET的环境。如果将listen_sock设为ET，则需要非壅闭轮询的方式accept。否则会导致同一时刻大量的客户端同时连接的时间，只能accept一次的问题。

tcp_socket.hpp

1. // 以下代码添加在 TcpSocket 类中
2. // 非阻塞 IO 接口
4. bool SetNoBlock()
5. {
6.         int fl = fcntl(fd_, F_GETFL);
7.         if (fl < 0)
8.         {
9.                 perror("fcntl F_GETFL");
10.                 return false;
11.         }
13.         int ret = fcntl(fd_, F_SETFL, fl | O_NONBLOCK);
14.         if (ret < 0)
15.         {
16.                 perror("fcntl F_SETFL");
17.                 return false;
18.         }
20.         return true;
21. }
23. bool RecvNoBlock(std::string* buf) const
24. {
25.         // 对于非阻塞 IO 读数据, 如果 TCP 接受缓冲区为空, 就会返回错误
26.         // 错误码为 EAGAIN 或者 EWOULDBLOCK, 这种情况也是意料之中, 需要重试
27.         // 如果当前读到的数据长度小于尝试读的缓冲区的长度, 就退出循环
28.         // 这种写法其实不算特别严谨(没有考虑粘包问题)
29.         buf->clear();
30.         char tmp[1024 * 10] = { 0 };
31.         for (;;)
32.         {
33.                 ssize_t read_size = recv(fd_, tmp, sizeof(tmp) - 1, 0);
34.                 if (read_size < 0)
35.                 {
36.                         if (errno == EWOULDBLOCK || errno == EAGAIN)
37.                         {
38.                                 continue;
39.                         }
40.                         perror("recv");
41.                         return false;
42.                 }
44.                 if (read_size == 0)
45.                 {
46.                         // 对端关闭, 返回 false
47.                         return false;
48.                 }
50.                 tmp[read_size] = '\0';
51.                 *buf += tmp;
52.                 if (read_size < (ssize_t)sizeof(tmp) - 1)
53.                 {
54.                         break;
55.                 }
56.         }
58.         return true;
59. }
61. bool SendNoBlock(const std::string& buf) const
62. {
63.         // 对于非阻塞 IO 的写入, 如果 TCP 的发送缓冲区已经满了, 就会出现出错的情况
64.         // 此时的错误号是 EAGAIN 或者 EWOULDBLOCK. 这种情况下不应放弃治疗
65.         // 而要进行重试
66.         ssize_t cur_pos = 0; // 记录当前写到的位置
67.         ssize_t left_size = buf.size();
68.         for (;;)
69.         {
70.                 ssize_t write_size = send(fd_, buf.data() + cur_pos,
71.                         left_size, 0);
72.                 if (write_size < 0)
73.                 {
74.                         if (errno == EAGAIN || errno == EWOULDBLOCK)
75.                         {
76.                                 // 重试写入
77.                                 continue;
78.                         }
80.                         return false;
81.                 }
83.                 cur_pos += write_size;
84.                 left_size -= write_size;
85.                 // 这个条件说明写完需要的数据了
86.                 if (left_size <= 0)
87.                 {
88.                         break;
89.                 }
90.         }
92.         return true;
93. }

复制代码

tcp_epoll_server.hpp

1. ///
2. // 封装一个 Epoll ET 服务器
3. // 修改点:
4. // 1. 对于 new sock, 加上 EPOLLET 标记
5. // 2. 修改 TcpSocket 支持非阻塞读写
6. // [注意!] listen_sock 如果设置成 ET, 就需要非阻塞调用 accept 了
7. ///
9. #pragma once
11. #include <vector>
12. #include <functional>
13. #include <sys/epoll.h>
14. #include "tcp_socket.hpp"
16. typedef std::function<void(const std::string&, std::string* resp)> Handler;
18. class Epoll
19. {
20. public:
21.         Epoll()
22.         {
23.                 epoll_fd_ = epoll_create(10);
24.         }
26.         ~Epoll()
27.         {
28.                 close(epoll_fd_);
29.         }
31.         bool Add(const TcpSocket& sock, bool epoll_et = false) const
32.         {
33.                 int fd = sock.GetFd();
34.                 printf("[Epoll Add] fd = %d\n", fd);
36.                 epoll_event ev;
37.                 ev.data.fd = fd;
38.                 if (epoll_et)
39.                 {
40.                         ev.events = EPOLLIN | EPOLLET;
41.                 }
42.                 else
43.                 {
44.                         ev.events = EPOLLIN;
45.                 }
47.                 int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev);
48.                 if (ret < 0)
49.                 {
50.                         perror("epoll_ctl ADD");
51.                         return false;
52.                 }
54.                 return true;
55.         }
57.         bool Del(const TcpSocket& sock) const
58.         {
59.                 int fd = sock.GetFd();
60.                 printf("[Epoll Del] fd = %d\n", fd);
62.                 int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, NULL);
63.                 if (ret < 0)
64.                 {
65.                         perror("epoll_ctl DEL");
66.                         return false;
67.                 }
69.                 return true;
70.         }
72.         bool Wait(std::vector<TcpSocket>* output) const
73.         {
74.                 output->clear();
75.                 epoll_event events[1000];
77.                 int nfds = epoll_wait(epoll_fd_, events, sizeof(events) / sizeof(events[0]), -1);
78.                 if (nfds < 0)
79.                 {
80.                         perror("epoll_wait");
81.                         return false;
82.                 }
84.                 // [注意!] 此处必须是循环到 nfds, 不能多循环
85.                 for (int i = 0; i < nfds; ++i)
86.                 {
87.                         TcpSocket sock(events[i].data.fd);
88.                         output->push_back(sock);
89.                 }
91.                 return true;
92.         }
94. private:
95.         int epoll_fd_;
96. };
98. class TcpEpollServer
99. {
100. public:
101.         TcpEpollServer(const std::string& ip, uint16_t port) : ip_(ip),
102.                 port_(port)
103.         {
104.         }
106.         bool Start(Handler handler)
107.         {
108.                 // 1. 创建 socket
109.                 TcpSocket listen_sock;
110.                 CHECK_RET(listen_sock.Socket());
112.                 // 2. 绑定
113.                 CHECK_RET(listen_sock.Bind(ip_, port_));
115.                 // 3. 监听
116.                 CHECK_RET(listen_sock.Listen(5));
118.                 // 4. 创建 Epoll 对象, 并将 listen_sock 加入进去
119.                 Epoll epoll;
120.                 epoll.Add(listen_sock);
122.                 // 5. 进入事件循环
123.                 for (;;)
124.                 {
125.                         // 6. 进行 epoll_wait
126.                         std::vector<TcpSocket> output;
127.                         if (!epoll.Wait(&output))
128.                         {
129.                                 continue;
130.                         }
132.                         // 7. 根据就绪的文件描述符的种类决定如何处理
133.                         for (size_t i = 0; i < output.size(); ++i)
134.                         {
135.                                 if (output[i].GetFd() == listen_sock.GetFd())
136.                                 {
137.                                         // 如果是 listen_sock, 就调用 accept
138.                                         TcpSocket new_sock;
139.                                         listen_sock.Accept(&new_sock);
140.                                         epoll.Add(new_sock, true);
141.                                 }
142.                                 else
143.                                 {
144.                                         // 如果是 new_sock, 就进行一次读写
145.                                         std::string req, resp;
146.                                         bool ret = output[i].RecvNoBlock(&req);
147.                                         if (!ret)
148.                                         {
149.                                                 // [注意!!] 需要把不用的 socket 关闭
150.                                                 // 先后顺序别搞反. 不过在 epoll 删除的时候其实就已经关闭 socket 了
151.                                                 epoll.Del(output[i]);
152.                                                 output[i].Close();
153.                                                 continue;
154.                                         }
156.                                         handler(req, &resp);
157.                                         output[i].SendNoBlock(resp);
158.                                         printf("[client %d] req: %s, resp: %s\n",
159.                                                 output[i].GetFd(),
160.                                                 req.c_str(), resp.c_str());
161.                                 } // end for
162.                         } // end for (;;)
163.                 }
165.                 return true;
166.         }
168. private:
169.         std::string ip_;
170.         uint16_t port_;
171. };

复制代码

---

感谢各位大佬支持！！！

互三啦！！！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。