【网络编程开发】11.IO模型 12.IO多路复用

11.IO模型

什么是IO:

IO 是 Input/Output 的缩写，指的是输入和输出。在计算机当中，IO 操作通常指将数据从一个设备或文件中读取到计算机内存中，或将内存中的数据写入设备或文件中。这些设备可以包括硬盘驱动器、网卡、键盘、屏幕等。

通常用户进程中的一个完整I/O分为两个阶段：
用户进程空间→内核空间
内核空间→设备空间
I/O分为内存I/O、网络I/O和磁盘I/O三种

IO操作的两个阶段

Linux中进程无法直接操作I/O设备，其必须通过系统调用请求内核来协助完成I/O操作。
内核会为每个I/O设备维护一个缓冲区。
对于一个输入操作来说，进程I/O系统调用后，内核会先看缓冲区中有没有相应的缓存数据，没有的话再到设备（比如网卡设备）中读取（因为设备I/O一般速度较慢，需要等待）；
内核缓冲区有数据则直接复制到用户进程空间。
所以，对于一个网络输入操作通常包括两个不同阶段：

1. 等待网络数据到达网卡，把数据从网卡读取到内核缓冲区，准备好数据。
2. 从内核缓冲区复制数据到用户进程空间。

网络I/O的本质是对socket的读取，socket在Linux系统中被抽象为流，I/O可以理解为对流的操作。
网络I/O的模型可分为两种：

• 异步I/O(asynchronous I/O)
• 同步I/O(synchronous I/O)

同步I/O又包括

• 阻塞I/O(blocking I/O)
• 非阻塞I/O(non-blocking I/O)
• 多路复用I/O(multiplexing I/O)
• 信号驱动I/O(signal-driven I/O)

强调一下：信号驱动I/O属于同步I/O，原因往后看。
信号驱动I/O和异步I/O只作概念性的讲解，不作为学习重点。

五种I/O模型

阻塞I/O(blocking I/O)

对于一个套接字上的输入操作，第一步通常涉及等待数据从网络中到达，当所有等待分组到达时，它被复制到内核中的某个缓冲区。第二步是把数据从内核缓冲区复制到应用程序缓冲区。
同步阻塞I/O模型是最常用、最简单的模型。在Linux中，默认情况下，所有套接字都是阻塞的。下面我们以阻塞套接字的recvfrom的调用图来说明阻塞，如图所示

非阻塞I/O(non-blocking I/O)

非阻塞的recvform系统调用之后，进程并没有被阻塞，内核马上返回给进程，如果数据还没准备好，此时会返回一个error（EAGAIN或EWOULDBLOCK）。
进程在返回之后，可以先处理其他的业务逻辑，稍后再发起recvform系统调用。
采用轮询的方式检查内核数据，直到数据准备好。再拷贝数据到进程，进行数据处理。
在Linux下，可以通过设置套接字选项使其变为非阻塞。非阻塞的套接字的recvfrom操作如图所示

可以看到前三次调用recvfrom请求时，并没有数据返回，内核返回errno(EWOULDBLOCK)，并不会阻塞进程。
当第四次调用recvfrom时，数据已经准备好了，于是将它从内核空间拷贝到程序空间，处理数据。
在非阻塞状态下，I/O执行的等待阶段并不是完全阻塞的，但是第二个阶段依然处于一个阻塞状态（调用者将数据从内核拷贝到用户空间，这个阶段阻塞)。

多路复用I/O(multiplexing I/O)

I/O多路复用的好处在于单个进程就可以同时处理多个网络连接的I/O。它的基本原理是不再由应用程序自己监视连接，而由内核替应用程序监视文件描述符。
以select函数为例，当用户进程调用了select，那么整个进程会被阻塞，而同时，kernel会“监视”所有select负责的socket，当任何一个socket中的数据准备好，select就会返回。
这个时候用户进程再调用read操作，将数据从内核拷贝到用户进程，如下图所示。

信号驱动I/O(signal-driven I/O)

该模型允许socket进行信号驱动I/O，并注册一个信号处理函数，进程继续运行并不阻塞。当数据准备好时，进程会收到一个SIGIO信号，可以在信号处理函数中调用I/O操作函数处理数据，如图所示

注意：虽然信号驱动IO在注册完信号处理函数以后，就可以做其他事情了。但是第二阶段拷贝数据的过程当中进程依然是被阻塞的，而后要介绍的异步IO是完全不会阻塞进程的，所以信号驱动虽然具有异步的特点，但依然属于异步IO

异步I/O(asynchronous I/O)

