Shiina Labs
返回博客
Shiina Labs

Windows 下 IOCP 的简单使用

windowscppnetworkingiocpserver

一句话结论：IOCP 是 Windows 上的完成端口模型。核心套路是：用 WSASocketW（带 WSA_FLAG_OVERLAPPED）建监听 socket → 创建 IOCP → 把每个连接 socket 关联到 IOCP → 提交 WSARecv → 在事件线程里用 GetQueuedCompletionStatus 取完成事件并处理。

背景：为什么用 IOCP

IOCP 是 Microsoft 在 Windows 下推荐的完成端口模型。要在 Windows 上做一个性能还过得去的网络服务器，IOCP 基本绕不开。

和 Linux 的 epoll、FreeBSD 的 kqueue 相比，触发模型不一样：

  • epoll / kqueue：通常在缓冲区有内容时触发事件（epoll 默认是水平触发 LT，EPOLLET 才是边缘触发 ET）。
  • IOCP：提交重叠 I/O，等 I/O 完成（读到或写到一定字节数）后才触发完成事件。

勘误：原文把 epoll 说成“ET 边缘触发模式”不准确——epoll 同时支持 LT（默认）和 ET。

关于 IOCP 里 TCP 连接的复用还有个坑，但不是本文讨论的范畴。

核心函数

因为 IOCP 需要重叠 I/O，所以涉及的关键 API 有以下这些。

WSASocketW

SOCKET WSAAPI WSASocketW(
  [in] int                 af,
  [in] int                 type,
  [in] int                 protocol,
  [in] LPWSAPROTOCOL_INFOW lpProtocolInfo,
  [in] GROUP               g,
  [in] DWORD               dwFlags
);

创建监听 socket 要用它，重点在最后一个参数：dwFlags 设置为 WSA_FLAG_OVERLAPPED，表示这个 socket 参与重叠 I/O。

勘误：原文说“重叠 I/O 可以理解为一次 TCP 连接的标识符或 GUID”并不准确。WSA_FLAG_OVERLAPPED 是启用重叠 I/O，不是连接标识符。

ioctlsocket

int ioctlsocket(
  [in]      SOCKET s,
  [in]      long   cmd,
  [in, out] u_long *argp
);

用来设置非阻塞 I/O，这里不多展开。

CreateIoCompletionPort

HANDLE WINAPI CreateIoCompletionPort(
  _In_     HANDLE    FileHandle,
  _In_opt_ HANDLE    ExistingCompletionPort,
  _In_     ULONG_PTR CompletionKey,
  _In_     DWORD     NumberOfConcurrentThreads
);

创建 IOCP 内核对象并返回句柄。

  • FileHandle：要关联的文件/套接字句柄（第一次传 INVALID_HANDLE_VALUE 来新建端口）。
  • ExistingCompletionPort：一个已存在的 IOCP 句柄（第一次传 nullptr）。
  • CompletionKey：完成键（本文不深入使用）。
  • NumberOfConcurrentThreads：允许同时处理完成事件的最大线程数上限，不是线程池总大小。

PostQueuedCompletionStatus

BOOL WINAPI PostQueuedCompletionStatus(
  _In_     HANDLE       CompletionPort,
  _In_     DWORD        dwNumberOfBytesTransferred,
  _In_     ULONG_PTR    dwCompletionKey,
  _In_opt_ LPOVERLAPPED lpOverlapped
);

向工作线程投递一个事件。本文用它来发退出标识：把第二参数设为 -1。

GetQueuedCompletionStatus

BOOL GetQueuedCompletionStatus(
  [in]  HANDLE       CompletionPort,
        LPDWORD      lpNumberOfBytesTransferred,
  [out] PULONG_PTR   lpCompletionKey,
  [out] LPOVERLAPPED *lpOverlapped,
  [in]  DWORD        dwMilliseconds
);

用于接收 PostQueuedCompletionStatus / WSARecv / WSASend 所投递的完成事件。比 Post 版多最后一个超时参数。

WSARecv / WSASend

int WSAAPI WSARecv(
  [in]      SOCKET                             s,
  [in, out] LPWSABUF                           lpBuffers,
  [in]      DWORD                              dwBufferCount,
  [out]     LPDWORD                            lpNumberOfBytesRecvd,
  [in, out] LPDWORD                            lpFlags,
  [in]      LPWSAOVERLAPPED                    lpOverlapped,
  [in]      LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine
);

