C++ 跨平台适配层构建：设计一套同时支撑 Windows IOCP 与 Linux io_uring 的 C++ 高性能网络通信内核

各位技术同仁，下午好。

今天，我们将深入探讨一个在高性能网络服务领域极具挑战性与实用价值的主题：如何构建一个同时支持 Windows IOCP (I/O Completion Port) 和 Linux io_uring 的 C++ 跨平台高性能网络通信内核。在现代云原生和分布式系统中，网络通信的效率直接决定了服务的响应速度和吞吐量。面对 Windows 和 Linux 这两大主流操作系统截然不同的异步 I/O 模型，设计一个统一、高效且易于使用的抽象层，是每一个追求极致性能的 C++ 开发者梦寐以求的目标。

本文将从 IOCP 和 io_uring 的基本原理出发，详细阐述其各自的优势与特点，进而深入探讨如何设计一个通用的跨平台适配层，包含核心接口、平台特定实现、内存管理策略以及错误处理机制。我们将以讲座的形式，辅以代码示例，力求逻辑严谨、表述清晰，希望能为大家在构建高性能网络服务时提供有益的参考。

1. 异步 I/O：高性能网络通信的基石

在网络编程中，同步 I/O 模型的阻塞特性严重限制了服务器的并发能力。当一个请求到达时，如果服务器线程需要等待 I/O 操作完成（如从磁盘读取数据或向网络发送数据），它将无法处理其他请求，导致性能瓶颈。异步 I/O 模型正是为了解决这一问题而生，它允许应用程序发起 I/O 操作后立即返回，并在 I/O 完成时通过某种机制通知应用程序。

Windows 上的 IOCP 和 Linux 上的 io_uring 是两种各自平台下最先进、最高效的异步 I/O 模型。它们都提供了事件驱动的、非阻塞的 I/O 范式，但其内部机制和编程接口却大相径庭。

1.1 Windows I/O Completion Port (IOCP)

IOCP 是 Windows 操作系统提供的一种高效的并发 I/O 模型。它的核心思想是将多个 I/O 操作的完成事件聚合到一个端口上，并由一个或多个工作线程从该端口获取完成事件并进行处理。

IOCP 的主要特点：

• 线程池模型： IOCP 通常与一个工作线程池结合使用。当 I/O 操作完成时，系统会将完成通知放入 IOCP 队列，工作线程从队列中取出通知并执行相应的回调。
• 并发性管理： IOCP 能够智能地管理并发线程的数量，避免过多的线程竞争，从而减少上下文切换的开销。它通过关联的完成端口限制了同时运行的处理完成事件的线程数。
• 用户数据关联： 每个异步 I/O 操作都可以关联一个 OVERLAPPED 结构体和一个自定义的 PerHandleData 或 PerIOData 结构体，方便在完成时检索上下文信息。
• 通用性： IOCP 不仅限于网络 I/O，也适用于文件 I/O 等其他异步操作。

IOCP 的工作流程：

1. 创建 I/O Completion Port。
2. 创建服务器监听套接字，并将其与 I/O Completion Port 关联。
3. 为每个客户端连接套接字也与 I/O Completion Port 关联。
4. 在套接字上发起异步读写操作（WSARecv、WSASend、AcceptEx 等），并传入 OVERLAPPED 结构体。
5. 创建工作线程池，每个线程调用 GetQueuedCompletionStatus 等待 I/O 完成事件。
6. 当 I/O 操作完成时，系统将完成通知放入 IOCP 队列，GetQueuedCompletionStatus 返回，工作线程处理事件。

1.2 Linux io_uring

io_uring 是 Linux 5.1 内核引入的一种全新的异步 I/O 接口，旨在提供比 epoll 等现有机制更高效、功能更强大的异步 I/O 能力，特别是在高并发、低延迟场景下表现卓越。

io_uring 的主要特点：

• 零拷贝（Zero-copy）和批处理： io_uring 通过在用户空间和内核空间之间共享环形缓冲区（Submission Queue – SQ 和 Completion Queue – CQ），实现了零拷贝的数据传输，并支持一次性提交多个 I/O 请求（批处理），极大地减少了系统调用次数和上下文切换开销。
• 完全异步： 几乎所有的 I/O 操作都可以通过 io_uring 以异步方式执行，包括文件操作、网络操作（TCP/UDP）、定时器等。
• 用户态轮询（SQPOLL）： 支持用户态轮询模式，在特定场景下可以进一步降低延迟，避免系统调用。
• 强大的功能： 除了基本的读写，还支持 accept、connect、splice、sendmsg、recvmsg 等多种网络操作，以及文件注册、缓冲区注册等高级功能。

io_uring 的工作流程：

1. 通过 io_uring_setup 创建一个 io_uring 实例，获得 SQ 和 CQ 的文件描述符。
2. 通过 mmap 将 SQ 和 CQ 映射到用户空间，形成共享内存环形缓冲区。
3. 用户应用程序将 I/O 请求（io_uring_sqe）填充到 SQ 中。
4. 通过 io_uring_enter 或 io_uring_submit 系统调用通知内核有新的请求。
5. 内核执行 I/O 操作。
6. 操作完成后，内核将完成事件（io_uring_cqe）填充到 CQ 中。
7. 用户应用程序从 CQ 中获取完成事件，并处理结果。

2. 跨平台适配层设计哲学

面对 IOCP 和 io_uring 如此巨大的差异，设计一个统一的跨平台适配层，其核心挑战在于：如何在不牺牲各自平台原生性能优势的前提下，提供一个简洁、一致且高效的编程接口。

我们的设计哲学将围绕以下几点展开：

1. 抽象统一接口： 向上层应用提供一套统一的、平台无关的异步 I/O 接口，隐藏底层操作系统差异。
2. 事件驱动模型： 核心是事件循环（Event Loop），负责提交 I/O 请求、等待 I/O 完成并分发完成事件。
3. 上下文关联： 允许用户将自定义数据与每个 I/O 操作关联，以便在操作完成时恢复上下文。
4. 资源管理： 统一的内存管理（特别是缓冲区）和套接字/文件描述符管理。
5. 性能优先： 尽可能利用 IOCP 和 io_uring 的批处理、零拷贝等高级特性。
6. 错误处理： 统一的错误报告机制。

2.1 核心抽象：IOContext 和 IOLoop

在我们的设计中，两个核心的抽象是 IOContext 和 IOLoop。

• IOContext (I/O 上下文)： 代表一个待处理或已完成的异步 I/O 操作。它包含了操作的类型（读、写、接受连接、连接等）、目标套接字、数据缓冲区、长度、以及一个用户回调函数或用户数据指针。IOContext 将是连接上层应用逻辑和底层平台特定 I/O 机制的关键。
• IOLoop (I/O 循环)： 这是事件循环的核心。它提供提交 I/O 操作、运行事件循环以及停止事件循环的接口。IOLoop 内部将根据操作系统类型，实例化不同的平台特定实现（WindowsIOLoop 或 LinuxIOLoop）。

统一的 IOContext 结构体

#include <functional>
#include <vector>
#include <memory>

// 定义各种I/O操作类型
enum class IOOperationType {
    READ,
    WRITE,
    ACCEPT,
    CONNECT,
    CLOSE,
    // ... 其他操作类型
};

// 异步I/O操作完成时的回调函数类型
// 参数：操作类型、用户数据、操作结果（字节数或错误码）、错误信息
using IOCompletionCallback = std::function<void(
    IOOperationType type,
    void* userData,
    long result, // 实际传输字节数或错误码
    int errorCode // 操作系统原生错误码
)>;

