C++ 跨平台适配层抽象：构建一套同时兼容 Windows IOCP 与 Linux epoll 的高性能 C++ 网络通信内核

欢迎来到本次技术讲座，主题是“C++ 跨平台适配层抽象：构建一套同时兼容 Windows IOCP 与 Linux epoll 的高性能 C++ 网络通信内核”。在当今互联互通的世界中，高性能网络通信是几乎所有服务端应用的核心。然而，操作系统底层网络I/O模型的多样性，尤其是Windows上的I/O完成端口（IOCP）与Linux上的epoll，给跨平台应用开发带来了巨大的挑战。直接使用条件编译（#ifdef _WIN32）会导致代码冗余、难以维护和测试。

本次讲座的目标是深入探讨如何设计并实现一个优雅、高效的C++抽象层，将IOCP和epoll的底层差异封装起来，为上层应用提供一套统一、简洁且高性能的网络I/O事件处理机制。我们将从底层原理出发，逐步构建起这个抽象层，并讨论其关键设计考量和实现细节。

1. 跨平台高性能网络通信的挑战与机遇

现代网络服务对并发连接数、吞吐量和延迟都有极高的要求。为了满足这些需求，操作系统提供了各自的高效I/O复用机制。

Windows I/O完成端口 (IOCP)
IOCP是Windows上处理大量并发I/O操作的黄金标准。它基于异步I/O和线程池模型，能够有效地避免“一个连接一个线程”的资源消耗问题。当一个异步I/O操作完成时，操作系统会将一个完成包（completion packet）投递到与IOCP关联的完成队列中，并由一个或多个工作线程从队列中获取并处理。IOCP的优势在于其天然的线程池管理和负载均衡能力，可以根据系统负载动态调整工作线程数量。

Linux epoll
epoll是Linux上处理大量并发I/O事件的利器。它取代了传统的select和poll，以其更高的性能和可伸缩性而闻名。epoll通过将感兴趣的文件描述符（包括socket）注册到一个内核事件表中，并仅在这些文件描述符上有事件发生时才通知应用程序，从而避免了每次调用都需要遍历所有文件描述符的开销。epoll支持边缘触发（ET）和水平触发（LT）两种模式，其中ET模式通常用于追求极致性能的场景。

挑战
尽管IOCP和epoll都提供了高性能的I/O复用能力，但它们的API设计、工作模式和事件模型截然不同。

• API差异：函数名称、参数、返回值、数据结构完全不同。
• 事件模型：IOCP是“完成”驱动的，应用程序提交一个异步操作，等待其完成；epoll是“就绪”驱动的，应用程序查询哪些文件描述符已准备好进行读写。
• 线程模型：IOCP内建线程池和完成队列，倾向于多线程处理；epoll通常在一个或少数几个线程中轮询事件，再分发给工作线程或直接处理。
• 资源管理：各自对socket句柄、事件上下文的管理方式不同。

机遇
通过构建一个精心设计的抽象层，我们可以：

• 提高代码可移植性：一套代码库可以编译运行在Windows和Linux上，无需大量平台特定代码。
• 降低开发复杂性：为上层应用提供统一、简洁的API，开发者无需关心底层I/O模型的细节。
• 确保高性能：在抽象层下充分利用各自平台的原生高性能I/O机制。
• 提升维护性：平台相关的逻辑集中在抽象层内部，便于管理和更新。

2. 核心概念：IOCP与epoll再探

在深入抽象层设计之前，我们有必要回顾一下IOCP和epoll的关键机制。

2.1 Windows I/O完成端口 (IOCP)

IOCP的核心思想是将异步I/O操作的完成通知与一个工作线程池关联起来。

工作原理

1. 创建完成端口：使用 CreateIoCompletionPort 创建一个I/O完成端口句柄。
2. 关联文件句柄：将一个或多个可支持异步I/O的文件句柄（如socket）与完成端口关联起来。每次关联时可以指定一个CompletionKey，用于在I/O完成时识别是哪个句柄的完成事件。
3. 发起异步I/O：应用程序调用异步I/O函数，如WSARecv、WSASend、AcceptEx等，并传入一个OVERLAPPED结构体。这些函数会立即返回，I/O操作在后台进行。
4. 等待I/O完成：一个或多个工作线程调用 GetQueuedCompletionStatus 函数，阻塞等待I/O操作完成。
5. 处理完成通知：当一个异步I/O操作完成时，系统会将一个完成包放入完成端口的队列。GetQueuedCompletionStatus 返回，工作线程获取完成包，其中包含CompletionKey、OVERLAPPED结构体指针、传输的字节数和错误码，应用程序据此处理完成的I/O事件。
6. 投递自定义完成包：应用程序还可以使用 PostQueuedCompletionStatus 主动向完成端口投递自定义的完成包，用于实现内部消息传递或优雅关闭。

关键数据结构

• HANDLE hCompletionPort: 完成端口句柄。
• SOCKET s: 关联的socket句柄。
• ULONG_PTR CompletionKey: 关联socket的上下文标识。
• OVERLAPPED overlapped: 用于异步I/O操作的结构体，必须在操作期间保持有效。通常会嵌入到自定义的结构体中，以便在完成时获取更多上下文信息。

IOCP的优势

• 高效的线程管理：操作系统负责管理工作线程的唤醒和睡眠，避免了忙等待。
• 无锁队列：内核层面实现的完成队列通常是无锁的，性能极高。
• 负载均衡：多个工作线程可以同时从队列中获取完成包，自动实现负载均衡。

2.2 Linux epoll

epoll是Linux下一种高效的I/O事件通知机制。

工作原理

