煮酒与君饮

这个屌丝很懒，什么也没留下！

热门标签

Linux高级IO

作者：煮酒与君饮 | 2024-07-19 21:43:14

踩

Linux高级IO

五种IO模型

举一个钓鱼的例子

张三：拿了1个鱼竿，死死的盯着浮漂，什么也不做，当浮漂动了就挥动鱼竿将鱼钓上来。（阻塞IO）
李四：拿了1个鱼竿，定期观察浮漂，如果有鱼上钩就挥动鱼竿将鱼钓上来，否则继续去做其他事情。（非阻塞IO）
王五：拿了1个鱼竿，在鱼竿顶部绑一个铃铛，如果铃铛响了就证明有鱼上钩就挥动鱼竿将鱼钓上来，如果没响就继续去做其他事情。（信号驱动IO）
赵六：拿了100个鱼竿，并排将鱼竿插好，走过来，在走过去遍历的查询鱼竿，如果有鱼上钩就挥动鱼竿将鱼钓上来，如果没有就一直遍历。（多路复用/多路转接IO）
田七：田七是一个有钱的老板，他给了自己的司机一个鱼竿，让司机去钓鱼，当鱼钓上来时再打电话告诉田七来拿鱼，而田七自己则开车去做其他事情去了。（异步IO）

阻塞IO: 在内核将数据准备好之前, 系统调用会一直等待. 所有的套接字, 默认都是阻塞方式.

在这里插入图片描述

非阻塞IO: 如果内核还未将数据准备好, 系统调用仍然会直接返回, 并且返回EWOULDBLOCK错误码

非阻塞IO往往需要程序员循环的方式反复尝试读写文件描述符, 这个过程称为轮询. 这对CPU来说是较大的浪费,一般只有特定场景下才使用.

在这里插入图片描述

信号驱动IO: 内核将数据准备好的时候, 使用SIGIO信号通知应用程序进行IO操作

在这里插入图片描述

IO多路转接: 虽然从流程图上看起来和阻塞IO类似. 实际上最核心在于IO多路转接能够同时等待多个文件描述符的就绪状态.

在这里插入图片描述

异步IO: 由内核在数据拷贝完成时, 通知应用程序(而信号驱动是告诉应用程序何时可以开始拷贝数据).

在这里插入图片描述

设置为非阻塞

//将fd文件描述符设置为非阻塞状态
void SetNoBlock(int fd)
{
    int fl = fcntl(fd, F_GETFL);
    if (fl < 0)
    {
        perror("fcntl");
        return;
    }
    fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}
1
2
3
4
5
6
7
8
9
10
11

使用F_GETFL将当前的文件描述符的属性取出来(这是一个位图).
然后再使用F_SETFL将文件描述符设置回去. 设置回去的同时, 加上一个O_NONBLOCK参数.

轮询方式读取标准输入

#include <iostream>
#include <unistd.h>
#include <fcntl.h>
#include <cstdio>
#include <cstring>

void SetNoBlock(int fd)
{
    int fl = fcntl(fd, F_GETFL);
    if (fl < 0)
    {
        perror("fcntl");
        return;
    }
    fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}

int main()
{
    char buffer[64];
    SetNoBlock(0);
    while (true)
    {
        printf(">>");
        //fflush(stdout);
        ssize_t n = read(0, buffer, sizeof(buffer) - 1);
        if (n > 0)
        {
            buffer[n - 1] = 0;
            std::cout << "echo#" << buffer << std::endl;
        }
        else if (n == 0) // ctrl(按键) + d
        {
            std::cout << "end file" << std::endl;
            break;
        }
        else
        {
            //一旦低层没有数据就绪，以出错的形式返回，但是不算真正的出错。
            if (errno == EAGAIN || errno == EWOULDBLOCK)
            {

                // 低层数据没有准备好，希望你下次继续来检测
                sleep(1);//在这里你可以处理其他事件
                continue;
            }
            else if (errno == EINTR)
            {
                // 这次IO被信号中断，也需要重新读取
                continue;
            }
            else
                std::cout << "read error??" << strerror(errno) << "error code: " << errno << std::endl;
            break;
        }
    }

    return 0;
}
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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

