【Linux高性能服务器编程】I/O复用的高级应用

左手的ㄟ右手 2023-09-25 20:40 45阅读 0赞

### 文章目录 ###

*  一、基于 select 的非阻塞 connect
 *  二、基于 poll 的聊天室程序
 *   *  2.1 客户端
     *  2.2 服务器
 *  三、基于 epoll 实现同时处理 TCP 和 UDP 服务

--------------------

# 一、基于 select 的非阻塞 connect #

`connect`系统调用的 man 手册中有如下一段内容：

> EINPROGERESS  
> The socket is nonblocking and the connection cannot be completed immediately. It is possible to select(2) or poll(2) for completion by selecting the socket for writing. After select(2) indicates writability, use getsockopt(2) to read the SO\_ERROR option at level SOL\_SOCKET to determine whether connect() completed successfully (SO\_ERROR is zero) or unsuccessfully (SO\_ERROR is one of the usual error codes listed here, explaining the reason for the failure).

这段话描述了 `connect` 出错时的一种 `errno` 值：`EINPROGRESS`。这种错误发生在对非阻塞了`socket`调用了`connect`，而连接又没有建立时。根据man文档的解释，在这种情况下我们可以调用`select`、`poll`等函数来监听这个连接失败的`socket`上的可写事件。当`select`、`poll`等函数返回后，再利用`getsockopt`来读取错误码并清除该`socket`上的错误。如果错误码是0，表示连接成功，否则连接失败。

通过上面描述的非阻塞 `connect` 方式，我们就能同时发起多个连接并一起等待，下面是非阻塞`connect`的一种实现方式。

**首先是设置文件描述符为非阻塞状态：**

// 设置文件描述符为非阻塞
    int setnonblocking(int fd)
    {
        
        int old_opt = fcntl(fd, F_GETFL);
        int new_opt = old_opt | O_NONBLOCK;
        fcntl(fd, F_SETFL, new_opt);
        // 返回以前的文件描述符状态
        return old_opt;
    }

先使用`fcntl`函数获取并保存`sockfd`描述符的状态，然后再使用`fcntl`函数将其设置为非阻塞的，然后将原状态返回，便于在建立连接成功后恢复`sockfd`的状态。

**实现unblock\_connect函数：**

// 非阻塞连接函数，参数是服务器IP、端口以及超时时间。成功返回处于连接状态的socket，失败则返回-1.
    int unblock_connect(const std::string &ip, uint16_t port, int time)
    {
        
        int ret = 0;
        struct sockaddr_in addr;
        bzero(&addr, sizeof(addr));
        addr.sin_family = AF_INET;
        inet_pton(AF_INET, ip.c_str(), &addr.sin_addr);
        addr.sin_port = htons(port);
    
        int sockfd = socket(AF_INET, SOCK_STREAM, 0);
        // 设置套接字描述符为非阻塞，并保存其原来的状态
        int fdopt = setnonblocking(sockfd);
        ret = connect(sockfd, (const sockaddr *)&addr, sizeof(addr));
        if (ret == 0)
        {
        
            // 连接成功，将sockfd恢复原来的状态，然后返回
            std::cout << "connect with server immediately!" << std::endl;
            fcntl(sockfd, F_SETFL, fdopt);
            return sockfd;
        }
        else if (errno != EINPROGRESS)
        {
        
            // 如果连接没有立即建立，那么只有errno为EINPROGRESS时才表示连接还在继续，否则出错返回
            std::cout << "unblock connect not support!" << std::endl;
            close(sockfd);
            return -1;
        }
        else
        {
        
            // 继续连接
    
            // 写文件描述符集
            fd_set writefds;
            FD_ZERO(&writefds);
            // 将sockfd添加到writefds中
            FD_SET(sockfd, &writefds);
            timeval timeout;
            timeout.tv_sec = time;
            timeout.tv_usec = 0;
    
            ret = select(sockfd + 1, nullptr, &writefds, nullptr, &timeout);
            if (ret <= 0)
            {
        
                // select超时或出错，立即返回
                std::cout << "connect timeout!" << std::endl;
                close(sockfd);
                return -1;
            }
    
            if (!FD_ISSET(sockfd, &writefds))
            {
        
                std::cout << "no events on sockfd found!" << std::endl;
                close(sockfd);
                return -1;
            }
    
            int error = 0;
            socklen_t len = sizeof(error);
            // 调用 getsockopt来获取并清除sockfd上的错误
            if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &error, &len) < 0)
            {
        
                std::cout << "get socket option failed!" << std::endl;
                close(sockfd);
                return -1;
            }
    
            // 错误码不为0表示连接出错
            if (error != 0)
            {
        
                std::cout << "conntion failed after select with the error: " << error << std::endl;
                close(sockfd);
                return -1;
            }
    
            // 连接成功，将sockfd恢复原来的状态，然后返回
            std::cout << "connect successfully after select with the sockfd: " << sockfd << std::endl;
            fcntl(sockfd, F_SETFL, fdopt);
            return sockfd;
        }
    }