1. 创建epoll实例：使用 epoll_create1 创建一个epoll实例的文件描述符。
2. 注册/修改/删除事件：使用 epoll_ctl 向epoll实例中添加、修改或删除感兴趣的文件描述符及其事件类型（如EPOLLIN表示可读，EPOLLOUT表示可写）。
   • epoll_event 结构体：定义了事件类型和用户数据（epoll_data_t），后者通常用于存储指向自定义上下文对象的指针。
3. 等待事件：应用程序调用 epoll_wait 函数，阻塞等待注册的文件描述符上发生事件。
4. 处理就绪事件：epoll_wait 返回后，会填充一个 epoll_event 结构体数组，其中包含所有就绪的事件。应用程序遍历这个数组，根据epoll_data.ptr获取上下文，并处理相应的I/O事件。

关键数据结构

• int epoll_fd: epoll实例的文件描述符。
• int sock_fd: 注册到epoll的socket文件描述符。
• struct epoll_event event: 用于epoll_ctl和epoll_wait。
  • uint32_t events: 感兴趣的事件类型，如EPOLLIN | EPOLLOUT | EPOLLET。
  • epoll_data_t data: 用户数据，通常是void* ptr，指向应用程序定义的上下文。

epoll的优势

• O(1) 复杂度：事件列表的查找和维护在内核中以O(1)复杂度完成，不受注册文件描述符数量的影响。
• 高效通知：只通知应用程序有事件发生的文件描述符，避免了遍历所有文件描述符的开销。
• 边缘触发 (ET) 模式：只在状态发生变化时通知一次，需要应用程序一次性读写完所有数据，可以减少不必要的系统调用。

3. 抽象层设计：统一事件模型

构建跨平台网络通信内核的关键在于定义一个统一的事件模型和一套统一的API。我们的目标是让上层应用无需知道底层使用的是IOCP还是epoll。

3.1 核心组件抽象

为了实现这一目标，我们需要抽象出以下几个核心组件：

1. IIOEventHandler (I/O事件处理器接口)

   • 定义了所有I/O事件回调的纯虚函数接口。
   • 例如：on_read_complete, on_write_complete, on_accept_complete, on_connect_complete, on_error, on_close。
   • 所有需要处理I/O事件的实体（如TcpConnection、TcpServer）都将实现此接口。

2. IIOEventLoop (I/O事件循环接口)

   • 这是抽象层的核心，负责管理底层的I/O复用机制。
   • 定义了注册/注销I/O事件处理器、启动/停止事件循环等通用API。
   • 例如：register_socket(NativeSocketHandle fd, IIOEventHandler* handler, EventType events)
   • unregister_socket(NativeSocketHandle fd)
   • post_task(std::function<void()> task) (用于在事件循环线程中执行自定义任务)
   • run(), stop()

3. IOContext (I/O操作上下文)

   • 用于在I/O操作完成时，将事件与特定的IIOEventHandler及其相关数据关联起来。
   • 在Windows上，这会包含OVERLAPPED结构体。
   • 在Linux上，这会通过epoll_data.ptr指向。

4. NativeSocketHandle

   • 一个平台无关的socket句柄类型别名。
   • 在Windows上是SOCKET，在Linux上是int。

3.2 统一的事件类型

尽管IOCP和epoll的事件通知机制不同，但它们最终都代表了类似的I/O状态变化。我们可以定义一套统一的事件类型：

统一事件类型	Windows IOCP 对应机制	Linux epoll 对应机制	描述
READ_COMPLETE	WSARecv 完成	EPOLLIN 事件触发后，读取操作完成	数据已成功读取到缓冲区中
WRITE_COMPLETE	WSASend 完成	EPOLLOUT 事件触发后，写入操作完成	数据已成功从缓冲区发送出去
ACCEPT_COMPLETE	AcceptEx 完成	EPOLLIN 事件触发后，accept 操作完成	新连接已建立
CONNECT_COMPLETE	ConnectEx 完成	EPOLLOUT (或非阻塞连接成功) 事件触发后，连接操作完成	异步连接已建立或失败
ERROR_OCCURRED	GetQueuedCompletionStatus 返回错误或I/O操作失败	EPOLLERR / EPOLLHUP 事件触发或I/O操作失败	发生I/O错误或连接断开
CUSTOM_TASK	PostQueuedCompletionStatus 投递自定义任务	eventfd 触发后，处理自定义任务	用于在事件循环线程中执行非I/O任务或关闭信号

3.3 抽象层的核心接口定义

// common/NativeTypes.h
#ifdef _WIN32
#include <winsock2.h>
#pragma comment(lib, "ws2_32.lib")
using NativeSocketHandle = SOCKET;
#define INVALID_NATIVE_SOCKET INVALID_SOCKET
#define NATIVE_SOCKET_ERROR SOCKET_ERROR
#define GET_LAST_SOCKET_ERROR WSAGetLastError
#else
#include <sys/socket.h>
#include <unistd.h>
using NativeSocketHandle = int;
#define INVALID_NATIVE_SOCKET -1
#define NATIVE_SOCKET_ERROR -1
#define GET_LAST_SOCKET_ERROR errno
#endif

// common/EventTypes.h
enum class EventType {
    READ_COMPLETE,
    WRITE_COMPLETE,
    ACCEPT_COMPLETE,
    CONNECT_COMPLETE,
    ERROR_OCCURRED,
    CUSTOM_TASK,
    // ... 其他可能的事件，如 TIMER, SIGNAL 等
};