相对于同步I/O，异步I/O不是按顺序执行。用户进程进行aio_read系统调用之后，就可以去处理其他逻辑了，无论内核数据是否准备好，都会直接返回给用户进程，不会对进程造成阻塞。这是因为aio_read只向内核递交申请，并不关心有没有数据。
等到数据准备好了，内核直接复制数据到进程空间，然后内核向进程发送通知，此时数据已经在用户空间了，可以对数据进行处理。

五种I/O模型比较

前四种I/O模型都是同步I/O操作，它们的区别在于第一阶段，而第二阶段是一样的：在数据从内核复制到应用缓冲区期间（用户空间)，进程阻塞于recvfrom调用。
相反，异步I/O模型在等待数据和接收数据的这两个阶段都是非阻塞的，可以处理其他的逻辑，用户进程将整个I/O操作交由内核完成，内核完成后会发送通知。在此期间，用户进程不需要检查I/O操作的状态，也不需要主动拷贝数据。
在了解了Linux的I/O模型之后，我们就可以进行服务器设计了。

12.IO多路复用

I/O多路复用的好处在于单个进程就可以同时处理多个网络连接的I/O。它的基本原理是不再由应用程序自己监视连接，而由内核替应用程序监视文件描述符。

在Linux系统中，select、poll和epoll是三种常用的I/O多路复用技术，它们用于处理多个I/O流，以实现高效的并发服务器设计。

特性	select函数	poll函数	epoll函数族
支持平台	几乎所有类Unix系统	Unix及类Unix系统	主要为Linux系统
数据结构	位图（限制文件描述符数量）	链表（无文件描述符数量限制）	红黑树（高效管理事件）
文件描述符限制	通常最多1024个	无限制	无限制
拷贝开销	每次调用时拷贝整个集合	每次调用时拷贝整个集合	通过回调机制，避免拷贝
返回就绪描述符	需要遍历所有描述符识别就绪状态	直接返回就绪状态的描述符	高效返回只包含就绪事件的描述符
并发性能	低至中等（因位图扫描）	中等（因链表扫描）	高（高效的事件通知和数据结构）
触发模式	不支持	不支持	支持水平触发和边缘触发
API复杂度	简单直观	类似select但更灵活	功能丰富但使用相对复杂
适用场景	小规模并发服务器或客户端	中规模并发服务器	大规模并发服务器，尤其是网络服务

select函数

1. 原型：

   #include <sys/select.h>
   int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);
   

2. 功能：监视文件描述符集合，等待其中任意一个文件描述符准备好进行I/O操作。

3. 参数：

   • nfds：文件描述符集合中最大的文件描述符值加1。
   • readfds：需要监视可读状态的文件描述符集合。
   • writefds：需要监视可写状态的文件描述符集合。
   • exceptfds：需要监视异常状态的文件描述符集合。
   • timeout：设置select函数的超时时间。
     • NULL，永久阻塞。
     • 0，非阻塞。

4. 返回值：

   • 成功：返回准备好的文件描述符个数。
   • 超时：返回0。
   • 出错：返回-1，并设置errno。

5. fd_set:

   • 表示文件描述符集合的数据结构
   • 在fd_set中，每个文件描述符都对应一个位，如果该位为1，则表示对应的文件描述符处于准备好的状态；如果该位为0，则表示对应的文件描述符未准备好。
   • fd_set提供了一些宏操作来方便地对文件描述符集合进行操作:
     • FD_ZERO(fd_set *set)：清空文件描述符集合。
     • FD_SET(int fd, fd_set *set)：将指定的文件描述符添加到集合中。
     • FD_CLR(int fd, fd_set *set)：从集合中移除指定的文件描述符。
     • FD_ISSET(int fd, fd_set *set)：检查指定的文件描述符是否在集合中。
     • FD_COPY(fd_set *src, fd_set *dst)：复制源文件描述符集合到目标文件描述符集合。

6. struct timeval：

   • struct timeval {long    tv_sec;         /* 秒 */long    tv_usec;        /* 微秒 */
     };
     

select实现多路复用

sever.c