// 统一的I/O上下文结构体
struct IOContext {
    IOOperationType type;           // 操作类型
    int socketFd;                   // 目标套接字（Windows: SOCKET, Linux: int）
    std::vector<char> buffer;       // 数据缓冲区
    size_t bytesToTransfer;         // 期望传输的字节数
    size_t bytesTransferred;        // 实际传输的字节数
    IOCompletionCallback callback;  // 完成回调
    void* userData;                 // 用户自定义数据指针

    // 平台特定的数据，通过联合体或继承实现
    // Windows: OVERLAPPED 结构体
    // Linux: io_uring_sqe 的 user_data 指针指向此结构
    // 这种设计允许我们将 IOContext 作为 OVERLAPPED 的一部分，或者其 user_data 指针。
    // 更安全的做法是使用一个公共基类，然后派生出平台特定子类，并在其中包含平台特定字段。
    // 为了简化，这里我们假设 IOContext 可以被平台特定实现“嵌入”或“指向”。

#ifdef _WIN32
    OVERLAPPED overlapped;          // Windows特有的OVERLAPPED结构
    // 为方便从 OVERLAPPED 恢复 IOContext，通常将 IOContext 嵌入到自定义结构中
    // 如下面所示的 PerIOData
#else
    // Linux io_uring 不需要额外的结构体，只需将此 IOContext 的地址作为 user_data
    // 在 io_uring_cqe_get_data() 中取回
#endif

    IOContext() :
        type(IOOperationType::READ),
        socketFd(-1),
        bytesToTransfer(0),
        bytesTransferred(0),
        userData(nullptr)
    {
#ifdef _WIN32
        memset(&overlapped, 0, sizeof(overlapped));
#endif
    }

    // 禁用拷贝和赋值，因为 IOContext 经常被作为指针传递和管理
    IOContext(const IOContext&) = delete;
    IOContext& operator=(const IOContext&) = delete;
};

// Windows特有的结构体，将IOContext嵌入其中，方便从OVERLAPPED指针恢复
#ifdef _WIN32
struct PerIOData {
    OVERLAPPED overlapped;
    IOContext* ioContext; // 指向实际的IOContext
    // 其他Windows特有数据...
    PerIOData() : ioContext(nullptr) {
        memset(&overlapped, 0, sizeof(overlapped));
    }
};
#endif

统一的 IOLoop 接口

// IOLoop.h
#pragma once

#include "IOContext.h"

class IOLoop {
public:
    virtual ~IOLoop() = default;

    // 初始化I/O循环
    virtual bool Init() = 0;

    // 提交一个I/O操作
    virtual bool SubmitIO(std::shared_ptr<IOContext> context) = 0;

    // 运行事件循环，处理完成事件
    // blockUntilEvents: 是否阻塞直到有事件发生
    virtual void Run(bool blockUntilEvents = true) = 0;

    // 停止事件循环
    virtual void Stop() = 0;

    // 获取平台原生的错误码
    virtual int GetLastError() const = 0;

    // 注册一个套接字/文件描述符到I/O循环中
    // 某些平台可能需要显式注册，例如Windows IOCP需要关联句柄
    // io_uring虽然不需要，但可以作为统一接口提供
    virtual bool RegisterHandle(int socketFd) = 0;

    // 注销一个套接字/文件描述符
    virtual bool UnregisterHandle(int socketFd) = 0;
};

2.2 平台特定实现

接下来，我们将为 IOLoop 接口提供 Windows IOCP 和 Linux io_uring 的具体实现。

2.2.1 Windows IOCPIOLoop 实现

IOCPIOLoop 将使用 CreateIoCompletionPort 创建完成端口，并通过工作线程池调用 GetQueuedCompletionStatus 来获取完成事件。

// WindowsIOLoop.h
#pragma once

#ifdef _WIN32
#include "IOLoop.h"
#include <windows.h>
#include <vector>
#include <thread>
#include <mutex>
#include <map>

// Windows特有的I/O上下文扩展
// 在Windows中，OVERLAPPED是操作的“身份”
// 我们需要从OVERLAPPED指针恢复到我们的IOContext
struct WindowsIOContextWrapper {
    OVERLAPPED overlapped;
    std::shared_ptr<IOContext> ioContext;

    WindowsIOContextWrapper(std::shared_ptr<IOContext> ctx)
        : ioContext(std::move(ctx)) {
        memset(&overlapped, 0, sizeof(overlapped));
    }

    // 静态方法，用于从 OVERLAPPED 指针获取 WindowsIOContextWrapper
    static WindowsIOContextWrapper* FromOverlapped(OVERLAPPED* ov) {
        return CONTAINING_RECORD(ov, WindowsIOContextWrapper, overlapped);
    }
};

class IOCPIOLoop : public IOLoop {
public:
    IOCPIOLoop(int numWorkerThreads = std::thread::hardware_concurrency());
    ~IOCPIOLoop() override;

    bool Init() override;
    bool SubmitIO(std::shared_ptr<IOContext> context) override;
    void Run(bool blockUntilEvents = true) override;
    void Stop() override;
    int GetLastError() const override;
    bool RegisterHandle(int socketFd) override;
    bool UnregisterHandle(int socketFd) override;

private:
    HANDLE _ioCompletionPort;
    std::vector<std::thread> _workerThreads;
    std::atomic<bool> _running;
    int _numWorkerThreads;
    std::mutex _contextMutex;
    // 存储所有待处理的IOContext，以便在完成时查找和清理
    // 实际项目中可能需要更复杂的管理策略，例如使用IOContext的ID
    std::map<WindowsIOContextWrapper*, std::shared_ptr<IOContext>> _pendingContexts;

    // 工作线程函数
    void WorkerThreadFunc();

    // 处理完成事件
    void HandleCompletion(DWORD bytesTransferred, ULONG_PTR completionKey, OVERLAPPED* overlapped);
};

#endif // _WIN32

// WindowsIOLoop.cpp
#ifdef _WIN32
#include "WindowsIOLoop.h"
#include <iostream>

#pragma comment(lib, "ws2_32.lib") // Link with ws2_32.lib

IOCPIOLoop::IOCPIOLoop(int numWorkerThreads)
    : _ioCompletionPort(INVALID_HANDLE_VALUE),
      _running(false),
      _numWorkerThreads(numWorkerThreads)
{
    if (_numWorkerThreads <= 0) {
        _numWorkerThreads = 1;
    }
}

IOCPIOLoop::~IOCPIOLoop() {
    Stop();
    if (_ioCompletionPort != INVALID_HANDLE_VALUE) {
        CloseHandle(_ioCompletionPort);
    }
}

bool IOCPIOLoop::Init() {
    // 创建一个I/O完成端口
    _ioCompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
    if (_ioCompletionPort == NULL) {
        std::cerr << "CreateIoCompletionPort failed: " << GetLastError() << std::endl;
        return false;
    }

    // 启动工作线程
    _running = true;
    for (int i = 0; i < _numWorkerThreads; ++i) {
        _workerThreads.emplace_back(&IOCPIOLoop::WorkerThreadFunc, this);
    }

    return true;
}