阻塞I/O模式

检查条件是否就绪：当read系统调用被调用时，它会检查是否有数据可读。如果数据不可用（即，没有数据到达或者缓冲区中没有数据），read会暂时挂起调用它的进程，直到数据变得可用。
如果条件就绪，就唤起当前进程：一旦数据到达或缓冲区中有数据可供读取，read系统调用会“唤醒”挂起的进程，继续执行，并将数据从文件描述符的底层缓冲区复制到用户提供的缓冲区中。如果read成功读取数据，它会返回读取的字节数。

非阻塞I/O模式

检查条件是否就绪：read系统调用仍然会检查是否有数据可读。但是，如果数据不可用，它不会挂起调用它的进程。
立即返回：如果数据不可用，read会立即返回一个错误，通常设置errno为EAGAIN或EWOULDBLOCK。这意味着调用者需要在稍后再次尝试读取，或者使用其他机制（如select、poll或epoll）来监测文件描述符上的活动。

在非阻塞模式下，read不会“唤醒”进程，因为进程在read调用中不会被挂起。相反，进程需要主动检查文件描述符的状态，以确定何时进行读取操作。

I/O多路转接之select

认识select

一定要以某种形式，一次等待多个fd (仅仅是等待，没有拷贝)
哪一个fd或者那些个fd就绪，用户需要知道的
select 要等的的多个fd，一定有少量或者全部都是准备好的，一般是少量

在这里插入图片描述

select调用失败时，错误码可能被设置为：

EBADF 文件描述词为无效的或该文件已关闭
EINTR 此调用被信号所中断
EINVAL 参数n 为负值。
ENOMEM 核心内存不足

关于fd_set结构体

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;
1
2
3
4
5
6
7
8
9
10
11
12

其实这个结构就是一个整数数组, 更严格的说, 是一个 “位图”. 使用位图中对应的位来表示要监视的文件描述符
提供了一组操作fd_set的接口, 来比较方便的操作位图

void FD_CLR(int fd, fd_set *set); // 用来清除描述词组set中相关fd 的位
int FD_ISSET(int fd, fd_set *set); // 用来测试描述词组set中相关fd 的位是否为真
void FD_SET(int fd, fd_set *set); // 用来设置描述词组set中相关fd的位
void FD_ZERO(fd_set *set); // 用来清除描述词组set的全部位
1
2
3
4

select服务器
select服务器，使用时需要程序员自己维护一个第三方的数组，来进行已经获得的sock(文件描述符)进行管理。

// Sock.hpp
#pragma once
#include <iostream>
#include <string>
#include <cstdlib>
#include <cstring>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <unistd.h>

class Sock
{
public:
    Sock(int ret = -1)
    :_sock(ret)
    {}

    void Socket()
    {
        _sock = socket(AF_INET, SOCK_STREAM, 0);
        if(-1 == _sock)
        {
            std::cout<< "socket error" << std::endl;
            exit(-1);
        }
    }

    void Bind(const uint16_t& port)
    {
        struct sockaddr_in local;
        memset(&local, 0, sizeof(local));//强烈建议做清空
        local.sin_family = AF_INET;//通信方式
        local.sin_port = htons(port);//端口号
        local.sin_addr.s_addr = INADDR_ANY;//ip地址

        if(bind(_sock, (struct sockaddr*)&local, sizeof(local)) < 0)
        {
            std::cout<< "bind error" << std::endl;
            exit(-1);
        }
    }

    void Listen()
    {
        if(listen(_sock, 0) < 0)
        {
            std::cout<< "listen error" << std::endl;
            exit(-1);
        }
    }

