快捷搜索:  汽车  科技

io操作底层用的什么设计模式(高性能异步io机制)

io操作底层用的什么设计模式(高性能异步io机制)接下来,进行同步 epoll 与异步 io_uring 服务器的测试对比,代码见 liburing 测试代码epoll 与 io_uring 事件的区别编写的 posix.fio 配置文件如下[global] thread=1 group_reporting=1 direct=1 verify=0 time_based=1 runtime=10 bs=16K size=16384 iodepth=64 rw=randwrite filename=Cat ioengine=io_uring [test] stonewall description="variable bs"实验结果:iops:psync 8k io_uring 19.0k,由此可以看出异步 io 的性能优势。服务器性能测试方法

io_uring 是 Linux 内核 5.10 引入的异步 io 接口。相比起用户态的DPDK、SPDK,io_uring 作为内核的一部分,通过 mmap 的方式实现用户和内核共享内存,并基于 memory barrier 在这块内存上实现了两个无锁环形队列: submission queue ring(sq) 和 completion queue ring(cq)。 sq 用于用户程序向内核提交 IO 任务,内核执行完成的任务会放入cq,用户程序从 cq 获取结果。在提交任务和返回任务结果时,用户程序和内核共用环形队列中的数据,不再需要额外的数据拷贝。此外,io_uring 还提供了两种轮询 Polling 模式,可以避免提交任务时的系统调用,以及io完成后的中断通知。

1、性能测试1.1、FIO

iops 是指单位时间内系统能处理的I/O请求数量,用于存储设备性能测试。这里我们使用硬盘性能辅助测试工具 FIO,来直观感受异步 io: io_uring 的性能优势。

# 安装 fio sudo apt install fio # 运行方式 fio job_file

需要通过编写一个配置文件来预定义 FIO 将要以什么样的模式来执行任务。

FIO 的基本参数:

  • rw readwrite:定义 IO 类型。随机读 randread、随机写 randwrite、顺序读 read、顺序写 write、顺序读写 rw readwrite ,随机混合读写 randrw
  • bs blocksize:IO 的块大小。默认 4k
  • size: IO 传输的数据大小
  • ioengine:IO 引擎。同步模式psync、异步模式io_uring
  • iodepth:I/O 引擎若使用异步模式,保持队列深度
  • direct: 是否使用非缓冲 io ,默认 false 缓冲 io

编写的 posix.fio 配置文件如下

[global] thread=1 group_reporting=1 direct=1 verify=0 time_based=1 runtime=10 bs=16K size=16384 iodepth=64 rw=randwrite filename=Cat ioengine=io_uring [test] stonewall description="variable bs"

实验结果:iops:psync 8k io_uring 19.0k,由此可以看出异步 io 的性能优势。

1.2、rust_echo_benc

服务器性能测试方法

  • 连接数
  • 每个请求连接的大小
  • 持续时间

epoll 与 io_uring 事件的区别

  • epoll 设置完后,不更改。
  • io_uring 设置一次,触发一次。

接下来,进行同步 epoll 与异步 io_uring 服务器的测试对比,代码见 liburing 测试代码

# 安装 rust_echo_benc git clone https://github.com/haraldh/rust_echo_bench.git cargo run --release # 测试 cargo run --release -- --address "127.0.0.1:9999" --number 1000 --duration 60 --length 512

实验结果:在网络 io 方面,io_uring并不明显。在磁盘 io 方面,io_uring 具有一定的优势。

2、io_uring

io_uring 提供了三个系统调用接口 io_uring_setup、io_uring_enter、io_uring_register

2.1、io_uring_setup

在 kernel 中创建:

  • 提交队列 SQ:里面每一项是 sqe(submission queue event),描述1个任务
  • 完成队列 CQ:里面每一项是 cqe(completion queue event),描述1个任务返回结果
  • 提交队列项 SQEs 数组(Submission Queue Entries)

io操作底层用的什么设计模式(高性能异步io机制)(1)

SQ 和 CQ 采用 Ringbuffer 的结构,有 head 和 tail 两个成员,head = tail 时队列为空。每个节点保存的是 SQEs 数组的偏移量,实际的请求保存在 SQEs 数组中,这样就可以批量提交一组 SQEs 上不连续的请求。SQ 和 CQ 本身没有提供锁等同步机制,向 SQ中放入 sqe,从 CQ 中取出 cqe,都需要通过 memory barrier 来实现。

