Linux 高性能服务器编程：I/O复用一-Aet

Bingliaolong 2021-12-11 12:33 Aet | 隐藏边栏 | 抢沙发 | 7 0

文章评分 3 次，平均分 5.0 ：

概述

I/O复用使得程序能同时监听多个文件描述符，这对提高程序的性能很重要。
通常，网络程序在下列情况下需要使用I/O复用技术：
1. 客户端程序要同时处理多个socket。
2. 客户端程序要同时处理用户输入和网络连接。
3. TCP服务器要同时处理监听socket和连接socket。
4. 服务器要同时处理TCP请求和UDP请求。
5. 服务器要同时监听多个端口，或者处理多种服务。

select系统调用

select

select系统调用的用途是，在一段时间内，监听用户感兴趣的文件描述符上的可读、可写或异常等事件。
1. nfds参数指定被监听的文件描述符的总数。它通常被设置为select监听的所有文件描述符中最大值加1，因为文件描述符是从0开始计数的。
2. readfds，writefds和exceptfds参数分别指向可读、可写和异常事件对应的文件描述符集合。
应用程序调用select时，通过这3个参数传入自己感兴趣的文件描述符。
select调用返回时，内核将修改它们来通知应用程序哪些文件描述符已经就绪。

#include <sys/select.h>

int select(int nfds, fd_set* readfds, fd_set* writefds, fd_set* exceptfds, struct timeval* timeout);

#include <typesizes.h>

#define __FD_SETSIZE 1024

#include <sys/select.h>

#define FD_SETSIZE __FD_SETSIZE

typedef long int __fd_mask;

#undef __NFDBITS

#define _NFDBITS (8 * (int)sizeof(___fd_mask))

typedef struct

{

#ifdef __USE_XOPEN

__fd_mask fds_bits[ __FD_SETSIZE / __NFDBITS ];

#define __FDS_BITS(set) ((set)->fds_bits)

#else

__fd_mask fds_bits[ __FD_SETSIZE / __NFDBITS ];

#define __FDS_BITS(set) ((set)->__fds_bits)

#endif

};

可以看到，fd_set结构体仅包含一个整形数组，该数组的每一个元素的每一位标记一个文件描述符。
fd_set能容纳的文件描述符的数量由FD_SETSIZE指定，这就限制了select能同时处理的文件描述符的总量。

#include <sys/select.h>

// 清除fdset的所有位

FD_ZERO(fd_set* fdset);

// 设置fdset的位fd

FD_SET(int fd, fd_set* fdset);

// 清除fdset的位fd

FD_CLR(int fd, fd_set* fdset);

// 测试fdset的位fd是否被设置

int FD_ISSET(int fd, fd_set* fdset);

timeout参数用来设置select函数的超时时间。它是一个timeval结构类型的指针，采用指针参数是因为内核将修改它以告诉应用程序select等待了多久。
不过不能完全信任select调用返回的timeout值，比如调用失败时timeout的值是不确定的。
1. 这两个参数都传0，select将立即返回。
2. 如果给timeout传NULL，select将一直阻塞，直到某个文件描述符就绪。

struct timeout

{

long tv_sec; // 秒数

long tv_usec; // 微秒数

};

select成功时返回就绪（可读、可写和异常）的文件描述符的总数。
如果在超时时间内没有任何文件描述符就绪，select将返回0。
select失败时返回-1并设置errno。
如果在select等待期间，程序接收到信号，则select立即返回-1，并设置errno为EINTR。

文件描述符就绪条件

在网络编程中，下列情况下socket可读：
1. socket内核接收缓冲区的字节数大于或等于其低水位标记SO_RCVLOWAT。
  此时可以无阻塞地读socket，并且读操作返回的字节数大于0。
2. socket通信的对方关闭连接。
  此时对该socket的读操作将返回0。
3. 监听socket上有新的连接请求。
4. socket上有未处理的错误。
  此时可以用getsockopt来读取和清除该错误。
下列情况下socket可写：
1. socket内核发送缓冲区中的可用字节数大于或等于其低水位标记SO_SNDLOWAT。
  此时可以无阻塞的写该socket，并且写操作返回的字节数大于0。
2. socket的写操作被关闭。
  对写操作被关闭的socket执行写操作将触发一个SIGPIPE信号。
3. socket使用非阻塞connect连接成功或失败（超时）之后。
4. socket上有未处理的错误。
  此时可以使用getsockopt来读取和清楚该错误。
