新一代异步IO框架 io_uring ｜ 得物技术

#include <unistd.h>                                                  #include <sys/types.h> #include <sys/stat.h> #include <fcntl.h> 
//  返回值:成功返回新分配的文件描述符，出错返回-1并设置errno                                                  int open(const char *pathname, int flags);int open(const char *pathname, int flags, mode_t mode);
// 返回值:成功返回0，出错返回-1并设置errno                                                  int close(int fd); 
// 返回值:成功返回读取的字节数，出错返回-1并设置errno，如果在调read之前已到达文件末尾，则这次read返回0                                                  ssize_t read(int fd, void *buf, size_t count);   
// 返回值:成功返回写入的字节数，出错返回-1并设置errno                                                     ssize_t write(int fd, const void *buf, size_t count);

// 成功执行时，mmap()返回被映射区的指针。失败时，mmap()返回MAP_FAILED[其值为(void *)-1]，// error被设为以下的某个值：// 1 EACCES：访问出错// 2 EAGAIN：文件已被锁定，或者太多的内存已被锁定// 3 EBADF：fd不是有效的文件描述词// 4 EINVAL：一个或者多个参数无效// 5 ENFILE：已达到系统对打开文件的限制// 6 ENODEV：指定文件所在的文件系统不支持内存映射// 7 ENOMEM：内存不足，或者进程已超出最大内存映射数量// 8 EPERM：权能不足，操作不允许// 9 ETXTBSY：已写的方式打开文件，同时指定MAP_DENYWRITE标志//10 SIGSEGV：试着向只读区写入//11 SIGBUS：试着访问不属于进程的内存区void *mmap(void *start, size_t length, int prot, int flags, int fd, off_t offset);
// 成功执行时，munmap()返回0。失败时，munmap返回-1，error返回标志和mmap一致；// 该调用在进程地址空间中解除一个映射关系，addr是调用mmap()时返回的地址，len是映射区的大小；int munmap( void * addr, size_t len )
// 进程在映射空间的对共享内容的改变并不直接写回到磁盘文件中，往往在调用munmap（）后才执行该操作。// 如果期望内存的数据变化能够立刻反应到磁盘上，可以通过调用msync()实现。int msync( void *addr, size_t len, int flags )

// 创建一个 SQ 和一个 CQ，queue size 至少 entries 个元素// 返回一个文件描述符，随后用于在这个 io_uring 实例上执行操作。// 参数p 有两个作用：// 1.作为入参：应用用来配置 io_uring 的一些行为// 2.作为出参：内核返回的 SQ/CQ 地址信息等也通过它带回来。int io_uring_setup(u32 entries, struct io_uring_params *p);

// 注册用于异步 I/O 的文件或用户缓冲区（files or user buffers）：int io_uring_register(unsigned int fd, unsigned int opcode, void *arg, unsigned int nr_args);

// 用于初始化和完成I/O，使用共享的 SQ 和 CQ。单次调用同时提交新的 I/O 请求和等待 I/O 完成操作// fd 是 io_uring_setup() 返回的文件描述符；// to_submit 指定了 SQ 中提交的 I/O 数量；// 默认模式下如果指定了 min_complete，会等待这个数量的 I/O 事件完成再返回；// 轮询模式（2种）：//   0：要求内核返回当前以及完成的所有 events，无阻塞；//   非0：如果有事件完成，内核仍然立即返回；如果没有完成事件，内核会 poll，等待指定的次数完成，或者这个进程的时间片用完。int io_uring_enter(unsigned int fd, unsigned int to_submit, unsigned int min_complete, unsigned int flags, sigset_t *sig);