    int Accept(std::string* clientIp, uint16_t* clientPort)
    {
        struct sockaddr_in temp;
        socklen_t len = sizeof(temp);
        int sock = accept(_sock, (struct sockaddr*)&temp, &len);
        if(sock < 0)
        {

            std::cout<< "accept error" << std::endl;
            exit(-1);
        }
        else
        {
            *clientIp = inet_ntoa(temp.sin_addr);
            *clientPort = ntohs(temp.sin_port);
        }

        return sock;
    }
    // 输入: const &
    // 输出: *
    // 输入输出: &
    int Connect(std::string& serverIp, uint16_t& serverPort)
    {
        struct sockaddr_in server;
        memset(&server, 0, sizeof(server));
        server.sin_family = AF_INET;
        server.sin_port = htons(serverPort);
        server.sin_addr.s_addr = inet_addr(serverIp.c_str());

        return connect(_sock, (struct sockaddr *)&server, sizeof(server));
    }
    int Fd()
    {
        return _sock;
    }
    void Close()
    {
        if (_sock != -1)
            close(_sock);
    }
    ~Sock()
    {}
private:
    int _sock;
};
//selectServer.hpp
#pragma once
#include <iostream>
#include <sys/select.h>
#include "Sock.hpp"

const static uint16_t gport = 8888;
typedef int type_t;
static const int defaultfd = -1;
class selectServer
{
    static const int N = sizeof(fd_set) *8 ;
public:
    selectServer(uint16_t port = gport)
    :_port(port)
    {}
    void InitServer()
    {
        _listenSock.Socket();
        _listenSock.Bind(_port);
        _listenSock.Listen();

        for(int i = 0; i < N; ++i)
        {
            fdarray[i] = defaultfd;
        }
    }
    void  Accepter()
    {
        std::cout<< "Accepter();" << std::endl;
        std::string clientip;
        uint16_t clientport;
        int sock = _listenSock.Accept(&clientip,&clientport);
        if(sock < 0)
        {
            return;
        }
        int pos = 1;
        for(; pos < N; ++pos)
        {
            if(fdarray[pos] == defaultfd)
            {
                fdarray[pos] = sock;
                break;
            }
        }
        if(pos >= N)
        {
            close(sock);
            std::cerr << "fd_set的位图已经超过了其大小"<<std::endl;
        }
        

    }
    void HandlerEvent(fd_set &rfds)
    {
        for(int i = 0; i < N; ++i)
        {
            if(fdarray[i] == defaultfd)
                continue;
            
            if((fdarray[i] == _listenSock.Fd()) && (FD_ISSET(fdarray[i], &rfds)))//将监听套接字文件准备就绪
            {
                
                Accepter();
            }
            else if((fdarray[i] != _listenSock.Fd()) && (FD_ISSET(fdarray[i], &rfds)))
            {
                int fd = fdarray[i];
                char buffer[1024];
                ssize_t s = recv(fd, buffer, sizeof(buffer) - 1, 0);
                if (s > 0)
                {
                    buffer[s-1] = 0;
                    std::cout << "client# " << buffer << std::endl;

                    // 发送回去也要被select管理的,TODO
                    std::string echo = buffer;
                    echo += " [select server echo]";
                    send(fd, echo.c_str(), echo.size(), 0);
                }
                else
                {
                    if (s == 0)
                    {
                        close(fdarray[i]);
                        fdarray[i] = defaultfd;
                    }
                    else
                    {
                        std::cerr<<errno<< ":"<< strerror(errno) << std::endl;
                    }
                }
            }
        }
    }
    void Start()
    {
        // 1. 这里我们能够直接获取新的链接吗？
        // 2. 最开始的时候，我们的服务器是没有太多的sock的，甚至只有一个sock！listensock
        // 3. 在网络中， 新连接到来被当做 读事件就绪！
        fdarray[0] = _listenSock.Fd();

        while(true)
        {
            // 因为rfds是一个输入输出型参数，注定了每次都要对rfds进行重置，重置，必定要知道我历史上都有哪些fd?fdarray[]
            // 因为服务器在运行中，sockfd的值一直在动态变化，所以maxfd也一定在变化， maxfd是不是也要进行动态更新， fdarray[]
            fd_set rfds;
            FD_ZERO(&rfds);
            int maxfd = fdarray[0];
            for (int i = 0; i < N; ++i)
            {
                if(fdarray[i] == defaultfd)
                    continue;
                
                FD_SET(fdarray[i], &rfds);
                if(maxfd < fdarray[i])
                {
                    maxfd = fdarray[i];
                }  
            }
            int n = select(maxfd + 1,&rfds,nullptr,nullptr,nullptr);
            switch(n)
            {
            case 0:
                std::cout << "没有文件描述符，就绪" << std::endl;
                break;
            case -1:
                std::cerr<<errno<< ":"<< strerror(errno) << std::endl;
                break;
            default:
                std::cout << "有一个就绪事件发生了" << std::endl;
                HandlerEvent(rfds);
                DebugPrint();
                break;
            }
        }
    }