// common/IIOEventHandler.h
class IIOEventLoop; // 前向声明

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

    // 处理 I/O 完成事件
    // sock_fd: 触发事件的socket句柄
    // type: 事件类型
    // bytes_transferred: 传输的字节数 (适用于读写完成)
    // error_code: 错误码 (0表示成功)
    // context_data: 与事件相关的自定义上下文数据 (例如，指向 IOContext 的指针)
    virtual void on_io_event(NativeSocketHandle sock_fd, EventType type,
                             size_t bytes_transferred, int error_code, void* context_data) = 0;

    // 当处理器被注册到事件循环时调用
    virtual void on_registered(IIOEventLoop* loop) {}

    // 当处理器从事件循环中注销时调用
    virtual void on_unregistered(IIOEventLoop* loop) {}
};

// common/IIOEventLoop.h
#include <functional>
#include <memory>
#include <vector>

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

    // 注册一个socket及其事件处理器。
    // 在Windows上，会关联到IOCP。
    // 在Linux上，会添加到epoll实例。
    // initial_events: 在Linux上是 EPOLLIN | EPOLLOUT 等，在Windows上通常忽略。
    virtual bool register_socket(NativeSocketHandle fd, IIOEventHandler* handler, uint32_t initial_events = 0) = 0;

    // 从事件循环中注销一个socket。
    virtual void unregister_socket(NativeSocketHandle fd) = 0;

    // 启动事件循环。会阻塞直到 stop() 被调用。
    virtual void run() = 0;

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

    // 异步投递一个任务到事件循环线程执行。
    // 适用于在非事件循环线程中安全地与事件循环交互。
    virtual bool post_task(std::function<void()> task) = 0;

    // 投递一个I/O完成事件到事件循环。
    // 主要用于IOCP模型，但在epoll模型中也可用于模拟完成事件或内部通知。
    virtual bool post_io_event(NativeSocketHandle fd, EventType type,
                               size_t bytes_transferred, int error_code, void* context_data) = 0;
};

// IOContext 抽象 (具体实现会包含平台特定结构)
// common/IOContext.h
struct IOContext {
    IIOEventHandler* handler; // 哪个处理器负责此事件
    NativeSocketHandle sock_fd; // 关联的socket
    EventType operation_type; // 此次I/O操作的类型 (例如，READ, WRITE, ACCEPT, CONNECT)

    // 平台特定的数据成员，通过匿名union或条件编译来管理
#ifdef _WIN32
    OVERLAPPED overlapped; // Windows IOCP 需要
    // 额外的缓冲区指针、长度等，根据具体操作存储
    WSABUF wsa_buf;
#else
    // Linux epoll 不直接使用此结构体，但其信息可以通过 epoll_data.ptr 传递
    // 或者当 epoll_wait 返回事件后，通过 handler 和 sock_fd 重新构建上下文
    // 但为了统一，我们可以让epoll的事件回调也接收一个IOContext*。
    // 额外的缓冲区指针、长度等
    char* buffer;
    size_t buffer_len;
#endif

    // 构造函数等
    IOContext(IIOEventHandler* h, NativeSocketHandle fd, EventType op_type)
        : handler(h), sock_fd(fd), operation_type(op_type) {
#ifdef _WIN32
        memset(&overlapped, 0, sizeof(overlapped));
        memset(&wsa_buf, 0, sizeof(wsa_buf));
#endif
    }
};

4. 平台特定实现

现在，我们来详细看看 IIOEventLoop 接口在Windows和Linux上的具体实现。

4.1 Windows IOCPEventLoop 实现

IOCPEventLoop 将封装 CreateIoCompletionPort、GetQueuedCompletionStatus 和相关异步I/O函数。

关键点

• 完成端口句柄：一个 HANDLE 类型的成员变量。
• 工作线程池：维护一组线程，它们都调用 GetQueuedCompletionStatus 等待I/O完成。
• OVERLAPPED_EX 结构：自定义一个继承自OVERLAPPED的结构体，用于在I/O完成时携带更多上下文信息，尤其是指向 IOContext 的指针。
• Socket与CompletionKey的映射：需要一个机制来存储 CompletionKey 到 IIOEventHandler 的映射，因为 GetQueuedCompletionStatus 返回的是 CompletionKey。

// windows/IOCPEventLoop.h
#include "../common/IIOEventLoop.h"
#include "../common/NativeTypes.h"
#include "../common/IOContext.h"
#include <map>
#include <thread>
#include <vector>
#include <atomic>
#include <mutex>
#include <queue>

// 扩展 OVERLAPPED 结构，用于携带更多上下文信息
struct OVERLAPPED_EX : public OVERLAPPED {
    IOContext* io_context; // 指向我们自定义的 IOContext
    // ... 其他可能需要的字段
};

class IOCPEventLoop : public IIOEventLoop {
public:
    IOCPEventLoop();
    ~IOCPEventLoop();

    bool register_socket(NativeSocketHandle fd, IIOEventHandler* handler, uint32_t initial_events = 0) override;
    bool unregister_socket(NativeSocketHandle fd) override;
    void run() override;
    void stop() override;
    bool post_task(std::function<void()> task) override;
    bool post_io_event(NativeSocketHandle fd, EventType type,
                       size_t bytes_transferred, int error_code, void* context_data) override;

private:
    void worker_thread_func(); // IOCP工作线程函数
    void cleanup_socket_resources(NativeSocketHandle fd); // 清理socket资源

    HANDLE _iocp_handle;
    std::vector<std::thread> _worker_threads;
    std::atomic<bool> _running;
    std::map<NativeSocketHandle, IIOEventHandler*> _fd_to_handler; // 映射 socket 到处理器
    std::mutex _handler_map_mutex; // 保护 _fd_to_handler

    // 用于 post_task 的队列
    std::queue<std::function<void()>> _task_queue;
    std::mutex _task_queue_mutex;
};

// windows/IOCPEventLoop.cpp
#include "IOCPEventLoop.h"
#include <iostream>