函数返回1个 fd 用于 io_uring 管理。用户将 fd 以 mmap 的方式映射到内存,实现了用户态和内核态的共享内存。

/* - 参数1 entries:期望的 sq 长度。默认cq长度是sq的两倍 - 参数2 params: 配置io_uring,内核返回的 sq/cq 配置信息也通过它带回来 */ int io_uring_setup(unsigned entries struct io_uring_params *params) struct io_uring_params { __u32 sq_entries; __u32 cq_entries; __u32 flags; __u32 sq_thread_cpu; __u32 sq_thread_idle; __u32 resv[5]; struct io_sqring_offsets sq_off; struct io_cqring_offsets cq_off; };相关视频推荐

io管理只有epoll吗,io_uring是不是更好的选择

linux多线程之epoll原理剖析与reactor原理及应用

linux下的epoll实战揭秘——支撑亿级IO的底层基石

需要C/C Linux服务器架构师学习资料加qun812855908获取(资料包括C/C ,Linux,golang技术,Nginx,ZeroMQ,MySQL,Redis,fastdfs,MongoDB,ZK,流媒体,CDN,P2P,K8S,Docker,TCP/IP,协程,DPDK,ffmpeg等),免费分享

io操作底层用的什么设计模式(高性能异步io机制)(2)

2.2、io_uring_enter

调用时,执行两个操作

  • 提交 IO 请求:把 sqe 的索引尾插到 SQ 中,调用io_uring_enter提交到内核
  • 等待 IO 完成:内核将完成的 IO 放到 CQ 中,用户轮询 CQ 来等待结果

io操作底层用的什么设计模式(高性能异步io机制)(3)

/* - 参数1 fd:io_uring_setup返回的fd - 参数2 to_submit: 一次提交多少个 sqe 到内核 - 参数3 min_complete: 要求内核至少等待min_complete个任务完成再返回 - 参数4 flags:接口控制行为,IORING_ENTER_GETEVENTS */ int io_uring_enter(unsigned int fd u32 to_submit u32 min_complete u32 flags);2.3、io_uring_register

注册用于异步 I/O 的文件或用户缓冲区

对于文件, 保持内核长时间持有该文件的索引。每次通过 sqe 向内核传递一个 fd,内核都需要通过 fd 找到对应的文件索引,完成该sqe 处理后,则将该索引释放。对于高 iops 的场景,这个开销会拖慢请求的速度。通过预先注册一组已经打开的文件。

对于缓冲区,保持内存的长期映射。内核在读写前进行page map,读写完成后,执行unmap。类似的,通过预注册,来避免多次的 map 和 unmap。

/* - 参数1 fd:io_uring_setup返回的fd - 参数2 opcode: 注册类型。 文件类型: IORING_REGISTER_FILES; 用户缓冲类型 buffer: IORING_REGISTER_BUFFERS - 参数3 arg: 文件类型: 指向一个fd数组; 用户缓冲类型:指向一个struct iovec的数组。 - 参数4 nr_args:arg数组的长度 */ int io_uring_register(unsigned int fd unsigned int opcode void *arg unsigned int nr_args);2.4、使用方法:cat 程序为例

接下来,基于 io_uring 的系统调用接口进行封装,实现自定义的 uring_cat 程序