bool IOCPIOLoop::SubmitIO(std::shared_ptr<IOContext> context) {
    if (!_running || context == nullptr) {
        return false;
    }

    // 为Windows I/O准备上下文
    auto wrapper = std::make_unique<WindowsIOContextWrapper>(context);
    // 将wrapper的地址作为PerHandleData (CompletionKey)，这样在GetQueuedCompletionStatus时可以取回
    // 也可以将wrapper的地址作为overlapped参数，通过CONTAINING_RECORD宏恢复
    // 这里我们直接将wrapper的overlapped成员作为参数
    OVERLAPPED* pOverlapped = &wrapper->overlapped;

    // 将wrapper添加到待处理列表
    {
        std::lock_guard<std::mutex> lock(_contextMutex);
        _pendingContexts[wrapper.get()] = context; // 存储shared_ptr
    }
    // 释放 unique_ptr 的所有权，由 IOCP 管理生命周期，直到完成
    WindowsIOContextWrapper* rawWrapper = wrapper.release();

    BOOL ret = FALSE;
    DWORD bytes = 0;
    int error = 0;

    switch (context->type) {
        case IOOperationType::READ: {
            WSABUF wsaBuf;
            wsaBuf.buf = context->buffer.data();
            wsaBuf.len = static_cast<ULONG>(context->bytesToTransfer);
            DWORD flags = 0;
            ret = WSARecv(static_cast<SOCKET>(context->socketFd), &wsaBuf, 1, &bytes, &flags, pOverlapped, NULL);
            if (ret == FALSE && (error = WSAGetLastError()) != WSA_IO_PENDING) {
                std::cerr << "WSARecv failed: " << error << std::endl;
                // 立即处理错误，并从pendingContexts中移除
                HandleCompletion(0, 0, pOverlapped); // 模拟失败完成
                return false;
            }
            break;
        }
        case IOOperationType::WRITE: {
            WSABUF wsaBuf;
            wsaBuf.buf = context->buffer.data();
            wsaBuf.len = static_cast<ULONG>(context->bytesToTransfer);
            ret = WSASend(static_cast<SOCKET>(context->socketFd), &wsaBuf, 1, &bytes, 0, pOverlapped, NULL);
            if (ret == FALSE && (error = WSAGetLastError()) != WSA_IO_PENDING) {
                std::cerr << "WSASend failed: " << error << std::endl;
                HandleCompletion(0, 0, pOverlapped);
                return false;
            }
            break;
        }
        case IOOperationType::ACCEPT: {
            // AcceptEx 需要一个预先创建的套接字作为新连接的套接字
            // 以及一个足够大的缓冲区来接收地址信息
            // 假设 context->userData 包含了 AcceptEx 所需的额外参数和缓冲区
            // 这里为了简化，省略 AcceptEx 的复杂性
            // 实际应用中，需要一个单独的结构来管理AcceptEx的复杂参数
            // 这里只是一个示意
            std::cerr << "AcceptEx not fully implemented for example." << std::endl;
            // 模拟失败
            HandleCompletion(0, 0, pOverlapped);
            return false;
        }
        case IOOperationType::CONNECT: {
            // ConnectEx 同样复杂，需要预先创建套接字并绑定
            std::cerr << "ConnectEx not fully implemented for example." << std::endl;
            // 模拟失败
            HandleCompletion(0, 0, pOverlapped);
            return false;
        }
        default:
            std::cerr << "Unsupported IO operation type: " << static_cast<int>(context->type) << std::endl;
            HandleCompletion(0, 0, pOverlapped);
            return false;
    }
    return true;
}

void IOCPIOLoop::Run(bool blockUntilEvents) {
    // 在IOCP中，Run方法通常只是一个启动工作线程的信号，
    // 实际的事件处理在工作线程中完成。
    // 如果需要主线程也参与事件处理，可以像工作线程一样调用GetQueuedCompletionStatus。
    // 为了简化，这里假设所有事件都在工作线程中处理。
    if (!_running) {
        if (!Init()) {
            std::cerr << "IOCPIOLoop failed to initialize." << std::endl;
            return;
        }
    }
    // 主线程可以等待所有工作线程结束
    for (auto& thread : _workerThreads) {
        if (thread.joinable()) {
            // 在实际应用中，主线程可能需要执行其他任务，而不是在这里阻塞
            // 这里的join只是为了确保所有线程正常退出
        }
    }
}

void IOCPIOLoop::Stop() {
    if (_running.exchange(false)) {
        // 向IOCP投递足够多的退出事件，以唤醒所有工作线程
        for (int i = 0; i < _numWorkerThreads; ++i) {
            PostQueuedCompletionStatus(_ioCompletionPort, 0, NULL, NULL);
        }
        for (auto& thread : _workerThreads) {
            if (thread.joinable()) {
                thread.join();
            }
        }
        _workerThreads.clear();

        // 清理所有悬挂的IOContext
        // 注意：这里只是简单清理，实际应用中需要确保所有I/O操作都被取消或完成
        std::lock_guard<std::mutex> lock(_contextMutex);
        for (auto const& [wrapperPtr, contextPtr] : _pendingContexts) {
            delete wrapperPtr; // 释放 rawWrapper
        }
        _pendingContexts.clear();
    }
}

int IOCPIOLoop::GetLastError() const {
    return ::GetLastError();
}

bool IOCPIOLoop::RegisterHandle(int socketFd) {
    if (_ioCompletionPort == INVALID_HANDLE_VALUE) {
        std::cerr << "IOCP not initialized." << std::endl;
        return false;
    }
    // 将套接字与完成端口关联。CompletionKey可以用于标识句柄。
    // 这里我们简单地使用句柄本身作为CompletionKey。
    HANDLE hResult = CreateIoCompletionPort(reinterpret_cast<HANDLE>(socketFd), _ioCompletionPort, reinterpret_cast<ULONG_PTR>(socketFd), 0);
    if (hResult == NULL) {
        std::cerr << "CreateIoCompletionPort for socket failed: " << GetLastError() << std::endl;
        return false;
    }
    return true;
}

bool IOCPIOLoop::UnregisterHandle(int socketFd) {
    // IOCP没有显式的注销API，通常在句柄关闭时，其关联的I/O操作会自动取消或失败。
    // 对于已经提交但未完成的I/O操作，可能需要使用 CancelIoEx。
    // 这里简化处理，不实现显式注销逻辑。
    return true;
}

void IOCPIOLoop::WorkerThreadFunc() {
    DWORD bytesTransferred;
    ULONG_PTR completionKey;
    OVERLAPPED* pOverlapped;
    while (_running) {
        // 等待I/O完成事件
        BOOL success = GetQueuedCompletionStatus(
            _ioCompletionPort,
            &bytesTransferred,
            &completionKey, // PerHandleData
            &pOverlapped,   // PerIOData
            INFINITE        // 永远阻塞直到有事件
        );

        if (pOverlapped == NULL) { // 收到NULL，表示退出信号
            if (!_running) break; // 确认是停止信号
            continue; // 否则可能是错误，继续等待
        }

        // 处理完成事件
        HandleCompletion(bytesTransferred, completionKey, pOverlapped);
    }
}