#include "net.h" 
#include <sys/select.h> 
#define MAX_SOCK_FD 1024 // 定义最大文件描述符数量为1024int main(int argc, char *argv[])
{int i, ret, fd, newfd; fd_set set, tmpset; Addr_in clientaddr;socklen_t clientlen = sizeof(Addr_in); // 定义客户端地址长度clientlen/*检查参数，小于3个 直接退出进程*/Argment(argc, argv);/*创建已设置监听模式的套接字*/fd = CreateSocket(argv);FD_ZERO(&set);FD_ZERO(&tmpset);FD_SET(fd, &set);while(1){ tmpset = set;if( (ret = select(MAX_SOCK_FD, &tmpset, NULL, NULL, NULL)) < 0) // 调用select函数监听文件描述符集合tmpset中的文件描述符ErrExit("select"); if(FD_ISSET(fd, &tmpset) ){/*接收客户端连接，并生成新的文件描述符*/if( (newfd = accept(fd, (Addr *)&clientaddr, &clientlen) ) < 0) perror("accept");printf("[%s:%d]已建立连接\n", inet_ntoa(clientaddr.sin_addr), ntohs(clientaddr.sin_port)); // 输出客户端地址和端口号FD_SET(newfd, &set); // 将新的文件描述符加入文件描述符集合set}else{ for(i = fd + 1; i < MAX_SOCK_FD; i++){ if(FD_ISSET(i, &tmpset)){ if( DataHandle(i) <= 0){ if( getpeername(i, (Addr *)&clientaddr, &clientlen) )perror("getpeername"); printf("[%s:%d]断开连接\n", inet_ntoa(clientaddr.sin_addr),ntohs(clientaddr.sin_port)); // 输出客户端地址和端口号FD_CLR(i, &set); // 从文件描述符集合set中移除文件描述符i}}}}}return 0; // 程序正常结束，返回0
}

socket.c

#include "net.h"void Argment(int argc, char *argv[]){if(argc < 3){fprintf(stderr, "%s<addr><port>\n", argv[0]);exit(0);}
}
int CreateSocket(char *argv[]){/*创建套接字*/int fd = socket(AF_INET, SOCK_STREAM, 0);if(fd < 0)ErrExit("socket");/*允许地址快速重用*/int flag = 1;if( setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag) ) )perror("setsockopt");/*设置通信结构体*/Addr_in addr;bzero(&addr, sizeof(addr) );addr.sin_family = AF_INET;addr.sin_port = htons( atoi(argv[2]) );/*绑定通信结构体*/if( bind(fd, (Addr *)&addr, sizeof(Addr_in) ) )ErrExit("bind");/*设置套接字为监听模式*/if( listen(fd, BACKLOG) )ErrExit("listen");return fd;
}
int DataHandle(int fd){char buf[BUFSIZ] = {};Addr_in peeraddr;socklen_t peerlen = sizeof(Addr_in);if( getpeername(fd, (Addr *)&peeraddr, &peerlen) )perror("getpeername");int ret = recv(fd, buf, BUFSIZ, 0);if(ret < 0)perror("recv");if(ret > 0){printf("[%s:%d]data: %s\n", inet_ntoa(peeraddr.sin_addr), ntohs(peeraddr.sin_port), buf);}return ret;
}

net.h

#ifndef _NET_H_
#define _NET_H_#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <strings.h>
#include <errno.h>typedef struct sockaddr Addr;
typedef struct sockaddr_in Addr_in;
#define BACKLOG 5
#define ErrExit(msg) do { perror(msg); exit(EXIT_FAILURE); } while(0)void Argment(int argc, char *argv[]);
int CreateSocket(char *argv[]);
int DataHandle(int fd);#endif

成功建立连接

poll 函数

与select函数的功能类似

1. 原型：

   #include <poll.h>
   int poll(struct pollfd *fds, nfds_t nfds, int timeout);
   

2. 功能：监视一组文件描述符的I/O状态，等待它们中的一个或多个变为可读、可写或异常状态。

3. 参数：

   • fds：一个指向pollfd结构体数组的指针，该数组包含了需要监视的文件描述符及其对应的事件。
   • nfds：fds数组中的元素个数。
   • timeout：等待的最长时间（以毫秒为单位），如果设置为0，则表示立即返回；如果设置为负数，则表示无限期等待。

4. 返回值：

   • 如果成功，返回发生事件的文件描述符个数；
   • 如果超时，返回0；
   • 如果出错，返回-1。

5. struct pollfd：

   • struct pollfd {int fd;         // 文件描述符short events;   // 注册的事件short revents;  // 返回的事件
     };
     

   • fd：这是文件描述符，即需要被监视的句柄。

   • events：这是一个位掩码，定义了我们关心的文件描述符的事件类型。常用的事件类型有：

     • POLLIN：表示文件描述符可读。
     • POLLOUT：表示文件描述符可写。
     • POLLPRI：表示文件描述符有紧急数据（带外数据）可读。
     • POLLERR：表示文件描述符发生错误。
     • POLLHUP：表示文件描述符挂起。

   • revents：这是一个输出参数，当poll返回时，它指出了文件描述符上实际发生了哪些事件。