int WSAAPI WSASend(
  [in]  SOCKET                             s,
  [in]  LPWSABUF                           lpBuffers,
  [in]  DWORD                              dwBufferCount,
  [out] LPDWORD                            lpNumberOfBytesSent,
  [in]  DWORD                              dwFlags,
  [in]  LPWSAOVERLAPPED                    lpOverlapped,
  [in]  LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine
);

就是 recv / send 的 WSA 版本，多了几个参数。

自定义结构：IOType 与 IOContext

enum class IOType {
    Read,
    Write
};

IOType 用于标识操作类型，本例只处理 Read。

#define MaxBufferSize 1024

struct IOContext {
    OVERLAPPED overlapped{};
    WSABUF wsaBuf{MaxBufferSize, buffer};
    CHAR buffer[MaxBufferSize]{};
    IOType type{};
    SOCKET socket = INVALID_SOCKET;
    DWORD nBytes = 0;
};

IOContext 的第一个成员必须是 OVERLAPPED，因为 GetQueuedCompletionStatus 返回的指针会被强转成 OVERLAPPED*，这样同一个指针就能访问到整个上下文。

基本流程

  1. WSAStartup 初始化 WSA。
  2. 用 WSASocketW 建监听 socket（WSA_FLAG_OVERLAPPED），bind + listen。
  3. CreateIoCompletionPort 建 IOCP。
  4. 启动接入线程 AcceptWorkerThread（负责 accept + 关联 IOCP + 投首个 WSARecv）。
  5. 启动 NumberOfThreads 个事件线程 EventWorkerThread（负责 GetQueuedCompletionStatus 处理完成事件）。
  6. 用 PostQueuedCompletionStatus(hIOCP, -1, ...) 通知线程退出，join 等线程结束。

完整代码（带勘误注释）

原逻辑保持不变，只在出问题的地方加了 // [勘误] 注释。代码以 GitHub 仓库为准。

GitHub - SHIINASAMA/IOCP

/// \author kaoru
/// \file main.cpp
/// \version 0.2
/// \date 2022.7.21
/// \brief IOCP 示例

#include <WinSock2.h>
#pragma comment(lib, "ws2_32.lib")
#include <WS2tcpip.h>
#include <atomic>
#include <cstdint>
#include <thread>
#include <vector>

// [勘误] 原代码这里 1024 * 0.5 = 512；与上文 #define MaxBufferSize 1024 不一致
constexpr static size_t MaxBufferSize = 1024 * 0.5;
constexpr static size_t NumberOfThreads = 1;

HANDLE hIOCP = INVALID_HANDLE_VALUE;
SOCKET serverSocket = INVALID_SOCKET;
std::vector<std::thread> threadGroup;
std::atomic_bool isShutdown{false};

// 此线程用于不断接收连接，并 Post 一次 Read 事件
void AcceptWorkerThread();
// 此线程用于不断处理 AcceptWorkerThread 所 Post 过来的事件
void EventWorkerThread();

// 用于标识事件的类型
enum class IOType {
    Read,
    Write
};

struct IOContext {
    OVERLAPPED overlapped{};
    WSABUF wsaBuf{MaxBufferSize, buffer};
    CHAR buffer[MaxBufferSize]{};
    IOType type{};
    SOCKET socket = INVALID_SOCKET;
    DWORD nBytes = 0;
};