// io_uring 结构体中包含需要使用到的 SQ和CQ ，以及需要关联的文件FD， 和相关的配置参数falgs; struct io_uring {        struct io_uring_sq sq;        struct io_uring_cq cq;        unsigned flags;        int ring_fd;};
struct io_uring_sq {        unsigned *khead;        unsigned *ktail;        unsigned *kring_mask;        unsigned *kring_entries;        unsigned *kflags;        unsigned *kdropped;        unsigned *array;        struct io_uring_sqe *sqes;
        unsigned sqe_head;        unsigned sqe_tail;
        size_t ring_sz;        void *ring_ptr;};
struct io_uring_cq {        unsigned *khead;        unsigned *ktail;        unsigned *kring_mask;        unsigned *kring_entries;        unsigned *koverflow;        struct io_uring_cqe *cqes;
        size_t ring_sz;        void *ring_ptr;};
// 用户初始化 io_uring。该方法中包含了内存空间的初始化以及mmap 调用,entries：队列深度 int io_uring_queue_init(unsigned entries, struct io_uring *ring, unsigned flags);
// 为了提交IO请求，需要获取里面queue的一个空闲项struct io_uring_sqe *io_uring_get_sqe(struct io_uring *ring);
// 非系统调用，准备阶段，和libaio封装的io_prep_writev一样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一样void io_uring_prep_readv(struct io_uring_sqe *sqe, int fd, const struct iovec *iovecs, unsigned nr_vecs, off_t offset)
// 提交sq的entry，不会阻塞等到其完成，内核在其完成后会自动将sqe的偏移信息加入到cq，在提交时需要加锁int io_uring_submit(struct io_uring *ring);
// 提交sq的entry，阻塞等到其完成，在提交时需要加锁。int io_uring_submit_and_wait(struct io_uring *ring, unsigned wait_nr);
// 非系统调用 遍历时，可以获取cqe的datavoid *io_uring_cqe_get_data(const struct io_uring_cqe *cqe)
// 清理io_uringvoid io_uring_queue_exit(struct io_uring *ring);

#include <stdio.h>#include <fcntl.h>#include <string.h>#include <stdlib.h>#include <sys/types.h>#include <sys/stat.h>#include <unistd.h>#include "liburing.h"
#define QD        4int main(int argc, char *argv[]){        struct io_uring ring;        int i, fd, ret, pending, done;        struct io_uring_sqe *sqe;        struct io_uring_cqe *cqe;        struct iovec *iovecs;        struct stat sb;        ssize_t fsize;        off_t offset;        void *buf;
        ret = io_uring_queue_init(QD, &ring, 0);        if (ret < 0) {                fprintf(stderr, "queue_init: %s\n", strerror(-ret)); return 1;        }        fd = open(argv[1], O_RDONLY | O_DIRECT);        fsize = 0;        iovecs = calloc(QD, sizeof(struct iovec));        for (i = 0; i < QD; i++) {                if (posix_memalign(&buf, 4096, 4096))                        return 1;                iovecs[i].iov_base = buf;                iovecs[i].iov_len = 4096;                fsize += 4096;        }
        offset = 0;        i = 0;        do {                sqe = io_uring_get_sqe(&ring);                if (!sqe)  break;                io_uring_prep_readv(sqe, fd, &iovecs[i], 1, offset);                offset += iovecs[i].iov_len;                i++;                if (offset > sb.st_size) break;        } while (1);
        ret = io_uring_submit(&ring);        if (ret < 0) {                fprintf(stderr, "io_uring_submit: %s\n", strerror(-ret)); return 1;        } else if (ret != i) {                fprintf(stderr, "io_uring_submit submitted less %d\n", ret); return 1;        }
        done = 0;        pending = ret;        fsize = 0;        for (i = 0; i < pending; i++) {                ret = io_uring_wait_cqe(&ring, &cqe);                if (ret < 0) {                        fprintf(stderr, "io_uring_wait_cqe: %s\n", strerror(-ret));return 1;                }                done++;                ret = 0;                if (cqe->res != 4096 && cqe->res + fsize != sb.st_size) {                        fprintf(stderr, "ret=%d, wanted 4096\n", cqe->res);                        ret = 1;                }                fsize += cqe->res;                io_uring_cqe_seen(&ring, cqe);                if (ret) break;        }        printf("Submitted=%d, completed=%d, bytes=%lu\n", pending, done, (unsigned long) fsize);        close(fd);        io_uring_queue_exit(&ring);        return 0;}