void IOCPIOLoop::HandleCompletion(DWORD bytesTransferred, ULONG_PTR completionKey, OVERLAPPED* pOverlapped) {
    int errorCode = 0;
    if (pOverlapped == NULL && bytesTransferred == 0) {
        // 这是一个PostQueuedCompletionStatus(0, 0, NULL, NULL)的退出信号
        // 或者是一个错误，但GetQueuedCompletionStatus通常不会返回NULL overlapped for error
        // 这里需要根据实际情况判断
        return;
    }

    // 从OVERLAPPED指针恢复我们自己的WindowsIOContextWrapper
    WindowsIOContextWrapper* wrapper = WindowsIOContextWrapper::FromOverlapped(pOverlapped);
    std::shared_ptr<IOContext> context;

    // 从pendingContexts中取出并移除
    {
        std::lock_guard<std::mutex> lock(_contextMutex);
        auto it = _pendingContexts.find(wrapper);
        if (it != _pendingContexts.end()) {
            context = it->second;
            _pendingContexts.erase(it);
        }
    }

    if (context) {
        context->bytesTransferred = bytesTransferred;
        if (!GetOverlappedResult(reinterpret_cast<HANDLE>(context->socketFd), pOverlapped, &bytesTransferred, FALSE)) {
            errorCode = GetLastError();
            std::cerr << "GetOverlappedResult failed with error: " << errorCode << std::endl;
        }

        if (context->callback) {
            context->callback(context->type, context->userData, bytesTransferred, errorCode);
        }
    } else {
        std::cerr << "Error: IOContext not found for completed operation." << std::endl;
    }

    // 释放 wrapper 对象
    delete wrapper;
}
#endif // _WIN32

IOCP 注意事项：

• OVERLAPPED 结构体的生命周期必须在 I/O 操作完成之前一直有效。
• WSARecv 和 WSASend 等函数在返回 FALSE 且 WSAGetLastError() 返回 WSA_IO_PENDING 时，表示操作已成功提交，将在未来完成。
• CreateIoCompletionPort 的 CompletionKey 参数可以用于在 GetQueuedCompletionStatus 返回时标识关联的句柄或自定义数据。这里我们通过 WindowsIOContextWrapper 结构体来关联 IOContext。

2.2.2 Linux IoUringIOLoop 实现

IoUringIOLoop 将使用 liburing 库来简化 io_uring 的编程接口。

// LinuxIOLoop.h
#pragma once

#ifndef _WIN32
#include "IOLoop.h"
#include <liburing.h>
#include <vector>
#include <thread>
#include <mutex>
#include <atomic>
#include <map>
#include <memory>

class IoUringIOLoop : public IOLoop {
public:
    IoUringIOLoop(unsigned int queueDepth = 128);
    ~IoUringIOLoop() override;

    bool Init() override;
    bool SubmitIO(std::shared_ptr<IOContext> context) override;
    void Run(bool blockUntilEvents = true) override;
    void Stop() override;
    int GetLastError() const override;
    bool RegisterHandle(int socketFd) override;
    bool UnregisterHandle(int socketFd) override;

private:
    io_uring _ring;
    unsigned int _queueDepth;
    std::atomic<bool> _running;
    std::thread _workerThread; // io_uring通常在一个线程中处理提交和完成
    std::mutex _contextMutex;
    // 存储所有待处理的IOContext，以便在完成时查找和清理
    std::map<void*, std::shared_ptr<IOContext>> _pendingContexts;

    // 工作线程函数
    void WorkerThreadFunc();

    // 提交一个SQE
    bool SubmitSQE(io_uring_sqe* sqe, std::shared_ptr<IOContext> context);
};

#endif // !_WIN32

// LinuxIOLoop.cpp
#ifndef _WIN32
#include "LinuxIOLoop.h"
#include <iostream>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/eventfd.h> // 用于在停止时唤醒io_uring_wait_cqe

IoUringIOLoop::IoUringIOLoop(unsigned int queueDepth)
    : _queueDepth(queueDepth),
      _running(false)
{}

IoUringIOLoop::~IoUringIOLoop() {
    Stop();
    io_uring_queue_exit(&_ring);
}

bool IoUringIOLoop::Init() {
    // 初始化io_uring
    int ret = io_uring_queue_init(_queueDepth, &_ring, 0); // 0表示默认flags
    if (ret < 0) {
        std::cerr << "io_uring_queue_init failed: " << strerror(-ret) << std::endl;
        return false;
    }

    _running = true;
    _workerThread = std::thread(&IoUringIOLoop::WorkerThreadFunc, this);
    return true;
}

bool IoUringIOLoop::SubmitIO(std::shared_ptr<IOContext> context) {
    if (!_running || context == nullptr) {
        return false;
    }

    io_uring_sqe* sqe = io_uring_get_sqe(&_ring);
    if (!sqe) {
        std::cerr << "io_uring_get_sqe failed: Submission Queue full." << std::endl;
        return false;
    }

    // 将IOContext的地址作为user_data，在完成时可以取回
    io_uring_sqe_set_data(sqe, context.get());

    // 将context添加到待处理列表
    {
        std::lock_guard<std::mutex> lock(_contextMutex);
        _pendingContexts[context.get()] = context; // 存储shared_ptr
    }

    switch (context->type) {
        case IOOperationType::READ: {
            io_uring_prep_read(sqe, context->socketFd, context->buffer.data(), context->bytesToTransfer, 0);
            break;
        }
        case IOOperationType::WRITE: {
            io_uring_prep_write(sqe, context->socketFd, context->buffer.data(), context->bytesToTransfer, 0);
            break;
        }
        case IOOperationType::ACCEPT: {
            // io_uring_prep_accept 需要一个地址结构体和长度
            // 假设 context->userData 指向一个 sockaddr_storage 结构
            sockaddr_storage* client_addr = static_cast<sockaddr_storage*>(context->userData);
            socklen_t* client_len = reinterpret_cast<socklen_t*>(context->buffer.data()); // 借用buffer存储长度
            io_uring_prep_accept(sqe, context->socketFd, reinterpret_cast<sockaddr*>(client_addr), client_len, 0);
            break;
        }
        case IOOperationType::CONNECT: {
            // io_uring_prep_connect 需要地址结构体
            sockaddr_storage* server_addr = static_cast<sockaddr_storage*>(context->userData);
            io_uring_prep_connect(sqe, context->socketFd, reinterpret_cast<sockaddr*>(server_addr), sizeof(sockaddr_storage));
            break;
        }
        case IOOperationType::CLOSE: {
             io_uring_prep_close(sqe, context->socketFd);
             break;
        }
        default:
            std::cerr << "Unsupported IO operation type: " << static_cast<int>(context->type) << std::endl;
            // 立即处理错误，并从pendingContexts中移除
            {
                std::lock_guard<std::mutex> lock(_contextMutex);
                _pendingContexts.erase(context.get());
            }
            return false;
    }

    // 提交到内核
    int ret = io_uring_submit(&_ring);
    if (ret < 0) {
        std::cerr << "io_uring_submit failed: " << strerror(-ret) << std::endl;
        // 提交失败，从pendingContexts中移除
        std::lock_guard<std::mutex> lock(_contextMutex);
        _pendingContexts.erase(context.get());
        return false;
    }
    return true;
}

void IoUringIOLoop::Run(bool blockUntilEvents) {
    if (!_running) {
        if (!Init()) {
            std::cerr << "IoUringIOLoop failed to initialize." << std::endl;
            return;
        }
    }
    if (_workerThread.joinable()) {
        _workerThread.join();
    }
}

void IoUringIOLoop::Stop() {
    if (_running.exchange(false)) {
        // 唤醒 worker thread
        // io_uring 没有直接的“退出事件”概念，可以通过 eventfd 或一个假的 I/O 请求来唤醒
        // 最简单的方法是提交一个 NOP 请求，并设置一个很小的超时
        // 更好的方法是使用 io_uring_enter 的 IORING_ENTER_DONTWAIT 配合信号量
        // 或使用 eventfd 注册到 io_uring，然后通过 eventfd 唤醒
        // 考虑到简化，这里我们依赖 _workerThread 循环条件
        // 实际中可能需要更优雅的唤醒机制
        // 例如，创建一个临时的SQE，设置其user_data为特殊值，然后提交
        io_uring_sqe* sqe = io_uring_get_sqe(&_ring);
        if (sqe) {
            io_uring_prep_nop(sqe);
            io_uring_sqe_set_data(sqe, nullptr); // 使用 nullptr 作为停止信号
            io_uring_submit(&_ring);
        }

        if (_workerThread.joinable()) {
            _workerThread.join();
        }

        // 清理所有悬挂的IOContext
        std::lock_guard<std::mutex> lock(_contextMutex);
        _pendingContexts.clear();
    }
}

