深入理解 Linux IO 模型（二）：多路复用——select

目录

多路复用 IO

select 的核心定位

select 接口

基于 Select 的多路复用服务器 —— EchoServer

SelectServer.hpp

Socket.hpp

select 特点

select 缺点

多路复用 IO

在上一篇文章中：深入理解 Linux IO 模型（一）：阻塞 IO、非阻塞 IO 与信号驱动 IO 中，我们介绍了阻塞 IO、非阻塞 IO 和信号驱动 IO。其中，非阻塞 IO 虽然能避免进程因为等待数据而阻塞，但如果需要同时管理大量文件描述符，就需要不断轮询这些文件描述符，造成大量的 CPU 浪费。

那么，有没有一种机制可以让我们只使用一个线程，就能够同时监控多个文件描述符，并在文件描述符就绪时通知进程呢？

多路复用 IO 正是为了解决这个问题而出现的。

多路复用 I/O：只使用一个执行流，就可以同时监控多个文件描述符。当一个或多个文件描述符对应的 IO 事件就绪时，多路复用机制会返回这些就绪的文件描述符，执行流就可以对这些文件描述符进行 I/O 操作。

补充一：

与非阻塞 IO 不同，多路复用 IO 不需要应用程序不断轮询每个文件描述符，而是将多个文件描述符交给内核统一监控。应用程序可以阻塞在多路复用接口上，直到至少有一个文件描述符就绪，再对就绪的文件描述符进行处理。

补充二：

执行流是程序执行指令的基本单位，本质上对应一个线程。

一个进程可以包含一个执行流或多个执行流，也就是一个线程或多个线程。

对于单线程进程来讲，一个进程可以被称为一个执行流。

所以说：多路复用 IO 可以让一个执行流同时监控多个文件描述符。

补充三：

事件就绪：对于 Socket 的接收缓冲区，如果接收缓冲区上有数据，就称为读事件就绪。对于 Socket 的发送缓冲区，如果发送缓冲区上有剩余空间，就称为写事件就绪。

对于一个新创建的 Socket，一般情况下，它的读事件不就绪，它的写事件就绪。

注：IO 由两部分组成：1. 等待事件就绪 2.将数据从内核空间拷贝到用户空间

多路复用 IO 只负责监控文件描述符是否就绪，即只会帮进程完成第一步，它并不提进程完成第二步，即不会将数据从内核空间拷贝到用户空间

在 Linux 提供了 select、poll、epoll 等多路复用机制。本文首先从最经典的 select 开始，理解 I/O 多路复用究竟解决了什么问题，以及 select 在编码上是如何实现多个文件描述符同时监控的。

select 的核心定位

select 是一种 IO 多路复用机制，用于同时监控一个或多个文件描述符的 I/O 事件是否就绪。

它本身只负责等待和检测文件描述符的事件是否就绪，并不会替应用程序完成真正的 IO 操作。

select 接口

#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

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

参数：

int nfds：需要监听的所有文件描述符的最大值 + 1 

例如：需要监听的文件描述符：3 5 7 8 9 12 

此处：nfds：12 + 1

struct timeval *timeout：输入输出型参数

输入参数：控制进程在 select 处的最长等待时间。

在 timeout 时间内，进程阻塞在 select 处等待文件描述符事件就绪，如果存在一个或多个文件描述符事件就绪，select 返回。超过了 timeout 时间，select 依旧返回。

输出参数：select 返回后，timeout 表示剩余的等待时间

struct timeval {                long    tv_sec;         /* seconds */                long    tv_usec;        /* microseconds */

};

long tv_sec：单位是秒

long tv_usec：单位是微秒

注：1 秒 == 1e3 毫秒 == 1e6 微秒  

例如：

timeout.tv_sec = 2;

timeout.tv_usec = 0;

select(…, &timeout);

意思是：当进程调用 select 时，开始计时，
如果在 2 S内没有文件描述符就绪，进程会在 select 返回，
此时 timeout.tv_sec = 0, timeout.tv_usec = 0;
如果在 1 S内有文件描述符就绪，进程会直接在 select 返回，
此时 timeout.tv_sec = 1, timeout.tv_usec = 0;

特殊情况：

timeout = nullptr 表示：没有设置超时时间，进程将一直阻塞在 select 处，直到至少有一个文件描述符事件就绪。等价于阻塞 IO 

timeout.tv_sec = 0, timeout.tv_usec = 0 表示：不进行等待，检查所有被监控的文件描述符。即使没有文件描述符就绪，依旧返回。等价于非阻塞 IO

返回值：

