异步I/O机制：从POSIX AIO到io_uring

#include "lib/common.h"
#include <aio.h>
 
const int BUF_SIZE = 512;
 
int main() {
    int err;
    int result_size;
 
    // 创建一个临时文件
    char tmpname[256];
    snprintf(tmpname, sizeof(tmpname), "/tmp/aio_test_%d", getpid());
    unlink(tmpname);
    int fd = open(tmpname, O_CREAT | O_RDWR | O_EXCL, S_IRUSR | S_IWUSR);
    if (fd == -1) {
        error(1, errno, "open file failed ");
    }
 
    char buf[BUF_SIZE];
    struct aiocb aiocb;
 
    // 初始化 buf 缓冲，写入数据 0xfafa
    memset(buf, 0xfa, BUF_SIZE);
    memset(&aiocb, 0, sizeof(struct aiocb));
    aiocb.aio_fildes = fd;
    aiocb.aio_buf = buf;
    aiocb.aio_nbytes = BUF_SIZE;
 
    // 开始异步写
    if (aio_write(&aiocb) == -1) {
        printf(" Error at aio_write(): %s\n", strerror(errno));
        close(fd);
        exit(1);
    }
 
    // 轮询等待异步写完成
    while (aio_error(&aiocb) == EINPROGRESS) {
        printf("writing... \n");
    }
 
    // 判断写入结果
    err = aio_error(&aiocb);
    result_size = aio_return(&aiocb);
    if (err != 0 || result_size != BUF_SIZE) {
        printf(" aio_write failed() : %s\n", strerror(err));
        close(fd);
        exit(1);
    }
 
    // 准备异步读
    char buffer[BUF_SIZE];
    struct aiocb cb;
    cb.aio_nbytes = BUF_SIZE;
    cb.aio_fildes = fd;
    cb.aio_offset = 0;
    cb.aio_buf = buffer;
 
    // 开始异步读
    if (aio_read(&cb) == -1) {
        printf(" air_read failed() : %s\n", strerror(err));
        close(fd);
    }
 
    // 轮询等待异步读完成
    while (aio_error(&cb) == EINPROGRESS) {
        printf("Reading... \n");
    }
 
    // 判断读结果
    int numBytes = aio_return(&cb);
    if (numBytes != -1) {
        printf("Success.\n");
    } else {
        printf("Error.\n");
    }
 
    close(fd);
    return 0;
}

struct aiocb {
   int       aio_fildes;       /* File descriptor */
   off_t     aio_offset;       /* File offset */
   volatile void  *aio_buf;     /* Location of buffer */
   size_t    aio_nbytes;       /* Length of transfer */
   int       aio_reqprio;      /* Request priority offset */
   struct sigevent    aio_sigevent;     /* Signal number and value */
   int       aio_lio_opcode;       /* Operation to be performed */
};

#include <liburing.h>
#include <netinet/in.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
 
#define ENTRIES_LENGTH  4096
#define BUFFER_SIZE     2048
 
// 为每个连接维护的读缓冲区
struct conn_info {
    int fd;
    struct iovec iov;
};
 
int main(int argc, char *argv[]) {
    unsigned short port = 43211;
    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
 
    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
 
    bind(sockfd, (struct sockaddr *)&addr, sizeof(addr));
    listen(sockfd, 1024);
 
    // 初始化 io_uring 实例
    struct io_uring ring;
    io_uring_queue_init(ENTRIES_LENGTH, &ring, 0);
 
    // 注册 accept 请求
    struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
    io_uring_prep_accept(sqe, sockfd, NULL, NULL, 0);
    io_uring_submit(&ring);
 
    char buffers[ENTRIES_LENGTH][BUFFER_SIZE];
    struct conn_info conns[ENTRIES_LENGTH];
 
    while (1) {
        struct io_uring_cqe *cqe;
        io_uring_wait_cqe(&ring, &cqe);
 
        struct conn_info *conn_i = (struct conn_info *)io_uring_cqe_get_data(cqe);
        int res = cqe->data;
 
        if (conn_i == NULL) {
            // accept 完成，获取新连接
            int connfd = cqe->res;
            io_uring_cqe_seen(&ring, cqe);
 
            // 继续注册 accept
            sqe = io_uring_get_sqe(&ring);
            io_uring_prep_accept(sqe, sockfd, NULL, NULL, 0);
            io_uring_submit(&ring);
 
            // 注册新连接的 recv 请求
            int bid = connfd;
            conns[bid].fd = connfd;
            conns[bid].iov.iov_base = buffers[bid];
            conns[bid].iov.iov_len = BUFFER_SIZE;
 
            sqe = io_uring_get_sqe(&ring);
            io_uring_prep_recv(sqe, connfd, buffers[bid], BUFFER_SIZE, 0);
            io_uring_sqe_set_data(sqe, &conns[bid]);
            io_uring_submit(&ring);
        } else {
            // recv 完成
            int n = cqe->res;
            io_uring_cqe_seen(&ring, cqe);
 
            if (n <= 0) {
                close(conn_i->fd);
                continue;
            }
 
            // 注册 send 请求（Echo 回写）
            sqe = io_uring_get_sqe(&ring);
            io_uring_prep_send(sqe, conn_i->fd, conn_i->iov.iov_base, n, 0);
            io_uring_sqe_set_data(sqe, conn_i);
            io_uring_submit(&ring);
 
            // 再次注册 recv
            sqe = io_uring_get_sqe(&ring);
            io_uring_prep_recv(sqe, conn_i->fd, conn_i->iov.iov_base, BUFFER_SIZE, 0);
            io_uring_sqe_set_data(sqe, conn_i);
            io_uring_submit(&ring);
        }
    }
 
    io_uring_queue_exit(&ring);
    return 0;
}