int IoUringIOLoop::GetLastError() const {
    return errno;
}

bool IoUringIOLoop::RegisterHandle(int socketFd) {
    // io_uring 提供了 io_uring_register_files 用于文件描述符注册，
    // 以避免在每个SQE中传递文件描述符，减少开销。
    // 这里为了简化，我们直接在SQE中传递socketFd。
    // 如果需要注册，可以使用 io_uring_register_files。
    // 例如：
    // int fds[] = {socketFd};
    // io_uring_register_files(&_ring, fds, 1);
    // 并在SQE中通过 IORING_OP_READ_FIXED 或 IORING_OP_WRITE_FIXED 使用文件描述符索引。
    return true; // io_uring不需要显式注册
}

bool IoUringIOLoop::UnregisterHandle(int socketFd) {
    // io_uring_unregister_files 对应注册，这里同样不实现。
    return true; // io_uring不需要显式注销
}

void IoUringIOLoop::WorkerThreadFunc() {
    while (_running) {
        io_uring_cqe* cqe;
        // 阻塞等待完成事件
        int ret = io_uring_wait_cqe(&_ring, &cqe);
        if (ret < 0) {
            if (ret == -EINTR) {
                continue; // 被信号中断，继续等待
            }
            std::cerr << "io_uring_wait_cqe failed: " << strerror(-ret) << std::endl;
            // 致命错误，退出循环
            _running = false;
            break;
        }

        if (cqe == nullptr) { // 通常不会发生，除非是io_uring_wait_cqe自身错误
            continue;
        }

        // 检查是否是停止信号
        if (io_uring_cqe_get_data(cqe) == nullptr) {
            io_uring_cq_advance(&_ring, 1); // 消费掉这个CQE
            continue;
        }

        // 从user_data恢复IOContext
        IOContext* rawContextPtr = static_cast<IOContext*>(io_uring_cqe_get_data(cqe));
        std::shared_ptr<IOContext> context;

        // 从pendingContexts中取出并移除
        {
            std::lock_guard<std::mutex> lock(_contextMutex);
            auto it = _pendingContexts.find(rawContextPtr);
            if (it != _pendingContexts.end()) {
                context = it->second;
                _pendingContexts.erase(it);
            }
        }

        if (context) {
            long result = cqe->res; // 操作结果：传输字节数或错误码
            int errorCode = 0;
            if (result < 0) {
                errorCode = -result; // io_uring 错误码通常是负数
                std::cerr << "IO operation failed: " << strerror(errorCode) << std::endl;
            }
            context->bytesTransferred = (result > 0) ? result : 0;

            if (context->callback) {
                context->callback(context->type, context->userData, result, errorCode);
            }
        } else {
            std::cerr << "Error: IOContext not found for completed operation (user_data: " << rawContextPtr << ")." << std::endl;
        }

        io_uring_cq_advance(&_ring, 1); // 消费掉这个CQE
    }
}
#endif // !_WIN32

io_uring 注意事项：

• liburing 库极大地简化了 io_uring 的使用。
• io_uring_sqe_set_data 是将用户数据与请求关联的关键，完成时通过 io_uring_cqe_get_data 取回。
• io_uring_submit 提交请求到内核，io_uring_wait_cqe 阻塞等待完成事件。
• 错误码是负数，需要转换为正数 errno。
• io_uring 支持缓冲区注册（io_uring_register_buffers）和文件描述符注册（io_uring_register_files），以实现零拷贝和减少系统调用，这在高性能场景下非常重要。在我们的示例中，为了简化，没有使用这些高级特性，但实际生产环境中应考虑。

3. 跨平台网络通信抽象层

在核心 I/O 循环之上，我们需要构建一个更高级的网络通信抽象层，例如 NetworkSocket 或 Connection。

// NetworkSocket.h
#pragma once

#include "IOLoop.h" // 包含抽象的IOLoop接口
#include <string>
#include <memory>
#include <stdexcept>

// 平台无关的Socket句柄类型
#ifdef _WIN32
using SocketHandle = SOCKET;
#else
using SocketHandle = int;
#endif

class NetworkSocket : public std::enable_shared_from_this<NetworkSocket> {
public:
    // 构造函数：接受IOLoop实例和可选的已存在socketFd
    NetworkSocket(std::shared_ptr<IOLoop> ioLoop, SocketHandle existingSocket = -1);
    ~NetworkSocket();

    // 创建一个TCP监听套接字
    bool Listen(const std::string& ip, int port, int backlog = 128);

    // 接受一个新连接
    // 当连接接受时，`onAcceptCallback`会被调用，传入新的NetworkSocket
    using AcceptCallback = std::function<void(std::shared_ptr<NetworkSocket> newSocket, int errorCode)>;
    bool AsyncAccept(AcceptCallback onAcceptCallback);

    // 连接到远程服务器
    using ConnectCallback = std::function<void(bool success, int errorCode)>;
    bool AsyncConnect(const std::string& ip, int port, ConnectCallback onConnectCallback);

    // 异步读取数据
    using ReadCallback = std::function<void(const char* data, size_t bytesRead, int errorCode)>;
    bool AsyncRead(char* buffer, size_t bytesToRead, ReadCallback onReadCallback);

    // 异步写入数据
    using WriteCallback = std::function<void(size_t bytesWritten, int errorCode)>;
    bool AsyncWrite(const char* data, size_t bytesToWrite, WriteCallback onWriteCallback);

    // 关闭套接字
    void Close();

    SocketHandle GetSocketFd() const { return _socketFd; }
    bool IsOpen() const { return _socketFd != -1; }

private:
    std::shared_ptr<IOLoop> _ioLoop;
    SocketHandle _socketFd;
    // 用于管理正在进行的I/O上下文
    // 实际项目中可能需要更复杂的并发控制
    std::map<void*, std::shared_ptr<IOContext>> _pendingIOContexts; // key是原始IOContext指针

    // 内部完成回调，将通用IOContext转换为NetworkSocket特定的回调
    void HandleIOCompletion(IOOperationType type, void* userData, long result, int errorCode);

    // 用于Accept操作的临时套接字和地址信息
    // Windows的AcceptEx和Linux的io_uring_prep_accept参数不同，需要各自处理
    // 这里简化，AcceptEx需要一个预先创建的socket，io_uring_prep_accept会返回新的socket
#ifdef _WIN32
    SocketHandle _acceptSocket; // For AcceptEx
#endif
};

// 辅助函数：初始化Winsock
#ifdef _WIN32
static bool InitWinsock() {
    WSADATA wsaData;
    return WSAStartup(MAKEWORD(2, 2), &wsaData) == 0;
}
static void CleanupWinsock() {
    WSACleanup();
}
#endif

// NetworkSocket.cpp
#include "NetworkSocket.h"
#include <iostream>

#ifdef _WIN32
#include <winsock2.h>
#include <ws2tcpip.h>
#else
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#endif