6. nfds_t:

   • typedef unsigned long int nfds_t;
     

poll实现多路复用

sever.c

#include "net.h"
#include <poll.h>#define MAX_SOCK_FD 1024
int main(int argc, char *argv[])
{int i, j, fd, newfd;nfds_t nfds = 1;struct pollfd fds[MAX_SOCK_FD] = {};Addr_in addr;socklen_t addrlen = sizeof(Addr_in);/*检查参数，小于3个 直接退出进程*/Argment(argc, argv);/*创建已设置监听模式的套接字*/fd = CreateSocket(argv);fds[0].fd = fd;fds[0].events = POLLIN;while(1){if( poll(fds, nfds, -1) < 0)ErrExit("poll");for(i = 0; i < nfds; i++){/*接收客户端连接，并生成新的文件描述符*/if(fds[i].fd == fd && fds[i].revents & POLLIN){if( (newfd = accept(fd, (Addr *)&addr, &addrlen) ) < 0)perror("accept");fds[nfds].fd = newfd;fds[nfds++].events = POLLIN;printf("[%s:%d][nfds=%lu] connection successful.\n", inet_ntoa(addr.sin_addr), ntohs(addr.sin_port), nfds);}/*处理客户端数据*/if(i > 0 && fds[i].revents & POLLIN){if(DataHandle(fds[i].fd) <= 0){if( getpeername(fds[i].fd, (Addr *)&addr, &addrlen) < 0)perror("getpeername");printf("[%s:%d][fd=%d] exited.\n", inet_ntoa(addr.sin_addr), ntohs(addr.sin_port), fds[i].fd);close(fds[i].fd);for(j=i; j<nfds-1; j++)fds[j] = fds[j+1];nfds--;i--;}}}}close(fd);return 0;
}

socket.c

#include "net.h"void Argment(int argc, char *argv[]){if(argc < 3){fprintf(stderr, "%s<addr><port>\n", argv[0]);exit(0);}
}
int CreateSocket(char *argv[]){/*创建套接字*/int fd = socket(AF_INET, SOCK_STREAM, 0);if(fd < 0)ErrExit("socket");/*允许地址快速重用*/int flag = 1;if( setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag) ) )perror("setsockopt");/*设置通信结构体*/Addr_in addr;bzero(&addr, sizeof(addr) );addr.sin_family = AF_INET;addr.sin_port = htons( atoi(argv[2]) );/*绑定通信结构体*/if( bind(fd, (Addr *)&addr, sizeof(Addr_in) ) )ErrExit("bind");/*设置套接字为监听模式*/if( listen(fd, BACKLOG) )ErrExit("listen");return fd;
}
int DataHandle(int fd){char buf[BUFSIZ] = {};Addr_in peeraddr;socklen_t peerlen = sizeof(Addr_in);if( getpeername(fd, (Addr *)&peeraddr, &peerlen) )perror("getpeername");int ret = recv(fd, buf, BUFSIZ, 0);if(ret < 0)perror("recv");if(ret > 0){printf("[%s:%d]data: %s\n", inet_ntoa(peeraddr.sin_addr), ntohs(peeraddr.sin_port), buf);}return ret;
}

net.h

#ifndef _NET_H_
#define _NET_H_#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <strings.h>
#include <errno.h>typedef struct sockaddr Addr;
typedef struct sockaddr_in Addr_in;
#define BACKLOG 5
#define ErrExit(msg) do { perror(msg); exit(EXIT_FAILURE); } while(0)void Argment(int argc, char *argv[]);
int CreateSocket(char *argv[]);
int DataHandle(int fd);#endif

成功实现多路复用

epoll函数族

epoll函数族用于高效的I/O事件管理，特别适用于高并发服务器应用

头文件: #include <sys/epoll.h>

1. epoll_create:

   • int epoll_create(int size);
     

   • 功能：创建一个epoll实例，并返回一个文件描述符作为该epoll实例的标识。

   • 参数size：之前用于定义事件队列的大小，但在Linux 2.6以后的版本中已被忽略。通常设置为0。

   • 返回值：

     • 成功：返回一个非负的文件描述符。
     • 失败：返回-1，并设置errno。

2. epoll_ctl:

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

   • 功能：向epoll实例中添加、修改或删除文件描述符及其相关事件。

   • 参数：

     • epfd：epoll实例的文件描述符。
     • op：操作类型，可以是EPOLL_CTL_ADD（添加新的文件描述符）、EPOLL_CTL_MOD（修改已注册的文件描述符的事件）或EPOLL_CTL_DEL（删除一个文件描述符）。
     • fd：要操作的文件描述符。
     • event：指向epoll_event结构的指针，用于指定事件类型和文件描述符的数据。

   • 返回值：

     • 成功：返回0。
     • 失败：返回-1，并设置errno。