    void DebugPrint()
    {
        std::cout << "fdarray[]: ";
        for (int i = 0; i < N; i++)
        {
            if (fdarray[i] == defaultfd)
                continue;
            std::cout << fdarray[i] << " ";
        }
        std::cout << "\n";
        sleep(1);
    }
private:
    uint16_t _port;
    Sock _listenSock;
    type_t fdarray[N];
};
//test.cc
#include <iostream>
#include <memory>
#include "selectServer.hpp"

int main()
{
    std::unique_ptr<selectServer> svr(new selectServer());
    svr->InitServer();
    svr->Start();
    
    return 0;
}
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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267

select缺点

每次调用select, 都需要手动设置fd集合, 从接口使用角度来说也非常不便.
每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大
同时每次调用select都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也很大
select支持的文件描述符数量太小.

I/O多路转接之poll

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

struct pollfd {
 int fd;         /* 文件描述符 */
 short events;   /* 需要监视该文件描述符上的那些事件 */
 short revents;  /* poll函数返回时告知用户该文件描述符上的那些事件，已经就绪了 */
};
1
2
3
4
5
6
7
8

参数说明

fds是一个poll函数监听的结构列表. 每一个元素中, 包含了三部分内容: 文件描述符, 监听的事件集合, 返
回的事件集合.
nfds表示fds数组的长度.
timeout表示poll函数的超时时间, 单位是毫秒(ms) （-1：永久阻塞）（0：非阻塞）（>0：在规定时间内阻塞等待，超时后非阻塞一次）

events和revents的取值:

事件	描述	是否可作为输入	是否可作为输入
POLLIN	数据（包括普通数据和优先数据）可读	是	是
POLLRDNORM	普通数据可读	是	是
POLLRDBAND	优先级带数据可读（Linux不支持）	是	是
POLLPRI	高优先级数据可读，比如TCP带外数据	是	是
POLLOUT	数据（包括普通数据和优先数据）可写	是	是
POLLWRNORM	普通数据可写	是	是
POLLWRBAND	优先级带数据可写	是	是
POLLRDHUP	TCP连接被对方关闭，或者对方关闭了写操作，它由GNU引入	是	是
POLLERR	错误	否	是
POLLHUP	挂起。比如管道的写端被关闭后，读端描述符上将收到POLLHUP事件	否	是
POLLNVAL	文件描述符没有打开	否	是

这些取值实际都是以宏的方式进行定义的，它们的二进制序列当中有且只有一个比特位是1，且为1的比特位是各不相同的。
在这里插入图片描述

返回结果

返回值小于0, 表示出错;
返回值等于0, 表示poll函数等待超时;
返回值大于0, 表示poll由于监听的文件描述符就绪而返回.

poll服务器

//Sock.hpp
#pragma once
#include <iostream>
#include <string>
#include <cstdlib>
#include <cstring>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <unistd.h>

class Sock
{
public:
    Sock(int ret = -1)
    :_sock(ret)
    {}

    void Socket()
    {
        _sock = socket(AF_INET, SOCK_STREAM, 0);
        if(-1 == _sock)
        {
            std::cout<< "socket error" << std::endl;
            exit(-1);
        }
    }

    void Bind(const uint16_t& port)
    {
        struct sockaddr_in local;
        memset(&local, 0, sizeof(local));//强烈建议做清空
        local.sin_family = AF_INET;//通信方式
        local.sin_port = htons(port);//端口号
        local.sin_addr.s_addr = INADDR_ANY;//ip地址

        if(bind(_sock, (struct sockaddr*)&local, sizeof(local)) < 0)
        {
            std::cout<< "bind error" << std::endl;
            exit(-1);
        }
    }