IOCPEventLoop::IOCPEventLoop() : _iocp_handle(nullptr), _running(false) {
    // 初始化 Winsock
    WSADATA wsaData;
    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0) {
        throw std::runtime_error("WSAStartup failed");
    }

    _iocp_handle = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
    if (_iocp_handle == NULL) {
        WSACleanup();
        throw std::runtime_error("CreateIoCompletionPort failed");
    }
}

IOCPEventLoop::~IOCPEventLoop() {
    stop();
    if (_iocp_handle) {
        CloseHandle(_iocp_handle);
        _iocp_handle = nullptr;
    }
    WSACleanup();
}

bool IOCPEventLoop::register_socket(NativeSocketHandle fd, IIOEventHandler* handler, uint32_t initial_events) {
    // 将 socket fd 与 IOCP 关联起来。CompletionKey 可以是 fd 或者 handler 地址。
    // 这里我们使用 fd 作为 CompletionKey，便于 GetQueuedCompletionStatus 返回时识别。
    HANDLE result = CreateIoCompletionPort((HANDLE)fd, _iocp_handle, (ULONG_PTR)fd, 0);
    if (result == NULL) {
        std::cerr << "CreateIoCompletionPort for socket failed: " << GET_LAST_SOCKET_ERROR() << std::endl;
        return false;
    }

    std::lock_guard<std::mutex> lock(_handler_map_mutex);
    _fd_to_handler[fd] = handler;
    handler->on_registered(this); // 通知 handler 已注册
    return true;
}

bool IOCPEventLoop::unregister_socket(NativeSocketHandle fd) {
    std::lock_guard<std::mutex> lock(_handler_map_mutex);
    auto it = _fd_to_handler.find(fd);
    if (it != _fd_to_handler.end()) {
        it->second->on_unregistered(this); // 通知 handler 已注销
        _fd_to_handler.erase(it);
        cleanup_socket_resources(fd); // 清理资源，例如关闭socket
        return true;
    }
    return false;
}

void IOCPEventLoop::run() {
    if (_running.exchange(true)) {
        return; // 已经运行
    }

    // 启动与CPU核心数相同数量的工作线程
    unsigned int num_threads = std::thread::hardware_concurrency();
    if (num_threads == 0) num_threads = 2; // 至少两个线程

    for (unsigned int i = 0; i < num_threads; ++i) {
        _worker_threads.emplace_back(&IOCPEventLoop::worker_thread_func, this);
    }

    // 主线程也可以作为工作线程，或者做其他事情。这里我们让它也作为工作线程。
    worker_thread_func();
}

void IOCPEventLoop::stop() {
    if (!_running.exchange(false)) {
        return; // 已经停止
    }

    // 向所有工作线程投递一个特殊的完成包，使其退出
    for (size_t i = 0; i < _worker_threads.size(); ++i) {
        PostQueuedCompletionStatus(_iocp_handle, 0, 0, NULL);
    }
    // 同时向主线程投递一个，如果主线程也在 worker_thread_func() 中
    PostQueuedCompletionStatus(_iocp_handle, 0, 0, NULL);

    for (std::thread& t : _worker_threads) {
        if (t.joinable()) {
            t.join();
        }
    }
    _worker_threads.clear();
}

bool IOCPEventLoop::post_task(std::function<void()> task) {
    if (!_running) return false;

    {
        std::lock_guard<std::mutex> lock(_task_queue_mutex);
        _task_queue.push(std::move(task));
    }
    // 投递一个特殊的完成包来唤醒一个工作线程处理任务
    return PostQueuedCompletionStatus(_iocp_handle, 0, (ULONG_PTR)nullptr, nullptr); // CompletionKey 为 nullptr 代表自定义任务
}

bool IOCPEventLoop::post_io_event(NativeSocketHandle fd, EventType type,
                                 size_t bytes_transferred, int error_code, void* context_data) {
    if (!_running) return false;

    // 直接投递一个I/O完成事件。这主要用于模拟或内部通信。
    // context_data 应该是指向 OVERLAPPED_EX 的指针
    return PostQueuedCompletionStatus(_iocp_handle, (DWORD)bytes_transferred, (ULONG_PTR)fd, (LPOVERLAPPED)context_data);
}

void IOCPEventLoop::worker_thread_func() {
    DWORD bytes_transferred = 0;
    ULONG_PTR completion_key = 0;
    LPOVERLAPPED lpOverlapped = nullptr;

    while (_running) {
        BOOL ret = GetQueuedCompletionStatus(
            _iocp_handle,
            &bytes_transferred,
            &completion_key,
            &lpOverlapped,
            INFINITE // 永久等待
        );

        if (!ret && lpOverlapped == nullptr) {
            // GetQueuedCompletionStatus 失败且 lpOverlapped 为 NULL，通常是 IOCP 本身关闭或严重错误
            // 或者 PostQueuedCompletionStatus 投递了 NULL (用于退出线程)
            if (!_running) { // 检查是否是停止信号
                break;
            }
            std::cerr << "GetQueuedCompletionStatus failed (no overlapped): " << GET_LAST_SOCKET_ERROR() << std::endl;
            continue;
        }

        // 处理自定义任务
        if (completion_key == (ULONG_PTR)nullptr && lpOverlapped == nullptr) {
            std::function<void()> task;
            {
                std::lock_guard<std::mutex> lock(_task_queue_mutex);
                if (!_task_queue.empty()) {
                    task = _task_queue.front();
                    _task_queue.pop();
                }
            }
            if (task) {
                task();
            }
            continue; // 继续等待下一个事件
        }

        // 处理 I/O 完成事件
        OVERLAPPED_EX* ov_ex = static_cast<OVERLAPPED_EX*>(lpOverlapped);
        if (!ov_ex || !ov_ex->io_context) {
            std::cerr << "Invalid OVERLAPPED_EX or IOContext received." << std::endl;
            continue;
        }

        IOContext* ctx = ov_ex->io_context;
        NativeSocketHandle sock_fd = ctx->sock_fd;
        IIOEventHandler* handler = ctx->handler;

        int error_code = 0;
        if (!ret) { // I/O操作本身失败
            error_code = GET_LAST_SOCKET_ERROR();
        }

        if (handler) {
            handler->on_io_event(sock_fd, ctx->operation_type, bytes_transferred, error_code, ctx);
        }
        // 注意：IOContext 的内存管理需要在 on_io_event 内部处理，通常是释放或回收。
        // 例如，如果在一个TcpConnection的SendBuffer中，SendBuffer完成时会释放对应的IOContext。
    }
}