NetworkSocket::NetworkSocket(std::shared_ptr<IOLoop> ioLoop, SocketHandle existingSocket)
    : _ioLoop(std::move(ioLoop)),
      _socketFd(existingSocket)
{
#ifdef _WIN32
    _acceptSocket = INVALID_SOCKET;
#endif
    if (_socketFd == -1) {
        _socketFd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
#ifdef _WIN32
        if (_socketFd == INVALID_SOCKET) {
#else
        if (_socketFd < 0) {
#endif
            throw std::runtime_error("Failed to create socket.");
        }
    }
    _ioLoop->RegisterHandle(_socketFd);
}

NetworkSocket::~NetworkSocket() {
    Close();
}

bool NetworkSocket::Listen(const std::string& ip, int port, int backlog) {
    if (_socketFd == -1) return false;

    sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    inet_pton(AF_INET, ip.c_str(), &addr.sin_addr);

    if (bind(_socketFd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) == -1) {
        std::cerr << "Bind failed: " << _ioLoop->GetLastError() << std::endl;
        Close();
        return false;
    }

    if (listen(_socketFd, backlog) == -1) {
        std::cerr << "Listen failed: " << _ioLoop->GetLastError() << std::endl;
        Close();
        return false;
    }
    return true;
}

bool NetworkSocket::AsyncAccept(AcceptCallback onAcceptCallback) {
    if (_socketFd == -1) return false;

    // 创建一个新的socket用于AcceptEx，或者作为io_uring_prep_accept的输出
#ifdef _WIN32
    _acceptSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (_acceptSocket == INVALID_SOCKET) {
        std::cerr << "Failed to create accept socket: " << _ioLoop->GetLastError() << std::endl;
        return false;
    }
    // 还需要为 AcceptEx 准备一个额外的缓冲区来接收地址信息等，这里简化
#endif

    auto context = std::make_shared<IOContext>();
    context->type = IOOperationType::ACCEPT;
    context->socketFd = _socketFd; // 监听套接字
    context->userData = new AcceptCallback(onAcceptCallback); // 将回调函数存储在userData中

#ifdef _WIN32
    // Windows AcceptEx 需要一个预先创建的socket作为新连接的socket
    // 并需要一个特殊的缓冲区来接收地址信息
    // 这里为了简化，假设已经处理了 AcceptEx 的复杂参数
    // 实际需要 GetAcceptExSockaddrs 来解析地址
    // 还需要 GetQueuedCompletionStatus 的 CompletionKey 来标识新连接socket
    // 这里我们直接将新连接socket作为userData传递给回调
    // 这是一个简化的 AcceptEx 模拟，并非完整实现
    // AcceptEx 函数原型复杂，需要多个参数
    // BOOL AcceptEx(
    //     SOCKET sListenSocket,
    //     SOCKET sAcceptSocket,
    //     PVOID lpOutputBuffer,
    //     DWORD dwReceiveDataLength,
    //     DWORD dwLocalAddressLength,
    //     DWORD dwRemoteAddressLength,
    //     LPDWORD lpdwBytesReceived,
    //     LPOVERLAPPED lpOverlapped
    // );
    // 这里我们无法直接调用 AcceptEx，因为它需要一个输出缓冲区
    // 而是通过 IOCPIOLoop 的 SubmitIO 提交一个 IOOperationType::ACCEPT，
    // 在 IOCPIOLoop 内部处理 AcceptEx 的细节。
    // 为了让示例跑通，我们在此处直接模拟一个成功的回调，实际需要更复杂的集成
    std::cerr << "Windows AcceptEx is complex and needs more integration with IOCPIOLoop." << std::endl;
    // 模拟成功
    auto self = shared_from_this();
    _ioLoop->SubmitIO(context); // 提交一个Accept请求
    // 假设IOCPIOLoop内部会处理AcceptEx，并在完成后调用HandleIOCompletion
#else
    // Linux io_uring_prep_accept 会在完成时返回新的 socket fd
    sockaddr_storage* client_addr = new sockaddr_storage(); // 存储客户端地址
    socklen_t* client_len = new socklen_t(sizeof(sockaddr_storage)); // 存储地址长度
    context->userData = client_addr; // 存储地址
    context->buffer.resize(sizeof(socklen_t)); // 借用buffer存储长度，实际上io_uring_prep_accept直接用socklen_t*
    *(reinterpret_cast<socklen_t*>(context->buffer.data())) = sizeof(sockaddr_storage);

    if (!_ioLoop->SubmitIO(context)) {
        delete static_cast<AcceptCallback*>(context->userData);
        delete client_addr;
        delete client_len;
        return false;
    }
#endif
    _pendingIOContexts[context.get()] = context;
    return true;
}

bool NetworkSocket::AsyncConnect(const std::string& ip, int port, ConnectCallback onConnectCallback) {
    if (_socketFd == -1) return false;

    sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    inet_pton(AF_INET, ip.c_str(), &addr.sin_addr);

    auto context = std::make_shared<IOContext>();
    context->type = IOOperationType::CONNECT;
    context->socketFd = _socketFd;
    context->userData = new ConnectCallback(onConnectCallback);
    // 连接操作需要目标地址
    context->buffer.assign(reinterpret_cast<char*>(&addr), reinterpret_cast<char*>(&addr) + sizeof(addr));

    if (!_ioLoop->SubmitIO(context)) {
        delete static_cast<ConnectCallback*>(context->userData);
        return false;
    }
    _pendingIOContexts[context.get()] = context;
    return true;
}

bool NetworkSocket::AsyncRead(char* buffer, size_t bytesToRead, ReadCallback onReadCallback) {
    if (_socketFd == -1) return false;

    auto context = std::make_shared<IOContext>();
    context->type = IOOperationType::READ;
    context->socketFd = _socketFd;
    context->buffer.assign(buffer, buffer + bytesToRead); // 拷贝到内部buffer
    context->bytesToTransfer = bytesToRead;
    context->userData = new ReadCallback(onReadCallback);
    context->callback = std::bind(&NetworkSocket::HandleIOCompletion, shared_from_this(),
                                   std::placeholders::_1, std::placeholders::_2,
                                   std::placeholders::_3, std::placeholders::_4);

    if (!_ioLoop->SubmitIO(context)) {
        delete static_cast<ReadCallback*>(context->userData);
        return false;
    }
    _pendingIOContexts[context.get()] = context;
    return true;
}

bool NetworkSocket::AsyncWrite(const char* data, size_t bytesToWrite, WriteCallback onWriteCallback) {
    if (_socketFd == -1) return false;

    auto context = std::make_shared<IOContext>();
    context->type = IOOperationType::WRITE;
    context->socketFd = _socketFd;
    context->buffer.assign(data, data + bytesToWrite); // 拷贝数据
    context->bytesToTransfer = bytesToWrite;
    context->userData = new WriteCallback(onWriteCallback);
    context->callback = std::bind(&NetworkSocket::HandleIOCompletion, shared_from_this(),
                                   std::placeholders::_1, std::placeholders::_2,
                                   std::placeholders::_3, std::placeholders::_4);

    if (!_ioLoop->SubmitIO(context)) {
        delete static_cast<WriteCallback*>(context->userData);
        return false;
    }
    _pendingIOContexts[context.get()] = context;
    return true;
}

void NetworkSocket::Close() {
    if (_socketFd != -1) {
        _ioLoop->UnregisterHandle(_socketFd);
#ifdef _WIN32
        closesocket(_socketFd);
#else
        close(_socketFd);
#endif
        _socketFd = -1;
    }
#ifdef _WIN32
    if (_acceptSocket != INVALID_SOCKET) {
        closesocket(_acceptSocket);
        _acceptSocket = INVALID_SOCKET;
    }
#endif
    // 清理所有悬挂的I/O上下文
    // 实际中可能需要取消这些操作
    _pendingIOContexts.clear();
}

void NetworkSocket::HandleIOCompletion(IOOperationType type, void* userData, long result, int errorCode) {
    // 从 _pendingIOContexts 中移除对应的 context，因为它已经完成了
    // 注意：这里需要一个机制来从 userData 恢复原始的 shared_ptr<IOContext>
    // 在 io_uring 中，userData 就是原始的 IOContext 指针
    // 在 IOCP 中，我们通过 WindowsIOContextWrapper 间接存储
    // 为了简化，我们假设 userData 是指向回调函数的指针，且其生命周期由 IOContext 管理
    // 在此示例中，我们直接从 userData 恢复回调并调用

    // 假设 userData 指向的是动态分配的回调对象
    if (userData) {
        switch (type) {
            case IOOperationType::READ: {
                auto* cb = static_cast<ReadCallback*>(userData);
                // IOContext的buffer中包含了数据
                (*cb)(_pendingIOContexts[static_cast<IOContext*>(userData)]->buffer.data(), result, errorCode);
                delete cb;
                break;
            }
            case IOOperationType::WRITE: {
                auto* cb = static_cast<WriteCallback*>(userData);
                (*cb)(result, errorCode);
                delete cb;
                break;
            }
            case IOOperationType::ACCEPT: {
                auto* cb = static_cast<AcceptCallback*>(userData);
#ifdef _WIN32
                // Windows AcceptEx 完成后，需要知道新的socketfd
                // 暂时模拟一个新socket，实际需要从lpOutputBuffer解析
                SocketHandle newSocketFd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); // 模拟
                auto newSocket = std::make_shared<NetworkSocket>(_ioLoop, newSocketFd);
                (*cb)(newSocket, errorCode);
                closesocket(_acceptSocket); // 关闭AcceptEx的临时socket
                _acceptSocket = INVALID_SOCKET;
#else
                // io_uring_prep_accept 的结果就是新的 socket fd
                SocketHandle newSocketFd = static_cast<SocketHandle>(result);
                // 同时，userData 存储了 sockaddr_storage*
                sockaddr_storage* client_addr = static_cast<sockaddr_storage*>(_pendingIOContexts[static_cast<IOContext*>(userData)]->userData);
                // 可以在这里使用 client_addr
                auto newSocket = std::make_shared<NetworkSocket>(_ioLoop, newSocketFd);
                (*cb)(newSocket, errorCode);
                delete client_addr; // 释放存储的地址
#endif
                delete cb;
                break;
            }
            case IOOperationType::CONNECT: {
                auto* cb = static_cast<ConnectCallback*>(userData);
                (*cb)(errorCode == 0, errorCode);
                delete cb;
                break;
            }
            default:
                std::cerr << "Unhandled IO completion type: " << static_cast<int>(type) << std::endl;
                break;
        }
    }
    // 移除已完成的IOContext
    _pendingIOContexts.erase(static_cast<IOContext*>(userData));
}