    void Listen()
    {
        if(listen(_sock, 0) < 0)
        {
            std::cout<< "listen error" << std::endl;
            exit(-1);
        }
    }

    int Accept(std::string* clientIp, uint16_t* clientPort)
    {
        struct sockaddr_in temp;
        socklen_t len = sizeof(temp);
        int sock = accept(_sock, (struct sockaddr*)&temp, &len);
        if(sock < 0)
        {

            std::cout<< "accept error" << std::endl;
            exit(-1);
        }
        else
        {
            *clientIp = inet_ntoa(temp.sin_addr);
            *clientPort = ntohs(temp.sin_port);
        }

        return sock;
    }
    // 输入: const &
    // 输出: *
    // 输入输出: &
    int Connect(std::string& serverIp, uint16_t& serverPort)
    {
        struct sockaddr_in server;
        memset(&server, 0, sizeof(server));
        server.sin_family = AF_INET;
        server.sin_port = htons(serverPort);
        server.sin_addr.s_addr = inet_addr(serverIp.c_str());

        return connect(_sock, (struct sockaddr *)&server, sizeof(server));
    }
    int Fd()
    {
        return _sock;
    }
    void Close()
    {
        if (_sock != -1)
            close(_sock);
    }
    ~Sock()
    {}
private:
    int _sock;
};
//pollServer.hpp
#pragma once
#include <iostream>
#include <sys/poll.h>
#include "Sock.hpp"

const static uint16_t gport = 8888;
typedef struct pollfd type_t;
const static int defaultfd = -1;
const static int N = 1024;
const static short defaultevent = 0;
class pollServer
{
    
public:
    pollServer(uint16_t port = gport)
    :_port(port)
    {}
    void InitServer()
    {
        _listenSock.Socket();
        _listenSock.Bind(_port);
        _listenSock.Listen();

        fdarray = new type_t[N];
        for(int i = 0; i < N; ++i)
        {
            fdarray[i].fd = defaultfd;
            fdarray[i].events = defaultevent;
            fdarray[i].revents = defaultevent;
        }
    }
    void  Accepter()
    {
        
        std::string clientip;
        uint16_t clientport;
        int sock = _listenSock.Accept(&clientip,&clientport);
        if(sock < 0)
        {
            return;
        }
        int pos = 1;
        for(; pos < N; ++pos)
        {
            if(fdarray[pos].fd == defaultfd)
            {
                fdarray[pos].fd = sock;
                fdarray[pos].events = POLLIN;
                fdarray[pos].revents = defaultevent;
                break;
            }
        }
        if(pos >= N)
        {
            close(sock);
            std::cerr << "已经超过N了其大小"<<std::endl;
        }
        

    }
    void HandlerEvent()
    {
        for(int i = 0; i < N; ++i)
        {
            if(fdarray[i].fd == defaultfd)
                continue;
            
            if((fdarray[i].fd == _listenSock.Fd()) && ((fdarray[i].revents) & POLLIN))//将监听套接字文件准备就绪
            {
                
                Accepter();
            }
            else if((fdarray[i].fd != _listenSock.Fd()) && ((fdarray[i].revents) & POLLIN))
            {
                int fd = fdarray[i].fd;
                char buffer[1024];
                ssize_t s = recv(fd, buffer, sizeof(buffer) - 1, 0);
                if (s > 0)
                {
                    buffer[s-1] = 0;
                    std::cout << "client# " << buffer << std::endl;

                    // 发送回去也要被select管理的,TODO
                    std::string echo = buffer;
                    echo += " [select server echo]";
                    send(fd, echo.c_str(), echo.size(), 0);
                }
                else
                {
                    if (s == 0)
                    {
                        close(fdarray[i].fd);
                        fdarray[i].fd = defaultfd;
                        fdarray[i].events = defaultevent;
                        fdarray[i].revents = defaultevent;
                    }
                    else
                    {
                        std::cerr<<errno<< ":"<< strerror(errno) << std::endl;
                    }
                }
            }
        }
    }
    void Start()
    {
        fdarray[0].fd = _listenSock.Fd();
        fdarray[0].events = POLLIN;
        while(true)
        {
           //poll内部会对struct pollfd结构体内的文件描述符做甄别，-1跳过。
            int n = poll(fdarray,N,-1);
            switch(n)
            {
            case 0:
                std::cout << "没有文件描述符，就绪" << std::endl;
                break;
            case -1:
                std::cerr<<errno<< ":"<< strerror(errno) << std::endl;
                break;
            default:
                std::cout << "有一个就绪事件发生了" << std::endl;
                HandlerEvent();
                DebugPrint();
                break;
            }
        }
    }

