高性能网络编程：io_uring 与内存优化技术详解

# 检查系统大页配置
cat /proc/meminfo | grep Huge
# 配置大页
echo 20 > /proc/sys/vm/nr_hugepages  # 分配20个大页

# 检查内存锁定限制
ulimit -l
# 修改限制（需要root权限）
echo "* soft memlock unlimited" >> /etc/security/limits.conf

struct IOBuffer {
    char* data;
    size_t size;

    explicit IOBuffer(size_t s) : size(s) {
        // 1. 尝试使用大页内存
        data = static_cast<char*>(mmap(nullptr, size,
            PROT_READ | PROT_WRITE,
            MAP_PRIVATE | MAP_ANONYMOUS | MAP_HUGETLB,
            -1, 0));

        if (data == MAP_FAILED) {
            // 2. 回退到普通内存 + 预填充
            data = static_cast<char*>(mmap(nullptr, size,
                PROT_READ | PROT_WRITE,
                MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE,
                -1, 0));

            if (data == MAP_FAILED) {
                throw std::runtime_error("Failed to allocate memory");
            }
        }

        // 3. 尝试锁定内存
        if (mlock(data, size) != 0) {
            LOG_WARN("Failed to lock memory: {}", strerror(errno));
        }
    }

    ~IOBuffer() {
        if (data != MAP_FAILED && data != nullptr) {
            munlock(data, size);
            munmap(data, size);
        }
    }
};

// 传统 epoll 模型
int epoll_fd = epoll_create1(0);
struct epoll_event events[MAX_EVENTS];

// 每个 I/O 操作都需要系统调用
read(fd, buffer, len);    // 系统调用
write(fd, data, len);     // 系统调用
epoll_wait(epoll_fd, events, MAX_EVENTS, timeout);  // 系统调用

struct io_uring ring;
struct io_uring_sqe *sqe;
struct io_uring_cqe *cqe;

// 批量提交 I/O 请求
for (int i = 0; i < n_requests; i++) {
    sqe = io_uring_get_sqe(&ring);
    io_uring_prep_read(sqe, fds[i], buffers[i], len, offset);
    sqe->user_data = i;  // 标识请求
}

// 一次系统调用提交所有请求
io_uring_submit(&ring);

struct IOContext {
    int fd;
    IOBuffer buffer;
    std::function<void(const char*, size_t)> callback;
};

class WebSocketClient {
private:
    struct io_uring ring;
    std::vector<IOContext> contexts;
    static constexpr int QUEUE_DEPTH = 256;
    // ...
};

void WebSocketClient::processMultipleConnections() {
    struct io_uring_params params = {};
    params.flags = IORING_SETUP_SQPOLL;
    params.sq_thread_cpu = cpu_core_;

    // 初始化 io_uring
    io_uring_queue_init_params(QUEUE_DEPTH, &ring, &params);

    // 为每个连接提交读请求
    for (auto& ctx : contexts) {
        struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
        io_uring_prep_read(sqe, ctx.fd, ctx.buffer.data, ctx.buffer.size, 0);
        sqe->user_data = reinterpret_cast<__u64>(&ctx);
    }

    // 一次提交所有请求
    io_uring_submit(&ring);

    // 处理完成事件
    while (running_) {
        struct io_uring_cqe *cqe;
        int ret = io_uring_wait_cqe(&ring, &cqe);
        
        if (ret == 0) {
            IOContext *ctx = reinterpret_cast<IOContext*>(cqe->user_data);
            if (cqe->res > 0) {
                // 处理数据
                ctx->callback(ctx->buffer.data, cqe->res);
                
                // 提交新的读请求
                struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
                io_uring_prep_read(sqe, ctx->fd, ctx->buffer.data, 
                                 ctx->buffer.size, 0);
                sqe->user_data = cqe->user_data;
                io_uring_submit(&ring);
            }
            io_uring_cqe_seen(&ring, cqe);
        }
    }
}

void submitBatchRequests() {
    int pending = 0;
    for (auto& ctx : contexts) {
        struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
        io_uring_prep_read(sqe, ctx.fd, ctx.buffer.data, ctx.buffer.size, 0);
        sqe->user_data = reinterpret_cast<__u64>(&ctx);
        pending++;
        
        // 达到批次大小时提交
        if (pending == BATCH_SIZE) {
            io_uring_submit(&ring);
            pending = 0;
        }
    }
    
    // 提交剩余请求
    if (pending > 0) {
        io_uring_submit(&ring);
    }
}

struct alignas(64) IOContext {  // 缓存行对齐
    int fd;
    IOBuffer buffer;
    std::function<void(const char*, size_t)> callback;
    char padding[CACHE_LINE_SIZE - sizeof(fd) - sizeof(buffer) 
                - sizeof(callback)];
};

struct IOStats {
    std::atomic<uint64_t> total_requests{0};
    std::atomic<uint64_t> completed_requests{0};
    std::atomic<uint64_t> total_bytes{0};
    std::atomic<uint64_t> error_count{0};
    
    void recordRequest() { total_requests++; }
    void recordCompletion(size_t bytes) {
        completed_requests++;
        total_bytes += bytes;
    }
    void recordError() { error_count++; }
};

void monitorPerformance() {
    while (running_) {
        auto start_stats = io_stats;
        std::this_thread::sleep_for(std::chrono::seconds(1));
        auto end_stats = io_stats;
        
        uint64_t requests_per_sec = end_stats.completed_requests 
                                   - start_stats.completed_requests;
        uint64_t bytes_per_sec = end_stats.total_bytes 
                                - start_stats.total_bytes;
        
        LOG_INFO("IO Stats: {} req/s, {} MB/s", 
                 requests_per_sec, bytes_per_sec / (1024 * 1024));
    }
}