NetworkSocket 注意事项：

• NetworkSocket 负责套接字的生命周期管理和高级网络操作的封装。
• AsyncAccept 和 AsyncConnect 在 Windows 和 Linux 上存在显著差异，尤其是在参数传递和结果获取方面。示例中对 AcceptEx 和 io_uring_prep_accept 的处理进行了简化，实际生产代码需要更细致的适配。
• _pendingIOContexts 用于存储 shared_ptr<IOContext>，确保在 I/O 操作完成之前，相关的上下文对象不会被释放。当操作完成并回调后，再将其从 _pendingIOContexts 中移除。
• userData 的管理需要非常小心，这里我们假设它指向动态分配的回调对象，并在回调结束后释放。在更复杂的场景中，userData 可能是指向一个更高级的 Connection 对象的指针，由 Connection 对象来管理其生命周期。

4. 内存管理与性能考量

高性能网络通信对内存管理和性能优化有着极高的要求。

4.1 缓冲区池（Buffer Pool）

频繁的 new/delete 或 malloc/free 会导致内存碎片和性能下降。为避免这种情况，应实现一个统一的缓冲区池，预分配固定大小的内存块，并在 I/O 操作完成后回收利用。

// 简单的缓冲区池示例
#include <vector>
#include <mutex>
#include <queue>
#include <memory>

class BufferPool {
public:
    BufferPool(size_t bufferSize, size_t initialCount)
        : _bufferSize(bufferSize) {
        for (size_t i = 0; i < initialCount; ++i) {
            _buffers.push(std::make_unique<std::vector<char>>(_bufferSize));
        }
    }

    std::shared_ptr<std::vector<char>> AcquireBuffer() {
        std::lock_guard<std::mutex> lock(_mutex);
        if (_buffers.empty()) {
            // 如果池为空，可以动态创建新缓冲区，或者返回nullptr表示资源不足
            // 生产环境中应有更复杂的策略，例如扩容或等待
            return std::make_shared<std::vector<char>>(_bufferSize);
        }
        std::shared_ptr<std::vector<char>> buffer = std::move(_buffers.front());
        _buffers.pop();
        return buffer;
    }

    void ReleaseBuffer(std::shared_ptr<std::vector<char>> buffer) {
        if (buffer && buffer->size() == _bufferSize) {
            std::lock_guard<std::mutex> lock(_mutex);
            _buffers.push(std::move(buffer));
        }
    }

private:
    size_t _bufferSize;
    std::queue<std::shared_ptr<std::vector<char>>> _buffers;
    std::mutex _mutex;
};

结合 IOContext 使用：
IOContext 可以持有一个从 BufferPool 获取的 shared_ptr<std::vector<char>>。当 IOContext 完成并被销毁时，其持有的 shared_ptr 会自动释放对缓冲区的引用，如果这是最后一个引用，缓冲区会被归还到 BufferPool。

4.2 零拷贝（Zero-copy）

• io_uring 优势： io_uring 通过 io_uring_register_buffers 和 IORING_OP_READ_FIXED/IORING_OP_WRITE_FIXED 提供了真正的零拷贝机制。应用程序可以预先注册一组缓冲区，内核可以直接在这些缓冲区中进行读写，避免了数据在用户态和内核态之间的复制。
• IOCP 考量： IOCP 本身不直接提供零拷贝，但可以通过用户模式的内存映射文件（File Mapping）或使用 WSARecvMsg/WSASendMsg 配合 MSG_ZEROCOPY 标志（如果网卡驱动支持）来模拟或实现类似的效果。

4.3 批处理（Batching）

• io_uring 优势： io_uring 的环形缓冲区天生支持批处理。应用程序可以将多个 I/O 请求一次性填充到 SQ 中，然后通过一次 io_uring_submit 系统调用提交所有请求。这显著减少了系统调用开销。
• IOCP 考量： IOCP 没有直接的批处理 API，但可以通过一次性提交多个异步 I/O 操作（如多个 WSARecv）来模拟批处理，然后由 GetQueuedCompletionStatus 统一处理完成事件。

4.4 线程模型

• IOCP： 典型的 IOCP 模型是创建一个与 CPU 核心数相匹配的工作线程池。GetQueuedCompletionStatus 会智能地分发完成事件到这些线程，避免过多的线程同时运行，从而减少上下文切换。
• io_uring： 通常一个 io_uring 实例由一个线程负责提交和轮询完成事件。对于多核系统，可以创建多个 io_uring 实例，每个实例绑定一个线程，实现并行处理。或者，一个 io_uring 实例可以在一个线程中进行提交，而在另一个或多个线程中进行轮询（如果使用 IORING_SETUP_SQPOLL ）。