void IOCPEventLoop::cleanup_socket_resources(NativeSocketHandle fd) {
    if (fd != INVALID_NATIVE_SOCKET) {
        // 关闭 socket 句柄
        closesocket(fd);
    }
    // 在这里也可以进行其他与socket相关的资源清理
}

OVERLAPPED_EX 的重要性
OVERLAPPED_EX 结构体是IOCP的精髓之一。应用程序发起异步I/O时，会传入一个 OVERLAPPED 结构体指针。当I/O完成时，GetQueuedCompletionStatus 会返回这个指针。通过将 OVERLAPPED 嵌入到我们自定义的 OVERLAPPED_EX 中，并在其中加入 IOContext* 指针，我们可以在I/O完成时，直接获取到与该操作相关的上下文信息（如是哪个 TcpConnection 的读操作完成，以及读到的缓冲区信息等），而无需额外的查找开销。

4.2 Linux EpollEventLoop 实现

EpollEventLoop 将封装 epoll_create1、epoll_ctl 和 epoll_wait。

关键点

• epoll 实例句柄：一个 int 类型的成员变量。
• 事件循环线程：通常是一个单独的线程，调用 epoll_wait。
• eventfd：用于实现 post_task 和 stop 机制。当需要向事件循环线程发送内部信号时，写入 eventfd，epoll_wait 会被唤醒。
• epoll_event::data.ptr 映射：将 IIOEventHandler* 或 IOContext* 直接存储到 epoll_event 的 data.ptr 中，以便在事件就绪时直接获取。
• 边缘触发 (ET) 模式：通常使用ET模式以获得最高性能，这意味着当EPOLLIN或EPOLLOUT事件触发时，应用程序必须一次性读写完所有数据，直到 recv/send 返回 EAGAIN/EWOULDBLOCK。

// linux/EpollEventLoop.h
#include "../common/IIOEventLoop.h"
#include "../common/NativeTypes.h"
#include "../common/IOContext.h"
#include <map>
#include <thread>
#include <atomic>
#include <vector>
#include <mutex>
#include <queue>
#include <sys/epoll.h>

class EpollEventLoop : public IIOEventLoop {
public:
    EpollEventLoop();
    ~EpollEventLoop();

    bool register_socket(NativeSocketHandle fd, IIOEventHandler* handler, uint32_t initial_events = 0) override;
    bool unregister_socket(NativeSocketHandle fd) override;
    void run() override;
    void stop() override;
    bool post_task(std::function<void()> task) override;
    bool post_io_event(NativeSocketHandle fd, EventType type,
                       size_t bytes_transferred, int error_code, void* context_data) override;

private:
    void event_loop_thread_func(); // epoll事件循环线程函数
    void handle_eventfd_read(); // 处理 eventfd 上的事件

    int _epoll_fd;
    NativeSocketHandle _event_fd; // 用于 post_task 和 stop 信号
    std::thread _event_thread;
    std::atomic<bool> _running;
    std::map<NativeSocketHandle, IIOEventHandler*> _fd_to_handler; // 映射 socket 到处理器
    std::mutex _handler_map_mutex; // 保护 _fd_to_handler

    // 用于 post_task 的队列
    std::queue<std::function<void()>> _task_queue;
    std::mutex _task_queue_mutex;
};

// linux/EpollEventLoop.cpp
#include "EpollEventLoop.h"
#include <iostream>
#include <fcntl.h>
#include <errno.h>
#include <sys/eventfd.h>

EpollEventLoop::EpollEventLoop() : _epoll_fd(-1), _event_fd(-1), _running(false) {
    _epoll_fd = epoll_create1(EPOLL_CLOEXEC);
    if (_epoll_fd == -1) {
        throw std::runtime_error("epoll_create1 failed");
    }

    // 创建 eventfd 用于内部通信 (post_task, stop)
    _event_fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
    if (_event_fd == -1) {
        close(_epoll_fd);
        throw std::runtime_error("eventfd failed");
    }

    // 将 eventfd 注册到 epoll 实例
    epoll_event event;
    event.events = EPOLLIN | EPOLLET; // 边缘触发
    event.data.ptr = nullptr; // 特殊标记，表示是 eventfd 事件
    if (epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, _event_fd, &event) == -1) {
        close(_event_fd);
        close(_epoll_fd);
        throw std::runtime_error("epoll_ctl add eventfd failed");
    }
}

EpollEventLoop::~EpollEventLoop() {
    stop();
    if (_epoll_fd != -1) {
        close(_epoll_fd);
        _epoll_fd = -1;
    }
    if (_event_fd != -1) {
        close(_event_fd);
        _event_fd = -1;
    }
}