在该函数中，首先创建`socket`套接字，并将其设置为非阻塞的，然后调用`connect`函数，如果连接成功，恢复`sockfd`的原状态后返回；如果不能立即连接成功，只有返回的`errno`是`EINPROGRESS`才表示连接还在进行，否则直接出错返回。

当返回的`errno`是`EINPROGRESS`时，调用`select`多路复用系统调用对`sockfd`的写事件进行监听。监听成功后，调用`getsockopt`函数来获取并清除`sockfd`上的错误。如果错误码是0，则表示在调用`select`函数后建立连接成功，恢复`sockfd`的原状态后返回；否则出错返回。

**`main`函数逻辑：**

int main(int argc, char* argv[])
    {
        
        if(argc != 3)
        {
        
            std::cout << "usage: " << argv[0] << " ip port" << std::endl;
            return -1;
        }
    
        const std::string ip = argv[1];
        uint16_t port = atoi(argv[2]);
    
        int sockfd = unblock_connect(ip, port, 10);
        if(sockfd < 0)
        {
        
            std::cout << "unblock connect failed!" << std::endl;
            return -1;
        }
        else
        {
        
            std::cout << "连接成功，可以向服务器发起请求了..." << std::endl;
        }
    
        close(sockfd);
        return 0;
    }

# 二、基于 poll 的聊天室程序 #

这里基于poll系统调用实现一个简单的聊天室程序，以阐述如何使用I/O复用技术来同时处理网络连接给用户输入。该聊天室程序能让所有的用户同时在线群聊，它分为客户端和服务端两个部分。

其中客户端有两个功能：一是从标准输入终端读入用户数据，并将用户数据发送至服务器；二是往标准输出终端打印服务器发送给它的数据。服务器的功能是接收客户端数据，并将接收到的数据发送给每一个登录到该服务器上的客户端（数据发送者除外）。

**效果展示如下：**

以下是客户端和服务端的代码。

## 2.1 客户端 ##

客户端程序使用 poll 同时监听用户输入和网络连接，并利用 splice 函数将用户输入的内容直接定向到网络连接上以发送之，从而实现数据的零拷贝，提高了程序的运行效率。客户端代码如下：

