IO多路轉接之epoll

文章目錄

• IO多路轉接之epoll
• • 一、epool
  • 二、基于epoll實現服務器（LT）
  • 三、**基于epoll實現服務器（LT）**

一、epool

是為處理大批量句柄而作了改進的pol

• 1.相關函數

a.

int epoll_create(int size);

創建一個epoll的句柄.自從linux2.6.8之后，size參數是被忽略的.
用完之后, 必須調用close()關閉

官方的講：創建一個epoll的句柄其實呢，它是創建了一個epoll模型：1. 在操作系統內核構建一個紅黑樹節點 ： 表示要關心的哪個文件描述符的事件key鍵 ：用文件描述符作為key鍵2. 在操作系統內核構建一個回調機制 作用：就是減少了操作系統的開銷（不用操作系統再去輪詢的找就緒事件）有這么一個機制告訴我們，我們所關心的文件描述符的時間已經就緒3. 在操作系統內核構建一個就緒隊列如何構建的：有了回調機制，告訴了我們所關心的文件描述符的事件已經就緒接下來就是把該文件描述符拷貝到就緒隊列中；等我們處理的時候就不用輪詢的去找就緒事件，而是 從就緒隊列的開始找epoll_wait（） 的返回值（>0,成功的情況下）這么大的一個區間， 這段區間就是當前的就緒事件這三個組在一起的返回值是fd(文件描述符)

b.

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event) 參數作用

epfd	epoll_create()的返回值(epoll的句柄)
op	表示動作，用三個宏來表示，
fd	是需要監聽的fd
struct epoll_event *event	是告訴內核需要監聽什么事

op的三個宏

• EPOLL_CTL_ADD ：注冊新的fd到epfd中
• EPOLL_CTL_MOD ：修改已經注冊的fd的監聽事件
• EPOLL_CTL_DEL ：從epfd中刪除一個fd

struct epoll_event的結構：

typedef union epoll_data {void *ptr;int fd;uint32_t u32;uint64_t u64;} epoll_data_t;struct epoll_event {uint32_t events; /* Epoll events */epoll_data_t data; /* User data variable */};

events可以是下面幾個宏：

• EPOLLIN : 表示對應的文件描述符可以讀 (包括對端SOCKET正常關閉);
• EPOLLOUT : 表示對應的文件描述符可以寫;
• EPOLLPRI : 表示對應的文件描述符有緊急的數據可讀 (這里應該表示有帶外數據到來);
• EPOLLERR :表示對應的文件描述符發生錯誤;
• EPOLLHUP : 表示對應的文件描述符被掛斷;
• EPOLLET : 將EPOLL設為邊緣觸發(EdgeTriggered)模式, 這是相對于水平觸發(Level Triggered)來說的.
• EPOLLONESHOT：只監聽一次事件,當監聽完這次事件之后, 如果還需要繼續監聽這個socket的話, 需要 再次把這個socket加入到EPOLL隊列里

epoll的事件注冊函數：要關心哪個文件描述符的事件1. 第一個參數是epoll_create()的返回值：一個文件描述符；2.第二個參數表示動作，三個宏來表示：EPOLL_CTL_ADD：注冊新的fd到epfd中；EPOLL_CTL_MOD：修改已經注冊的fd的監聽事件；EPOLL_CTL_DEL：從epfd中刪除一個fd；3. 第三個參數是需要監聽的fd ；4. 第四個參數是告訴內核需要監聽什么事 .events可以是以下幾個宏的集合： EPOLLIN ：表示對應的文件描述符可以讀（包括對端SOCKET正常關閉）； EPOLLOUT：表示對應的文件描述符可以寫； EPOLLPRI：表示對應的文件描述符有緊急的數據可讀（這個應該表示有帶外數據到來）； EPOLLERR：表示對應的文件描述符發?錯誤； EPOLLHUP：表示對應的文件描述符被掛斷； EPOLLET： 將EPOLL設為邊緣觸發(Edge Triggered)模式， 這是相對于水平觸發(LevelTriggered)來說的。 EPOLLONESHOT：只監聽一次事件，當監聽完這次事件之后，如果還需要繼續 監聽這個socket的話，需要再次把這個socket加入到EPOLL隊列中。

c.

int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout); 參數作用