int main() {
    // 初始化 Windows 网络库
    WSAData data{};
    WSAStartup(MAKEWORD(2, 2), &data);

    struct sockaddr_in address {};
    address.sin_family = AF_INET;
    address.sin_port = htons(8080);
    inet_pton(AF_INET, "127.0.0.1", &address.sin_addr);

    // 初始化 Socket
    unsigned long ul = 1;
    serverSocket = WSASocketW(AF_INET, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
    if (INVALID_SOCKET == serverSocket) {
        // [勘误] perror 读的是 errno，Winsock 错误应改用 WSAGetLastError() / FormatMessage
        perror("FAILED TO CREATE SERVER SOCKET");
        closesocket(serverSocket);
        exit(-1);
    }
    // [勘误] 把监听 socket 设成非阻塞后，下面 AcceptWorkerThread 里的 accept 会空转（见下）
    if (SOCKET_ERROR == ioctlsocket(serverSocket, FIONBIO, &ul)) {
        perror("FAILED TO SET NONBLOCKING SOCKET");
        closesocket(serverSocket);
        exit(-2);
    }
    if (SOCKET_ERROR == bind(serverSocket, (const struct sockaddr *) &address, sizeof(address))) {
        perror("FAILED TO BIND ADDRESS");
        closesocket(serverSocket);
        exit(-3);
    }
    if (SOCKET_ERROR == listen(serverSocket, SOMAXCONN)) {
        perror("FAILED TO LISTEN SOCKET");
        closesocket(serverSocket);
        exit(-4);
    }

    // 初始化 IOCP
    hIOCP = CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, NumberOfThreads);
    if (INVALID_HANDLE_VALUE == hIOCP) {
        perror("FAILED TO CREATE IOCP HANDLE");
        closesocket(serverSocket);
        exit(-5);
    }

    // 初始化工作线程
    for (size_t i = 0; i < NumberOfThreads; i++) {
        threadGroup.emplace_back(std::thread(EventWorkerThread));
    }

    void *lpCompletionKey = nullptr;
    auto acceptThread = std::thread(AcceptWorkerThread);
    getchar();
    // 按任意键进入退出程序
    isShutdown = true;
    // 有多少个线程就发送 post 多少次，让工作线程收到事件并主动退出
    for (size_t i = 0; i < NumberOfThreads; i++) {
        PostQueuedCompletionStatus(hIOCP, -1, (ULONG_PTR) lpCompletionKey, nullptr);
    }
    acceptThread.join();
    for (auto &thread: threadGroup) {
        thread.join();
    }

    WSACleanup();
    return 0;
}

void AcceptWorkerThread() {
    while (!isShutdown) {
        // 开始监听接入
        // [勘误] 监听 socket 是非阻塞的，没有连接时 accept 会立刻返回 WSAEWOULDBLOCK，
        //        这里 continue 会让线程满速空转（busy-wait）。监听 socket 不必非阻塞。
        SOCKET clientSocket = accept(serverSocket, nullptr, nullptr);
        if (INVALID_SOCKET == clientSocket) continue;

        unsigned long ul = 1;
        if (SOCKET_ERROR == ioctlsocket(clientSocket, FIONBIO, &ul)) {
            shutdown(clientSocket, SD_BOTH);
            closesocket(clientSocket);
            continue;
        }

        if (nullptr == CreateIoCompletionPort((HANDLE) clientSocket, hIOCP, 0, 0)) {
            shutdown(clientSocket, SD_BOTH);
            closesocket(clientSocket);
            continue;
        }

        DWORD nBytes = MaxBufferSize;
        DWORD dwFlags = 0;
        auto ioContext = new IOContext;
        ioContext->socket = clientSocket;
        ioContext->type = IOType::Read;
        auto rt = WSARecv(clientSocket, &ioContext->wsaBuf, 1, &nBytes, &dwFlags, &ioContext->overlapped, nullptr);
        auto err = WSAGetLastError();
        if (SOCKET_ERROR == rt && ERROR_IO_PENDING != err) {
            // 发生不为 ERROR_IO_PENDING 的错误
            shutdown(clientSocket, SD_BOTH);
            closesocket(clientSocket);
            delete ioContext;
        }
    }
}