#include <iostream>
    #include <string>
    #include <cstring>
    #include <cassert>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <sys/types.h>
    #include <sys/socket.h>
    #include <unistd.h>
    #include <poll.h>
    #include <fcntl.h>
    
    #define BUFFER_SIZE 1024
    
    using namespace std;
    
    int main(int argc, char *argv[])
    {
        
        if (argc != 3)
        {
        
            cout << "usage: " << argv[0] << " ip port" << endl;
            return -1;
        }
    
        const string ip = argv[1];
        uint16_t port = atoi(argv[2]);
    
        sockaddr_in server_addr;
        bzero(&server_addr, sizeof(server_addr));
        server_addr.sin_family = AF_INET;
        inet_pton(AF_INET, ip.c_str(), &server_addr.sin_addr);
        server_addr.sin_port = htons(port);
    
        int sockfd = socket(AF_INET, SOCK_STREAM, 0);
        assert(sockfd >= 0);
    
        if (connect(sockfd, (const sockaddr *)&server_addr, sizeof(server_addr)) < 0)
        {
        
            cerr << "connect failed!" << endl;
            close(sockfd);
            return -1;
        }
    
        pollfd fds[2];
    
        // 注册文件描述符0（标准输入） 和 sockfd 文件描述符的可读事件
        fds[0].fd = 0;
        fds[0].events = POLLIN;
        fds[0].revents = 0;
    
        fds[1].fd = sockfd;
        fds[1].events = POLLIN | POLLRDHUP;
        fds[1].revents = 0;
    
        char read_buf[BUFFER_SIZE];
        int pipefd[2];
        int ret = 0;
        ret = pipe(pipefd);
        assert(ret != -1);
    
    
        while (true)
        {
        
            ret = poll(fds, 2, -1);
            if(ret < 0)
            {
        
                cout << "poll falied!" << endl;
                break;
            }
    
            if(fds[1].revents & POLLRDHUP)
            {
        
                cout << "server close the connection!" << endl;
            }
            else if(fds[1].revents & POLLIN)
            {
        
                memset(read_buf, '\0', BUFFER_SIZE);
                ssize_t s = recv(fds[1].fd, read_buf, BUFFER_SIZE - 1, 0);
                read_buf[s] = '\0';
                cout << read_buf << endl;
            }
    
            if(fds[0].revents & POLLIN)
            {
        
                //使用splice函数将用户输入的数据直接写到 sockfd 上（零拷贝）
                ret = splice(0, nullptr, pipefd[1], nullptr, 32768, SPLICE_F_MORE | SPLICE_F_MOVE);
                ret = splice(pipefd[0], nullptr, sockfd, nullptr, 32768, SPLICE_F_MORE | SPLICE_F_MOVE);
            }
        }
    
        close(sockfd);
        return 0;
    }

## 2.2 服务器 ##

服务器程序使用poll同时管理监听 socket 和连接 socket，并且使用牺牲空间换取时间的策略来提高服务器的性能。服务器代码如下：

