Linux异步io机制 io_uring

简介: Linux异步io机制 io_uring

io_uring作为2019年的后起之秀,为linux异步网络编程新增一把倚天大剑,让我们简单学习一下!

数据结构:

a. sq (submition queue):提交队列,一个存放待执行事件的环形队列

b. cq (completion queue): 完成队列,存放已经完成的事件的环形队列

:这两个队列是用户态和内核态之间共享的内存,使用mmap实现

c. sqe :sq中的一项

struct io_uring_sqe { // sq中的一项

__u64 user_data; // 8 btyes
}
d. cqe:cq中的一项
struct io_uring_cqe { // cq中的一项
__u64 user_data; // 8 btyes
res; // 任务函数的返回值 eg. recv返回收到数据的长度、accept返回clientfd
}

io_uring基本工作流程如下:(参照上图)

1.初始化ring,其中包含了sq和cq

2.向提交队列sq里注册(提交)一个事件(任务)sqe

3.将sq中的若干事件(任务)提交给内核处理(submit)

4.内核将处理成功的事件cqe放到完成队列cq里

5.用户取出cq里的完成事件cqe

需要注意的是:

注 : sq里注册的任务sqe一旦交给内核处理,就从sq里被移除了,是一次性的,和epoll不同

注 : cq里的 completion 事件cqe,不主动清除会一直存在

注 : cq队列里都是执行成功的任务,没有执行失败的任务,不用判断recv()返回值 < 0的情况

接下来进行接口分析

io_uring 的三个系统调用liburing 中的封装

  • io_uring_setup (初始化ring)被封装为 io_uring_queue_init_params(entries, &ring, &params);
  • io_uring_register(从sqe向sq里注册一个任务sqe)被封装为 io_uring_prep_recv,io_uring_prep_accept等
  • io_uring_enter(将sq里的任务提交给内核处理)被封装为 io_uring_submit(&ring)

初始化:

struct io_uring_params params; // 用来初始化 ring
memset(&params, 0, sizeof(params)); // 初始化 params
struct io_uring ring; // sq and cq 两个环形队列
// 下面使用了系统调用io_uring_setup : 初始化ring
io_uring_queue_init_params(1024, &ring, &params); // 初始化两个环形队列:submition queue 和 completion queue

注册(添加)任务(事件):

struct io_uring_sqe *sqe = io_uring_get_sqe(ring); // 获取sq中下一个可用的位置sqe
  struct user_data user_data = {  // sqe->user_data
    .fd = sockfd,
    .event = EVENT_ACCEPT,
  };
  
  io_uring_prep_accept(sqe, sockfd, (struct sockaddr*)addr, addrlen, flags); // 向sq里加入一个任务sqe
  memcpy(&sqe->user_data, &user_data, sizeof(struct user_data)); // 设置任务sqe的属性

提交事件给内核处理:

// 将sq里的任务提交给内核处理
        io_uring_submit(&ring);

获取完成的事件:

struct io_uring_cqe *cqe; // 指向cqe(cq中的一项)的指针
        // 等待sq里的任务完成,并返回一个结果
        io_uring_wait_cqe(&ring, &cqe); // 这个函数会阻塞当前线程,直到至少一个 cqe 可用为止。
                                        // 一旦有一个 cqe 可用,它就会将其存储在提供的指针cqe中,并返回
        
        // 从完成队列cq获取若干内核处理完成的结果,到cqes数组里,不会阻塞
        struct io_uring_cqe *cqes[128];
        int nready = io_uring_peek_batch_cqe(&ring, cqes, 128);  // 和 epoll_wait() 类似
/*
  io_uring_wait_cqe 和 io_uring_peek_batch_cqe 一个是阻塞获取一个结果,一个是非阻塞获取多个结果
*/

移除cq中的完成事件:

// 处理完结果后,将完成的任务从cq中移除
        io_uring_cq_advance(&ring, nready);

最后附上一个io_uring实现的tcpserver,注释很详细

#include <stdio.h>
#include <liburing.h>
#include <netinet/in.h>
#include <string.h>
#include <unistd.h>
/* 
 * io_uring 的三个系统调用在 liburing 中的封装:
 *
 * io_uring_setup (初始化ring) -->被封装为 io_uring_queue_init_params(entries, &ring, &params); 
 *
 * io_uring_register(从sqe向sq里注册一个任务sqe) -->被封装为 io_uring_prep_recv,io_uring_prep_send,io_uring_prep_accept
 * 
 * io_uring_enter(将sq里的任务提交给内核处理)-->被封装为 io_uring_submit(&ring)
 * 
 * 
 * 
 * 注 : sq里注册的任务sqe一旦交给内核处理,就从sq里被移除了,是一次性的,和epoll不同
 * 注 : cq里的 completion 事件cqe,不主动清除会一直存在
 * 
 * 注 : ##cq队列里都是执行成功的任务##,没有执行失败的任务,不用判断recv()返回值 < 0的情况
 * 
 * 注 : sq和cq都是环形队列,是用户态和内核态的共享内存空间,用mmap实现 (submition queue & completion queue )
 * 
 * 注 : sqe(submition queue entry), cqe(completion queue entry) --> 这两个分别代表sq、cq中的一项
 * 
 */