void EventWorkerThread() {
    IOContext *ioContext = nullptr;
    DWORD lpNumberOfBytesTransferred = 0;
    void *lpCompletionKey = nullptr;

    DWORD dwFlags = 0;
    DWORD nBytes = MaxBufferSize;

    while (true) {
        BOOL bRt = GetQueuedCompletionStatus(
                hIOCP,
                &lpNumberOfBytesTransferred,
                (PULONG_PTR) &lpCompletionKey,
                (LPOVERLAPPED *) &ioContext,
                INFINITE);

        if (!bRt) continue;

        // 收到 PostQueuedCompletionStatus 发出的退出指令
        if (lpNumberOfBytesTransferred == -1) break;

        // [勘误] 客户端断开时 WSARecv 会以 0 字节完成；这里直接 continue，
        //        既没 closesocket 也没 delete ioContext，会泄漏 socket 和内存
        if (lpNumberOfBytesTransferred == 0) continue;

        // 读到，或者写入的字节总数
        ioContext->nBytes = lpNumberOfBytesTransferred;
        // 处理对应的事件
        switch (ioContext->type) {
            case IOType::Read: {
                // 输出读取到的内容
                setbuf(stdout, nullptr);
                puts(ioContext->buffer);
                // fflush(stdout);
                // closesocket(ioContext->socket);
                // delete ioContext;
                // ioContext = nullptr;
                if (lpNumberOfBytesTransferred == MaxBufferSize) {
                    puts("post read request again");
                    int nRt = WSARecv(
                            ioContext->socket,
                            &ioContext->wsaBuf,
                            1,
                            &nBytes,
                            &dwFlags,
                            &(ioContext->overlapped),
                            nullptr);
                    auto e = WSAGetLastError();
                    // [勘误] e != WSAGetLastError() 永远为 false（中间没有其它 WSA 调用）。
                    //        原意应为 e != ERROR_IO_PENDING
                    if (SOCKET_ERROR == nRt && e != WSAGetLastError()) {
                        // 读取发生错误
                        closesocket(ioContext->socket);
                        delete ioContext;
                        ioContext = nullptr;
                    }
                }
            }
            // [勘误] 这里缺一个 break，Read 分支会掉进 Write 分支（Write 是空操作所以无害）
            case IOType::Write: {
                // 此项目没有这方面的需求，故不处理
                break;
            }
        }
    }
}

勘误说明

  • perror：perror 读的是 C 的 errno，Winsock 错误要用 WSAGetLastError()（配合 FormatMessage）。这里所有 perror 都打印不出真实错误。
  • e != WSAGetLastError()：两次调用之间没有任何 Winsock 调用，所以这个条件恒为 false，错误清理分支永远不会执行；原意应是 e != ERROR_IO_PENDING。
  • 监听 socket 非阻塞 + accept 空转：FIONBIO 令监听 socket 非阻塞，没有连接时 accept 立即返回 WSAEWOULDBLOCK，continue 会让线程满速空转。
  • 0 字节断连泄漏：客户端断开时完成事件为 0 字节，代码直接 continue，既没 closesocket 也没 delete ioContext，连接和内存都会泄漏。
  • switch 掉分支：Read 分支结尾缺 break，会掉进 Write（因 Write 为空操作所以无害，但属于隐患）。
  • MaxBufferSize 不一致：前面 #define MaxBufferSize 1024，完整代码里却是 1024 * 0.5（=512）。

总结

其实使用 IOCP 主要就这几步：

  • 初始化 Socket 和 IOCP
  • 启动 Accept 处理线程
  • 启动 Event 处理线程
  • 设计一个靠谱的线程退出方式

Accept 线程不是必须的，也可以直接由主线程来做，但通常不推荐。Accept 线程可以理解成“事件的投递者”：接入连接后投递首个读事件；Event 线程负责处理完成端口交给它的事件。

相关阅读

常见问题

epoll、kqueue 和 IOCP 有什么不同？

epoll / kqueue 通常在缓冲区有内容时触发事件（epoll 默认是水平触发 LT，EPOLLET 才是边缘触发 ET）；IOCP 是提交重叠 I/O，等 I/O 完成（读/写到一定字节数）后才通知完成事件。

为什么监听 socket 要用 WSASocketW 创建？

参与重叠 I/O 的 socket 需要开启 WSA_FLAG_OVERLAPPED，WSASocketW 可以带这个标志创建出支持重叠 I/O 的句柄。

IOContext 的第一个成员为什么必须是 OVERLAPPED？

GetQueuedCompletionStatus 返回的指针会被强转成 OVERLAPPED*，把 OVERLAPPED 放在结构体首位，就能用同一个指针访问到整个 IOContext。

CreateIoCompletionPort 的 NumberOfConcurrentThreads 是什么？

它限制同时处理完成事件的最大线程数（上限），不是线程池总大小；并行处理线程数超过它时，其余线程会排队等待。

关于作者