网络程序中，select能处理的异常只有一种：
1. socket上接收到带外数据。

处理带外数据

socket上接收到普通数据和带外数据都将使socket返回,但socket处于不同的就绪状态：
1. 接收到普通数据处于可读状态
2. 接收到带外数据处于异常状态

// 同时接收普通数据和带外数据

#include <sys/types.h>

#include <sys/socket.h>

#include <netinet/in.h>

#include <arpa/inet.h>

#include <assert.h>

#include <stdio.h>

#include <unistd.h>

#include <errno.h>

#include <string.h>

#include <fcntl.h>

#include <stdlib.h>

int main(int argc, char* argv[]) {

if (argc <= 2) {

printf("usage:%s ip_address port_number\n", basename(argv[0]));

return 1;

}

const char* ip = argv[1];

int port = atoi(argv[0]);

int ret = 0;

struct sockaddr_in address;

bzero(&address, sizeof(address));

address.sin_family = AF_INET;

inet_pton(AF_INET, ip, &address.sin_addr);

address.sin_port = htons(port);

int listenfd = socket(PF_INET, SOCKET_STREAM, 0);

assert(listenfd > 0);

ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address));

assert(ret != -1);

ret = listen(listenfd, 5);

assert(ret != -1);

struct sockaddr_in client_address;

socklen_t client_addrlength = sizeof(client_address);

int connfd = accept(listenfd, (struct sockaddr*)&client_address, &client_addrlength);

if (connfd < 0) {

printf("errno is %d\n", errno);

close(listenfd);

}

char buf[1024];

fd_set read_fds;

fd_set except_fds;

FD_ZERO(&read_fds);

FD_ZERO(&except_fds);

while(1) {

memset(buf, '\0', sizeof(buf));

// 每次调用select前都需要重新在read_fds和except_fds中设置文件描述符connfd

// 因为事件发生之后,文件描述符集合将被内核修改

FD_SET(connfd, &read_fds);

FD_SET(connfd, &except_fds);

ret = select(connfd+1, &read_fds, NULL, &except_fds, NULL);

if (ret < 0) {

printf("selection failure\n");

break;

}

//对于可读事件,采用普通的recv函数读取数据

if (FD_ISSET(connfd, &read_fds)) {

ret = recv(connfd, buf, sizeof(buf)-1, 0);

if (ret <= 0) {

breadk;

}

printf("get %d bytes of normal data is %s\n", ret, buf);

}

else if (FD_ISSET(connfd, &except_fds)) {

ret = recv(connfd, buf, sizeof(buf)-1, MSG_OOB);

if (ret <= 0) {

break;

}

printf("get %d bytes of oob data %n", ret, buf);

}

close(connfd);

close(listenfd);

return 0;

}

poll系统调用

poll系统调用和select系统调用类似,也是在指定时间内沦陷一定数量的文件描述符,以测试其中是否有就绪者。
fd成员指定文件描述符
events成员告诉poll监听fd上的哪些事件，它是一系列事件的按位或
revents成员则由内核修改，以通知应用程序fd上实际上发生了哪些事件（下图）。
nfds参数指定被监听事件集合fds的大小
timeout参数指定poll的超时值，单位是毫秒。当timeout为-1时，poll调用将永远阻塞，直到某个事件发生；当timeout为0时，poll调用将立即返回。
poll系统调用的返回值的含义与select相同。

#include <poll.h>

struct pollfd {

int fd; // 文件描述符

short events; // 注册的事件

short revents; // 实际发生的事件，由内核填充

};

1	typedef unsigned long int nfds_t;

上表中：POLLRDNORM，POLLRDBAND，POLLWRNORM，POLLWRBAND由XOPEN规范定义。
它们实际上是将POLLIN事件和POLLOUT事件分得更细，以区别对待普通数据和优先数据。但Linux并不完全支持它们。
通常，应用程序需要recv调用的返回值来区分接收到的是有效数据还是对方关闭连接的请求，并做相应的处理。
不过，自Linux2.6.17开始，GNU为poll系统调用增加了一个POLLRDHUP事件，它在socket上接收到对方关闭连接的请求之后触发。这为我们区分上述两种情况提供了一种更简单的方式。但使用POLLRDHUP事件时，我们需要在代码最开始处定义_GNU_SOURCE。

epoll系统调用