#include <iostream>
    #include <string>
    #include <cstring>
    #include <cassert>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <unistd.h>
    #include <fcntl.h>
    #include <sys/types.h>
    #include <sys/socket.h>
    #include <poll.h>
    
    #define USER_LIMIT 5     // 最大用户数量
    #define BUFFER_SIZE 1024 // 读缓冲区的大小
    #define FD_LIMIT 65535   // 文件描述符数量限制
    
    // 客户端数据
    struct client_data
    {
        
        sockaddr_in address;        // 客户端socket
        char *write_buf;            // 待写到客户端的数据的位置
        char read_buf[BUFFER_SIZE]; // 从客户端读入的数据
    };
    
    using namespace std;
    
    int setnonblocking(int fd)
    {
        
        int old_opt = fcntl(fd, F_GETFL);
        int new_opt = old_opt | O_NONBLOCK;
        fcntl(fd, F_SETFL, new_opt);
        return old_opt;
    }
    
    int main(int argc, char *argv[])
    {
        
        if (argc != 3)
        {
        
            cout << "usage: " << argv[0] << " ip port" << endl;
            return -1;
        }
    
        string ip = argv[1];
        uint16_t port = atoi(argv[2]);
    
        int ret = 0;
        sockaddr_in addr;
        bzero(&addr, sizeof(addr));
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        inet_pton(AF_INET, ip.c_str(), &addr.sin_addr);
    
        int listenfd = socket(AF_INET, SOCK_STREAM, 0);
        int opt = 1;
        setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));
        assert(listenfd >= 0);
    
        ret = bind(listenfd, (const sockaddr *)&addr, sizeof(addr));
        assert(ret != -1);
    
        ret = listen(listenfd, 5);
        assert(ret != -1);
    
        // 创建users数组，分配FD_LIMIT个client_data对象。
        // 每个可能的socket连接都可以获得这样一个对象，并且socket的值可以直接原来索引socket对应的client_data对象，
        // 这是将socket和客户数据关联的简单而高效的方法。
        client_data *users = new client_data[FD_LIMIT];
    
        // 尽管分配了足够多的client_data对象，但为了提高poll的性能，仍然有必要限制用户数量
        pollfd fds[USER_LIMIT + 1];
        // 当前的用户数量
        int user_count = 0;
        // 初始化fds
        for (int i = 0; i <= USER_LIMIT; ++i)
        {
        
            fds[i].fd = -1;
            fds[i].events = 0;
        }
    
        // 添加listenfd到fds
        fds[0].fd = listenfd;
        fds[0].events = POLLIN | POLLERR;
        fds[0].revents = 0;
    
        while (true)
        {
        
            ret = poll(fds, user_count + 1, -1);
            if (ret < 0)
            {
        
                cerr << "poll failed!" << endl;
                break;
            }
    
            for (int i = 0; i <= user_count; ++i)
            {
        
                // 如果此时是监听套接字就绪，则处理新连接
                if ((fds[i].fd == listenfd) && (fds[i].revents & POLLIN))
                {
        
                    sockaddr_in client_addr;
                    socklen_t len = sizeof(client_addr);
                    int connfd = accept(listenfd, (sockaddr *)&client_addr, &len);
                    if (connfd < 0)
                    {
        
                        cerr << "accept failed! error: " << errno << "msg: " << strerror(errno) << endl;
                        continue;
                    }
    
                    // 如果请求太多，则关闭该连接
                    if (user_count >= USER_LIMIT)
                    {
        
                        const string info = "too many users\n";
                        cout << info;
                        send(connfd, info.c_str(), info.size(), 0);
                        close(connfd);
                    }
    
                    // 对于新连接，同时修改users和fds数组，以保证users[connfd] 对应于 新连接文件描述符connfd的客户端数据
                    ++user_count;
                    users[connfd].address = client_addr;
                    setnonblocking(connfd);
                    fds[user_count].fd = connfd;
                    fds[user_count].events = POLLIN | POLLERR | POLLRDHUP;
                    fds[user_count].revents = 0;
    
                    cout << "comes a new user, now have " << user_count << " users!" << endl;
                }
                // 如果connfd出错
                else if (fds[i].revents & POLLERR)
                {
        
                    cout << "get an error from socket: " << fds[i].fd << endl;
                    char errors[1024];
                    memset(errors, '\0', sizeof(errors));
                    socklen_t len = sizeof(errors);
                    if (getsockopt(fds[i].fd, SOL_SOCKET, SO_ERROR, &errors, &len) < 0)
                    {
        
                        cout << "get socket option failed!" << endl;
                    }
                    cout << "socket errors: " << string(errors) << endl;
                    continue;
                }
                // 如果客户端关闭连接
                else if (fds[i].revents & POLLRDHUP)
                {
        
                    // 客户端关闭连接，服务端也关闭连接，用户总数减1
                    // 这里直接将数组中的最后一个用户替代当前用户
                    users[fds[i].fd] = users[fds[user_count].fd];
                    close(fds[i].fd);
                    fds[i] = fds[user_count];
                    --i;
                    --user_count;
                    cout << "a user left!" << endl;
                }
                // connfd 读事件就绪
                else if (fds[i].revents & POLLIN)
                {
        
                    int connfd = fds[i].fd;
                    memset(users[connfd].read_buf, '\0', BUFFER_SIZE);
                    ret = recv(connfd, users[connfd].read_buf, BUFFER_SIZE-1, 0);
                    if (ret < 0)
                    {
        
                        // 如果读取出错，则关闭连接
                        if (errno != EAGAIN)
                        {
        
                            users[fds[i].fd] = users[fds[user_count].fd];
                            close(fds[i].fd);
                            fds[i] = fds[user_count];
                            --i;
                            --user_count;
                        }
                    }
                    if (ret == 0)
                    {
        
                    }
                    else
                    {
        
                        users[connfd].read_buf[ret] = '\0';
                        cout << "get " << ret << " bytes of clent data : " << users[connfd].read_buf << ". socket: " << connfd << endl;
                        // 如果收到客户端数据，则通知其他socket准备写数据
                        for (int j = 1; j <= user_count; ++j)
                        {
        
                            if (fds[j].fd == connfd)
                            {
        
                                continue;
                            }
    
                            fds[j].events |= ~POLLIN;
                            fds[j].events |= POLLOUT;
                            users[fds[j].fd].write_buf = users[connfd].read_buf;
                        }
                    }
                }
                // connfd 写事件就绪
                else if (fds[i].revents & POLLOUT)
                {
        
                    int connfd = fds[i].fd;
                    if (!users[connfd].write_buf)
                    {
        
                        continue;
                    }
    
                    ret = send(connfd, users[connfd].write_buf, strlen(users[connfd].write_buf), 0);
                    users[connfd].write_buf = nullptr;
    
                    // 写完数据要重新注册 fds[i] 上的可读事件
                    fds[i].events |= ~POLLOUT;
                    fds[i].events |= POLLIN;
                }
                else
                {
        
                    //...
                    // cout << "something happend!" << endl;
                }
            }
        }
    
        delete[] users;
        close(listenfd);
        return 0;
    }