// gcc -o uring_cat uring_cat.c // ./uring_cat filename #include <stdio.h> #include <stdlib.h> #include <sys/stat.h> #include <sys/ioctl.h> #include <sys/syscall.h> #include <sys/mman.h> #include <sys/uio.h> #include <linux/fs.h> #include <fcntl.h> #include <unistd.h> #include <string.h> #include <linux/io_uring.h> #define URING_QUEUE_DEPTH 1024 #define BLOCK_SZ 1024 // sqring struct app_io_sq_ring { unsigned *head; unsigned *tail; unsigned *ring_mask; unsigned *ring_entries; unsigned *flags; unsigned *array; }; // cqring struct app_io_cq_ring { unsigned *head; unsigned *tail; unsigned *ring_mask; unsigned *ring_entries; struct io_uring_cqe *cqes; }; // 提交器: cq sq sqe struct submitter { int ring_fd; struct app_io_sq_ring sq_ring; struct app_io_cq_ring cq_ring; struct io_uring_sqe *sqes; }; ------------------- struct file_info { off_t file_sz; struct iovec iovecs[]; }; ------------------- // 利用系统调用执行 io_uring_setup 流程 // 1、int 0x80 中断信号 // 2、mv arg1 eax // 3、mv arg2 ebx // 4、call sys_call_table: sys_call_table[__NR_io_uring_setup] int io_uring_setup(unsigned entries struct io_uring_params *p) { return (int) syscall(__NR_io_uring_setup entries p); } int io_uring_enter(int ring_fd unsigned int to_submit unsigned int min_complete unsigned int flags) { return (int) syscall(__NR_io_uring_enter ring_fd to_submit min_complete flags NULL 0); } int app_setup_uring(struct submitter *s) { struct io_uring_params p; memset(&p 0 sizeof(p)); // 创建sq cq sqes s->ring_fd = io_uring_setup(URING_QUEUE_DEPTH &p); if (s->ring_fd < 0) return -1; // 获取初始的sq cq的大小,sq_off cq_off起始偏移地址 int sring_sz = p.sq_off.array p.sq_entries * sizeof(unsigned); int cring_sz = p.cq_off.cqes p.cq_entries * sizeof(struct io_uring_cqe); // io_uring特性:IORING_FEAT_SINGLE_MMAP:内核通过一次mmap完成sq cq的映射 // 即sq,cq共用1块内存,则两者大小必须设置相同 if (p.features & IORING_FEAT_SINGLE_MMAP) { if (cring_sz > sring_sz) { sring_sz = cring_sz; } cring_sz = sring_sz; } // 1、将 sq 的映射到用户空间,sq_ptr 指向sq首地址 void *sq_ptr = mmap(0 sring_sz PROT_READ|PROT_WRITE MAP_SHARED|MAP_POPULATE s->ring_fd IORING_OFF_SQ_RING); if (sq_ptr == MAP_FAILED) return -1; // 2、将 cq 的映射到用户空间,cq_ptr 指向cq首地址 void *cq_ptr; // 若共用一块内存,则两个指针指向相同 if (p.features & IORING_FEAT_SINGLE_MMAP) { cq_ptr = sq_ptr; } else { // 若使用两块内存,则重新对cq进行mmap, cq_ptr = mmap(0 sring_sz PROT_READ|PROT_WRITE MAP_SHARED|MAP_POPULATE s->ring_fd IORING_OFF_CQ_RING); if (cq_ptr == MAP_FAILED) return -1; } struct app_io_sq_ring *sring = &s->sq_ring; struct app_io_cq_ring *cring = &s->cq_ring; sring->head = sq_ptr p.sq_off.head; sring->tail = sq_ptr p.sq_off.tail; sring->ring_mask = sq_ptr p.sq_off.ring_mask; sring->ring_entries = sq_ptr p.sq_off.ring_entries; sring->flags = sq_ptr p.sq_off.flags; sring->array = sq_ptr p.sq_off.array; // 3、将 seqs 映射到用户空间 s->sqes = mmap(0 p.sq_entries * sizeof(struct io_uring_sqe) PROT_READ | PROT_WRITE MAP_SHARED | MAP_POPULATE s->ring_fd IORING_OFF_SQES); if (s->sqes == MAP_FAILED) { return 1; } cring->head = cq_ptr p.cq_off.head; cring->tail = cq_ptr p.cq_off.tail; cring->ring_mask = cq_ptr p.cq_off.ring_mask; cring->ring_entries = cq_ptr p.cq_off.ring_entries; cring->cqes = cq_ptr p.cq_off.cqes; return 0; } off_t get_file_size(int fd) { struct stat st; if(fstat(fd &st) < 0) { perror("fstat"); return -1; } if (S_ISBLK(st.st_mode)) { unsigned long long bytes; if (ioctl(fd BLKGETSIZE64 &bytes) != 0) { perror("ioctl"); return -1; } return bytes; } else if (S_ISREG(st.st_mode)) return st.st_size; return -1; } void output_to_console(char *buf int len) { while (len--) { fputc(*buf stdout); } } void read_from_cq(struct submitter *s) { struct file_info *fi; struct app_io_cq_ring *cring = &s->cq_ring; struct io_uring_cqe *cqe; unsigned head = *cring->head; while (1) { //read_barrier(); if (head == *cring->tail) break; cqe = &cring->cqes[head & *s->cq_ring.ring_mask]; fi = (struct file_info*)cqe->user_data; if (cqe->res < 0) { fprintf(stderr "Error: %d\n" cqe->res); } int blocks = fi->file_sz / BLOCK_SZ; if (fi->file_sz % BLOCK_SZ) blocks ; int i = 0; while ( i < blocks) { output_to_console(fi->iovecs[i].iov_base fi->iovecs[i].iov_len); printf("------------------------i : %d blocks: %d\n" i blocks); } head ; printf("head: %d tail: %d blocks: %d\n" head *cring->tail blocks); } *cring->head = head; printf("exit read_from_cq\n"); //write_barrier(); } int submit_to_sq(char *file_path struct submitter *s) { int filefd = open(file_path O_RDONLY); if (filefd < 0) { return -1; } struct app_io_sq_ring *sring = &s->sq_ring; off_t filesz = get_file_size(filefd); if (filesz < 0) return -1; off_t bytes_remaining = filesz; int blocks = filesz / BLOCK_SZ; if (filesz % BLOCK_SZ) blocks ; struct file_info *fi = malloc(sizeof(struct file_info) sizeof(struct iovec) * blocks); if (!fi) return -2; fi->file_sz = filesz; unsigned current_block; while (bytes_remaining) { off_t bytes_to_read = bytes_remaining; if (bytes_to_read > BLOCK_SZ) bytes_to_read = BLOCK_SZ; fi->iovecs[current_block].iov_len = bytes_to_read; void *buf; if (posix_memalign(&buf BLOCK_SZ BLOCK_SZ)) { return 1; } fi->iovecs[current_block].iov_base = buf; current_block ; bytes_remaining -= bytes_to_read; } unsigned next_tail = 0 tail = 0 index = 0; next_tail = tail = *sring->tail; next_tail ; index = tail & *s->sq_ring.ring_mask; struct io_uring_sqe *sqe = &s->sqes[index]; sqe->fd = filefd; sqe->flags = 0; sqe->opcode = IORING_OP_READV; sqe->addr = (unsigned long)fi->iovecs; sqe->len = blocks; sqe->off = 0; sqe->user_data = (unsigned long long)fi; sring->array[index] = index; tail = next_tail; if (*sring->tail != tail) { *sring->tail = tail; } int ret = io_uring_enter(s->ring_fd 1 1 IORING_ENTER_GETEVENTS); if (ret < 0) { return 1; } return 0; } int main(int argc char *argv[]) { struct submitter *s = malloc(sizeof(struct submitter)); if (!s) { perror("malloc"); return -1; } memset(s 0 sizeof(struct submitter)); // 1、setup if (app_setup_uring(s)) return 1; int i = 1; for (i = 1;i < argc;i ) { // 2、submit if (submit_to_sq(argv[i] s)) { //fprintf(stderr "Error reading file\n"); return 1; } read_from_cq(s); } return 0; }3、liburing