bool EpollEventLoop::register_socket(NativeSocketHandle fd, IIOEventHandler* handler, uint32_t initial_events) {
    epoll_event event;
    event.events = initial_events | EPOLLERR | EPOLLHUP | EPOLLET; // 总是关注错误和挂断，使用ET模式
    event.data.ptr = handler; // 将 handler 指针作为用户数据

    if (epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, fd, &event) == -1) {
        std::cerr << "epoll_ctl add failed for fd " << fd << ": " << GET_LAST_SOCKET_ERROR() << std::endl;
        return false;
    }

    std::lock_guard<std::mutex> lock(_handler_map_mutex);
    _fd_to_handler[fd] = handler;
    handler->on_registered(this); // 通知 handler 已注册
    return true;
}

bool EpollEventLoop::unregister_socket(NativeSocketHandle fd) {
    // 从 epoll 实例中删除
    if (epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, fd, nullptr) == -1) {
        // 如果 socket 已经关闭，可能会失败，但这不是致命错误
        std::cerr << "epoll_ctl del failed for fd " << fd << ": " << GET_LAST_SOCKET_ERROR() << std::endl;
    }

    std::lock_guard<std::mutex> lock(_handler_map_mutex);
    auto it = _fd_to_handler.find(fd);
    if (it != _fd_to_handler.end()) {
        it->second->on_unregistered(this); // 通知 handler 已注销
        _fd_to_handler.erase(it);
        close(fd); // 关闭 socket 句柄
        return true;
    }
    return false;
}

void EpollEventLoop::run() {
    if (_running.exchange(true)) {
        return; // 已经运行
    }
    _event_thread = std::thread(&EpollEventLoop::event_loop_thread_func, this);
    // 主线程可以做其他事情，或者等待事件循环线程
}

