epoll中et+多執行緒模式中很重要的EPOLL_ONESHOT實驗
阿新 • • 發佈：2019-02-05
因為et模式需要迴圈讀取，但是在讀取過程中，如果有新的事件到達，很可能觸發了其他執行緒來處理這個socket，那就亂了。
EPOLL_ONESHOT就是用來避免這種情況。注意在一個執行緒處理完一個socket的資料，也就是觸發EAGAIN errno時候，就應該重置EPOLL_ONESHOT的flag，這時候，新到的事件，就可以重新進入觸發流程了。
注：EPOLL_ONESHOT的原理其實是，每次觸發事件之後，就將事件註冊從fd上清除了，也就不會再被追蹤到；下次需要用epoll_ctl的EPOLL_CTL_MOD來手動加上才行。
伺服器程式碼如下：
#include <stdio.h> #include<stdlib.h> #include <unistd.h> #include <errno.h> #include <string.h> #include <fcntl.h> #include <assert.h> #include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include <sys/epoll.h> #include <pthread.h> #defineMAX_EVENT_NUMBER 1024 #define BUFFER_SIZE 1024 struct fds { int epollfd; int sockfd; }; int setnonblocking(int fd) { int old_option = fcntl(fd, F_GETFL); int new_option = old_option | O_NONBLOCK; fcntl(fd, F_SETFL, new_option); return old_option; } voidaddfd(int epollfd, int fd, bool oneshot) { epoll_event event; event.data.fd = fd; event.events = EPOLLIN | EPOLLET; if (oneshot) { event.events |= EPOLLONESHOT; } epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event); setnonblocking(fd); } void reset_oneshot(int epollfd, int fd) { epoll_event event; event.data.fd = fd; event.events = EPOLLIN | EPOLLET | EPOLLONESHOT; epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &event); } void* worker(void *arg) { int sockfd = ((fds*)arg)->sockfd; int epollfd = ((fds*)arg)->epollfd; pthread_t pid = pthread_self(); printf("start new thread %u to recv data on fd: %d\n", pid, sockfd); char buf[BUFFER_SIZE]; memset(buf, '\0', BUFFER_SIZE); while(1) { int ret = recv(sockfd, buf, BUFFER_SIZE-1, 0); if (ret == 0) { close(sockfd); printf("foreiner closed the connection\n"); break; } else if (ret < 0) { if (errno == EAGAIN) { reset_oneshot(epollfd, sockfd); printf("EAGAIN read later\n"); break; } } else { buf[ret] = '\0'; printf("thread %u get content: %s\n", pid, buf); printf("thread %u about to sleep\n", pid); sleep(5); printf("thread %u back from sleep\n", pid); } } //printf("end thread %u receiving data on fd: %d\n", pid, sockfd); } int main(int argc, char *argv[]) { if (argc <= 1) { printf("usage: %s port_number [ip_address]\n", basename(argv[0])); return 1; } int port = atoi(argv[1]); int ret = 0; sockaddr_in address; bzero(&address, sizeof(address)); address.sin_family = AF_INET; if (argc >= 3) { const char *ip =argv[2]; inet_pton(AF_INET, ip, &address.sin_addr); } else { address.sin_addr.s_addr = INADDR_ANY; } address.sin_port = htons(port); int listenfd = socket(PF_INET, SOCK_STREAM, 0); assert(listenfd >= 0); ret = bind(listenfd, (sockaddr*)&address, sizeof(address)); assert(ret != -1); ret = listen(listenfd, 5); assert(ret != -1); epoll_event events[MAX_EVENT_NUMBER]; int epollfd = epoll_create(5); assert(epollfd != -1); addfd(epollfd, listenfd, false); while(1) { int ret = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1); if (ret < 0) { printf("epoll failure\n"); break; } for (int i=0; i<ret; i++) { int sockfd = events[i].data.fd; if (sockfd == listenfd) { sockaddr_in client_address; socklen_t client_addrlength = sizeof(client_address); int connfd = accept(listenfd, (sockaddr*)&client_address, &client_addrlength); addfd(epollfd, connfd, true); printf("new connection is added to epollfd\n"); } else if (events[i].events & EPOLLIN) { pthread_t thread; fds fds_for_new_worker; fds_for_new_worker.epollfd = epollfd; fds_for_new_worker.sockfd = sockfd; // new thread pthread_create(&thread, NULL, worker, (void*)&fds_for_new_worker); } else { printf("something else happened\n"); } } } close(listenfd); return 0; }
以下是使用telnet客戶端傳送的文字，勻速敲入程式碼：
$telnet 127.0.0.1 12346 Trying 127.0.0.1... Connected to 127.0.0.1. Escape character is '^]'. hi1 hi2 hi3 hi4 hi5 hi6 hi7 hi8 Connection closed by foreign host.
以下是伺服器的執行和輸出：
$./epoll_oneshot 12346 new connection is added to epollfd start new thread 1734051584 to recv data on fd: 5 thread 1734051584 get content: hi1 thread 1734051584 about to sleep thread 1734051584 back from sleep thread 1734051584 get content: hi2 hi3 thread 1734051584 about to sleep thread 1734051584 back from sleep thread 1734051584 get content: hi4 hi5 hi6 thread 1734051584 about to sleep thread 1734051584 back from sleep thread 1734051584 get content: hi7 thread 1734051584 about to sleep thread 1734051584 back from sleep EAGAIN read later start new thread 1723561728 to recv data on fd: 5 thread 1723561728 get content: hi8 thread 1723561728 about to sleep thread 1723561728 back from sleep EAGAIN read later ^C
最後用Ctrl+C來結束伺服器。
可以看出，在hi7文字和hi8文字之間，伺服器收到了EAGAIN，表示讀取告一段落。而之後的執行緒id也換成了新執行緒id。在hi7之前，因為每次伺服器sleep結束之後，都還有沒有讀完的資料，所以執行緒id始終沒有變，始終是同一個執行緒處理資料。
另外，要注意的是，EPOLL_ONESHOT既可以在et下也可以在lt下設定。效果是一樣的，都是同一個fd上面的相同事件只會觸發一次。上面是et的例子，對於lt，如果設定了EPOLL_ONESHOT，也是需要把資料讀完，然後重置event。而不能像原始lt程式設計方式那樣依賴於事件通知來讀取資料了。