#include <errno.h>#include <fcntl.h>#include <netinet/in.h>#include <stdio.h>#include <stdlib.h>#include <string.h>#include <strings.h>#include <sys/poll.h>#include <sys/socket.h>#include <unistd.h>
#include "liburing.h"
#define MAX_CONNECTIONS     4096#define BACKLOG             512#define MAX_MESSAGE_LEN     2048#define BUFFERS_COUNT       MAX_CONNECTIONS
void add_accept(struct io_uring *ring, int fd, struct sockaddr *client_addr, socklen_t *client_len, unsigned flags);void add_socket_read(struct io_uring *ring, int fd, unsigned gid, size_t size, unsigned flags);void add_socket_write(struct io_uring *ring, int fd, __u16 bid, size_t size, unsigned flags);void add_provide_buf(struct io_uring *ring, __u16 bid, unsigned gid);
enum {    ACCEPT,    READ,    WRITE,    PROV_BUF,};
typedef struct conn_info {    __u32 fd;    __u16 type;    __u16 bid;} conn_info;
char bufs[BUFFERS_COUNT][MAX_MESSAGE_LEN] = {0};int group_id = 1337;
int main(int argc, char *argv[]) {    if (argc < 2) {        printf("Please give a port number: ./io_uring_echo_server [port]\n");        exit(0);    }
    // some variables we need    int portno = strtol(argv[1], NULL, 10);    struct sockaddr_in serv_addr, client_addr;    socklen_t client_len = sizeof(client_addr);
    // setup socket    int sock_listen_fd = socket(AF_INET, SOCK_STREAM, 0);    const int val = 1;    setsockopt(sock_listen_fd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val));
    memset(&serv_addr, 0, sizeof(serv_addr));    serv_addr.sin_family = AF_INET;    serv_addr.sin_port = htons(portno);    serv_addr.sin_addr.s_addr = INADDR_ANY;
    // bind and listen    if (bind(sock_listen_fd, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) {        perror("Error binding socket...\n");        exit(1);    }    if (listen(sock_listen_fd, BACKLOG) < 0) {        perror("Error listening on socket...\n");        exit(1);    }    printf("io_uring echo server listening for connections on port: %d\n", portno);
    // initialize io_uring    struct io_uring_params params;    struct io_uring ring;    memset(&params, 0, sizeof(params));
    if (io_uring_queue_init_params(2048, &ring, &params) < 0) {        perror("io_uring_init_failed...\n");        exit(1);    }
    // check if IORING_FEAT_FAST_POLL is supported    if (!(params.features & IORING_FEAT_FAST_POLL)) {        printf("IORING_FEAT_FAST_POLL not available in the kernel, quiting...\n");        exit(0);    }
    // check if buffer selection is supported    struct io_uring_probe *probe;    probe = io_uring_get_probe_ring(&ring);    if (!probe || !io_uring_opcode_supported(probe, IORING_OP_PROVIDE_BUFFERS)) {        printf("Buffer select not supported, skipping...\n");        exit(0);    }    free(probe);
    // register buffers for buffer selection    struct io_uring_sqe *sqe;    struct io_uring_cqe *cqe;
    sqe = io_uring_get_sqe(&ring);    io_uring_prep_provide_buffers(sqe, bufs, MAX_MESSAGE_LEN, BUFFERS_COUNT, group_id, 0);
    io_uring_submit(&ring);    io_uring_wait_cqe(&ring, &cqe);    if (cqe->res < 0) {        printf("cqe->res = %d\n", cqe->res);        exit(1);    }    io_uring_cqe_seen(&ring, cqe);
    // add first accept SQE to monitor for new incoming connections    add_accept(&ring, sock_listen_fd, (struct sockaddr *)&client_addr, &client_len, 0);
    // start event loop    while (1) {        io_uring_submit_and_wait(&ring, 1);        struct io_uring_cqe *cqe;        unsigned head;        unsigned count = 0;
        // go through all CQEs        io_uring_for_each_cqe(&ring, head, cqe) {            ++count;            struct conn_info conn_i;            memcpy(&conn_i, &cqe->user_data, sizeof(conn_i));
            int type = conn_i.type;            if (cqe->res == -ENOBUFS) {                fprintf(stdout, "bufs in automatic buffer selection empty, this should not happen...\n");                fflush(stdout);                exit(1);            } else if (type == PROV_BUF) {                if (cqe->res < 0) {                    printf("cqe->res = %d\n", cqe->res);                    exit(1);                }            } else if (type == ACCEPT) {                int sock_conn_fd = cqe->res;                // only read when there is no error, >= 0                if (sock_conn_fd >= 0) {                    add_socket_read(&ring, sock_conn_fd, group_id, MAX_MESSAGE_LEN, IOSQE_BUFFER_SELECT);                }
                // new connected client; read data from socket and re-add accept to monitor for new connections                add_accept(&ring, sock_listen_fd, (struct sockaddr *)&client_addr, &client_len, 0);            } else if (type == READ) {                int bytes_read = cqe->res;                int bid = cqe->flags >> 16;                if (cqe->res <= 0) {                    // read failed, re-add the buffer                    add_provide_buf(&ring, bid, group_id);                    // connection closed or error                    close(conn_i.fd);                } else {                    // bytes have been read into bufs, now add write to socket sqe                    add_socket_write(&ring, conn_i.fd, bid, bytes_read, 0);                }            } else if (type == WRITE) {                // write has been completed, first re-add the buffer                add_provide_buf(&ring, conn_i.bid, group_id);                // add a new read for the existing connection                add_socket_read(&ring, conn_i.fd, group_id, MAX_MESSAGE_LEN, IOSQE_BUFFER_SELECT);            }        }
        io_uring_cq_advance(&ring, count);    }}
