选取合适的 demo 是头等大事

 

* 以协程为目标，涉及到的新语法会简单说明，不涉及的不旁征博引

 

* 若语法的原理非常简单，也会简单展开讲讲，有利于透过现象看本质，用起来更得心应手

 

上一篇文章里不光引入了基于事件的调度器，还说明了如何开启多文件并行、await_suspend 与试读的关系、singalfd 用于完美退出等话题，如果没有这些内容铺垫，看本文时会有很多地方难以理解，还没看过的小伙伴，墙裂建议先看那篇。

 

工具还是之前介绍过的 Compile Explorer，这里不再用到 C++ Insights ，主要是它不支持 Windows 平台，其实 Compiler Explorer 也只是编译，运行的话还是不太行，因为它的环境不支持像文件、网络之类的异步 IO，需要用户自行搭建开发环境。

 

基于完成端口的 IO 多路复用

上文中提到了 Unix 系统中多路复用接口的发展历程：分别经历了 select -> poll -> epoll/kqueue，Windows 则通过完成端口一统江山，其实它俩调用方式差不太多：

 

  epoll IOCP

初始化 epoll_create 

 

CreateIoCompletionPort

 

关联句柄 epoll_ctl 

 

CreateIoCompletionPort

 

等待并获取下一个事件 epoll_wait 

 

GetQueuedCompletionStatus

 

投递事件 n/a (self pipe trick) PostQueuedCompletionStatus

销毁 close CloseHandle

而在可等待对象上，IOCP 则丰富的多：

 

* 文件 I/O 事件​​

* 文件系统变更

* 套接字（Socket）事件​​

* 命名管道（Named Pipe）事件​​

* 设备 I/O 事件​​

* 定时器事件（结合 Waitable Timer）​​

 

这方面能与它相提并论的恐怕只有 kqueue 了。有了上面的铺垫再参考之前 epoll 的实现，直接上 demo 源码：

 

#include <coroutine>

#include <unordered_map>

#include <windows.h>

#include <vector>

#include <stdexcept>

#include <iostream>

#include <sstream>

#include <memory>

 

struct Task {

    struct promise_type {

        Task get_return_object() { return {}; }

        std::suspend_never initial_suspend() { return {}; }

        std::suspend_never final_suspend() noexcept { return {}; }

        void return_void() {}

        void unhandled_exception() { std::terminate(); }

    };

};

 

class IocpScheduler {

private:

    HANDLE iocp_handle;

    std::unordered_map<HANDLE, std::coroutine_handle<>> io_handles;

 

public:

    IocpScheduler() {

        iocp_handle = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);

        if (iocp_handle == NULL) {

            throw std::runtime_error("CreateIoCompletionPort failed");

        }

    }

 

    ~IocpScheduler() {

        CloseHandle(iocp_handle);

    }

 

    void register_io(HANDLE file_handle, std::coroutine_handle<> handle) {

        if (io_handles.find(file_handle) == io_handles.end()) {

            io_handles[file_handle] = handle;

 

            if (CreateIoCompletionPort(file_handle, iocp_handle, (ULONG_PTR)file_handle, 0) == NULL) {

                throw std::runtime_error("CreateIoCompletionPort failed to associate file handle");

            }

        }

    }

 

    void run() {

        while (true) {

            DWORD bytes_transferred = 0;

            ULONG_PTR completion_key = 0;

            LPOVERLAPPED overlapped = nullptr;

 

            BOOL success = GetQueuedCompletionStatus(

                iocp_handle,

                &bytes_transferred,

                &completion_key,

                &overlapped,

                INFINITE);

 

            if (completion_key != 0) {

                HANDLE ready_handle = (HANDLE)completion_key;

                if (auto it = io_handles.find(ready_handle); it != io_handles.end()) {

                    it->second.resume();

                }

            }

        }

    }

};

 

struct AsyncReadAwaiter {

    IocpScheduler& sched;

    HANDLE file_handle;

    std::unique_ptr<char[]> buffer;

    DWORD buffer_size;

    OVERLAPPED overlapped;