    void DebugPrint()
    {
        std::cout << "fdarray[]: ";
        for (int i = 0; i < N; i++)
        {
            if (fdarray[i].fd == defaultfd)
                continue;
            std::cout << fdarray[i].fd << " ";
        }
        std::cout << "\n";
        sleep(1);
    }
    ~pollServer()
    {
        _listenSock.Close();
        if(fdarray)
            delete[] fdarray;
    }
private:
    uint16_t _port;
    Sock _listenSock;
    type_t* fdarray;
};
//test.cc
#include <iostream>
#include <memory>
#include "pollServer.hpp"

int main()
{
    // fd_set fd;
    // std::cout << sizeof(fd) * 8<< std::endl;
    std::unique_ptr<pollServer> svr(new pollServer());
    svr->InitServer();
    svr->Start();
    
    return 0;
}
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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266

poll的缺点
poll中监听的文件描述符数目增多时

和select函数一样，poll返回后，需要轮询pollfd来获取就绪的描述符.
每次调用poll都需要把大量的pollfd结构从用户态拷贝到内核中.
同时连接的大量客户端在一时刻可能只有很少的处于就绪状态, 因此随着监视的描述符数量的增长, 其效率也会线性下降.

I/O多路转接之epoll

epoll_create

epoll_create函数用于创建一个epoll文件描述符，这是使用epoll机制的起点。

int epoll_create(int size);
1

参数说明：

size：自从Linux2.6.8之后，size参数是被忽略的，但size的值必须设置为大于0的值。

返回值：

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

epoll_ctl

对添加到红黑树中做节点，进行增改删

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

参数说明：

epfd：指定的epoll模型，就是epoll_create的返回值。
op：表示具体的动作，用三个宏来表示。
fd：需要监视的文件描述符。
event：如果需要监视该文件描述符上的哪些事件就传参，如果不需要就设置空指针。

第二个参数的取值:

EPOLL_CTL_ADD ：注册新的fd到epfd中；
EPOLL_CTL_MOD ：修改已经注册的fd的监听事件；
EPOLL_CTL_DEL ：从epfd中删除一个fd

struct epoll_event结构如下:

typedef union epoll_data
{
  void *ptr;
  int fd;
  uint32_t u32;
  uint64_t u64;
} epoll_data_t;

struct epoll_event
{
  uint32_t events;	/* Epoll events */
  epoll_data_t data;	/* User data variable */
} __EPOLL_PACKED;
1
2
3
4
5
6
7
8
9
10
11
12
13

events的常用取值如下：

EPOLLIN：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）。
EPOLLOUT：表示对应的文件描述符可以写。
EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）。
EPOLLERR：表示对应的文件描述符发送错误。
EPOLLHUP：表示对应的文件描述符被挂断，即对端将文件描述符关闭了。
EPOLLET：将epoll的工作方式设置为边缘触发（Edge Triggered）模式。
EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听该文件描述符的话，需要重新将该文件描述符添加到epoll模型中。

epoll_wait

从就绪队列中获取就绪事件

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

参数说明：

epfd：指定的epoll模型，就是epoll_create的返回值。
epoll将会把发生的事件赋值到events数组中 (events不可以是空指针，内核只负责把数据复制到这个events数组中，不会去帮助我们在用户态中分配内存).
maxevents告之内核这个events有多大，这个 maxevents的值不能大于创建epoll_create()时的size.
参数timeout是超时时间 (毫秒，0会立即返回，-1是永久阻塞).

返回值：

如果函数调用成功，返回对应I/O上已准备好的文件描述符数目，如返回0表示已超时, 返回小于0表示函数失败.

epoll工作原理