# 三、基于 epoll 实现同时处理 TCP 和 UDP 服务 #

以上的两个服务器程序都只能监听一个端口，在实际应用中，有不少程序能够同时监听多个端口，比如超级服务`inetd`和 android 的调试服务`adbd`。

从`bind`系统调用的参数来看，一个`socket`只能与一个`socket`地址绑定，即一个`socket`只能监听一个端口。因此，如果一个程序想要监听多个端口，就必须创建多个`socket`，并将它们分别绑定到各个端口上。这样一来服务器程序就需要同时管理多个监听`socket`，I/O复用技术就有了用武之地了。另外，即使在同一个端口，如果服务器要同时处理该端口上的TCP和UDP请求，则也需要创建两个不同的`socket`：一个是流`socket`，另一个是数据报`socket`，如何把它们两个绑定到同一个端口上。

以下的回射服务器就能同时处理一个端口上的TCP和UDP请求：

#include <iostream>
    #include <string>
    #include <cstring>
    #include <cassert>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <sys/types.h>
    #include <sys/socket.h>
    #include <sys/epoll.h>
    #include <unistd.h>
    #include <fcntl.h>
    
    #define MAX_EVENT_NUMBER 1024
    #define TCP_BUFFER_SIZE 512
    #define UDP_BUFFER_SIZE 1024
    
    using namespace std;
    
    int setnonblocking(int fd)
    {
        
        int old_opt = fcntl(fd, F_GETFL);
        fcntl(fd, F_SETFL, old_opt | O_NONBLOCK);
        return old_opt;
    }
    
    void addfd(int epollfd, int fd)
    {
        
        epoll_event event;
        event.data.fd = fd;
        event.events = EPOLLIN | EPOLLET;
        epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);
        setnonblocking(fd);
    }
    
    int main(int argc, char *argv[])
    {
        
        if (argc != 3)
        {
        
            cout << "usage " << argv[0] << " ip port" << endl;
            return 1;
        }
    
        const string ip = argv[1];
        uint16_t port = atoi(argv[2]);
    
        int ret = 0;
        sockaddr_in address;
        bzero(&address, sizeof(address));
        address.sin_family = AF_INET;
        inet_pton(AF_INET, ip.c_str(), &address.sin_addr);
        address.sin_port = htons(port);
    
        // 创建 TCP socket，并将其绑定到端口 port 上
        int listenfd = socket(AF_INET, SOCK_STREAM, 0);
        assert(listenfd >= 0);
    
        ret = bind(listenfd, (const sockaddr *)&address, sizeof(address));
        assert(ret != -1);
    
        ret = listen(listenfd, 5);
        assert(ret != -1);
    
        // 创建 UDP socket，并将其绑定到端口 port 上
        bzero(&address, sizeof(address));
        address.sin_family = AF_INET;
        inet_pton(AF_INET, ip.c_str(), &address.sin_addr);
        address.sin_port = htons(port);
    
        int udpfd = socket(AF_INET, SOCK_DGRAM, 0);
        assert(listenfd >= 0);
    
        ret = bind(udpfd, (const sockaddr *)&address, sizeof(address));
        assert(ret != -1);
    
        epoll_event events[MAX_EVENT_NUMBER];
        int epollfd = epoll_create(5);
        assert(epollfd != -1);
    
        // 注册 TCP socket 和 UDP socket 上的可读事件
        addfd(epollfd, listenfd);
        addfd(epollfd, udpfd);
    
        while (true)
        {
        
            int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);
            if (number < 0)
            {
        
                cerr << "epoll_wait failed!" << endl;
                break;
            }
    
            for (int i = 0; i < number; ++i)
            {
        
                int sockfd = events[i].data.fd;
                uint32_t revent = events[i].events;
                if (revent & EPOLLIN)
                {
        
                    if (sockfd == listenfd)
                    {
        
                        // listenfd就绪，处理新的TCP连接
                        sockaddr_in peer;
                        socklen_t len = sizeof(peer);
                        int connfd = accept(listenfd, (sockaddr *)&peer, &len);
                        assert(connfd >= 0);
                        addfd(epollfd, connfd);
                    }
                    else if (sockfd == udpfd)
                    {
        
                        // udpfd读事件就绪，接收并回发数据
                        char buffer[UDP_BUFFER_SIZE];
                        memset(buffer, '\0', UDP_BUFFER_SIZE);
    
                        sockaddr_in peer;
                        socklen_t len = sizeof(peer);
    
                        ret = recvfrom(udpfd, buffer, UDP_BUFFER_SIZE - 1, 0, (sockaddr *)&peer, &len);
                        if (ret > 0)
                        {
        
                            buffer[ret] = '\0';
                            cout << "client by UDP -> server# " << buffer << endl;
                            sendto(udpfd, buffer, UDP_BUFFER_SIZE - 1, 0, (const sockaddr *)&peer, len);
                        }
                    }
                    else
                    {
        
                        // 普通connfd 读事件就绪
                        char buffer[TCP_BUFFER_SIZE];
                        memset(buffer, '\0', TCP_BUFFER_SIZE);
                        while (true)
                        {
        
                            memset(buffer, '\0', TCP_BUFFER_SIZE);
                            ret = recv(sockfd, buffer, TCP_BUFFER_SIZE - 1, 0);
                            if (ret < 0)
                            {
        
                                if (errno == EAGAIN || errno == EWOULDBLOCK)
                                    break;
    
                                close(sockfd);
                                break;
                            }
                            else if (ret == 0)
                            {
        
                                cout << "client quit!" << endl;
                                close(sockfd);
                            }
                            else
                            {
        
                                buffer[ret] = '\0';
                                cout << "client by TCP -> server# " << buffer << endl;
                                send(sockfd, buffer, TCP_BUFFER_SIZE - 1, 0);
                            }
                        }
                    }
                }
                else
                {
        
                    //...
                }
            }
        }
    
        return 0;
    }

启动程序，使用`netstat -nltup | grep multiport`命令就可以看到该程序那个同时处理TCP和UDP请求了：  
![a82463fab85348a8819fe7a7d44e8f5a.png][]

[a82463fab85348a8819fe7a7d44e8f5a.png]: https://img-blog.csdnimg.cn/a82463fab85348a8819fe7a7d44e8f5a.png