    DWORD bytes_read;

 

    AsyncReadAwaiter(IocpScheduler& s, HANDLE file, DWORD size)

        : sched(s), file_handle(file), buffer_size(size), bytes_read(0) {

        buffer = std::make_unique<char[]>(size);

        ZeroMemory(&overlapped, sizeof(OVERLAPPED));

    }

 

    bool await_ready() const {

        return false;

    }

 

    void await_suspend(std::coroutine_handle<> h) {

        sched.register_io(file_handle, h);

        

        if (!ReadFile(file_handle, buffer.get(), buffer_size, &bytes_read, &overlapped)) {

            DWORD error = GetLastError();

            if (error != ERROR_IO_PENDING) {

                std::stringstream ss;

                ss << "ReadFile failed, error " << error;

                throw std::runtime_error(ss.str());

            }

        }

    }

 

    std::string await_resume() {

        DWORD bytes_transferred = 0;

        if (!GetOverlappedResult(file_handle, &overlapped, &bytes_transferred, FALSE)) {

            DWORD error = GetLastError();

            std::stringstream ss;

            ss << "GetOverlappedResult failed, error " << error;

            throw std::runtime_error(ss.str());

        }

 

        return std::string(buffer.get(), bytes_transferred);

    }

};

 

Task async_read_file(IocpScheduler& sched, const char* path) {

    HANDLE file_handle = CreateFileA(

        path,

        GENERIC_READ,

        FILE_SHARE_READ,

        NULL,

        OPEN_EXISTING,

        FILE_FLAG_OVERLAPPED,

        NULL);

 

    if (file_handle == INVALID_HANDLE_VALUE) {

        std::stringstream ss;

        ss << "CreateFile failed, error " << GetLastError();

        throw std::runtime_error(ss.str());

    }

 

    while (true) {

        auto data = co_await AsyncReadAwaiter(sched, file_handle, 4096);

        std::cout << "Read " << data.size() << " bytes\n";

        if (data.size() == 0) {

            break;

        }

    }

 

    CloseHandle(file_handle);

}

 

int main(int argc, char* argv[]) {

    if (argc < 2) {

        std::cout << "Usage: sample file_path" << std::endl;

        return 1;

    }

 

    IocpScheduler scheduler;

    async_read_file(scheduler, argv[1]);

    scheduler.run();

    return 0;

}

先看编译：

 

image

 

Compile Explorer 中指定最新的 msvc 编译器和 C++20 选项可以编译通过，注意在 Windows 中选项指定的语法与 Unix 大相径庭，别弄错了。

 

一点一点降低版本尝试，发现能编译这段代码的最低版本是 msvc19.29，对应 vs16.11，如果你需要在本地安装测试环境的话，稳妥起见安装 msvc19.30、对应 vs17.0 也就是 VS2022 比较好，如果本地只有 VS2019，需要升级到第五个也就是最后一个发行版才可以。

 

image

 

接下来创建一个简单的控制台应用包含上面的源文件，需要配置一下 C++ 语言标准：

 

image

 

就可以编译生成可执行文件了，在同目录准备一个文本文件 (test.txt) 进行测试：

 

PS D:\code\iocp_coroutine\Debug> .\iocp_coroutine.exe test.txt

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

...

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

Read 4096 bytes

Read 409

PS D:\code\iocp_coroutine\Debug>

居然死循环了。

 

指定偏移量

同样的代码逻辑，Unix 上没问题 Windows 上却死循环，主要原因是：前者底层使用的是管道，与 socket 之类相似是一个流 (stream)，因此没有读写偏移量的说法，每次从开头获取就可以了；后者使用的是文件，如果不指定偏移量，每次都会从位置 0 读取，有的读者可能问了，为何不能使用当前文件的读取位置呢？这是因为 Windows 上的多路复用底层是彻彻底底的异步架构，必需每次为 ReadFile 指定一个偏移量，而不能够使用当前文件的偏移量。

 

修复的方法很简单，为 ReadFile 的 overlapped 参数的 Offset & OffsetHigh 字段指定要读取数据的偏移量即可：

 