由于 io_uring 使用起来比较麻烦,作者封装了 io_uring 接口,创作了 liburing 库。

# 安装 liburing git clone https://github.com/axboe/liburing.git ./configure make && make install

3.1、liburing api

// 初始化io_uring,内部调用io_uring_setup int io_uring_queue_init_params(unsigned entries struct io_uring *ring struct io_uring_params *p); // 提交 sq 到内核,内核完成后移动到 cq,内部调用 io_uring_enter // 1、提交io请求:将sqe的偏移信息加入到sq,提交sq到内核,不阻塞等待其完成 // 2、等待io完成:内核在io完成后,自动将sqe的偏移信息加入到cq int io_uring_submit(struct io_uring *ring); // 等待io完成,获取cqe // 阻塞等待 unsigned io_uring_peek_batch_cqe(struct io_uring *ring struct io_uring_cqe **cqes unsigned count); // 不阻塞等待 int io_uring_wait_cqes(struct io_uring *ring struct io_uring_cqe **cqe_ptr unsigned wait_nr struct __kernel_timespec *ts sigset_t *sigmask); // 轮询 cq 队列,将 cq 队首后移动 nr 个 static inline void io_uring_cq_advance(struct io_uring *ring unsigned nr) // 和libaio封装的io_prep_writev一样 static inline void io_uring_prep_writev(struct io_uring_sqe *sqe int fd const struct iovec *iovecs unsigned nr_vecs off_t offset) // 和libaio封装的io_prep_readv一样 static inline void io_uring_prep_readv(struct io_uring_sqe *sqe int fd const struct iovec *iovecs unsigned nr_vecs off_t offset) // 销毁 io void io_uring_queue_exit(struct io_uring *ring);3.2、测试代码

利用 liburing 编写的简单测试 iouring_server