在这里插入图片描述
当某一进程调用epoll_create方法时，Linux内核会创建一个eventpoll结构体，这个结构体中有两个成员与epoll的使用方式密切相关，如下所示：

struct eventpoll{ 
 .... 
 /*红黑树的根节点，这颗树中存储着所有添加到epoll中的需要监控的事件*/ 
 struct rb_root rbr; 
 /*双链表中则存放着将要通过epoll_wait返回给用户的满足条件的事件*/ 
 struct list_head rdlist; 
 .... 
};
1
2
3
4
5
6
7
8

文件描述符就是红黑树的key值。
epoll会将监视的文件，添加到红黑树中做节点，通过epoll_ctl函数中的EPOLL_CTL_ADD ，EPOLL_CTL_MOD ，EPOLL_CTL_DEL对红黑树进行增改删。
epoll模型中操作系统会将监视的就绪文件，添加到就绪队列中，通过调用epoll_wait函数实际就是在从就绪队列当中获取已经就绪的事件。
当数据就绪时，会通过网卡将数据写到对应文件的缓冲区。然后由操作系统会将文件缓冲区内的数据写的文件中，执行完后操作系统会将该文件添加到就绪队列中，这样就不用通过遍历查找就绪文件了。

epoll服务器

//Sock.hpp
#pragma once
#include <iostream>
#include <string>
#include <cstdlib>
#include <cstring>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <unistd.h>

class Sock
{
public:
    Sock(int ret = -1)
    :_sock(ret)
    {}

    void Socket()
    {
        _sock = socket(AF_INET, SOCK_STREAM, 0);
        if(-1 == _sock)
        {
            std::cout<< "socket error" << std::endl;
            exit(-1);
        }
    }

    void Bind(const uint16_t& port)
    {
        struct sockaddr_in local;
        memset(&local, 0, sizeof(local));//强烈建议做清空
        local.sin_family = AF_INET;//通信方式
        local.sin_port = htons(port);//端口号
        local.sin_addr.s_addr = INADDR_ANY;//ip地址

        if(bind(_sock, (struct sockaddr*)&local, sizeof(local)) < 0)
        {
            std::cout<< "bind error" << std::endl;
            exit(-1);
        }
    }

    void Listen()
    {
        if(listen(_sock, 0) < 0)
        {
            std::cout<< "listen error" << std::endl;
            exit(-1);
        }
    }

    int Accept(std::string* clientIp, uint16_t* clientPort)
    {
        struct sockaddr_in temp;
        socklen_t len = sizeof(temp);
        int sock = accept(_sock, (struct sockaddr*)&temp, &len);
        if(sock < 0)
        {

            std::cout<< "accept error" << std::endl;
            exit(-1);
        }
        else
        {
            *clientIp = inet_ntoa(temp.sin_addr);
            *clientPort = ntohs(temp.sin_port);
        }

        return sock;
    }
    // 输入: const &
    // 输出: *
    // 输入输出: &
    int Connect(std::string& serverIp, uint16_t& serverPort)
    {
        struct sockaddr_in server;
        memset(&server, 0, sizeof(server));
        server.sin_family = AF_INET;
        server.sin_port = htons(serverPort);
        server.sin_addr.s_addr = inet_addr(serverIp.c_str());

        return connect(_sock, (struct sockaddr *)&server, sizeof(server));
    }
    int Fd()
    {
        return _sock;
    }
    void Close()
    {
        if (_sock != -1)
            close(_sock);
    }
    ~Sock()
    {}
private:
    int _sock;
};
//epollServer.hpp
#pragma once
#include <iostream>
#include <sys/epoll.h>
#include "Sock.hpp"

const static uint16_t gport = 8888;
#define GUNM 64