// 自定义事件类型 event
#define EVENT_ACCEPT    0
#define EVENT_READ    1
#define EVENT_WRITE   2
/*
 * struct io_uring_cqe { // cq中的一项
 *     ...
 *  
 *     __u64 user_data; // 8 btyes
 * 
 *     res;  // 任务函数的返回值   eg. recv返回收到数据的长度、accept返回clientfd
 *    
 * }
 * 
 * struct io_uring_sqe { // sq中的一项
 *     ...
 * 
 *     __u64 user_data; // 8 btyes
 * }
 * 
 */
// 自定义 user_data 保证 8 btyes
struct user_data {   // sqe & cqe --> user_data
  int fd;
  int event;
};
int init_server(unsigned short port) {    // create a listener sockfd
 
  int sockfd = socket(AF_INET, SOCK_STREAM, 0); 
  struct sockaddr_in serveraddr;  
  memset(&serveraddr, 0, sizeof(struct sockaddr_in)); 
  serveraddr.sin_family = AF_INET;  
  serveraddr.sin_addr.s_addr = htonl(INADDR_ANY); 
  serveraddr.sin_port = htons(port);  
  if (-1 == bind(sockfd, (struct sockaddr*)&serveraddr, sizeof(struct sockaddr))) {   
    perror("bind");   
    return -1;  
  } 
  listen(sockfd, 10);
  
  return sockfd;
}
#define ENTRIES_LENGTH    1024
#define BUFFER_LENGTH   1024
int set_event_recv(struct io_uring *ring, int sockfd,
                    void *buf, size_t len, int flags) { // 向sqe中添加一个recv任务
    struct io_uring_sqe *sqe = io_uring_get_sqe(ring); //获取sq中下一个可用的位置sqe
    struct user_data user_data = {
        .fd = sockfd,
        .event = EVENT_READ,
    };
    io_uring_prep_recv(sqe, sockfd, buf, len, flags); // 向sq里加入一个recv任务
    memcpy(&sqe->use_data, &user_data, sizeof(struct user_data)); // 设置recv任务的属性
}
int set_event_send(struct io_uring * ring, int sockfd,
                    void *buf, size_t len, int flags) { // 向sq里添加一个send任务
    struct io_uring_sqe *sqe = io_uring_get_sqe(ring); // 获取sq中下一个可用的位置sqe
    struct user_data user_data = {
        .fd = sockfd,
        .event = EVENT_WRITE,
    };
    io_uring_prep_send(sqe, sockfd, buf, len, flags); // 向sq里加入一个任务sqe
    memcpy(&sqe->user_data, &user_data, sizeof(struct user_data)); // 设置任务sqe的属性
}   
int set_event_accept(struct io_uring *ring, int sockfd, struct sockaddr *addr,
          socklen_t *addrlen, int flags) {
  struct io_uring_sqe *sqe = io_uring_get_sqe(ring); // 获取sq中下一个可用的位置sqe
  struct user_data user_data = {  // sqe->user_data
    .fd = sockfd,
    .event = EVENT_ACCEPT,
  };
  
  io_uring_prep_accept(sqe, sockfd, (struct sockaddr*)addr, addrlen, flags); // 向sq里加入一个任务sqe
  memcpy(&sqe->user_data, &user_data, sizeof(struct user_data)); // 设置任务sqe的属性
}
int main(int argc, char *argv[]) {
    // 服务端监听套接字
    unsigned short port = 9999;
  int sockfd = init_server(port);
    // 客户端套接字地址
    struct sockaddr_in clientaddr;  
  socklen_t len = sizeof(clientaddr);
    struct io_uring_params params; // 用来初始化 ring
    memset(&params, 0, sizeof(params)); // 初始化 params
    struct io_uring ring; // sq and cq 两个环形队列
    // 下面使用了系统调用io_uring_setup : 初始化ring
    io_uring_queue_init_params(1024, &ring, &params); // 初始化两个环形队列:submition queue 和 completion queue
    
    // 向提交队列sq里添加一个accept任务
    set_event_accept(&ring, sockfd, (struct sockaddr *)&clientaddr, &len, 0);
    char buffer[BUFFER_LENGTH] = {0};
    while (1) {
        // 将sq里的任务提交给内核处理
        io_uring_submit(&ring);
        // 存储内核处理完成的结果在 cq(completion queue)里
        struct io_uring_cqe *cqe; // 指向cqe(cq中的一项)的指针
        // 等待sq里的任务完成,并返回一个结果
        io_uring_wait_cqe(&ring, &cqe); // 这个函数会阻塞当前线程,直到至少一个 cqe 可用为止。
                                        // 一旦有一个 cqe 可用,它就会将其存储在提供的指针cqe中,并返回
        
        // 从完成队列cq获取若干内核处理完成的结果,到cqes数组里,不会阻塞
        struct io_uring_cqe *cqes[128];
        int nready = io_uring_peek_batch_cqe(&ring, cqes, 128);  // 和 epoll_wait() 类似
/*
  io_uring_wait_cqe 和 io_uring_peek_batch_cqe 一个是阻塞获取一个结果,一个是非阻塞获取多个结果
*/
        // 查看并处理结果
        int i = 0;
        for (i = 0; i < nready; i++) {
            struct io_uring_cqe *entries = cqes[i]; // 已经完成的事件cqe
            struct user_data result;
            // 获取完成事件的 user_data
            memcpy(&result, &entries->user_data, sizeof(struct user_data));
            if (result.event == EVENT_ACCEPT) { // accept任务完成
                // 重新在sq注册一个accept任务
                set_event_accept(&ring, sockfd, (struct sockaddr *)&clientaddr, &len, 0);
                int connfd = entries->res;// accept任务的执行结果:clientfd
                // 在sq注册一个任务:接收客户端数据
                set_event_recv(&ring, connfd, buffer, BUFFER_LENGTH, 0);
            } else if (result.event == EVENT_READ) { // read任务完成
                int ret = entries->res; // read的返回值
                if (ret == 0) { // 对方发送了fin包通知断开连接
                    close(result.fd);
                } else if (ret > 0) {
                    //  在sq注册一个任务:向客户端send数据
                    set_event_send(&ring, result.fd, buffer, ret, 0);
                } else if (result.event == EVENT_WRITE) { // write任务完成
                    int ret = entries->res; // write的返回值
                    // 在sq注册一个任务:接收客户端数据
                    set_event_recv(&ring, result.fd, buffer, BUFFER_LENGTH, 0);
                }
            }
        }
        // 处理完结果后,将完成的任务从cq中移除
        io_uring_cq_advance(&ring, nready);
    }
}


