[TOC] ## 1、流？I/O操作？阻塞？epoll? ### 一、流？I/O操作? 阻塞？ #### (1) 流 - 可以進行I/O操作的內核對象 - 文件、管道、套接字…… - 流的入口：文件描述符(fd) #### (2) I/O操作 所有對流的讀寫操作，我們都可以稱之為IO操作。 當一個流中， 在沒有數據read的時候，或者說在流中已經寫滿了數據，再write，我們的IO操作就會出現一種現象，就是阻塞現象，如下圖。   --- #### (3) 阻塞  ? **阻塞場景**: 你有一份快遞，家里有個座機，快遞到了主動給你打電話，期間你可以休息。  **非阻塞，忙輪詢場景**: 你性子比較急躁， 每分鐘就要打電話詢問快遞小哥一次， 到底有沒有到，快遞員接你電話要停止運輸，這樣很耽誤快遞小哥的運輸速度。 * 阻塞等待 空出大腦可以安心睡覺, 不影響快遞員工作（不占用CPU寶貴的時間片）。 - 非阻塞，忙輪詢 浪費時間，浪費電話費，占用快遞員時間（占用CPU，系統資源）。 很明顯，阻塞等待這種方式，對于通信上是有明顯優勢的， 那么它有哪些弊端呢？ ### 二、解決阻塞死等待的辦法 #### 阻塞死等待的缺點  ? 也就是同一時刻，你只能被動的處理一個快遞員的簽收業務，其他快遞員打電話打不進來，只能干瞪眼等待。那么解決這個問題，家里多買N個座機， 但是依然是你一個人接，也處理不過來，需要用影分身術創建都個自己來接電話(采用多線程或者多進程）來處理。 ? 這種方式就是沒有多路IO復用的情況的解決方案， 但是在單線程計算機時代(無法影分身)，這簡直是災難。 --- 那么如果我們不借助影分身的方式(多線程/多進程)，該如何解決阻塞死等待的方法呢？ #### 辦法一：非阻塞、忙輪詢  ```go while true { for i in 流[] { if i has 數據 { 讀 或者 其他處理 } } } ``` 非阻塞忙輪詢的方式，可以讓用戶分別與每個快遞員取得聯系，宏觀上來看，是同時可以與多個快遞員溝通(并發效果)、 但是快遞員在于用戶溝通時耽誤前進的速度(浪費CPU)。 --- #### 辦法二：select  我們可以開設一個代收網點，讓快遞員全部送到代收點。這個網店管理員叫select。這樣我們就可以在家休息了，麻煩的事交給select就好了。當有快遞的時候，select負責給我們打電話，期間在家休息睡覺就好了。 但select 代收員比較懶，她記不住快遞員的單號，還有快遞貨物的數量。她只會告訴你快遞到了，但是是誰到的，你需要挨個快遞員問一遍。 ```go while true { select(流[]); //阻塞 //有消息抵達 for i in 流[] { if i has 數據 { 讀 或者 其他處理 } } } ``` --- #### 辦法三：epoll  epoll的服務態度要比select好很多，在通知我們的時候，不僅告訴我們有幾個快遞到了，還分別告訴我們是誰誰誰。我們只需要按照epoll給的答復，來詢問快遞員取快遞即可。 ```go while true { 可處理的流[] = epoll_wait(epoll_fd); //阻塞 //有消息抵達，全部放在 “可處理的流[]”中 for i in 可處理的流[] { 讀 或者 其他處理 } } ``` --- ### 三、epoll？ - 與select，poll一樣，對I/O多路復用的技術 - 只關心“活躍”的鏈接，無需遍歷全部描述符集合 - 能夠處理大量的鏈接請求(系統可以打開的文件數目) ### 四、epoll的API #### (1) 創建EPOLL ```c /** * @param size 告訴內核監聽的數目 * * @returns 返回一個epoll句柄（即一個文件描述符） */ int epoll_create(int size); ``` 使用 ```c int epfd = epoll_create(1000); ```  創建一個epoll句柄，實際上是在內核空間，建立一個root根節點，這個根節點的關系與epfd相對應。 #### (2) 控制EPOLL ```c /** * @param epfd 用epoll_create所創建的epoll句柄 * @param op 表示對epoll監控描述符控制的動作 * * EPOLL_CTL_ADD(注冊新的fd到epfd) * EPOLL_CTL_MOD(修改已經注冊的fd的監聽事件) * EPOLL_CTL_DEL(epfd刪除一個fd) * * @param fd 需要監聽的文件描述符 * @param event 告訴內核需要監聽的事件 * * @returns 成功返回0，失敗返回-1, errno查看錯誤信息 */ int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event); struct epoll_event { __uint32_t events; /* epoll 事件 */ epoll_data_t data; /* 用戶傳遞的數據 */ } /* * events : {EPOLLIN, EPOLLOUT, EPOLLPRI, EPOLLHUP, EPOLLET, EPOLLONESHOT} */ typedef union epoll_data { void *ptr; int fd; uint32_t u32; uint64_t u64; } epoll_data_t; ``` 使用 ```c struct epoll_event new_event; new_event.events = EPOLLIN | EPOLLOUT; new_event.data.fd = 5; epoll_ctl(epfd, EPOLL_CTL_ADD, 5, &new_event); ``` ? 創建一個用戶態的事件，綁定到某個fd上，然后添加到內核中的epoll紅黑樹中。  #### (3) 等待EPOLL ```c /** * * @param epfd 用epoll_create所創建的epoll句柄 * @param event 從內核得到的事件集合 * @param maxevents 告知內核這個events有多大, * 注意: 值 不能大于創建epoll_create()時的size. * @param timeout 超時時間 * -1: 永久阻塞 * 0: 立即返回，非阻塞 * >0: 指定微秒 * * @returns 成功: 有多少文件描述符就緒,時間到時返回0 * 失敗: -1, errno 查看錯誤 */ int epoll_wait(int epfd, struct epoll_event *event, int maxevents, int timeout); ``` 使用 ```c struct epoll_event my_event[1000]; int event_cnt = epoll_wait(epfd, my_event, 1000, -1); ``` ? `epoll_wait`是一個阻塞的狀態，如果內核檢測到IO的讀寫響應，會拋給上層的epoll_wait, 返回給用戶態一個已經觸發的事件隊列，同時阻塞返回。開發者可以從隊列中取出事件來處理，其中事件里就有綁定的對應fd是哪個(之前添加epoll事件的時候已經綁定)。  #### (4) 使用epoll編程主流程骨架 ```c int epfd = epoll_crete(1000); //將 listen_fd 添加進 epoll 中 epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd,&listen_event); while (1) { //阻塞等待 epoll 中 的fd 觸發 int active_cnt = epoll_wait(epfd, events, 1000, -1); for (i = 0 ; i < active_cnt; i++) { if (evnets[i].data.fd == listen_fd) { //accept. 并且將新accept 的fd 加進epoll中. } else if (events[i].events & EPOLLIN) { //對此fd 進行讀操作 } else if (events[i].events & EPOLLOUT) { //對此fd 進行寫操作 } } } ``` ### 五、epoll的觸發模式 #### (1) 水平觸發   水平觸發的主要特點是，如果用戶在監聽`epoll`事件，當內核有事件的時候，會拷貝給用戶態事件，但是**如果用戶只處理了一次，那么剩下沒有處理的會在下一次epoll_wait再次返回該事件**。 這樣如果用戶永遠不處理這個事件，就導致每次都會有該事件從內核到用戶的拷貝，耗費性能，但是水平觸發相對安全，最起碼事件不會丟掉，除非用戶處理完畢。 ##### (2) 邊緣觸發   邊緣觸發，相對跟水平觸發相反，當內核有事件到達， 只會通知用戶一次，至于用戶處理還是不處理，以后將不會再通知。這樣減少了拷貝過程，增加了性能，但是相對來說，如果用戶馬虎忘記處理，將會產生事件丟的情況。 ### 六、簡單的epoll服務器(C語言) #### (1) 服務端 ```c #include <stdio.h> #include <stdlib.h> #include <ctype.h> #include <string.h> #include <unistd.h> #include <sys/types.h> #include <sys/socket.h> #include <arpa/inet.h> #include <sys/epoll.h> #define SERVER_PORT (7778) #define EPOLL_MAX_NUM (2048) #define BUFFER_MAX_LEN (4096) char buffer[BUFFER_MAX_LEN]; void str_toupper(char *str) { int i; for (i = 0; i < strlen(str); i ++) { str[i] = toupper(str[i]); } } int main(int argc, char **argv) { int listen_fd = 0; int client_fd = 0; struct sockaddr_in server_addr; struct sockaddr_in client_addr; socklen_t client_len; int epfd = 0; struct epoll_event event, *my_events; / socket listen_fd = socket(AF_INET, SOCK_STREAM, 0); // bind server_addr.sin_family = AF_INET; server_addr.sin_addr.s_addr = htonl(INADDR_ANY); server_addr.sin_port = htons(SERVER_PORT); bind(listen_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)); // listen listen(listen_fd, 10); // epoll create epfd = epoll_create(EPOLL_MAX_NUM); if (epfd < 0) { perror("epoll create"); goto END; } // listen_fd -> epoll event.events = EPOLLIN; event.data.fd = listen_fd; if (epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &event) < 0) { perror("epoll ctl add listen_fd "); goto END; } my_events = malloc(sizeof(struct epoll_event) * EPOLL_MAX_NUM); while (1) { // epoll wait int active_fds_cnt = epoll_wait(epfd, my_events, EPOLL_MAX_NUM, -1); int i = 0; for (i = 0; i < active_fds_cnt; i++) { // if fd == listen_fd if (my_events[i].data.fd == listen_fd) { //accept client_fd = accept(listen_fd, (struct sockaddr*)&client_addr, &client_len); if (client_fd < 0) { perror("accept"); continue; } char ip[20]; printf("new connection[%s:%d]\n", inet_ntop(AF_INET, &client_addr.sin_addr, ip, sizeof(ip)), ntohs(client_addr.sin_port)); event.events = EPOLLIN | EPOLLET; event.data.fd = client_fd; epoll_ctl(epfd, EPOLL_CTL_ADD, client_fd, &event); } else if (my_events[i].events & EPOLLIN) { printf("EPOLLIN\n"); client_fd = my_events[i].data.fd; // do read buffer[0] = '\0'; int n = read(client_fd, buffer, 5); if (n < 0) { perror("read"); continue; } else if (n == 0) { epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, &event); close(client_fd); } else { printf("[read]: %s\n", buffer); buffer[n] = '\0'; #if 1 str_toupper(buffer); write(client_fd, buffer, strlen(buffer)); printf("[write]: %s\n", buffer); memset(buffer, 0, BUFFER_MAX_LEN); #endif /* event.events = EPOLLOUT; event.data.fd = client_fd; epoll_ctl(epfd, EPOLL_CTL_MOD, client_fd, &event); */ } } else if (my_events[i].events & EPOLLOUT) { printf("EPOLLOUT\n"); /* client_fd = my_events[i].data.fd; str_toupper(buffer); write(client_fd, buffer, strlen(buffer)); printf("[write]: %s\n", buffer); memset(buffer, 0, BUFFER_MAX_LEN); event.events = EPOLLIN; event.data.fd = client_fd; epoll_ctl(epfd, EPOLL_CTL_MOD, client_fd, &event); */ } } } END: close(epfd); close(listen_fd); return 0; } ``` #### (2) 客戶端 ```c #include <stdio.h> #include <stdlib.h> #include <string.h> #include <strings.h> #include <sys/types.h> #include <sys/socket.h> #include <arpa/inet.h> #include <unistd.h> #include <fcntl.h> #define MAX_LINE (1024) #define SERVER_PORT (7778) void setnoblocking(int fd) { int opts = 0; opts = fcntl(fd, F_GETFL); opts = opts | O_NONBLOCK; fcntl(fd, F_SETFL); } int main(int argc, char **argv) { int sockfd; char recvline[MAX_LINE + 1] = {0}; struct sockaddr_in server_addr; if (argc != 2) { fprintf(stderr, "usage ./client <SERVER_IP>\n"); exit(0); } // 創建socket if ( (sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) { fprintf(stderr, "socket error"); exit(0); } // server addr 賦值 bzero(&server_addr, sizeof(server_addr)); server_addr.sin_family = AF_INET; server_addr.sin_port = htons(SERVER_PORT); if (inet_pton(AF_INET, argv[1], &server_addr.sin_addr) <= 0) { fprintf(stderr, "inet_pton error for %s", argv[1]); exit(0); } // 鏈接服務端 if (connect(sockfd, (struct sockaddr*) &server_addr, sizeof(server_addr)) < 0) { perror("connect"); fprintf(stderr, "connect error\n"); exit(0); } setnoblocking(sockfd); char input[100]; int n = 0; int count = 0; // 不斷的從標準輸入字符串 while (fgets(input, 100, stdin) != NULL) { printf("[send] %s\n", input); n = 0; // 把輸入的字符串發送 到 服務器中去 n = send(sockfd, input, strlen(input), 0); if (n < 0) { perror("send"); } n = 0; count = 0; // 讀取 服務器返回的數據 while (1) { n = read(sockfd, recvline + count, MAX_LINE); if (n == MAX_LINE) { count += n; continue; } else if (n < 0){ perror("recv"); break; } else { count += n; recvline[count] = '\0'; printf("[recv] %s\n", recvline); break; } } } return 0; } ```