class epollServer
{
    
public:
    epollServer(int epfd = -1,uint16_t port = gport)
    :_port(port)
    {}
    void InitServer()
    {
        _listenSock.Socket();
        _listenSock.Bind(_port);
        _listenSock.Listen();
        _epfd = epoll_create(1);
        if(_epfd < 0)
        {
            std::cerr<<errno<< "epoll_create errror:"<< strerror(errno) << std::endl;
            exit(-1);
        }

    }
    void  Accepter()
    {
        
    }
    void HandlerEvent(int n)
    {
        for(int i = 0; i < n; ++i)
        {
            int fd = fdarry[i].data.fd;
            uint32_t events = fdarry[i].events;
            if(events & EPOLLIN)
            {
                if(fd == _listenSock.Fd())//1.新连接到来
                {
                    std::string temp;
                    uint16_t port;
                    int sock = _listenSock.Accept(&temp, &port);
                    if(sock < 0)
                    {
                        continue;
                    }
                    std::cout<<"这个连接已经连上服务器了" << temp << ": " << sock << std::endl;
                    struct epoll_event event;
                    event.events = EPOLLIN;
                    event.data.fd = sock; // 用户数据，epoll低层不对该数据做任何修改，就是为了给未来就绪返回的。
                    epoll_ctl(_epfd,EPOLL_CTL_ADD,sock,&event);
                }
                else//2.读取事件就绪
                {
                    char buffer[1024];
                    ssize_t s = recv(fd, buffer, sizeof(buffer) - 1, 0);//我们目前无法保证读取到完整的报文。
                    if (s > 0)
                    {
                        buffer[s - 1] = 0;
                        std::cout << "client# " << buffer << std::endl;

                        // 发送回去也要被select管理的,TODO
                        std::string echo = buffer;
                        echo += " [select server echo]";
                        send(fd, echo.c_str(), echo.size(), 0);
                    }
                    else
                    {
                        if (s == 0)
                        {
                            //要先从epoll中移除，然后再关闭文件（必须的）
                            epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr);
                            close(fd);
                            std::cout << "要关闭的文件描述符是：" << fd << std::endl;
                        }
                        else
                        {
                            std::cerr << errno << ":" << strerror(errno) << std::endl;
                        }
                    }
                }
            }
        }
    }
    void Start()
    {
        
        //1.将listeSock添加到epoll中
        struct epoll_event event;
        event.events = EPOLLIN;
        event.data.fd = _listenSock.Fd();//用户数据，epoll低层不对该数据做任何修改，就是为了给未来就绪返回的。
        int n = epoll_ctl(_epfd,EPOLL_CTL_ADD,_listenSock.Fd(), &event);
        
        if(n > 0)
        {
            std::cerr<<errno<< "epoll_create errror:"<< strerror(errno) << std::endl;
            exit(-1);
        }
        while(true)
        {
            int n = epoll_wait(_epfd, fdarry,GUNM,-1);
            switch (n)
            {
            case 0:
                break;
            case -1:
                break;
            default:
                HandlerEvent(n);
                break;
            }
        }
    }

    ~epollServer()
    {
        _listenSock.Close();
       if(_epfd != -1)
       {
            close(_epfd);
       }
    }
private:
    uint16_t _port;
    Sock _listenSock;
    int _epfd;
   struct epoll_event fdarry[GUNM];
};
//test.cc
#include <iostream>
#include <memory>
#include "epollServer.hpp"

int main()
{
    std::unique_ptr<epollServer> svr(new epollServer());
    svr->InitServer();
    svr->Start();
    
    return 0;
}
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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243

epoll的优点(和 select 的缺点对应)

接口使用方便: 虽然拆分成了三个函数, 但是反而使用起来更方便高效. 不需要每次循环都设置关注的文件描述符, 也做到了输入输出参数分离开
数据拷贝轻量: 只在合适的时候调用EPOLL_CTL_ADD将文件描述符结构拷贝到内核中, 这个操作并不频繁(而select/poll都是每次循环都要进行拷贝)
事件回调机制: 避免使用遍历, 而是使用回调函数的方式, 将就绪的文件描述符结构加入到就绪队列中, epoll_wait 返回直接访问就绪队列就知道哪些文件描述符就绪. 这个操作时间复杂度O(1). 即使文件描述符数目很多, 效率也不会受到影响.
没有数量限制: 文件描述符数目无上限

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】

Linux高级IO

目录

五种IO模型

I/O多路转接之select

I/O多路转接之poll

I/O多路转接之epoll

epoll_create

epoll_ctl

epoll_wait

epoll工作原理