...

    struct AsyncReadAwaiter {

    IocpScheduler& sched;

    HANDLE file_handle;

    std::unique_ptr<char[]> buffer;

    DWORD buffer_size;

增加一个引用成员用来记录当前请求的偏移值，LARGE_INTEGER 可以理解为 uint64 的结构化表达

 

    LARGE_INTEGER &offset; 

    OVERLAPPED overlapped;

    DWORD bytes_read;

 

    AsyncReadAwaiter(IocpScheduler& s, HANDLE file, LARGE_INTEGER &off, DWORD size)

在构造函数中初始化新成员，这个值需要从外部传入，读取成功后更新之，以便跨等待对象使用

 

        : sched(s), file_handle(file), buffer_size(size), offset(off), bytes_read(0) {

        buffer = std::make_unique<char[]>(size);

        ZeroMemory(&overlapped, sizeof(OVERLAPPED));

    }

 

    bool await_ready() const {

        return false;

    }

 

    void await_suspend(std::coroutine_handle<> h) {

        sched.register_io(file_handle, h);

每次请求前设置 overlapped 的偏移字段，并增加调试日志输出以便观察

 

        overlapped.Offset = offset.LowPart; 

        overlapped.OffsetHigh = offset.HighPart; 

        std::cout << "ReadFile from " << offset.QuadPart << std::endl;

        if (!ReadFile(file_handle, buffer.get(), buffer_size, &bytes_read, &overlapped)) {

            DWORD error = GetLastError();

            if (error != ERROR_IO_PENDING) {

                std::stringstream ss;

                ss << "ReadFile failed, error " << error;

                throw std::runtime_error(ss.str());

            }

        }

    }

 

    std::string await_resume() {

        DWORD bytes_transferred = 0;

        if (!GetOverlappedResult(file_handle, &overlapped, &bytes_transferred, FALSE)) {

            DWORD error = GetLastError();

            std::stringstream ss;

            ss << "GetOverlappedResult failed, error " << error;

            throw std::runtime_error(ss.str());

        }

读取成功后，递增相应的偏移量

 

        offset.QuadPart += bytes_transferred; 

        return std::string(buffer.get(), bytes_transferred);

    }

};

 

Task async_read_file(IocpScheduler& sched, const char* path) {

    HANDLE file_handle = CreateFileA(

        path,

        GENERIC_READ,

        FILE_SHARE_READ,

        NULL,

        OPEN_EXISTING,

        FILE_FLAG_OVERLAPPED,

        NULL);

 

    if (file_handle == INVALID_HANDLE_VALUE) {

        std::stringstream ss;

        ss << "CreateFile failed, error " << GetLastError();

        throw std::runtime_error(ss.str());

    }

在外层循环中保存这个偏移量，以便可以持久化使用，初始值为 0

 

    LARGE_INTEGER offset = { 0 }; 

    while (true) {

        auto data = co_await AsyncReadAwaiter(sched, file_handle, offset, 4096);

        std::cout << "Read " << data.size() << " bytes\n";

        if (data.size() == 0) {

            break;

        }

    }

 

    CloseHandle(file_handle);

}

...

再次运行程序，可以输出读取的内容了：

 

PS D:\code\iocp_coroutine\Debug> .\iocp_coroutine.exe test.txt

ReadFile from 0

Read 4096 bytes

ReadFile from 4096

Read 456 bytes

ReadFile from 4552

但是额外的，也收到了一个崩溃提示：

 

image

 

处理文件 EOF

记得之前讲到协程体整个是包在编译的 try...catch 代码块中的，这里直接崩溃难道是 msvc 的异常处理没起作用？挂上调试器看看崩溃堆栈：

 

image

 