3. epoll_wait:

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

   • 功能：阻塞等待已注册的文件描述符上的事件发生。

   • 参数：

     • epfd：epoll实例的文件描述符。
     • events：指向epoll_event结构体数组的指针，用于存储发生的事件。
     • maxevents：可以接收的事件数量的最大值。
     • timeout：超时时间（以毫秒为单位），决定函数的阻塞行为。设置为0立即返回，设置为-1则无限期阻塞。

   • 返回值：

     • 成功：返回就绪事件的个数。
     • 超时：返回0。
     • 失败：返回-1，并设置errno。

epoll实现多路复用

sever.c

#include "net.h"
#include <sys/epoll.h>#define MAX_SOCK_FD 1024int main(int argc, char *argv[])
{int i, nfds, fd, epfd, newfd;Addr_in addr;socklen_t addrlen = sizeof(Addr_in);struct epoll_event tmp, events[MAX_SOCK_FD] = {};/*检查参数，小于3个 直接退出进程*/Argment(argc, argv);/*创建已设置监听模式的套接字*/fd = CreateSocket(argv);if( (epfd = epoll_create(1)) < 0)ErrExit("epoll_create");tmp.events = EPOLLIN;tmp.data.fd = fd;if( epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &tmp) )ErrExit("epoll_ctl");while(1) {if( (nfds = epoll_wait(epfd, events, MAX_SOCK_FD, -1) ) < 0)ErrExit("epoll_wait");printf("nfds = %d\n", nfds);for(i = 0; i < nfds; i++) {if(events[i].data.fd == fd){/*接收客户端连接，并生成新的文件描述符*/if( (newfd = accept(fd, (Addr *)&addr, &addrlen) ) < 0)perror("accept");printf("[%s:%d] connection.\n", inet_ntoa(addr.sin_addr), ntohs(addr.sin_port) );tmp.events = EPOLLIN;tmp.data.fd = newfd;if( epoll_ctl(epfd, EPOLL_CTL_ADD, newfd, &tmp) )ErrExit("epoll_ctl");}else{/*处理客户端数据*/if(DataHandle(events[i].data.fd) <= 0){if( epoll_ctl(epfd, EPOLL_CTL_DEL, events[i].data.fd, NULL) )ErrExit("epoll_ctl");if( getpeername(events[i].data.fd, (Addr *)&addr, &addrlen) )perror("getpeername");printf("[%s:%d] exited.\n", inet_ntoa(addr.sin_addr), ntohs(addr.sin_port) );close(events[i].data.fd);}}}}close(epfd);close(fd);return 0;
}

socket.c

#include "net.h"void Argment(int argc, char *argv[]){if(argc < 3){fprintf(stderr, "%s<addr><port>\n", argv[0]);exit(0);}
}
int CreateSocket(char *argv[]){/*创建套接字*/int fd = socket(AF_INET, SOCK_STREAM, 0);if(fd < 0)ErrExit("socket");/*允许地址快速重用*/int flag = 1;if( setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag) ) )perror("setsockopt");/*设置通信结构体*/Addr_in addr;bzero(&addr, sizeof(addr) );addr.sin_family = AF_INET;addr.sin_port = htons( atoi(argv[2]) );/*绑定通信结构体*/if( bind(fd, (Addr *)&addr, sizeof(Addr_in) ) )ErrExit("bind");/*设置套接字为监听模式*/if( listen(fd, BACKLOG) )ErrExit("listen");return fd;
}
int DataHandle(int fd){char buf[BUFSIZ] = {};Addr_in peeraddr;socklen_t peerlen = sizeof(Addr_in);if( getpeername(fd, (Addr *)&peeraddr, &peerlen) )perror("getpeername");int ret = recv(fd, buf, BUFSIZ, 0);if(ret < 0)perror("recv");if(ret > 0){printf("[%s:%d]data: %s\n", inet_ntoa(peeraddr.sin_addr), ntohs(peeraddr.sin_port), buf);}return ret;
}

net.h

#ifndef _NET_H_
#define _NET_H_#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <strings.h>
#include <errno.h>typedef struct sockaddr Addr;
typedef struct sockaddr_in Addr_in;
#define BACKLOG 5
#define ErrExit(msg) do { perror(msg); exit(EXIT_FAILURE); } while(0)void Argment(int argc, char *argv[]);
int CreateSocket(char *argv[]);
int DataHandle(int fd);#endif