内核事件表

epoll是Linux特有的I/O复用函数。
它在实现和使用上和select，poll有很大差异。
1. 首先，epoll使用一组函数来完成任务，而不是单个函数。
2. 其次，epoll把用户关心的文件描述符上的事件放在内核里的一个事件表中，从而无须像select和epoll那样每次调用都要重复传入文件描述符集或事件集。
3. 不同的是，epoll需要一个额外的文件描述符，来唯一标识内核中的这个事件表（使用下面的epoll_create来创建这个额外的文件描述符）。
epoll_create
1. size参数并不起作用，只是给内核一个提示，告诉它事件表需要多大。
2. 该函数返回的文件描述符将用作其他所有epoll系统调用的第一个参数，以指定要访问的内核事件表。

#include <sys/epoll.h>

int epoll_create(int size);

操作内核事件表的函数：
1. fd是要操作的文件描述符
2. op是指定操作类型
  EPOLL_CTL_ADD：往事件表中注册fd上的事件
  EPOLL_CTL_MOD：修改fd上的注册事件
  EPOLL_CTL_DEL：删除fd上的注册事件
3. 结构体events成员描述事件类型。epoll支持的事件类型和poll基本相同。但epoll有两个额外的事件类型：EPOLLET和EPOLLONESHOT。它们对epoll的高效运作非常关键。
4. epoll_data_t联合体，4个成员中使用最多的是fd，它指定事件所从属的目标文件描述符。
  ptr成员可用来指定与fd相关的用户数据。
  由于epoll_data_t是一个联合体，所以不能同时使用fd和ptr，如果要将文件描述符和用户数据关联起来，以实现快速的数据访问，只能使用其他手段，比如放弃使用epoll_data_t的fd成员，而在ptr指向的用户数据中包含fd。
5. epoll_ctl成功时返回0，失败则返回-1并设置errno。

#include <sys/epoll.h>

typedef union epoll_data {

void* ptr;

int fd;

uint32_t u32;

uint64_t u64;

} epoll_data_t;

struct epoll_event

{

__uint32_t events; // epoll事件

epoll_data_t data; // 用户数据

};

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

epoll_wait

epoll_wait系统调用的主要接口是epoll_wait函数。
它在一段超时时间内等待一组文件描述符上的事件。
调用成功时返回就绪的文件描述符的个数，失败时返回-1并设置errno。

#include <sys/epoll.h>

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

epoll_wait如果检测到事件，就将所有就绪事件从内核事件表（由epfd指定）中复制到它的第二个参数events指向的数据中。
这个数据只用于输出epoll_wait检测到的就绪事件，而不像select和poll的数组参数那样既用于传入用户注册的事件，又用于输出内核检测到的就绪事件。
这就极大地提高了应用程序索引就绪文件描述符的效率。

// 如何索引poll返回的就绪文件描述符

int ret = poll(fds, MAX_EVENT_NUMBER, -1);

// 必须遍历所有已注册文件描述符并找到其中的就绪者

for(int i = 0; i < MAX_EVENT_NUMBER; ++i) {

// 判断第i个文件描述符是否就绪

if (fds[i].revents & POLLIN) {

int sockfd = fds[i].fd;

// 处理sockfd

}

// 如何索引epoll返回的就绪文件描述符

int ret = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);

// 仅遍历就绪的ret个文件描述符

for(int i = 0; i < ret; i++) {

int sockfd = events[i].data.fd;

// sockfd肯定就绪，可以之间处理

}

LT和ET

epoll对文件描述符的操作有两种模式：
1. LT：水平触发
2. ET：边沿触发
LT模式是默认的工作模式，这种模式下的epoll相当于一个效率价高的poll。
当往epoll内核事件表中注册一个文件描述符上的EPOLLET事件时，epoll将以ET模式来操作该文件描述符。
ET模式是epoll的高效工作模式。
对于采用LT工作模式的文件描述符，当epoll_wait检测到其上有事件发生并将此事件通知应用程序后，应用程序可以不立即处理该事件。
这样，当应用程序下一次调用epoll_wait时，epoll_wait还会再次向应用程序通告此事件，直到该事件被处理。
而对于采用ET工作模式的文件描述符，当epoll_wait检测到其上有事件发生并将此事件通知应用程序后，应用程序必须立即注册该事件，因为后续的epoll_wait调用将不再向应用程序通知这一事件。
可见，ET模式在很大程度上降低了同一个epoll事件被重复触发的次数，因此效率要比LT效率高。

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