5. 错误处理与取消机制

错误处理：

• Windows： 使用 GetLastError() 和 WSAGetLastError() 获取错误码。
• Linux： 使用 errno 全局变量获取错误码。
• 统一： 适配层应提供一个统一的 GetLastError() 接口，并在内部根据平台返回对应的错误码。对于回调函数，可以约定传递一个整数错误码，0表示成功，非0表示失败。

取消机制：

• Windows IOCP： 使用 CancelIoEx 函数可以取消指定文件句柄上所有或某个 OVERLAPPED 结构体的未完成 I/O 操作。
• Linux io_uring： 使用 IORING_OP_ASYNC_CANCEL 操作可以异步取消一个或多个已提交的 I/O 请求。这需要在提交请求时为其分配一个 user_data 作为标识，以便后续取消。

在 IOLoop 接口中添加 CancelIO(std::shared_ptr<IOContext> context) 方法，并在平台特定实现中调用各自的取消 API。

6. 示例：简单的回声服务器

以下是一个使用我们设计的抽象层构建的简单回声服务器示例。

#include <iostream>
#include <vector>
#include <string>
#include <memory>
#include <thread>

#ifdef _WIN32
#include "WindowsIOLoop.h"
#else
#include "LinuxIOLoop.h"
#endif
#include "NetworkSocket.h"

// 负责管理客户端连接的类
class ClientConnection : public std::enable_shared_from_this<ClientConnection> {
public:
    ClientConnection(std::shared_ptr<NetworkSocket> socket)
        : _socket(std::move(socket)) {
        _buffer.resize(1024); // 接收缓冲区
    }

    void Start() {
        // 开始异步读取数据
        AsyncRead();
    }

private:
    std::shared_ptr<NetworkSocket> _socket;
    std::vector<char> _buffer;

    void AsyncRead() {
        auto self = shared_from_this();
        _socket->AsyncRead(_buffer.data(), _buffer.size(),
            [self](const char* data, size_t bytesRead, int errorCode) {
            if (errorCode == 0 && bytesRead > 0) {
                std::cout << "Received from client " << self->_socket->GetSocketFd()
                          << ": " << std::string(data, bytesRead) << std::endl;
                // 收到数据后，立即回传
                self->AsyncWrite(data, bytesRead);
            } else if (errorCode != 0) {
                std::cerr << "Client read error on socket " << self->_socket->GetSocketFd()
                          << ": " << errorCode << std::endl;
                self->_socket->Close();
            } else { // bytesRead == 0，表示连接关闭
                std::cout << "Client disconnected from socket " << self->_socket->GetSocketFd() << std::endl;
                self->_socket->Close();
            }
        });
    }

    void AsyncWrite(const char* data, size_t bytesToWrite) {
        auto self = shared_from_this();
        _socket->AsyncWrite(data, bytesToWrite,
            [self, bytesToWrite](size_t bytesWritten, int errorCode) {
            if (errorCode == 0 && bytesWritten == bytesToWrite) {
                // 写入成功，继续等待下一次读取
                self->AsyncRead();
            } else {
                std::cerr << "Client write error on socket " << self->_socket->GetSocketFd()
                          << ": " << errorCode << std::endl;
                self->_socket->Close();
            }
        });
    }
};

int main() {
#ifdef _WIN32
    if (!InitWinsock()) {
        std::cerr << "Failed to initialize Winsock." << std::endl;
        return 1;
    }
#endif

    std::shared_ptr<IOLoop> ioLoop;
#ifdef _WIN32
    ioLoop = std::make_shared<IOCPIOLoop>(4); // 4个工作线程
#else
    ioLoop = std::make_shared<IoUringIOLoop>(256); // io_uring 队列深度
#endif

    if (!ioLoop->Init()) {
        std::cerr << "Failed to initialize IOLoop." << std::endl;
        return 1;
    }

    auto serverSocket = std::make_shared<NetworkSocket>(ioLoop);
    if (!serverSocket->Listen("127.0.0.1", 8080)) {
        std::cerr << "Failed to listen on port 8080." << std::endl;
        return 1;
    }
    std::cout << "Server listening on port 8080..." << std::endl;

    // 接受新连接的lambda函数
    std::function<void(std::shared_ptr<NetworkSocket>, int)> acceptHandler;
    acceptHandler = [&serverSocket, &acceptHandler](std::shared_ptr<NetworkSocket> newSocket, int errorCode) {
        if (errorCode == 0) {
            std::cout << "Accepted new client connection on socket: " << newSocket->GetSocketFd() << std::endl;
            auto client = std::make_shared<ClientConnection>(newSocket);
            client->Start(); // 启动客户端连接的处理
        } else {
            std::cerr << "Accept error: " << errorCode << std::endl;
        }
        // 继续接受下一个连接
        serverSocket->AsyncAccept(acceptHandler);
    };

    // 启动异步接受第一个连接
    serverSocket->AsyncAccept(acceptHandler);

    // 运行I/O循环
    ioLoop->Run();

    std::cout << "Server stopped." << std::endl;

#ifdef _WIN32
    CleanupWinsock();
#endif
    return 0;
}

编译与运行：

• Windows (MSVC):

  cl /EHsc /std:c++17 /Fo:WindowsIOLoop.obj /c WindowsIOLoop.cpp
  cl /EHsc /std:c++17 /Fo:NetworkSocket.obj /c NetworkSocket.cpp
  cl /EHsc /std:c++17 /Fo:main.obj /c main.cpp
  link WindowsIOLoop.obj NetworkSocket.obj main.obj ws2_32.lib /OUT:echo_server.exe

  (注意：AcceptEx的复杂性在示例中简化，可能需要进一步实现细节)

• Linux (g++):

  g++ -std=c++17 -Wall -I/usr/local/include -L/usr/local/lib LinuxIOLoop.cpp NetworkSocket.cpp main.cpp -o echo_server -luring -lpthread

  (假设 liburing 已安装到 /usr/local 路径下)

7. 进一步的优化与高级特性

• 定时器： 引入一个统一的定时器管理机制，在 IOCP 中可以使用 SetTimerQueueTimer 或 CreateTimerQueueTimer，io_uring 则可以直接通过 IORING_OP_TIMEOUT 操作。
• 信号处理： 将信号事件也纳入 I/O 循环管理。
• 多线程模型： 考虑更复杂的线程模型，例如每个 IOLoop 实例一个线程，或者多个 IOLoop 实例共享一个工作线程池（但可能增加锁竞争）。
• UDP 支持： 扩展 NetworkSocket 以支持 UDP 异步收发。
• SSL/TLS 集成： 在 NetworkSocket 层之上集成 OpenSSL 或 BoringSSL 等库，提供加密通信。
• Metrics & Tracing： 收集性能指标和调用链追踪数据，以便于监控和调试。
• 连接池： 对于客户端，实现连接池以复用连接。

总结与展望

通过上述设计，我们成功构建了一个 C++ 跨平台高性能网络通信内核的初步框架，它能够有效地抽象 Windows IOCP 和 Linux io_uring 的底层差异，向上层应用提供统一、简洁的异步 I/O 编程接口。虽然示例代码为了简化略去了部分复杂性，但核心思想和架构模式已经清晰展现。

未来，随着 io_uring 在 Linux 生态中的普及和功能增强，以及 C++ 20/23 等新标准对协程的原生支持，异步网络编程将变得更加高效和易于编写。这个跨平台适配层正是通往这些先进技术的基础，它使得开发者能够专注于业务逻辑，而无需深陷操作系统底层 I/O 机制的泥沼。