返回值 > 0 ：在被监视的所有文件描述符中，事件就绪的文件描述符个数

返回值 == 0：在 timeout 时间内，没有文件描述事件就绪。对于 timeout = nullptr ，不存在这种情况。

返回值 == -1：select 调用出错

fd_set *readfds：输入输出型参数

/* fd_set for select and pselect. */
typedef struct
{
/* XPG4.2 requires this member name. Otherwise avoid the name
from the global namespace. */
#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 本质类型就是一个文件描述符集合，在内核中的实现就是一个位图。

输入参数：

比特位的编号：文件描述符的编号 

比特位的状态：文件描述符是否需要被监控，如果是 1，表示需要，否则表示不需要

用户告诉内核，你要帮我监控的文件描述符集合

输出参数：

比特位的编号：文件描述符的编号 

比特位的状态：文件描述符读事件是否就绪，如果是 1，表示事件就绪，否则表示读事件不就绪

内核告诉用户，被监控的文件描述符集合，哪些文件描述符读事件就绪，哪些文件描述符读事件没有就绪。

fd_set *writefds 表示写事件文件描述符集合 fd_set *exceptfds 表示异常文件描述符集合

位图的相关操作接口

void FD_CLR(int fd, fd_set *set);
int FD_ISSET(int fd, fd_set *set);
void FD_SET(int fd, fd_set *set);
void FD_ZERO(fd_set *set);

基于 Select 的多路复用服务器 —— EchoServer

SelectServer.hpp

#pragma once

#include "Socket.hpp"
#include <functional>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/select.h>

class SelectServer
{
const static int defaultfd = -1;
const static int default_backlog = 5;

const static int size = sizeof(fd_set) * 8;

public:
SelectServer(in_port_t port, int backlog = default_backlog)
: _port(port), _backlog(backlog), _listenfd(std::make_unique<TcpSocket>()), _isrunning(false)
{
_listenfd->BuildTcpListenSocket(port, backlog);
// 辅助数组初始化
for (int i = 0; i < size; ++i)
_rfds[i] = defaultfd;
_rfds[0] = _listenfd->Fd();
}

void Start()
{
_isrunning = true;
while (_isrunning)
{
fd_set rfds;
FD_ZERO(&rfds);
// 借助辅助数组找到历史所需要监控的文件描述符，并找到文件描述符的最大值
int maxfd = defaultfd;
for (int i = 0; i < size; ++i)
{
if (_rfds[i] == defaultfd)
continue;
FD_SET(_rfds[i], &rfds);
if (maxfd < _rfds[i])
maxfd = _rfds[i];
}
struct timeval timeout;
timeout.tv_sec = 1;
timeout.tv_usec = 0;
// 监控文件描述符的读事件就绪
int n = select(maxfd + 1, &rfds, nullptr, nullptr, &timeout);

switch (n)
{
case 0: // 阻塞 IO 不存在
LOG(LogLevel::INFO) << "select timeout";
continue;
break;
case -1:
LOG(LogLevel::FATAL) << "select error";
exit(SELECT_ERR);
break;
default:
// 有读事件就绪
HandleEvents(rfds);
break;
}
}
_isrunning = false;
}
// 事件派发器
void HandleEvents(fd_set &rfds)
{
for (int i = 0; i < size; ++i)
{
if (_rfds[i] == defaultfd)
continue;
// 要关心的文件描述符读事件就绪
if (FD_ISSET(_rfds[i], &rfds))
{
// 连接管理
if (_rfds[i] == _listenfd->Fd())
{
Accepter();
}
else // 数据处理
{
Recver(i);
}
}
}
}
// 连接管理器
void Accepter()
{
IntAddr client;
int fd = _listenfd->Accpet(&client);
if (fd < 0)
{
LOG(LogLevel::WARNING) << "accept error";
return;
}
// 添加到辅助数组中
int pos = 0;
while (pos < size)
{
if (_rfds[pos] == -1)
break;
++pos;
}
if (pos == size)
{
LOG(LogLevel::WARNING) << "rfds is full";
return;
}
_rfds[pos] = fd;
}

void Recver(int pos)
{
// 存在 TCP 粘包问题
char buffer[1024];
ssize_t n = recv(_rfds[pos], buffer, sizeof(buffer) – 1, 0);
if (n > 0)
{
buffer[n] = 0;
std::cout << "client say@ " << buffer << std::endl;
}
else if (n == 0)
{
LOG(LogLevel::INFO) << "client quit!";
close(_rfds[pos]);
_rfds[pos] = -1;
}
else
{
LOG(LogLevel::WARNING) << "recv error";
close(_rfds[pos]);
_rfds[pos] = -1;
}
}
void Stop()
{
_isrunning = false;
}

~SelectServer()
{
}

private:
in_port_t _port;
int _backlog;
std::unique_ptr<Socket> _listenfd;
bool _isrunning;
int _rfds[size];
};

Socket.hpp

#pragma once

#include "Log.hpp"
#include "Error.hpp"
#include "IntAddr.hpp"
#include <iostream>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <memory>
#include <cstdlib>

using namespace LogModule;

// 继承模板

class Socket
{
public:
virtual void CreateSockfd() = 0;
virtual void BindSockfd(in_port_t port) = 0;
virtual void ListenSockfd(int backlog) = 0;
virtual int Accpet(IntAddr *client_addr) = 0;
virtual int Connect(const IntAddr &server) = 0;
virtual void Close() = 0;
virtual ssize_t Recv(std::string *message) = 0;
virtual ssize_t Send(const std::string &message) = 0;
virtual int Fd() = 0;
void BuildTcpListenSocket(in_port_t port, int backlog)
{
CreateSockfd();
BindSockfd(port);
ListenSockfd(backlog);
}
void BuildTcpSocket()
{
CreateSockfd();
}

};

const static int default_sockfd = -1;

class TcpSocket : public Socket
{
public:
TcpSocket()
:_sockfd(default_sockfd)
{}
TcpSocket(int sockfd)
:_sockfd(sockfd)
{}
void CreateSockfd() override
{
_sockfd = ::socket(AF_INET, SOCK_STREAM, 0);
if(_sockfd < 0)
{
LOG(LogLevel::FATAL) << "socket error";
exit(SOCKET_ERR);
}
LOG(LogLevel::INFO) << "socket success";
}
void BindSockfd(in_port_t port) override
{
IntAddr server(port);
int n = ::bind(_sockfd, server.NetAddrPtr(), server.NetAddrLen());
if(n < 0)
{
LOG(LogLevel::FATAL) << "bind error";
exit(BIND_ERR);
}
LOG(LogLevel::INFO) << "bind success";
}
void ListenSockfd(int backlog) override
{
int n = ::listen(_sockfd, backlog);
if(n < 0)
{
LOG(LogLevel::FATAL) << "listen error";
exit(LISTEN_ERR);
}
LOG(LogLevel::INFO) << "listen success";
}
int Accpet(IntAddr *client_addr) override
{
struct sockaddr_in client;
socklen_t len = sizeof(client);
int sockfd = ::accept(_sockfd, CONV(client), &len);

return sockfd;
}
void Close() override
{
::close(_sockfd);
}

ssize_t Recv(std::string *message) override
{
char buffer[1024];
ssize_t n = ::recv(_sockfd, buffer, sizeof(buffer – 1), 0);
if(n > 0)
{
buffer[n] = 0;
*message += buffer;
}
return n;
}
ssize_t Send(const std::string &message) override
{
return ::send(_sockfd, message.c_str(), message.size(), 0);
}
int Connect(const IntAddr &server) override
{
return ::connect(_sockfd, server.NetAddrPtr(), server.NetAddrLen());
}

int Fd() override
{
return _sockfd;
}

~TcpSocket(){}
private:
int _sockfd;
};

select 特点

一个进程或者线程可以同时监听多个文件描述符，避免了为每个连接创建一个线程或者进程

当某个文件描述符事件就绪时，select 返回，程序再进行事件处理

select 缺点

（1）文件描述符数量存在限制

select 通常受 FD_SETSIZE 限制。Linux 中常见默认值是 1024，因此不适合大量连接的场景。

（2）每次调用 select 前都需要重新设置文件描述符集合

select 会修改传入的 fd_set，返回后其中只保留已经就绪的文件描述符。

因此，服务器通常需要使用辅助数组保存历史上需要监听的文件描述符，并在每次调用 select 前，根据辅助数组重新初始化 fd_set。

时间复杂度为 O(N)

（3）应用层需要遍历文件描述符集合

select 返回后，应用层并不知道具体哪些文件描述符就绪，需要遍历文件描述符集合，逐个判断哪些文件描述符发生了事件。

时间复杂度为 O(N)

（4）内核也需要遍历文件描述符

select 将需要监听的文件描述符集合交给内核后，内核需要遍历这些文件描述符，检查哪些文件描述符已经就绪。

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

因此，在 select 的参数列表中：nfds 的值为需要监听的文件描述符的最大值 + 1

时间复杂度 O(N)