#include <sys/types.h>

#include <sys/socket.h>

#include <netinet/in.h>

#include <arpa/inet.h>

#include <assert.h>

#include <stdio.h>

#include <unistd.h>

#include <errno.h>

#include <string.h>

#include <fcntl.h>

#include <stdlib.h>

#include <sys/epoll.h>

#include <pthread.h>

#define MAX_EVENT_MUBER 1024

#define BUFFER_SIZE 10

// 将文件描述符设置成非阻塞的

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;

}

// 将文件描述符fd上的EPOLLIN注册到epollfd指示的epoll内核事件表中

// 参数enable_et指定是否对fd启用ET模式

void addfd(int epollfd, int fd, bool enable_et)

{

epoll_event event;

event.data.fd = fd;

event.events = EPOLLIN;

if (enable_et) {

event.events |= EPOLLET;

}

epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);

setnonblocking(fd);

}

// LT 模式的工作流程

void lt(epoll_event* events, int number, int epollfd, int listenfd) {

char buf[BUFFER_SIZE];

for(int i = 0; i < number; i++) {

int sockfd = events[i].data.fd;

if (sockfd == listenfd) {

struct sockaddr_in client_address;

socklen_t client_addrlength = sizeof(client_address);

int connfd = accept(listenfd, (struct sockaddr*)&client_address,

&client_addrlength);

// 对connfd禁用ET模式

addfd(epollfd, connfd, false);

} else if(events[i].events & EPOLLIN) {

// 只要socket读缓冲区中还有未读的数据，这段代码就会被触发

printf("event rigger once\n");

memset(buf, '\0', BUFFER_SIZE);

int ret = recv(sockfd, buf, BUFFER_SIZE-1, 0);

if(ret <= 0) {

close(sockfd);

continue;

}

printf("get %d bytes of conteng %s\n", ret, buf);

} else {

printf("something else happened\n");

}

// ET模式的工作流程

void et(epoll_event* events, int number, int epollfd, int listenfd) {

char buf[BUFFER_SIZE];

for(int i = 0; i < number; i++) {

int sockfd = events[i].data.fd;

if (sockfd == listenfd) {

struct sockaddr_in client_address;

socklen_t client_addrlength = sizeof(client_address);

int connfd = accept(listenfd, (struct sockaddr*)&client_address,

&client_addrlength);

// 对connfd开始ET模式

addfd(epollfd, connfd, true);

} else if (events[i].events & EPOLLIN) {

// 这段代码不会被重复触发，所以我们循环读取数据，以确保把socket读缓冲区中的所有数据都读出

printf("event trigger once\n");

while(true) {

memset(buf, '\0', BUFFER_SIZE);

int ret = recv(sockfd, buf, BUFFER_SIZE-1, 0);

if (ret < 0) {

// 对于非阻塞I/O，下面的条件成立表示数据已经全部读取完毕

// 此后，epoll就能再次触发sockfd上EPOLLIN事件，以驱动下一次读操作

if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) {

printf("read later\n");

break;

}

close(sockfd);

break;

} else if (ret == 0) {

close(sockfd);

} else {

printf("get %d bytes of content %s\n", ret, buf);

}

} else {

printf("something else happened\n");

}

int main(int argc, char* argv[]) {

if (argc <= 2) {

printf("usage: %s ip_address port_number\n", basename(argv[0]));

return 1;

}

const char* ip = argv[1];

int port = atoi(argv[2]);

int ret = 0;

struct sockaddr_in address;

bzero(&address, sizeof(address));

address.sin_family = AF_INET;

inet_pton(AF_INET, ip, &address.sin_addr);

address.sin_port = htons(port);

int listenfd = socket(PF_INET, SOCK_STREAM, 0);

assert(listenfd >= 0);

ret = bind(listenfd, (struct 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, true);

while(true) {

int ret = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);

if (ret < 0) {

printf("epoll failure\n");

break;

}

// LT模式

lt(events, ret, epollfd, listenfd);

// ET模式

// et(events, ret, epollfd, listenfd);

}

close(listenfd);

return 0;

}

EPOLLONESHOT事件

即使使用ET模式，一个socket上的某个事件还是可能被触发多次。
这在并发程序中可能会引起一个问题。比如一个线程（或进程）在读取完某个socket上的数据后开始处理这些数据，而在数据处理过程中该socket上又有新的数据可读（EPOLLIN再次被触发），此时另外一个线程（或进程）被唤醒来读取这些新的数据。
于是，就出现了两个线程同时操作一个socket的局面。然而，这并不是我们期望的。我们期望的是，一个socket连接在任一时刻都只能被一个线程处理。
这一点，可以使用epoll的EPOLLONESHOT事件实现。
对于注册了EPOLLONESHOT事件的文件描述符，操作系统最多触发其上注册的一个可读、可写或者异常事件，且只触发一次，触发我们使用epoll_ctl函数重置该文件描述符上注册的EPOLLONESHOT事件。
这样，对于当一个线程在处理某个socket时，其他线程是不可能有机会操作该socket的。
反过来思考，注册了EPOLLONESHOT事件的socket一旦被某个线程处理完毕，该线程就应该立即重置这个socket上的EPOLLONESHOT事件，以确保这个socket下一次可读时，其EPOLLIN事件能被触发，进而让其他工作线程有机会继续处理这个socket。

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

#include <sys/types.h>

#include <sys/socket.h>

#include <netinet/in.h>

#include <arpa/inet.h>

#include <assert.h>

#include <stdio.h>

#include <unistd.h>

#include <errno.h>

#include <string.h>

#include <fcntl.h>

#include <stdlib.h>

#include <sys/epoll.h>

#Include <pthread.h>

#define MAX_EVENT_NUMBER 1024

#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;

}

// 将fd上的EPOLLIN和EPOLLET事件注册到epollfd指示的epoll内核事件表中

// 参数oneshot指定是否注册fd上的EPOLLONESHOT事件

void addfd(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);

}

// 重置fd上的事件，这样操作后，尽管fd上的EPOLLONESHOT事件被注册，但是操作系统仍然会触发fd上的EPOLLIN事件，且只触发一次

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;

printf("start new thread to receive data on fd: %d\n", sockfd);

char buf[BUFFER_SIZE];

memset(buf, '\0', BUFFER_SIZE);

// 循环读取socket上的数据，直到遇到了EAGAIN错误

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("read later\n");

break;

}

} else {

printf("get content: %s\n", buf);

// 休眠5s,模拟数据处理

sleep(5);

}