void EpollEventLoop::stop() {
    if (!_running.exchange(false)) {
        return; // 已经停止
    }
    // 通过 eventfd 发送信号唤醒 epoll_wait
    uint64_t one = 1;
    ssize_t ret = write(_event_fd, &one, sizeof(one));
    if (ret != sizeof(one)) {
        std::cerr << "Failed to write to eventfd for stopping: " << GET_LAST_SOCKET_ERROR() << std::endl;
    }

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

bool EpollEventLoop::post_task(std::function<void()> task) {
    if (!_running) return false;

    {
        std::lock_guard<std::mutex> lock(_task_queue_mutex);
        _task_queue.push(std::move(task));
    }
    // 写入 eventfd 唤醒事件循环线程
    uint64_t one = 1;
    ssize_t ret = write(_event_fd, &one, sizeof(one));
    if (ret != sizeof(one)) {
        std::cerr << "Failed to write to eventfd for posting task: " << GET_LAST_SOCKET_ERROR() << std::endl;
        return false;
    }
    return true;
}

bool EpollEventLoop::post_io_event(NativeSocketHandle fd, EventType type,
                                 size_t bytes_transferred, int error_code, void* context_data) {
    // EpollEventLoop 中不直接支持 post_io_event 来模拟 IOCP 的完成包
    // 如果需要，可以在这里将事件包装成任务 post 到 _task_queue
    // 或直接调用 handler 的 on_io_event (但在非事件循环线程调用可能不安全)
    // 最佳实践是，epoll模型下的I/O操作都是同步在事件循环线程中发起和处理的。
    // 如果需要跨线程触发，请使用 post_task。
    if (context_data) {
        // 在epoll模型下，通常只有在I/O操作（如recv/send）完成后才调用on_io_event
        // 这里的 context_data 可能是 IOContext*
        IIOEventHandler* handler = static_cast<IOContext*>(context_data)->handler;
        if (handler) {
            handler->on_io_event(fd, type, bytes_transferred, error_code, context_data);
            return true;
        }
    }
    return false;
}

void EpollEventLoop::handle_eventfd_read() {
    uint64_t value;
    // 必须读取 eventfd，否则它会持续触发 EPOLLIN
    if (read(_event_fd, &value, sizeof(value)) != sizeof(value)) {
        std::cerr << "Failed to read eventfd: " << GET_LAST_SOCKET_ERROR() << std::endl;
        return;
    }

    // 执行所有排队的任务
    std::queue<std::function<void()>> tasks_to_execute;
    {
        std::lock_guard<std::mutex> lock(_task_queue_mutex);
        tasks_to_execute.swap(_task_queue); // 批量移动任务，减少锁的持有时间
    }

    while (!tasks_to_execute.empty()) {
        tasks_to_execute.front()();
        tasks_to_execute.pop();
    }
}

void EpollEventLoop::event_loop_thread_func() {
    std::vector<epoll_event> events(128); // 预分配事件数组

    while (_running) {
        int num_events = epoll_wait(_epoll_fd, events.data(), events.size(), -1); // 永久等待

        if (num_events == -1) {
            if (GET_LAST_SOCKET_ERROR() == EINTR) {
                continue; // 被信号中断，重新等待
            }
            std::cerr << "epoll_wait failed: " << GET_LAST_SOCKET_ERROR() << std::endl;
            break; // 严重错误，退出循环
        }

        for (int i = 0; i < num_events; ++i) {
            epoll_event& event = events[i];
            NativeSocketHandle fd = (NativeSocketHandle)event.data.fd; // epoll_data 联合体，fd 和 ptr 可以互用

            // 处理 eventfd 上的事件 (内部任务或停止信号)
            if (event.data.ptr == nullptr) { // 约定 eventfd 的 data.ptr 为 nullptr
                handle_eventfd_read();
                if (!_running) { // 处理完任务后检查是否需要停止
                    break;
                }
                continue;
            }

            // 处理 socket 事件
            IIOEventHandler* handler = static_cast<IIOEventHandler*>(event.data.ptr);
            if (!handler) {
                std::cerr << "Handler for fd " << fd << " not found." << std::endl;
                continue;
            }

            if (event.events & (EPOLLERR | EPOLLHUP)) {
                // 发生错误或对端挂断
                handler->on_io_event(fd, EventType::ERROR_OCCURRED, 0, GET_LAST_SOCKET_ERROR(), nullptr);
                // 通常错误后会注销并关闭 socket
                // unregister_socket(fd); // 注意：这里直接调用 unregister_socket 会修改 _fd_to_handler 
                // 最好是 post 一个任务让事件循环结束后再清理
                continue;
            }

            // 处理可读事件
            if (event.events & EPOLLIN) {
                // 在 ET 模式下，必须一次性读完所有数据
                // 具体读取操作和 on_io_event 回调会在 TcpConnection 等实现中处理
                handler->on_io_event(fd, EventType::READ_COMPLETE, 0, 0, nullptr); // 字节数和错误码在具体实现中填充
            }
            // 处理可写事件
            if (event.events & EPOLLOUT) {
                // 在 ET 模式下，必须一次性写完所有数据
                handler->on_io_event(fd, EventType::WRITE_COMPLETE, 0, 0, nullptr); // 字节数和错误码在具体实现中填充
            }
        }
    }
}

eventfd 的重要性
在epoll模型中，eventfd 是实现跨线程通信的关键。它提供了一个文件描述符，可以像管道一样被写入和读取，并且可以注册到epoll实例中。当其他线程需要唤醒epoll_wait线程并让其执行某个任务时（如post_task或stop），只需向eventfd写入一个字节，epoll_wait就会被唤醒，从而处理对应的任务。这避免了使用互斥锁和条件变量唤醒事件循环线程的复杂性，并能与I/O事件统一处理。

5. 高层应用抽象：TcpConnection 与 TcpServer

有了底层 IIOEventLoop 的抽象，我们现在可以构建上层的 TcpConnection 和 TcpServer 类，它们将实现 IIOEventHandler 接口。

5.1 TcpConnection

一个 TcpConnection 对象代表一个已建立的TCP连接。它将负责该连接的读写操作。

// network/TcpConnection.h
#include "../common/IIOEventHandler.h"
#include "../common/NativeTypes.h"
#include "../common/IOContext.h"
#include <vector>
#include <string>
#include <deque>
#include <memory>

class TcpConnection : public IIOEventHandler, public std::enable_shared_from_this<TcpConnection> {
public:
    TcpConnection(NativeSocketHandle fd, IIOEventLoop* loop);
    ~TcpConnection();

    void start(); // 启动连接的读写
    void send_data(const std::string& data); // 发送数据

    // IIOEventHandler 接口实现
    void on_io_event(NativeSocketHandle sock_fd, EventType type,
                     size_t bytes_transferred, int error_code, void* context_data) override;
    void on_registered(IIOEventLoop* loop) override;
    void on_unregistered(IIOEventLoop* loop) override;

private:
    void do_read(); // 启动异步读操作
    void do_write(); // 启动异步写操作

    NativeSocketHandle _socket_fd;
    IIOEventLoop* _event_loop;
    std::vector<char> _read_buffer; // 接收缓冲区
    std::deque<std::string> _write_queue; // 发送队列
    bool _writing; // 标记当前是否有写操作正在进行

    // IOCP 专属的 IOContext 实例，用于读写操作
    // 注意：这里用 shared_ptr 是为了保证 IOContext 在异步操作完成前不会被释放
    // 实际生产中，会使用对象池或更精细的内存管理
    std::shared_ptr<IOContext> _read_context;
    std::shared_ptr<IOContext> _write_context;
};

TcpConnection::on_io_event 逻辑

• READ_COMPLETE:
  • 在Windows上，bytes_transferred 是实际读取的字节数。
  • 在Linux上，on_io_event 触发时，需要循环调用 recv 直到返回 EAGAIN/EWOULDBLOCK 或 0 (连接关闭)。
  • 处理接收到的数据，并可能再次调用 do_read 启动下一次读取。
• WRITE_COMPLETE:
  • 在Windows上，bytes_transferred 是实际发送的字节数。
  • 在Linux上，on_io_event 触发时，需要循环调用 send 直到返回 EAGAIN/EWOULDBLOCK 或发送队列为空。
  • 发送完一个数据包后，检查发送队列，如果还有数据，继续 do_write；否则清除 _writing 标志。
• ERROR_OCCURRED: 连接错误或关闭，通知上层应用，并注销 socket_fd。

5.2 TcpServer

TcpServer 负责监听端口，接受新连接，并为每个新连接创建一个 TcpConnection 对象。

// network/TcpServer.h
#include "../common/IIOEventHandler.h"
#include "../common/NativeTypes.h"
#include "../common/IOContext.h"
#include <functional>
#include <memory>
#include <map>

class TcpServer : public IIOEventHandler {
public:
    using NewConnectionCallback = std::function<void(std::shared_ptr<TcpConnection>)>;

    TcpServer(IIOEventLoop* loop, uint16_t port, NewConnectionCallback cb);
    ~TcpServer();

    bool start_listen(); // 启动监听

    // IIOEventHandler 接口实现 (用于监听socket)
    void on_io_event(NativeSocketHandle sock_fd, EventType type,
                     size_t bytes_transferred, int error_code, void* context_data) override;

private:
    void do_accept(); // 启动异步接受新连接操作

    IIOEventLoop* _event_loop;
    uint16_t _port;
    NativeSocketHandle _listen_socket_fd;
    NewConnectionCallback _new_conn_cb;

    // IOCP 专属的 IOContext 实例，用于 accept 操作
    std::shared_ptr<IOContext> _accept_context;
    // 在 Windows 上 AcceptEx 还需要一个用于新连接的 socket 句柄
#ifdef _WIN32
    NativeSocketHandle _accept_socket_fd;
#endif

    // 管理所有活跃的连接
    std::map<NativeSocketHandle, std::shared_ptr<TcpConnection>> _connections;
    std::mutex _connections_mutex; // 保护 _connections
};

TcpServer::on_io_event 逻辑

• ACCEPT_COMPLETE:
  • 在Windows上，AcceptEx 完成，一个新的连接已建立。
  • 在Linux上，EPOLLIN 事件触发在监听socket上，调用 accept 获取新连接。
  • 创建一个新的 TcpConnection 对象，注册到 IIOEventLoop，并调用 NewConnectionCallback 通知上层应用。
  • 再次调用 do_accept 启动下一次接受新连接操作。
• ERROR_OCCURRED: 监听socket错误，通常是致命错误，停止服务。

6. 内存管理与线程模型考量

6.1 内存管理

异步I/O操作通常需要缓冲区。在IOCP中，WSARecv和WSASend需要WSABUF结构体，它包含缓冲区指针和长度。在epoll中，recv和send也需要缓冲区。

• 缓冲区池：为避免频繁的内存分配和释放，可以实现一个缓冲区池，预先分配一批固定大小的缓冲区。
• IOContext 生命周期：IOContext 结构体（或其派生类，如OVERLAPPED_EX）的生命周期必须覆盖整个异步I/O操作期间。通常，它会被嵌入到 TcpConnection 的成员变量中，或者从对象池中获取。在 on_io_event 处理完成后，负责释放或回收。
• 智能指针：使用 std::shared_ptr 或 std::unique_ptr 管理 TcpConnection 和 IOContext 的生命周期，确保在异步操作完成前对象不会被销毁。

6.2 线程模型

IOCP的线程模型

• N个工作线程：通常启动与CPU核心数相同或两倍的工作线程。这些线程都调用 GetQueuedCompletionStatus。
• 无锁设计：IOCP的完成队列是内核无锁的，多个线程可以并发地从队列中获取完成包。
• 上下文切换：操作系统负责将完成包分发给空闲的工作线程，减少了应用程序层面的线程同步开销。

epoll的线程模型

• 单事件循环线程：最常见的模式是一个线程专门调用 epoll_wait，然后将就绪事件分发给其他工作线程处理，或者直接在事件循环线程中处理。
• 多事件循环线程：对于非常高的并发场景，可以通过 SO_REUSEPORT 选项在多个端口上监听，并为每个监听socket分配一个独立的epoll事件循环线程。或者，一个epoll实例管理所有socket，但将事件分发给一个线程池处理。
• eventfd 的角色：eventfd 在这里扮演了“安全队列”的角色，允许其他线程向事件循环线程提交任务，而无需直接操作事件循环的数据结构。

跨平台抽象层的线程模型兼容
我们的 IIOEventLoop 接口需要足够灵活，以适应这两种模型。

• IOCPEventLoop 天然支持多工作线程模型。
• EpollEventLoop 默认为单事件循环线程，但可以通过 post_task 将繁重或阻塞的任务提交给其他工作线程池处理，从而避免阻塞事件循环线程。

6.3 错误处理与优雅关闭

• 统一错误码：将平台特定的错误码（WSAGetLastError() / errno）映射到一套自定义的通用错误码，方便上层应用处理。
• 连接关闭：
  • IOCP：当 WSARecv 返回0字节或错误，表示连接关闭。需要调用 closesocket。
  • epoll：EPOLLHUP 或 EPOLLIN 触发 recv 返回0字节，表示连接关闭。需要调用 close。
  • 在 IIOEventHandler::on_io_event 中，当检测到连接关闭或错误时，应调用 IIOEventLoop::unregister_socket 来清理资源并从事件循环中移除该连接。
• 服务器关闭：当 IIOEventLoop::stop() 被调用时，所有注册的socket都应该被注销并关闭。

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

• 定时器管理：许多网络应用需要定时任务（如心跳、超时）。可以在 IIOEventLoop 中集成一个定时器管理器，利用 epoll_wait 的 timeout 参数或 SetWaitableTimer (Windows) / timerfd (Linux) 来实现。
• 零拷贝：在某些高性能场景下，可以考虑使用操作系统的零拷贝机制（如Linux的sendfile，Windows的TransmitFile），但这会增加抽象层的复杂性。
• 协议解析：抽象层只负责I/O事件通知和数据传输。协议解析（如HTTP、WebSocket）应在 TcpConnection 的 on_read_complete 回调中进行，通常会使用协议编解码器。
• SSL/TLS支持：集成OpenSSL等库实现加密通信。这通常会在 TcpConnection 内部实现，对底层I/O抽象层透明。
• 日志系统：一个健壮的日志系统对于调试和监控至关重要。

8. 总结与展望

本次讲座深入探讨了如何构建一个高性能C++跨平台网络通信内核的抽象层，它能够同时兼容Windows IOCP和Linux epoll。我们设计了IIOEventHandler和IIOEventLoop等核心接口，并详细阐述了它们在Windows和Linux平台上的具体实现。通过这种方式，上层应用可以享受到统一、简洁的API，同时底层能够充分利用操作系统原生的高效I/O复用机制。

这个抽象层不仅提高了代码的可移植性和可维护性，更为构建复杂、高性能的网络服务奠定了坚实的基础。未来的工作可以进一步完善错误处理、集成更多高级特性如零拷贝、SSL/TLS支持以及分布式系统的服务发现和负载均衡能力。

构建这样的跨平台网络内核是一项复杂而有益的任务，它要求开发者对底层操作系统I/O机制有深刻的理解。通过精心设计和模块化实现，我们能够打造出既强大又灵活的网络通信基础设施，为现代C++应用的开发提供强有力的支持。