// gcc -o iouring_server iouring_server.c -luring #include <liburing.h> #include <stdio.h> #include <string.h> #include <sys/socket.h> #include <netinet/in.h> #include <unistd.h> #define ENTRIES_LENGTH 4096 #define MAX_CONNECTIONS 1024 #define BUFFER_LENGTH 1024 char buf_table[MAX_CONNECTIONS][BUFFER_LENGTH] = {0}; // 传递的事件 enum { READ WRITE ACCEPT }; // 连接信息 struct conninfo { int connfd; // fd int type; // 事件类型 }; void set_read_event(struct io_uring *ring int fd void *buf size_t len int flags) { struct io_uring_sqe *sqe = io_uring_get_sqe(ring); // io_uring 读事件 io_uring_prep_recv(sqe fd buf len flags); struct conninfo ci = { .connfd = fd .type = READ }; memcpy(&sqe->user_data &ci sizeof(struct conninfo)); } void set_write_event(struct io_uring *ring int fd const void *buf size_t len int flags) { struct io_uring_sqe *sqe = io_uring_get_sqe(ring); // io_uring 写事件 io_uring_prep_send(sqe fd buf len flags); struct conninfo ci = { .connfd = fd .type = WRITE }; memcpy(&sqe->user_data &ci sizeof(struct conninfo)); } void set_accept_event(struct io_uring *ring int fd struct sockaddr *cliaddr socklen_t *clilen unsigned flags) { // 获取 sq 队列的空 sqe struct io_uring_sqe *sqe = io_uring_get_sqe(ring); // io_uring的accept事件:将fd放入到sqe里 io_uring_prep_accept(sqe fd cliaddr clilen flags); // 用于回调函数 struct conninfo ci = { .connfd = fd .type = ACCEPT }; memcpy(&sqe->user_data &ci sizeof(struct conninfo)); } int main() { int listenfd = socket(AF_INET SOCK_STREAM 0); if (listenfd == -1) return -1; struct sockaddr_in servaddr clientaddr; servaddr.sin_family = AF_INET; servaddr.sin_addr.s_addr = htonl(INADDR_ANY); servaddr.sin_port = htons(9999); if (-1 == bind(listenfd (struct sockaddr*)&servaddr sizeof(servaddr))) { return -2; } listen(listenfd 10); struct io_uring_params params; memset(¶ms 0 sizeof(params)); // 初始化队列,内部调用io_uring_setup struct io_uring ring; io_uring_queue_init_params(ENTRIES_LENGTH &ring ¶ms); socklen_t clilen = sizeof(clientaddr); set_accept_event(&ring listenfd (struct sockaddr*)&clientaddr &clilen 0); while (1) { // 封装 io_uring_enter // 1、提交io请求:将sqe的偏移信息加入到sq,提交sq到内核,不阻塞等待其完成 // 2、等待io完成:内核在io完成后,自动将sqe的偏移信息加入到cq io_uring_submit(&ring); // 从获取 cqe 的两种方式 // 1、阻塞等待io完成,获取 cqe struct io_uring_cqe *cqe; int ret = io_uring_wait_cqe(&ring &cqe); // 2、不阻塞等待io完成,没有cqe返回错误,获取 cqe struct io_uring_cqe *cqes[10]; int cqecount = io_uring_peek_batch_cqe(&ring cqes 10); int i = 0; unsigned count = 0; for (i = 0; i < cqecount; i) { cqe = cqes[i]; count ; struct conninfo ci; memcpy(&ci &cqe->user_data sizeof(ci)); if (ci.type == ACCEPT) { int connfd = cqe->res; char *buffer = buf_table[connfd]; set_read_event(&ring connfd buffer 1024 0); // io_uring 设置一次,触发一次 set_accept_event(&ring listenfd (struct sockaddr*)&clientaddr &clilen 0); } else if (ci.type == READ) { int bytes_read = cqe->res; if (bytes_read == 0) { close(ci.connfd); } else if (bytes_read < 0) { } else { char *buffer = buf_table[ci.connfd]; set_write_event(&ring ci.connfd buffer bytes_read 0); } } else if (ci.type == WRITE) { char *buffer = buf_table[ci.connfd]; set_read_event(&ring ci.connfd buffer 1024 0); } } // cq队列一次轮询完成后,因为cqe的取出,需要调整队首的位置,以便下次使用 io_uring_cq_advance(&ring count); } }

猜您喜欢: