简中 | English

co_context

co_context 是一个协程并发框架，提供可靠的性能和强易用性，让 C++ 初学者也能轻松写出高并发程序。 co_context 基于 Linux io_uring，其性能通常优于 epoll。

已有功能

1. 支持 read write accept timeout 等 io_uring 提供的所有系统调用，总计 74 个功能。
2. 并发支持: any, some, all, mutex, semaphore, condition_variable, channel。
3. 调度提示: yield, resume_on。
4. 取消 IO/协程：timeout, timeout_at, stop_token。

编译和运行

依赖

1. GCC 11.3.0 或 Clang 14 或更高。
2. [可选] mimalloc 从包管理器或源代码安装。
3. Linux 内核版本 >= 5.6，建议 >= 5.11，越新越好。
   • 运行 uname -r 即可查看你的内核版本。
   • 由于开发环境是 Linux 6.0，在其他版本下可能出现兼容性错误。如遇问题，请将报错发到issue，非常感谢！
   • docker 将继承宿主机的 Linux 内核版本。 因此，docker 无法解决 Linux 内核版本过低的问题。

编译命令

cmake -B build && cmake --build build -j
build/example/timer # 跑一个小示例

代码示例

基础用法

创建一个 io_context，用于运行协程：

    using namespace co_context;
    io_context context;

使用 task<> 定义一个 socket 监听协程。task<> 就像一个普通的 void 函数，但可以在里面使用 co_await。

task<> server(uint16_t port) {
    acceptor ac{inet_address{port}};
    for (int sockfd; (sockfd = co_await ac.accept()) >= 0;) {// 异步接受 client
        co_spawn(session(co_context::socket{sockfd})); // 每个连接生成一个 worker 任务
    }
}

继续使用 task<T> 描述业务逻辑，例如读取 socket 的内容并输出到 stdout：

task<> session(co_context::socket sock) {
    char buf[8192];
    int nr = co_await sock.recv(buf);

    // 不断接收字节流并打印到stdout
    while (nr > 0) {
        co_await lazy::write(STDOUT_FILENO, {buf, (size_t)nr});
        nr = co_await sock.recv(buf);
    }
}

task<T> 和普通 T 函数一样可以任意嵌套：

task<int> the_answer() {
    co_return 42;
}
task<> universe() {
    printf("The answer is %d\n", co_await the_answer());
    co_return;
}

如何写一个 main()：

int main(int argc, const char *argv[]) {
    if (argc < 3) {
        printf("Usage:\n  %s hostname port\n  %s -l port\n", argv[0], argv[0]);
        return 0;
    }

    io_context context; // 1. 定义一个 io_context

    int port = atoi(argv[2]);
    if (strcmp(argv[1], "-l") == 0) {
        context.co_spawn(server(port)); // 2. 至少创建一个 task<>
    } else {
        context.co_spawn(client(argv[1], port));
    }

    context.start(); // 3. 启动 io_context 线程
    context.join();  // 4. 需要时等待 io_context 线程

    return 0;
}

更多特性示例

一秒触发器

task<> my_clock() {
    for (int cnt = 0;;) {
        printf("Time = %d\n", cnt++);
        co_await timeout(1s);
    }
}

网络超时

秒懂的超时控制。

task<> session(co_context::socket peer) {
    char buf[8192];
    int nr = co_await timeout(peer.recv(buf), 3s); // 限时3秒

    while (nr > 0) {
        co_await lazy::write(STDOUT_FILENO, {buf, (size_t)nr});
        nr = co_await timeout(peer.recv(buf), 3s); // 限时3秒
    }

    log_error(-nr);
}

void log_error(int err) {
    switch (err) {
        case ECANCELED:
            log::e("timeout!\n");
            break;
        default:
            log::e("%s\n", strerror(err));
            break;
    }
}

负载均衡

每个 io_context 代表一个线程。要负载均衡，只需将 task<> 分配到合理的 io_context。

示例：echo_server_MT.cpp

借助 resume_on()，可以在运行时任意切换线程。

示例：resume_on.cpp

Generator

同步执行的生成器，可配合std::range。 需要 g++

co_context::generator<int> gen_iota(int x) {
    while (true)
        co_yield x++;
}

int main() {
    using namespace std::views;

    for (int x : gen_iota(1) | drop(5) | take(3)) {
        std::cout << x << " ";
    }
    // 输出 6 7 8

    return 0;
}

链接 IO

用 && 来链接两个 I/O。链接 I/O 可以减少重入内核态和调度器，增强性能表现。（默认只返回最后一个返回值。）

nr = co_await (
    peer.send({buf, (size_t)nr})
    && peer.recv(buf)
);

此例子利用 link_io 大幅增强 echo_server 的性能

channel（实验性）

借鉴自 Go 语言的阻塞队列。

示例：channel.cpp