printf("end thread receiving data on fd %d\n", sockfd);

}

int main(int argc, char* argv[]) {

if (argc <= 2) {

printf("usage %s ip_address port_number\n", basename(argv[0]));

return 1;

}

const char* ip = argv[1];

int port =- atoi(argv[2]);

int ret = 0;

struct sockaddr_in address;

bzero(&address, sizeof(address));

address.sin_family = AF_INET;

inet_pton(AF_INET, ip, &address.sin_addr);

address.sin_port = htons(port);

int listenfd = socket(PF_INET, SOCK_STREAM, 0);

assert(listenfd >= 0);

ret = bind(listenfd, (struct 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);

// 监听socket listenfd上是不能注册EPOLLONESHOT事件的，否则应用程序只能处理一个客户连接

// 因为后续的客户连接请求将不再触发listenfd上的EPOLLIN事件

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) {

struct sockaddr_in client_address;

socklen_t client_addrlength = sizeof(client_address);

int connfd = accept(listenfd, (struct sockaddr*)&client_address,

&client_addrlength);

// 对每个非监听文件描述符都注册EPOLLONESHOT事件

addfd(epollfd, connfd, true);

} 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;

// 启动一个新的工作线程为socket服务

pthread_create(&thread, NULL, worker, (void*)&fds_for_new_worker);

} else {

printf("something else happened\n");

}

clsoe(listenfd);

return 0;

}

概述
select系统调用
select
文件描述符就绪条件
处理带外数据
poll系统调用
epoll系统调用
内核事件表
epoll_wait
LT和ET
EPOLLONESHOT事件

epoll et io linux lt poll select

本文为原创文章，版权归Aet所有，欢迎分享本文，转载请保留出处！

Bingliaolong 关注：0 粉丝：0

Everything will be better.

概述

select系统调用

select

文件描述符就绪条件

处理带外数据

poll系统调用

epoll系统调用

内核事件表

epoll_wait

LT和ET

EPOLLONESHOT事件

你可能也喜欢

热评文章

发表评论

标签云集

2025 年 4 月
一	二	三	四	五	六	日
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