目录
相关文章
|
4天前
|
Linux 数据处理 C语言
【Linux】基础IO----系统文件IO & 文件描述符fd & 重定向(下)
【Linux】基础IO----系统文件IO & 文件描述符fd & 重定向(下)
22 0
|
4天前
|
Linux 编译器 C语言
【Linux】基础IO----理解缓冲区
【Linux】基础IO----理解缓冲区
18 0
【Linux】基础IO----理解缓冲区
|
11天前
|
缓存 网络协议 算法
【Linux系统编程】深入剖析:四大IO模型机制与应用(阻塞、非阻塞、多路复用、信号驱动IO 全解读)
在Linux环境下,主要存在四种IO模型,它们分别是阻塞IO(Blocking IO)、非阻塞IO(Non-blocking IO)、IO多路复用(I/O Multiplexing)和异步IO(Asynchronous IO)。下面我将逐一介绍这些模型的定义:
|
4天前
|
Linux C语言 C++
【Linux】基础IO----系统文件IO & 文件描述符fd & 重定向(上)
【Linux】基础IO----系统文件IO & 文件描述符fd & 重定向(上)
19 0
|
7天前
|
Java 大数据
解析Java中的NIO与传统IO的区别与应用
解析Java中的NIO与传统IO的区别与应用
|
15天前
|
存储 缓存 Java
Java零基础入门之IO流详解(二)
Java零基础入门之IO流详解(二)
|
20天前
|
Java 数据处理 开发者
揭秘Java IO流:字节流与字符流的神秘面纱!
【6月更文挑战第26天】Java IO流涵盖字节流和字符流,字节流处理二进制数据,如图像,由InputStream/OutputStream家族管理;字符流处理文本,基于Reader/Writer,适于文本文件。在文件复制示例中,字节流用FileInputStream/FileOutputStream,字符流用FileReader/FileWriter。选择流类型取决于数据类型和处理需求,文本文件优选字符流,二进制数据则选字节流。
|
8天前
|
Java 大数据
解析Java中的NIO与传统IO的区别与应用
解析Java中的NIO与传统IO的区别与应用
|
15天前
|
存储 缓存 Java
Java零基础入门之IO流详解(一)
Java零基础入门之IO流详解(一)
|
19天前
|
存储 Java
杭州 java IO流详解(借鉴-侵-删)
杭州 java IO流详解(借鉴-侵-删)
16 0