看起来是命中 promise 对象的 unhandle_exception，这里调用的 terminate 导致崩溃框弹出，而 unhandled_exception 恰恰是编译器捕获了 throw 抛出的异常，与直觉刚好相反。经过排查，唯一可能抛出异常的位置是这里：

 

    std::string await_resume() {

        DWORD bytes_transferred = 0;

        if (!GetOverlappedResult(file_handle, &overlapped, &bytes_transferred, FALSE)) {

            DWORD error = GetLastError();

            std::stringstream ss;

            ss << "GetOverlappedResult failed, error " << error;

这里加打一行日志

 

            std::cerr << ss.str() << std::endl;

            throw std::runtime_error(ss.str());

        }

        

        offset.QuadPart += bytes_transferred; 

        return std::string(buffer.get(), bytes_transferred);

    }

新的输出果然提示这里返回了错误：

 

PS D:\code\iocp_coroutine\Debug> .\iocp_coroutine.exe test.txt

ReadFile from 0

Read 4096 bytes

ReadFile from 4096

Read 456 bytes

ReadFile from 4552

GetOverlappedResult failed, error 38

错误码 38 对应的是 ERROR_HANDLE_EOF表示文件已到末尾，相比 epoll 管道不关心数据结尾的问题，IOCP 读文件还需要额外增加一些处理，另外在抛异常时，msvc 相比 clang 的显示不太友好，需要在抛出异常前补上 stderr 的打印，修复后的代码如下：

 

    std::string await_resume() {

        DWORD bytes_transferred = 0;

        if (!GetOverlappedResult(file_handle, &overlapped, &bytes_transferred, FALSE)) {

            DWORD error = GetLastError();

判断错误类型，如果是文件 EOF，直接返回空数据，上层会进行判断，从而退出读取循环

 

            if (error != ERROR_HANDLE_EOF) {

                std::stringstream ss;

                ss << "GetOverlappedResult failed, error " << error;

                std::cerr << ss.str() << std::endl;

                throw std::runtime_error(ss.str());

            }

            else {

                return ""; 

            }

        }

 

        offset.QuadPart += bytes_transferred; 

        return std::string(buffer.get(), bytes_transferred);

    }

下面是新的输出：

 

PS D:\code\iocp_coroutine\Debug> .\iocp_coroutine.exe test.txt

ReadFile from 0

Read 4096 bytes

ReadFile from 4096

Read 456 bytes

ReadFile from 4552

Read 0 bytes

# c++20
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

告别扫描文档处理烦恼：ScanTailor Advanced的智能文档优化解决方案

ScanTailor Advanced是一款专业的扫描文档后期处理工具，专为解决老旧文档数字化中的各种问题而设计。通过智能内容识别、批量处理和多线程优化，它能将倾斜、模糊、杂乱的扫描文档转化为整洁的电子文件，显著提升文档数字化效率。## 文档数字化的常见痛点与解决方案### 页面歪斜校正：智能算法自动修正传统扫描文档经常出现页面歪斜问题，手动调整每页耗时耗力。ScanTailor Adv

avatar AI Agent技术社区

SeedVR2终极指南：如何用单步扩散模型实现专业级视频修复

还在为老旧视频的模糊画质而烦恼吗？面对抖动严重的监控录像或褪色的家庭影像，你是否曾希望有一种简单高效的修复方案？SeedVR2正是为解决这些痛点而生的一站式视频修复工具。作为字节跳动推出的先进视频修复模型，它能够在单步推理中实现专业级的视频质量提升，彻底改变了传统视频修复的工作流程。## 核心价值：为什么选择SeedVR2进行视频修复？**传统视频修复的三大瓶颈**：1. **计算成本高

avatar AI Agent技术社区

终极指南：如何用Python自动化工具告别大麦抢票焦虑，轻松获取心仪演出门票

还记得那些让你心跳加速的抢票时刻吗？热门演唱会开票瞬间，你手忙脚乱地刷新页面，却眼睁睁看着"立即购买"按钮变灰，票在几秒内售罄。这种"拼手速"的抢票时代已经过去，现在有了基于Python开发的大麦自动抢票工具，让你告别手速焦虑，轻松获取周杰伦、五月天、周深等热门演出的门票。## 🔥 你的抢票痛点，我们都有解决方案### 场景一：热门演唱会一票难求"周杰伦演唱会开票3秒售罄，我连页面都没

avatar AI Agent技术社区