void add_accept(struct io_uring *ring, int fd, struct sockaddr *client_addr, socklen_t *client_len, unsigned flags) {    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);    io_uring_prep_accept(sqe, fd, client_addr, client_len, 0);    io_uring_sqe_set_flags(sqe, flags);
    conn_info conn_i = {        .fd = fd,        .type = ACCEPT,    };    memcpy(&sqe->user_data, &conn_i, sizeof(conn_i));}
void add_socket_read(struct io_uring *ring, int fd, unsigned gid, size_t message_size, unsigned flags) {    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);    io_uring_prep_recv(sqe, fd, NULL, message_size, 0);    io_uring_sqe_set_flags(sqe, flags);    sqe->buf_group = gid;
    conn_info conn_i = {        .fd = fd,        .type = READ,    };    memcpy(&sqe->user_data, &conn_i, sizeof(conn_i));}
void add_socket_write(struct io_uring *ring, int fd, __u16 bid, size_t message_size, unsigned flags) {    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);    io_uring_prep_send(sqe, fd, &bufs[bid], message_size, 0);    io_uring_sqe_set_flags(sqe, flags);
    conn_info conn_i = {        .fd = fd,        .type = WRITE,        .bid = bid,    };    memcpy(&sqe->user_data, &conn_i, sizeof(conn_i));}
void add_provide_buf(struct io_uring *ring, __u16 bid, unsigned gid) {    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);    io_uring_prep_provide_buffers(sqe, bufs[bid], MAX_MESSAGE_LEN, 1, gid, bid);
    conn_info conn_i = {        .fd = 0,        .type = PROV_BUF,    };    memcpy(&sqe->user_data, &conn_i, sizeof(conn_i));}