epfd	epoll_create函數的返回值
struct epoll_event *events	結構體，告訴內核需要監聽什么事
maxevents	告之內核這個events有多大，這個 maxevents的值不能大于創建epoll_create()時的size
timeout	是超時時間 (毫秒，0會立即返回，-1是永久阻塞）

返回 ：關心事件已經就緒的事件1. > 0 :滿足就緒條件的事件個數 2. 0 : 在規定的時間內沒有事件發生(超出timeout設置的時間) 3. -1 ：錯誤 原因由errno標識；此時中間三個參數的值變得不可預測。

• 2.epoll原理

當某一進程調用epoll_create方法時，Linux內核會創建一個eventpoll結構體，這個結構體中有兩個成員與epoll的使用方式密切相關

struct eventpoll{..../*紅黑樹的根節點，這顆樹中存儲著所有添加到epoll中的需要監控的事件*/struct rb_root rbr;/*雙鏈表中則存放著將要通過epoll_wait返回給用戶的滿足條件的事件*/struct list_head rdlist;.... };

• 每一個epoll對象都有一個獨立的eventpoll結構體，用于存放通過epoll_ctl方法向epoll對象中添加進來的事件.
• 這些事件都會掛載在紅黑樹中，如此，重復添加的事件就可以通過紅黑樹而高效的識別出來(紅黑樹的插 入時間效率是lgn，其中n為樹的高度).
• 而所有添加到epoll中的事件都會與設備(網卡)驅動程序建立回調關系，也就是說，當響應的事件發生時 會調用這個回調方法.
• 這個回調方法在內核中叫ep_poll_callback,它會將發生的事件添加到rdlist雙鏈表中
• 在epoll中，對于每一個事件，都會建立一個epitem結構體

struct epitem{struct rb_node rbn;//紅黑樹節點struct list_head rdllink;//雙向鏈表節點struct epoll_filefd ffd; //事件句柄信息struct eventpoll *ep; //指向其所屬的eventpoll對象struct epoll_event event; //期待發生的事件類型 }

• 當調用epoll_wait檢查是否有事件發生時，只需要檢查eventpoll對象中的rdlist雙鏈表中是否有epitem元素即可.
• 如果rdlist不為空，則把發生的事件拷貝到用戶態，同時將事件數量返回給用戶. 這個操作的時間復雜度 是O(1）

總結一下, epoll的使用過程就是三部曲:

• 調用epoll_create創建一個epoll句柄;

• 調用epoll_ctl, 將要監控的文件描述符進行注冊;

• 調用epoll_wait, 等待文件描述符就緒

• 3 . epoll的優點(和 select 的缺點對應)

• 接口使用方便: 雖然拆分成了三個函數, 但是反而使用起來更方便高效. 不需要每次循環都設置關注的文件描述符, 也做到了輸入輸出參數分離開

• 數據拷貝輕量: 只在合適的時候調用 EPOLL_CTL_ADD 將文件描述符結構拷貝到內核中, 這個操作并不頻繁(而select/poll都是每次循環都要進行拷貝)

• 事件回調機制: 避免使用遍歷, 而是使用回調函數的方式, 將就緒的文件描述符結構加入到就緒隊列中,epoll_wait 返回直接訪問就緒隊列就知道哪些文件描述符就緒. 這個操作時間復雜度O(1). 即使文件描述符數目很多, 效率也不會受到影響.

• 沒有數量限制: 文件描述符數目無上限

• 4.epoll的工作方式

• 水平觸發(LT)和邊緣觸發(ET)

假如有這樣一個例子:我們已經把一個tcp socket添加到epoll描述符這個時候socket的另一端被寫入了2KB的數據調用epoll_wait，并且它會返回. 說明它已經準備好讀取操作然后調用read, 只讀取了1KB的數據繼續調用epoll_wait......

• 水平觸發Level Triggered 工作模式
  epoll默認狀態下就是LT工作模式.

• 當epoll檢測到socket上事件就緒的時候, 可以不立刻進行處理. 或者只處理一部分.

• 如上面的例子, 由于只讀了1K數據, 緩沖區中還剩1K數據, 在第二次調用 epoll_wait 時, epoll_wait

• 仍然會立刻返回并通知socket讀事件就緒.

• 直到緩沖區上所有的數據都被處理完, epoll_wait 才不會立刻返回.

• 支持阻塞讀寫和非阻塞讀寫

• 邊緣觸發Edge Triggered工作模式

• 如果我們在第1步將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

• 5.理解ET模式和非阻塞文件描述符

使用 ET 模式的 epoll, 需要將文件描述設置為非阻塞. 這個不是接口上的要求, 而是 “工程實踐” 上的要求.

• 假設這樣的場景: 服務器接受到一個10k的請求, 會向客戶端返回一個應答數據. 如果客戶端收不到應答, 不會發送第二個10k請求
• 如果服務端寫的代碼是阻塞式的read, 并且一次只 read 1k 數據的話(read不能保證一次就把所有的數據都讀出來, 參考 man手冊的說明, 可能被信號打斷), 剩下的9k數據就會待在緩沖區中
• 此時由于 epoll 是ET模式, 并不會認為文件描述符讀就緒.
• epoll_wait 就不會再次返回. 剩下的 9k 數據會一直在緩沖區中. 直到下一次客戶端再給服務器寫數據. epoll_wait 才能返回

但是問題來了.

• 服務器只讀到1k個數據, 要10k讀完才會給客戶端返回響應數據.
• 客戶端要讀到服務器的響應, 才會發送下一個請求
• 客戶端發送了下一個請求, epoll_wait 才會返回, 才能去讀緩沖區中剩余的數據

所以, 為了解決上述問題(阻塞read不一定能一下把完整的請求讀完), 于是就可以使用非阻塞輪訓的方式來讀緩沖區,
保證一定能把完整的請求都讀出來

• 6.epoll的優點還有一個內存映射機制,這樣的說法正確嗎

內存映射機制： 內存直接把就緒隊列映射到用戶態，

但是我覺得這種說法是錯誤的。

就緒隊列是操作系統在管理

而操作系統就不會把自己的內部暴露給用戶態，如果暴露出去就不安全，

我們在使用epoll_wait時，會告訴了我們所關心的文件描述符的事件已經就緒，而這個時候有一個回調機制會告訴我們：所關心的哪個文件描述符的事件已經就緒不用操作系統一一去找，減少的操作系統的開銷；

接下來就是把該文件描述符拷貝到就緒隊列中；等我們處理的時候就不用輪詢的去找就緒事件，而是 從就緒隊列的開始找epoll_wait（） 的返回值（>0,成功的情況下）這么大的一個區間，這段區間就是當前的就緒事件

這個過程中，并沒有映射，如果有映射的話，再傳一個緩沖區，豈不是多此一舉

二、基于epoll實現服務器（LT）

#pragma once #include <vector> #include <functional> #include <sys/epoll.h> #include "tcp_socket.hpp"typedef std::function<void (const std::string&, std::string* resp)> Handler;class Epoll {public:Epoll() {epoll_fd_ = epoll_create(10);}~Epoll() {close(epoll_fd_);}bool Add(TcpSocket& sock) const {int fd = sock.GetFd();printf("[Epoll Add] fd = %d\n", fd);epoll_event ev;ev.data.fd = fd;ev.events = EPOLLIN;//設置為ET模式int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev);if (ret < 0) {perror("epoll_ctl ADD");return false;}return true;}bool Del(TcpSocket& sock) const {int fd = sock.GetFd();printf("[Epoll Del] fd = %d\n", fd);int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, NULL);if (ret < 0) {perror("epoll_ctl DEL");return false;}return true;}bool Wait(std::vector<TcpSocket>* output) const {output->clear();epoll_event events[1000];int nfds = epoll_wait(epoll_fd_, events, sizeof(events) / sizeof(events[0]), -1);if (nfds < 0) {perror("epoll_wait");return false;}// [注意!] 此處必須是循環到 nfds, 不能多循環for (int i = 0; i < nfds; ++i) {TcpSocket sock(events[i].data.fd);output->push_back(sock);}return true;}private:int epoll_fd_; };class TcpEpollServer {public:TcpEpollServer(const std::string& ip, uint16_t port) : ip_(ip), port_(port) {}bool Start(Handler handler) {//1. 創建 socketTcpSocket listen_sock;CHECK_RET(listen_sock.Socket());// 2. 綁定CHECK_RET(listen_sock.Bind(ip_, port_));// 3. 監聽CHECK_RET(listen_sock.Listen(5));// 4. 創建 Epoll 對象, 并將 listen_sock 加入進去Epoll epoll;epoll.Add(listen_sock);// 5. 進入事件循環for (;;) {// 6. 進行 epoll_waitstd::vector<TcpSocket> output;if (!epoll.Wait(&output)) {continue;}// 7. 根據就緒的文件描述符的種類決定如何處理for (size_t i = 0; i < output.size(); ++i) {if (output[i].GetFd() == listen_sock.GetFd()) {// 如果是 listen_sock, 就調用 acceptTcpSocket new_sock;listen_sock.Accept(&new_sock);epoll.Add(new_sock);}else {// 如果是 new_sock, 就進行一次讀寫std::string req, resp;bool ret = output[i].Recv(&req);if (!ret) {// [注意!!] 需要把不用的 socket 關閉// 先后順序別搞反. 不過在 epoll 刪除的時候其實就已經關閉 socket 了epoll.Del(output[i]);output[i].Close();continue;}handler(req, &resp);output[i].Send(resp);} // end for} // end for (;;)}return true;}private:std::string ip_;uint16_t port_; };

三、基于epoll實現服務器（LT）

#pragma once #include <vector> #include <functional> #include <sys/epoll.h> #include "tcp_socket.hpp"typedef std::function<void (const std::string&, std::string* resp)> Handler;//如果需要設置為非阻塞方式，需要在tcp_socket.hpp中提供非阻塞方式的recv和send接口 class Epoll {public:Epoll() {epoll_fd_ = epoll_create(10);}~Epoll() {close(epoll_fd_);}bool Add(TcpSocket& sock, bool epoll_et = false) const {int fd = sock.GetFd();printf("[Epoll Add] fd = %d\n", fd);epoll_event ev;ev.data.fd = fd;if (epoll_et)//如果為true，說明要設為非阻塞方式 {ev.events = EPOLLIN | EPOLLET;} else {ev.events = EPOLLIN;}int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev);if (ret < 0) {perror("epoll_ctl ADD");return false;}return true;}bool Del(TcpSocket& sock) const {int fd = sock.GetFd();printf("[Epoll Del] fd = %d\n", fd);int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, NULL);if (ret < 0) {perror("epoll_ctl DEL");return false;}return true;}bool Wait(std::vector<TcpSocket>* output) const {output->clear();epoll_event events[1000];int nfds = epoll_wait(epoll_fd_, events, sizeof(events) / sizeof(events[0]), -1);if (nfds < 0) {perror("epoll_wait");return false;}// [注意!] 此處必須是循環到 nfds, 不能多循環for (int i = 0; i < nfds; ++i) {TcpSocket sock(events[i].data.fd);output->push_back(sock);}return true;}private:int epoll_fd_; };class TcpEpollServer {public:TcpEpollServer(const std::string& ip, uint16_t port) : ip_(ip), port_(port) {}bool Start(Handler handler) {// 1. 創建 socketTcpSocket listen_sock;CHECK_RET(listen_sock.Socket());// 2. 綁定CHECK_RET(listen_sock.Bind(ip_, port_));// 3. 監聽CHECK_RET(listen_sock.Listen(5));// 4. 創建 Epoll 對象, 并將 listen_sock 加入進去Epoll epoll;epoll.Add(listen_sock);// 5. 進入事件循環for (;;) {// 6. 進行 epoll_waitstd::vector<TcpSocket> output;if (!epoll.Wait(&output)) {continue;}// 7. 根據就緒的文件描述符的種類決定如何處理for (size_t i = 0; i < output.size(); ++i) {if (output[i].GetFd() == listen_sock.GetFd()) {// 如果是 listen_sock, 就調用 acceptTcpSocket new_sock;listen_sock.Accept(&new_sock);epoll.Add(new_sock, true);} else {// 如果是 new_sock, 就進行一次讀寫std::string req, resp;bool ret = output[i].RecvNoBlock(&req);if (!ret) {// [注意!!] 需要把不用的 socket 關閉// 先后順序別搞反. 不過在 epoll 刪除的時候其實就已經關閉 socket 了epoll.Del(output[i]);output[i].Close();continue;}handler(req, &resp);output[i].SendNoBlock(resp);printf("[client %d] req: %s, resp: %s\n", output[i].GetFd(),req.c_str(), resp.c_str());} // end for} // end for (;;)}return true;}private:std::string ip_;uint16_t port_; };

總結

以上是生活随笔為你收集整理的IO多路转接之epoll的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。