有栈协程 vs 无栈协程

一、核心原理

有栈协程（Stackful Coroutine）

每个协程拥有独立的运行栈（通常在堆上分配，几 KB 到几 MB），切换时保存/恢复 CPU 寄存器上下文（rsp、rip、callee-saved 寄存器等），本质上是用户态线程。

切换过程（以 x86-64 为例）：

1. 把当前寄存器压入当前协程的栈
2. 把 rsp 指向目标协程的栈顶
3. 弹出目标协程之前保存的寄存器
4. ret 跳转到目标协程上次挂起的位置

关键特点：挂起点可以在任意调用深度。协程函数 A 调用 B，B 调用 C，C 里 yield()，整个 A→B→C 的调用链都保留在协程自己的栈上，恢复时直接接着跑。

典型实现：Go goroutine、Lua coroutine、libco/libgo/boost.context（ucontext/swapcontext、汇编切换）、Java 虚拟线程（Loom，混合式）。

示例 1：Go —— 在第 3 层调用深度挂起

package main

import "fmt"

// ★ 三个都是【普通函数】，没有任何 async / suspend 标记
func level3(ch chan int) {
    for i := 0; i < 3; i++ {
        fmt.Println("    level3: 准备让出", i)
        ch <- i //  ← 挂起点在第 3 层调用深度上
        fmt.Println("    level3: 恢复了")
    }
    close(ch)
}
func level2(ch chan int) { level3(ch) }
func level1(ch chan int) { level2(ch) }

func main() {
    ch := make(chan int)
    go level1(ch) // 起一个 goroutine（有栈协程）
    for v := range ch {
        fmt.Println("main 收到", v)
    }
}

    level3: 准备让出 0
main 收到 0
    level3: 恢复了
    level3: 准备让出 1
main 收到 1
    ...

关键点：ch <- i 挂起时，main→level1→level2→level3 这条完整的调用链原样保留在 goroutine 自己的栈上。恢复时直接从那一行往下走，中间三层帧一个都没丢。

level1 / level2 是完全普通的函数 —— 它们不需要知道自己身处协程中。这就是"无函数染色"。

示例 2：C++ —— 从裸 swapcontext 到标准三件套

和 Go 版本一样的三层调用结构。先看最小版（看清机制），再看真实的库怎么封装它。

最小版：直接用 swapcontext

#include <ucontext.h>
#include <cstdio>

static ucontext_t g_main, g_co;
static char       g_stack[64 * 1024];   // ★ 协程【自己的栈】，独立于主栈
static int        g_value;
static bool       g_done = false;

static void co_yield_(int v) {          // 可以在【任意深度】被调用
    g_value = v;
    swapcontext(&g_co, &g_main);        // 保存当前上下文 → 切回主上下文
}                                       // 恢复时从这一行的下一条继续

// ★ 三个都是普通函数，签名里没有任何协程标记
static void level3() {
    for (int i = 0; i < 3; ++i) {
        printf("    level3: 准备让出 %d\n", i);
        co_yield_(i);                   // ← 第 3 层挂起
        printf("    level3: 恢复了\n");
    }
}
static void level2() { level3(); }
static void level1() { level2(); }

static void co_entry() { level1(); g_done = true; }

int main() {
    getcontext(&g_co);
    g_co.uc_stack.ss_sp   = g_stack;    // 指定协程栈
    g_co.uc_stack.ss_size = sizeof(g_stack);
    g_co.uc_link          = &g_main;    // 协程函数返回后回到主上下文
    makecontext(&g_co, co_entry, 0);

    while (true) {
        swapcontext(&g_main, &g_co);    // resume
        if (g_done) break;
        printf("main 收到 %d\n", g_value);
    }
}

g++ -std=c++17 stackful.cpp -o stackful && ./stackful

    level3: 准备让出 0
main 收到 0
    level3: 恢复了
    level3: 准备让出 1
main 收到 1
    level3: 恢复了
    level3: 准备让出 2
main 收到 2
    level3: 恢复了

和 Go 版本结构完全一致 —— 挂起发生在第 3 层，三层调用帧全部保留在 g_stack 里。

注意：ucontext 只是演示用。它每次切换都会调用 sigprocmask 保存/恢复信号掩码，产生一次系统调用，实测比纯汇编切换慢 5–10 倍。macOS 已将其标记为废弃，Windows 上对应的是 Fiber API（CreateFiber / SwitchToFiber）。生产环境用 boost.context、libco、libaco 这类纯汇编实现。

包装成三件套：co_create / co_resume / co_yield

上面那版为了看清机制，直接把 swapcontext 写在业务代码里了。真实的库（Lua、libco、Python generator）都把它封装成非对称协程的经典三件套：

接口	语义	关键点
co_create	创建协程	只分配栈 + 配置入口，一行代码都不执行
co_resume	切进协程	首次调用 = 启动，之后 = 从上次 co_yield 处恢复
co_yield	让出	切回 resume 我的那个人（可以在任意调用深度调用）

实现

#include <ucontext.h>
#include <cstdio>
#include <cstdlib>
#include <cstdint>

struct Coro {
    ucontext_t ctx;              // 协程自己的上下文
    ucontext_t caller;           // ★ 谁 resume 我的 —— 非对称协程的关键字段
    char*      stack   = nullptr;
    size_t     size    = 0;
    void     (*fn)(void*) = nullptr;
    void*      arg     = nullptr;
    void*      transfer= nullptr; // resume / yield 之间来回传的值
    bool       done    = false;
};

static Coro* g_current = nullptr;          // 当前正在跑的协程

static void co_entry() {
    Coro* co = g_current;
    co->fn(co->arg);                       // 跑用户的函数
    co->done = true;
    swapcontext(&co->ctx, &co->caller);    // 结束，切回 resume 者，【永不返回】
}

// ① 创建 —— 只配置，不执行
Coro* co_create(void (*fn)(void*), void* arg, size_t stack_size = 64 * 1024) {
    Coro* co  = new Coro{};
    co->stack = (char*)malloc(stack_size);
    co->size  = stack_size;
    co->fn    = fn;
    co->arg   = arg;

    getcontext(&co->ctx);
    co->ctx.uc_stack.ss_sp   = co->stack;
    co->ctx.uc_stack.ss_size = stack_size;
    co->ctx.uc_link          = &co->caller;   // 兜底：函数返回后回到 resume 者
    makecontext(&co->ctx, co_entry, 0);       // ★ 只是布置好入口，此刻什么都没跑
    return co;
}

// ② 恢复 —— 切进协程；返回值是协程 yield 出来的东西
void* co_resume(Coro* co, void* in = nullptr) {
    if (co->done) return nullptr;
    Coro* prev   = g_current;
    g_current    = co;
    co->transfer = in;                        // 把值送进去
    swapcontext(&co->caller, &co->ctx);       // ★★ 保存"我"，切到协程
    g_current    = prev;                      // 协程 yield 回来了
    return co->transfer;                      // 取出协程送出来的值
}

// ③ 让出 —— 在【任意调用深度】都能调；返回值是下次 resume 送进来的东西
void* co_yield(void* out) {
    Coro* co     = g_current;
    co->transfer = out;                       // 把值送出去
    swapcontext(&co->ctx, &co->caller);       // ★★ 保存协程，切回 resume 者
    return co->transfer;                      // 被 resume 后从这里继续
}

bool co_done(Coro* co)    { return co->done; }
void co_destroy(Coro* co) { free(co->stack); delete co; }

命名提醒：co_yield 在 C++20 中是关键字，上面这段需要 -std=c++17 或更早才能编译（libco 是 C 库，没这个问题）。C++20 下改名成 co_yield_ 即可 —— 这正是上面最小版里那个 co_yield_ 末尾下划线的由来。

使用：还是那个三层调用

static void level3() {
    for (int i = 0; i < 3; ++i) {
        printf("    level3: 准备让出 %d\n", i);
        void* in = co_yield((void*)(intptr_t)i);      // ★ 第 3 层挂起，顺便传值
        printf("    level3: 恢复了，resume 传进来 %d\n", (int)(intptr_t)in);
    }
}
static void level2() { level3(); }                    // 普通函数
static void level1() { level2(); }                    // 普通函数
static void task(void*) { level1(); }

int main() {
    Coro* co = co_create(task, nullptr);
    printf("main: 协程已创建，此刻【一行代码都没跑】\n");

    int n = 100;
    while (true) {
        void* out = co_resume(co, (void*)(intptr_t)n);
        if (co_done(co)) break;
        printf("main: resume(%d) 得到 %d\n", n, (int)(intptr_t)out);
        ++n;
    }
    co_destroy(co);
    printf("main: 协程已结束\n");
}

g++ -std=c++17 coro3.cpp -o coro3 && ./coro3

main: 协程已创建，此刻【一行代码都没跑】
    level3: 准备让出 0
main: resume(100) 得到 0
    level3: 恢复了，resume 传进来 101
    level3: 准备让出 1
main: resume(101) 得到 1
    level3: 恢复了，resume 传进来 102
    level3: 准备让出 2
main: resume(102) 得到 2
    level3: 恢复了，resume 传进来 103
main: 协程已结束

三件套的执行流

main                                协程（跑在自己的 64KB 栈上）
 │
 ├─ co_create(task)  ────────────►  【只分配栈 + 布置入口，什么都不执行】
 │  「协程已创建」                    ctx.rip = co_entry, ctx.rsp = 栈顶
 │
 ├─ co_resume(co, 100)
 │    swapcontext(&caller, &ctx) ─►  co_entry → task → level1 → level2 → level3
 │                                   「准备让出 0」
 │  ◄──────── swapcontext(&ctx,&caller)  co_yield(0)      ← ★ 第 3 层挂起
 │  「resume(100) 得到 0」            （4 层调用帧原封不动留在协程栈上）
 │
 ├─ co_resume(co, 101)
 │    swapcontext ──────────────►   ★ 从 co_yield 内部的下一条指令继续
 │                                   co_yield 返回 101 →「恢复了，传进来 101」
 │                                   i=1 →「准备让出 1」
 │  ◄──────── co_yield(1)
 │  「resume(101) 得到 1」
 │
 │  ……（第 3 轮同理）……
 │
 ├─ co_resume(co, 103)  ─────────►  循环结束 → level3/2/1/task 层层返回
 │                                   co_entry: done = true
 │  ◄──────── swapcontext(&ctx,&caller)
 │  co_done() == true → break
 └─ co_destroy(co)

三个要点

① co_create 只配置不执行 —— 和无栈协程里 counter() 因为 initial_suspend 而立即返回是同一种"惰性"，但机制完全不同：这边是"布置好假栈帧等着被切进来"，那边是"挂起并 return"。

② caller 字段是"非对称"的全部含义 —— co_yield 之所以知道该切回谁，就是因为 co_resume 在切进来之前把自己存进了 co->caller。对称协程没有这个字段，切换时必须显式指定目标。

③ 三件套之上才是易用的 API —— 成熟的库都把这三个当内部原语藏起来，对外只暴露 go(fn) + channel + 同步原语（本文附录里的实现就是这么做的）。因为直接用 resume/yield 写业务，等于要求用户自己做调度。

各家的对应关系

	创建	恢复	让出	是否暴露给用户
Lua	coroutine.create	coroutine.resume	coroutine.yield	✅ 直接暴露
libco	co_create	co_resume	co_yield / co_yield_ct	✅ 直接暴露
Win32 Fiber	CreateFiber	SwitchToFiber	SwitchToFiber（对称）	✅
boost.context	fiber_context{fn}	.resume()	.resume()（对称）	✅ 但只做原语
libgo / coost	内部	内部	内部	❌ 只给 go(fn)
Go	内部	内部	隐式 + 抢占	❌ 只给 go f()

趋势很清楚：越成熟的库，越倾向于把三件套藏起来。

切换的本质：一段汇编

前面描述的四个步骤，实际就是这么一段代码（libco / boost.context 的核心）：

; void co_swap(Context* from, Context* to);
;   rdi = from, rsi = to        （System V AMD64）
co_swap:
    push rbp                    ; ① 把 callee-saved 寄存器压进【当前协程的栈】
    push rbx
    push r12
    push r13
    push r14
    push r15

    mov  [rdi], rsp             ; ② 把当前 rsp 存进 from->sp
    mov  rsp, [rsi]             ; ③ 把 rsp 指向【目标协程的栈顶】—— 换栈完成

    pop  r15                    ; ④ 弹出目标协程当初保存的寄存器
    pop  r14
    pop  r13
    pop  r12
    pop  rbx
    pop  rbp

    ret                         ; ⑤ 从目标协程的栈上弹出返回地址并跳过去
                                ;    —— 正是它上次调用 co_swap 的下一条指令

只需要保存 6 个 callee-saved 寄存器 + rsp —— 因为 caller-saved 寄存器按 ABI 约定本来就允许被函数调用破坏，而 co_swap 就是一次普通的函数调用。

整段没有任何系统调用，约 10–20 ns。这也解释了为什么"换栈"这件事必须写汇编、必须为每个架构各写一份（ARM64 要保存 x19–x28 和 x29/x30，寄存器名和数量都不同）。

对称 vs 非对称：resume / yield 只是换栈的两个方向

这是有栈协程 API 设计的第一个分岔口，定错了后面全要重写。

先看一个事实：两个方向用的是同一个函数

把前面三件套里的 co_resume 和 co_yield 并排放：

void* co_resume(Coro* co, void* in) {
    co->transfer = in;
    swapcontext(&co->caller, &co->ctx);    // 存"我"(main)，切到 co
    return co->transfer;
}

void* co_yield(void* out) {
    Coro* co = g_current;
    co->transfer = out;
    swapcontext(&co->ctx, &co->caller);    // 存"我"(co)，切到 caller
    return co->transfer;
}

把那两行对齐：

swapcontext(&co->caller, &co->ctx);      // resume：存 caller，加载 ctx
swapcontext(&co->ctx,    &co->caller);   // yield ：存 ctx，   加载 caller
            ^^^^^^^^^^^^^^^^^^^^^^^^
            参数对调 —— 这就是【全部】区别

两者都是"我停，你跑"，只是方向相反。 本文附录里那个库更直白 —— 两个方向调的是同一个汇编函数：

void Scheduler::resume(Coroutine* c) { co_switch(&main_.sp_, c->sp_);      }  // 切进协程
void detail::block()                 { co_switch(&self->sp_, S.main_.sp_); }  // 切回调度器

从 CPU 的角度，它分不清哪次是 resume 哪次是 yield。 co_switch 只做三件事：存 6 个寄存器 + mov rsp + ret。"resume"和"yield"是人类给这两个方向起的名字，不是机制差异 —— 开销也完全一样（都是 ~10–20 ns）。

顺带澄清一个常见误解：co_resume 不是"等待 co_yield"。调用它的瞬间，调用者这条执行流就被冻结了（现场存进 co->caller），CPU 完整地交给协程。单线程内始终只有一个执行流在跑，谈不上"等"。它更像 call，co_yield 更像 ret —— 唯一的区别是"返回"可以发生在函数中间，且栈帧保留、下次从那里继续。

唯一的不对称：谁决定目标

	co_resume	co_yield
机制	swapcontext	swapcontext（完全相同）
开销	~10–20 ns	~10–20 ns（一样）
目标由谁指定	调用者显式给：co_resume(co)	隐含的：就是 co->caller
方向	外 → 内	内 → 外

co_yield 不需要参数，是因为非对称模型规定了"yield 一定回到 resume 我的那个人" —— 而这个人早被 co_resume 存进 co->caller 了。

两种模型

非对称协程（asymmetric）—— Lua / libco / Python generator / 上面的示例

有明确的父子关系，靠 caller 字段维系：

struct Coro {
    ucontext_t ctx;
    ucontext_t caller;     // ★ 就是这个字段定义了"非对称"
};
co_resume(c);              // 主协程 → c
co_yield();                // c → 【自动回到 resume 它的那个】

对称协程（symmetric）—— boost.context

所有协程地位平等，没有"回到谁"，只有"去哪"。连 caller 字段都不需要：

struct Coro {
    ucontext_t ctx;
    char*      stack;
    // 没有 caller 字段！
};

static Coro* g_current;

void co_transfer_to(Coro* target) {          // 只有这一个函数
    Coro* self = g_current;
    g_current  = target;
    swapcontext(&self->ctx, &target->ctx);   // 我停，target 跑
}

用起来是这样 —— 可以在协程之间直接跳，不经过任何中间人：

// a 的函数体里：co_transfer_to(b);
// b 的函数体里：co_transfer_to(c);
// c 的函数体里：co_transfer_to(a);   ← 直接回到 a

代价是：你必须自己想清楚"接下来该谁跑"，否则控制流就丢了（没有 caller 兜底）。所以对称模型通常配一个调度器来回答这个问题。

对比

	非对称	对称
心智模型	像函数调用，容易理解	像 goto，灵活但绕
需要维护	caller 指针	无
接口	resume + yield 两个	transfer 一个
A → B 切换	两次：A → 调度器 → B	一次：A → B
控制流丢失风险	低（yield 总有地方去）	高（必须显式指定下一个）
适合	generator、简单场景、给最终用户用	调度器、高性能网络库
代表	Lua、libco、Python	boost.context、libgo 内部

"两次切换"是真实的性能差异

非对称模型下，协程 A 想让协程 B 跑，做不到直接切：

非对称：              对称：
   A                    A
   │ yield              │ transfer_to(B)
   ▼                    │
 调度器   ← 蹦床         │
   │ resume(B)          │
   ▼                    ▼
   B                    B

  两次换栈              一次换栈

中间那次"回到调度器再出去"是纯粹的浪费 —— 在每秒切换百万次的网络库里，这是实打实的开销。

一个跨语言的呼应：C++20 的 symmetric transfer

无栈协程遇到过一模一样的问题。C++20 早期，co_await 一个协程时也要经过"蹦床"：内层协程完成 → 返回给调度者 → 调度者再 resume 外层。除了多一次切换，还会让栈不断增长（每次 resume 都压一层帧），深度嵌套时直接爆栈。

C++17 之后引入的解法叫 symmetric transfer：

std::coroutine_handle<> await_suspend(std::coroutine_handle<> h) {
    return continuation_;      // ★ 返回一个 handle，编译器做【尾调用】直接跳过去
}                              //   不返回调用者，也就不压栈帧

本质上就是把非对称模型改成对称。 两种协程、隔了三十年，撞上的是同一个设计问题。

实践建议：底层对称，上层包非对称外壳

namespace detail { void co_switch(void** from, void* to); }   // 对称原语，内部用

void co_yield_();                    // 非对称外壳，给用户用
void co_resume(Coro* c);

libgo、coost 都是这么干的，本文附录里的实现也是：

• co_switch 是对称的（调度器和协程用同一个函数，只是参数对调）
• 对外暴露的是 co_go / co_chan / co_mutex，用户根本看不到 resume/yield

这样既拿到了对称模型的性能（调度器可以直接在协程间切），又给了用户最简单的心智模型。

无栈协程（Stackless Coroutine）

协程没有自己的栈，编译器把协程函数改写成一个状态机：

• 局部变量提升到堆上的"协程帧"（coroutine frame）结构体里
• 每个 co_await/yield 点变成一个状态编号
• 恢复就是调用编译器生成的那个 resume 函数（GCC 叫 actor），它开头一个 switch 按状态号跳到上次停下的位置
• 协程运行时借用调用者的栈

关键特点：挂起只能发生在协程函数本身。如果协程 A 调用普通函数 B，B 里无法挂起（B 不是协程，编译器没给它做状态机变换）。要挂起就必须 B 也是协程，A 通过 co_await B() 层层链式等待，这就是所谓的 "函数染色"（function coloring） 问题。

典型实现：C++20 coroutines、Rust async/await、Python async/generator、C# async、JavaScript async/await、Kotlin 挂起函数。

示例 1：C++20 —— 一个最小的 generator

#include <coroutine>
#include <exception>
#include <utility>
#include <cstdio>

template <class T>
struct Generator {
    struct promise_type {                       // ★ 编译器要求的"协程接口"
        T value;

        Generator get_return_object() {
            return Generator{ std::coroutine_handle<promise_type>::from_promise(*this) };
        }
        std::suspend_always initial_suspend() noexcept { return {}; }  // 创建后先挂起
        std::suspend_always final_suspend()   noexcept { return {}; }  // 结束后保持挂起
        std::suspend_always yield_value(T v) { value = v; return {}; } // co_yield 走这里
        void return_void() {}
        void unhandled_exception() { std::terminate(); }
    };

    std::coroutine_handle<promise_type> h;

    explicit Generator(std::coroutine_handle<promise_type> h_) : h(h_) {}
    Generator(Generator&& o) noexcept : h(std::exchange(o.h, {})) {}
    Generator(const Generator&) = delete;
    ~Generator() { if (h) h.destroy(); }         // ★ 协程帧要手动销毁

    bool next()        { h.resume(); return !h.done(); }
    T    value() const { return h.promise().value; }
};

// ★ 函数体里出现 co_yield / co_await / co_return，这个函数就是【协程】
Generator<int> counter() {
    for (int i = 0; i < 3; ++i) {
        printf("    counter: 准备让出 %d\n", i);
        co_yield i;                              // ← 挂起点【只能写在这个函数体里】
        printf("    counter: 恢复了\n");
    }
}

int main() {
    auto g = counter();
    while (g.next())
        printf("main 收到 %d\n", g.value());
}

g++ -std=c++20 stackless.cpp -o stackless && ./stackless
# GCC 10 需要额外加 -fcoroutines，GCC 11 起 -std=c++20 即可

    counter: 准备让出 0
main 收到 0
    counter: 恢复了
    counter: 准备让出 1
main 收到 1
    counter: 恢复了
    counter: 准备让出 2
main 收到 2
    counter: 恢复了

输出和上面有栈版本完全一样，但机制截然不同 —— 下面先拆开编译器到底把 counter() 变成了什么，再拿完整的执行流走一遍。

编译器做的状态机变换（完整拆解）

上面那段代码里，promise_type、coroutine_handle、co_yield、suspend_always 看着像一堆魔法。 这一节把它们逐个拆开，你会发现：全部加起来只是"一个堆上的结构体 + 两个函数指针 + 一个 switch"。

先建立坐标：编译器把 counter() 变成了什么

GCC 从一个协程函数生成三个函数（gcc/cp/coroutines.cc 开头的设计说明原文："We transform the user's function into three pieces"）：

名称	是什么
ramp	用户调用的那个 —— 就是符号 counter() 本身。建帧、构造 promise、拿返回对象，然后调一次 actor 把协程跑到第一个挂起点
actor	真正的函数体 + 状态机。GCC 源码的原话："An actor function that contains the state machine"。h.resume() 调的就是它
destroy	两行的 shim：把 _Coro_resume_index 最低位置 1（标记"走 destroy 路径"），然后调 actor。销毁代码本身在 actor 里，为什么还要它，见第 4 步

GCC 让 destroy 只是个 shim，是为了状态机只需优化一次 —— resume 和 destroy 两条路径都在 actor 里， 之后交给中端的 inline / DCE 分别优化。

三个函数的符号名由 gcc/cp/mangle.cc 生成：在原函数的修饰名后面追加 .actor / .destroy，nm -C 显示为：

0000000000000000 T counter()                                   ← ramp
000000000000014e t counter(_Z7counterv.Frame*) [clone .actor]   ← actor
00000000000004e6 t counter(_Z7counterv.Frame*) [clone .destroy] ← destroy

（GCC 15.2，Ubuntu 26.04，x86_64，g++ -std=c++20 -c stackless.cpp && nm -C stackless.o | grep counter。本文所有"实测"均指这个环境。）

本文写作 counter.actor / counter.destroy，这是符号名，不是合法的 C++ 标识符；下面的代码直接用它们指代那两个函数。

协程帧的类型也有真实名字：get_fn_local_identifier(orig_fn_decl, "Frame") 把原函数的修饰名 _Z7counterv 后面接上 .Frame，得到 _Z7counterv.Frame —— 上面 nm 输出里 actor 的参数类型就是它。字段名来自 coro_init_identifiers，跨挂起点存活的局部变量和 awaiter 也进帧，名字是"原名_序号_序号"。下面是实测的布局（g++ -g 后从 DWARF 读出的偏移）：

struct _Z7counterv.Frame {                                   // sizeof == 40
    void (*_Coro_resume_fn )(_Z7counterv.Frame*);            // 偏移 0   = &counter.actor
    void (*_Coro_destroy_fn)(_Z7counterv.Frame*);            // 偏移 8   = &counter.destroy
    Generator<int>::promise_type _Coro_promise;              // 偏移 16  （本例只有 int value，4 字节）
    unsigned short _Coro_resume_index;                       // 偏移 20  ★ 状态号
    unsigned short _Coro_frame_refcount;                     // 偏移 22  ramp 和函数体谁还在用这个帧，归零才 delete
    bool           _Coro_frame_needs_free;                   // 偏移 24  帧是不是 operator new 出来的
    bool           _Coro_initial_await_resume_called;        // 偏移 25  initial_suspend 的 await_resume 调过没
    std::suspend_always Is_1_1;                              // 偏移 26  initial_suspend() 返回的 awaiter
    int                 i_2_3;                               // 偏移 28  ★ 你写的 i —— 跨挂起点存活，进帧
    std::suspend_always Yd0_3_4;                             // 偏移 32  yield_value() 返回的 awaiter
    std::suspend_always Fs_1_5;                              // 偏移 33  final_suspend() 返回的 awaiter
};

前三个字段的位置是 ABI 强制的（同一段设计说明）：

The ABI mandates that pointers into the coroutine frame point to an area beginning with two function pointers (to the resume and destroy functions), these are immediately followed by the "promise object".

这正是 h.resume() 取偏移 0、h.destroy() 取偏移 8、h.promise() 取偏移 16 的依据。

_Coro_resume_index 的编号规则（build_actor_fn 里分派器的注释）：

0            还没启动 —— ramp 把它初始化成 0，然后调 actor
偶数 2,4,6…  resume 点：挂在第 N 号挂起点上，恢复时跳到标签 resume.N
奇数 3,5,7…  destroy 点：挂在第 N 号挂起点上被 destroy，|= 1 变成 N+1，跳到标签 destroy.N
1            特殊：unhandled_exception() 之后被 destroy，直接释放 promise 和帧

编号从 2 开始，每个挂起点占两个号（源码：data->index += 2）。本例三个挂起点：

挂起点	挂起时写入的 _Coro_resume_index	恢复标签
co_await initial_suspend()	2	resume.2
co_yield i	4	resume.4
co_await final_suspend()	6	resume.6（永远不会走到：此前已把 _Coro_resume_fn 清零）

actor 进来先看最低位：偶数走 resume 的 switch，奇数走 destroy 的 switch。

---

第 1 步：std::coroutine_handle<promise_type> h 到底是什么

它是一个类，唯一的数据成员是一个 void*，这个 void* 的值就是协程帧的地址。

static_assert(sizeof(std::coroutine_handle<promise_type>) == sizeof(void*));   // 8
static_assert(std::is_trivially_copyable_v<std::coroutine_handle<>>);          // 可 memcpy

libstdc++ 的定义（libstdc++-v3/include/std/coroutine，节选）：

namespace std {

template <>
struct coroutine_handle<void>
{
    constexpr coroutine_handle() noexcept { }
    constexpr coroutine_handle(nullptr_t) noexcept { }

    void resume()  const { __builtin_coro_resume (_M_fr_ptr); }
    void destroy() const { __builtin_coro_destroy(_M_fr_ptr); }
    bool done()    const noexcept { return __builtin_coro_done(_M_fr_ptr); }

    constexpr void* address() const noexcept { return _M_fr_ptr; }
    constexpr static coroutine_handle from_address(void* __a) noexcept
    { coroutine_handle __self; __self._M_fr_ptr = __a; return __self; }

    constexpr explicit operator bool() const noexcept { return bool(_M_fr_ptr); }

protected:
    void* _M_fr_ptr;                    // ★ 唯一的数据成员，值 = 协程帧地址
};

// ★ 注意：这是【另一个独立的类模板】，不是从上面那个继承来的，
//    它有自己的 _M_fr_ptr，两者靠下面的转换运算符打通。
template <typename _Promise>
struct coroutine_handle
{
    constexpr coroutine_handle() noexcept { }
    constexpr coroutine_handle(nullptr_t) noexcept { }

    void resume()  const { __builtin_coro_resume (_M_fr_ptr); }
    void destroy() const { __builtin_coro_destroy(_M_fr_ptr); }
    bool done()    const noexcept { return __builtin_coro_done(_M_fr_ptr); }

    constexpr void* address() const noexcept { return _M_fr_ptr; }
    constexpr static coroutine_handle from_address(void* __a) noexcept
    { coroutine_handle __self; __self._M_fr_ptr = __a; return __self; }

    // ★ 到无类型版本的转换：只是把那个 void* 复制过去
    constexpr operator coroutine_handle<>() const noexcept
    { return coroutine_handle<>::from_address(address()); }

    _Promise& promise() const
    {
        void* __t = __builtin_coro_promise(_M_fr_ptr, __alignof(_Promise), false);
        return *static_cast<_Promise*>(__t);
    }

    static coroutine_handle from_promise(_Promise& __p)
    {
        coroutine_handle __self;
        __self._M_fr_ptr = __builtin_coro_promise((char*) &__p,
                                                  __alignof(_Promise), true);
        return __self;
    }

private:
    void* _M_fr_ptr = nullptr;          // ★ 各自持有一份
};

}

所有方法都只做一件事：拿 _M_fr_ptr 去调一个 __builtin_coro_*。

注意分层：__builtin_coro_* 是编译器提供的内建函数，不属于标准库 —— 标准库只负责把成员函数一行转发过去，真正的展开由编译器完成。

GCC 在 gcc/coroutine-builtins.def 里声明了这四个（宏会给名字加 __builtin_coro_ 前缀）：

DEF_COROUTINE_BUILTIN (BUILT_IN_CORO_PROMISE, "promise", BT_FN_PTR_PTR_CONST_SIZE_BOOL, ...)
DEF_COROUTINE_BUILTIN (BUILT_IN_CORO_RESUME,  "resume",  BT_FN_VOID_PTR,                ...)
DEF_COROUTINE_BUILTIN (BUILT_IN_CORO_DESTROY, "destroy", BT_FN_VOID_PTR,                ...)
DEF_COROUTINE_BUILTIN (BUILT_IN_CORO_DONE,    "done",    BT_FN_BOOL_PTR,                ...)

翻译成签名就是：

void  __builtin_coro_resume  (void*);
void  __builtin_coro_destroy (void*);
bool  __builtin_coro_done    (void*);
void* __builtin_coro_promise (const void*, size_t align, bool from_promise);

下面逐个看这些内建被展开成了什么。

---

h.resume() —— 取帧偏移 0 的函数指针并调用

协程帧的前三个成员由 ABI 规定死（两个函数指针，紧跟着 promise 对象）；再往后的字段由编译器自行安排：

帧地址 + 0   →  _Coro_resume_fn   函数指针（指向 counter.actor）      ← ABI
帧地址 + 8   →  _Coro_destroy_fn  函数指针（指向 counter.destroy）    ← ABI
帧地址 + 16  →  _Coro_promise                                        ← ABI（偏移 = 16 向上对齐到 alignof(promise_type)）
再往后      →  _Coro_resume_index、跨挂起点的局部变量……              ← 编译器自行安排

这个偏移在 GCC 里就是一行乘法（gcc/coroutine-passes.cc，call_idx 对 resume 取 0、对 destroy 取 1）：

HOST_WIDE_INT psize  = TREE_INT_CST_LOW (TYPE_SIZE_UNIT (ptr_type_node));
HOST_WIDE_INT offset = call_idx * psize;

展开链条：

h.resume();

// ① 标准库：一行转发到编译器内建
void resume() const { __builtin_coro_resume(_M_fr_ptr); }

② 编译器展开这个内建：取【帧偏移 0】处的函数指针，调用它，参数是帧本身。

; ③ 实测：g++ -O1 编译 main，g.next() 里的 h.resume() 变成了这两条（rbx = _M_fr_ptr 的值，即帧地址）
mov  rdi, rbx                    ; 帧地址作为唯一参数
call [QWORD PTR [rbx]]           ; 取帧偏移 0 的函数指针并调用 —— 就是 counter.actor

这就是 counter.actor(帧) 被调用的全过程 —— 两条指令，没有虚函数、没有查表。actor 的签名是 void (*)(帧*)，所以帧地址放在第一个参数寄存器 rdi 里。

（-O0 下 coroutine_handle::resume() 不会被内联，会多一层普通调用和几条 load，最终执行的仍是这两条。）

h.destroy() 完全一样，只是取偏移 8。~Generator() 里那句 h.destroy() 实测编成：

mov  rdi, rbx
call [QWORD PTR 8[rbx]]          ; 取帧偏移 8 的函数指针并调用 —— counter.destroy

---

h.done() —— 检查帧偏移 0 是不是 null

协程走到 final_suspend 时，编译器会把帧偏移 0 的 resume 函数指针置为 nullptr —— 已结束的协程不能再 resume。所以 done() 就是把这一格和 null 比较。

// ① 标准库
bool done() const noexcept { return __builtin_coro_done(_M_fr_ptr); }

// ② GCC 的展开（gcc/coroutine-passes.cc）：读出帧里第一个指针，和 null 比较
tree done = fold_build2 (EQ_EXPR, boolean_type_node, d_ptr_tmp, null_pointer_node);

; ③ 实测：g.next() 里 return !h.done() 和 main 的 while 合并成了一次比较 + 跳转
cmp  QWORD PTR [rbx], 0          ; 帧偏移 0 == null ?
je   .L42                        ; 是 → 协程已结束，退出 while

---

h.promise() —— 帧地址加一个编译期常量偏移

这就是 handle 与 promise_type 关联起来的地方：

// ① 标准库
_Promise& promise() const {
    void* __t = __builtin_coro_promise(_M_fr_ptr, __alignof(_Promise), false);
    return *static_cast<_Promise*>(__t);
}

__builtin_coro_promise 被展开成一次编译期常量的指针加减（gcc/coroutine-passes.cc，dir 就是那个 from 参数）：

psize *= 2;                      /* Start with two pointers.  */
psize = ROUND_UP (psize, align); /* 向上对齐到 promise 的对齐要求 */
HOST_WIDE_INT offs = dir ? -psize : psize;

即：

offset = ROUND_UP(2 * sizeof(void*), alignof(Promise))
from == false  →  返回  帧地址 + offset        （帧 → promise）
from == true   →  返回  promise 地址 - offset  （promise → 帧）

alignof(promise_type)	offset
1 / 2 / 4 / 8 / 16	16
32	32
64	64

绝大多数情况就是 16。于是：

; 实测：main 里 g.value() → h.promise().value，折叠成一条指令（rbx = 帧地址）
mov  edx, DWORD PTR 16[rbx]         ; 直接读帧偏移 16 处的 promise.value，作为 printf 的参数

反方向 from_promise（get_return_object() 里用的就是它）就是减 16：promise 地址 - 16 = 帧地址，存进新 handle 的 _M_fr_ptr。

注意 alignof(Promise) 是这个偏移的唯一输入。 这也是为什么 coroutine_handle<>（不带类型参数）没有 promise() —— 它不知道 Promise 是谁，算不出偏移。

---

完整对照表

你写的	libstdc++ 转发到	编译器展开成	实测 main 的 -O1 汇编（rbx = 帧地址）
h.resume()	__builtin_coro_resume(p)	取帧[0]的函数指针并调用	mov rdi,rbx / call [QWORD PTR [rbx]]
h.destroy()	__builtin_coro_destroy(p)	取帧[8]的函数指针并调用	mov rdi,rbx / call [QWORD PTR 8[rbx]]
h.done()	__builtin_coro_done(p)	帧[0] == null ?	cmp QWORD PTR [rbx],0
h.address()	直接返回 _M_fr_ptr	无	无（就是那 8 字节本身）
from_address(p)	直接写 _M_fr_ptr	无	无
h.promise()	__builtin_coro_promise(p, align, false)	帧地址 + 16	mov edx, DWORD PTR 16[rbx]（连 .value 一起读了）
from_promise(pr)	__builtin_coro_promise(&pr, align, true)	promise 地址 − 16	在 counter() 内部，被折叠进 get_return_object

---

三层关系

   Generator（counter() 的返回类型，你自己定义的类）
   ┌────────────────────────────────────┐
   │  std::coroutine_handle<...>  h;    │  ① 一个类，sizeof == 8
   │        └─ void*  _M_fr_ptr ────────┼───┐  ② 唯一的数据成员，值 = 帧地址
   └────────────────────────────────────┘   │
                                            ▼  ③ 协程帧（堆上，编译器生成）
                                        ┌────────────────────────────────────────────┐
                               偏移 0   │ _Coro_resume_fn     → counter.actor        │
                               偏移 8   │ _Coro_destroy_fn    → counter.destroy      │
                               偏移 16  │ _Coro_promise       promise_type 的一个对象 │
                                        │ _Coro_resume_index  跑到哪了                │
                                        │ i_2_3               你写的 i，跨挂起点存活  │
                                        │ Is_1_1 / Yd0_3_4 / Fs_1_5   三个挂起点的 awaiter │
                                        └────────────────────────────────────────────┘

小结：coroutine_handle = 帧指针 + 一套"已知布局"的约定。没有虚函数，没有运行时类型信息，没有分配。

本节的实现细节核对自 GCC 主线源码，示例在 GCC 15.2 上实测：类定义见 libstdc++-v3/include/std/coroutine， 内建的声明见 gcc/coroutine-builtins.def，展开见 gcc/coroutine-passes.cc。

---

第 2 步：协程是怎么"出生"的 —— ramp function 逐步拆解

上面说了编译器生成 ramp / actor / destroy 三个函数。这一步专门拆开第一个 —— 你调用 counter() 时真正执行的那个 ramp function（GCC 源码：cp_coroutine_transform::build_ramp_function）。

★ counter() 这个符号编译后就是 ramp function，你写的 for 循环【不在里面】。 ramp 最后会调一次 counter.actor，但那次调用只跑到 initial_suspend 就挂起返回了，函数体一行都没碰到。

它一共做七件事，我们一步一步来。

---

2.1 分配协程帧

auto* f = (_Z7counterv.Frame*)operator new(40);     // dump 原文：operator new (.CO_FRAME (40, 0B))
f->_Coro_frame_needs_free = true;                    // 帧是 new 出来的，将来要 delete

f 是一个裸指针（GCC 里这个局部变量叫 _Coro_frameptr，下文为简短写作 f），存在 ramp function 的栈上。编译器从这一刻起就一直握着它。

假设 operator new 返回 0x5000，此刻堆上是这样：

   地址        协程帧（内容还是垃圾）
   0x5000  ┌──────────────────────────────────┐ ← f
           │ _Coro_resume_fn       （未填）    │
   0x5008  ├──────────────────────────────────┤
           │ _Coro_destroy_fn      （未填）    │
   0x5010  ├──────────────────────────────────┤ ← 偏移 16
           │ _Coro_promise         （未构造）  │
           ├──────────────────────────────────┤
           │ _Coro_resume_index / i / awaiter │
           └──────────────────────────────────┘

---

2.2 填好帧头部的两个函数指针

f->_Coro_resume_fn  = &counter.actor;       // 偏移 0
f->_Coro_destroy_fn = &counter.destroy;     // 偏移 8

源码里就是这两句（build_ramp_function，resumer / destroyer 是 actor 和 destroy 的函数声明）：

tree actor_addr = build1 (ADDR_EXPR, act_des_fn_ptr_type, resumer);
coro_build_and_push_artificial_var_with_dve (loc, coro_resume_fn_id, ..., actor_addr, deref_fp);
tree destroy_addr = build1 (ADDR_EXPR, act_des_fn_ptr_type, destroyer);
coro_build_and_push_artificial_var_with_dve (loc, coro_destroy_fn_id, ..., destroy_addr, deref_fp);

这两格加上紧随其后的 promise，是 ABI 规定死的前三个字段。后面 h.resume() / h.destroy() 就是取这两格来调用。

---

2.3 把参数拷贝进帧

// 本例 counter() 无参数，所以这一步是空的

★ 注意是"拷贝"。这就是"协程的引用参数会悬垂"那个经典坑的来源：

Task f(const std::string& s) {   // ✗ 帧里存的是【引用】，不是字符串本身
    co_await something();        //   挂起后调用者的临时 string 可能已经没了
    use(s);                      //   → 悬垂
}

这个坑不是无栈协程独有的。有栈协程同样有：

void task(void* arg) {                  // 有栈版本，arg 指向调用者的对象
    auto& s = *(std::string*)arg;
    co_yield();                         //   挂起期间调用者的 string 可能已经析构
    use(s);                             //   → 同样悬垂
}
co_resume(co_create(task, &std::string("tmp")));   // ✗ 临时对象在这个全表达式结束时销毁

两者的根源相同：协程的生命周期比一次函数调用长，而引用/指针指向的对象只活到调用者认为的"这次调用"结束。

区别只在于坑的显眼程度：

	有栈（co_create(fn, void* arg)）	无栈（C++20）
参数怎么传	你自己传 void*，一眼能看出是指针	写法和普通函数一模一样，引用参数看起来和普通函数一样安全
值参数安全吗	取决于库：入口函数只收 void*，值语义要自己包一层	安全：2.3 把值参数拷进帧，帧活多久它活多久
引用参数安全吗	不安全	不安全（帧里只存引用）
this / lambda 捕获	同样是指针，同样会悬垂	成员协程隐含 this，lambda 协程隐含闭包指针，都不进帧 —— 最常见的踩坑形式

C++20 之所以被单独拎出来说，是因为它把"这是一个会活很久的东西"这个事实藏在了普通函数调用的语法后面。

---

2.4 构造 promise，状态号清零

new (&f->_Coro_promise) promise_type();     // placement new，就地构造
f->_Coro_resume_index = 0;                  // 0 = "还没启动"

现在 f->_Coro_promise 的地址是 0x5010（偏移 16）。本例的 promise_type 只有一个 int 成员、构造函数是平凡的，所以 dump 里这一步没有任何代码；promise 有非平凡构造函数时才会看到构造调用。

偏移 16 是怎么来的？ = alignTo(2 * sizeof(void*), alignof(promise_type)) 也就是"把那两个函数指针占的 16 字节，向上对齐到 promise 的对齐要求"。 alignof(promise_type) ≤ 16 时，结果就是 16（绝大多数情况）。

---

2.5 ★★ 调用 get_return_object() —— handle 在这一刻诞生

auto _Coro_gro = f->_Coro_promise.get_return_object();
//               ^^^^^^^^^^^^^^^^ 注意：用的是【promise】，不是 h —— h 现在还不存在！

_Coro_gro 是 GCC 给这个变量起的名字（get-return-object），它就落在 ramp 的返回值槽里（dump：struct Generator _Coro_gro [value-expr: *<retval>]），不再拷贝一次。

你写的那个函数体：

Generator get_return_object() {
    return Generator{ std::coroutine_handle<promise_type>::from_promise(*this) };
}

逐步展开：

    this                     = &f->_Coro_promise = 0x5010
    from_promise(*this)      = (char*)this - 16  = 0x5000   ★ 就是 f！
    coroutine_handle{0x5000} ← handle 诞生
    Generator{ handle{0x5000} } ← 装进你的返回类型，存到 _Coro_gro

画成图：

   0x5000  ┌──────────────────────────────────┐ ◄─────┐
           │ _Coro_resume_fn                  │       │  ★ from_promise 算出来的
   0x5008  ├──────────────────────────────────┤       │     就是这个地址
           │ _Coro_destroy_fn                 │       │
   0x5010  ├──────────────────────────────────┤       │  减 16
           │ _Coro_promise  ← this 指这里     │ ──────┘
           ├──────────────────────────────────┤
           │ _Coro_resume_index / i / awaiter │
           └──────────────────────────────────┘

这一步之前 handle 不存在，所以编译器用的是它手里的 f，而不是 handle。get_return_object() 必须是 promise_type 的成员函数：编译器只调 promise 的成员函数、不额外传递帧指针，from_promise(*this) 靠 this 反推出帧地址。

---

2.6 调一次 actor

f->_Coro_frame_refcount = 1;    // ramp 正在用这个帧
counter.actor(f);               // ★ 源码注释："Start the coroutine body."
f->_Coro_frame_refcount -= 1;   // ramp 用完了

GCC 的 ramp 不自己展开 initial_suspend。co_await promise.initial_suspend() 被插在 actor 函数体的最前面（wrap_original_function_body），ramp 只是调一次 actor；actor 的分派器看到状态号 0 就跳到起点，执行的第一条语句就是这个 co_await，suspend_always 让它立刻 return 回到 ramp（展开见第 3 步 3.3）。

_Coro_frame_refcount 是帧的使用计数：ramp 调 actor 前置 1、返回后减 1；actor 进入时加 1、跑到头时减 1。谁把它减到 0，谁负责 operator delete 帧。正常情况下（initial_suspend 挂起）actor 在 ramp 之前退出，计数由 ramp 归零后由 ~Generator() → h.destroy() 走 destroy 路径释放；只有 initial_suspend 不挂起、函数体在 ramp 的这次调用里一路跑完时，才需要靠这个计数防止 actor 在 ramp 还没返回时就把帧删了。

如果 initial_suspend() 返回的是 suspend_never，await_ready() 就是 true，不挂起，直接往下跑进函数体 —— 那么 counter() 这一次调用就会一路执行到第一个 co_yield 才回来。这正是后面「执行顺序」那节要演示的差别。

---

2.7 把 2.5 存下的东西返回给调用者

return _Coro_gro;      // ← 就是 2.5 里 get_return_object() 的结果

main 里的 auto g = counter(); 拿到的就是它。

---

七步合起来

Generator<int> counter() {                                        // ← ramp function
    auto* f = (_Z7counterv.Frame*)operator new(40);                   // 2.1
    f->_Coro_frame_needs_free = true;
    f->_Coro_resume_fn  = &counter.actor;                             // 2.2
    f->_Coro_destroy_fn = &counter.destroy;
    /* 拷贝参数进帧（本例无参数） */                                    // 2.3
    /* 构造 promise（本例平凡，无代码） */                              // 2.4
    f->_Coro_resume_index = 0;
    auto _Coro_gro = f->_Coro_promise.get_return_object();            // 2.5 ★ handle 诞生
    f->_Coro_frame_refcount = 1;                                      // 2.6
    counter.actor(f);                     // → 跑到 initial_suspend 挂起，立刻回来
    f->_Coro_frame_refcount -= 1;
    return _Coro_gro;                                                 // 2.7
}

这就是 GCC 15.2 -fdump-tree-original 里 counter() 的全部内容，只去掉了异常处理（分配帧之后若抛异常：析构已构造的返回对象；_Coro_frame_refcount 为 0 就 operator delete 帧；重新抛出）。

从头到尾没有出现你写的 for 循环 —— 它在 counter.actor() 里 resume.2 之后，等第一次 h.resume() 才会被执行。

---

顺序为什么不能变：get_return_object 必须在 initial_suspend 之前

看 2.6 那一步 —— actor 在 initial_suspend 一挂起，ramp 就要立刻把返回值交给调用者（2.7）。 如果顺序反过来（先挂起再调 get_return_object），2.7 就没东西可返回了。

由此带来一个实际约束：不能在 get_return_object() 里访问函数体才会设置的东西。

struct promise_type {
    int value;                                  // co_yield 时才赋值
    Generator get_return_object() {
        printf("%d\n", value);                  // ✗ 此刻函数体一行没跑，value 是垃圾
        return Generator{ ... };
    }
};

---

from_promise 与 promise()：同一个偏移的两个方向

	用在哪	方向	什么时候用
from_promise(*this)	get_return_object() 里面	promise → 帧 (char*)this - 16	帧刚造好，handle 还不存在
h.promise()	g.value()	帧 → promise (char*)h + 16	handle 已经在手里了

两者调的是同一个编译器内建 __builtin_coro_promise(ptr, align, from)，最后一个参数就是方向开关。

第 3 步：counter.actor 逐步拆解 —— 它就是状态机

actor 是什么

先把"你写的函数体"贴在这里，下面反复对照的就是它：

Generator<int> counter() {
    for (int i = 0; i < 3; ++i) {
        printf("    counter: 准备让出 %d\n", i);
        co_yield i;                                 // ← 挂起点
        printf("    counter: 恢复了\n");
    }
}

actor 是一个普通函数：这个函数体，被编译器改写成"可以从中间 return 出去、下次调用再从那个位置接着跑"的形式。这种形式的函数就叫状态机。actor 是 GCC 给这个函数起的名字，状态机是它的形态。

一个普通函数只能从头跑到尾。要做到"从中间出去、再从中间回来"，只需要三样东西：

要素	在 actor 里是什么
① 一个记录"跑到哪了"的变量，放在函数外面	帧里的 _Coro_resume_index
② 函数开头一个 switch，按这个变量跳到上次停下的位置	3.1 分派器
③ 每个挂起点处：写变量、return；紧接着放一个标签，下次跳回来	3.2 co_await 的展开

对比一下：

// 普通函数：每次调用都从头开始
void f() { A; B; C; }

// 状态机：三次调用分别跑 A、B、C
void f(Frame* fr) {
    switch (fr->index) {              // ② 开头的 switch
        case 0: goto s0;
        case 1: goto s1;
        case 2: goto s2;
    }
s0: A;
    fr->index = 1; return;            // ③ 记下位置、出去
s1: B;                                //    ← 标签紧贴在上一句 return 后面：下次 goto s1 就等于"从 return 的下一行接着跑"
    fr->index = 2; return;
s2: C; return;
}

return 照常退出函数，s1: B; 这一行这次不会执行。它是留给下一次调用的：下一次进来，开头的 switch 读到 index == 1，goto s1 直接跳到这一行。三次调用逐行走一遍：

第 1 次 f(fr)   index == 0 → goto s0 → 跑 A → index = 1 → return（退出 f）
第 2 次 f(fr)   index == 1 → goto s1 → 跑 B → index = 2 → return（退出 f）   ← 跳过了 A
第 3 次 f(fr)   index == 2 → goto s2 → 跑 C → return

每次调用都从头进入 f，先走一遍开头的 switch；index 记着上次停在哪，switch 就 goto 到哪。这就是"跳到上次停下的地方"。

counter.actor 就是这个结构，只是：位置变量叫 _Coro_resume_index，switch 多了一张 destroy 用的表，每个"记位置 + return"被包在 co_await 的展开里，而 A、B、C 就是你写的 for 循环被挂起点切成的几段。

actor 的完整实现

先说清一件事，否则下面的代码会让人找不到 co_await：actor 里一共有三个 co_await，而 dump 是展开之后的结果。

• ① co_await promise.initial_suspend() —— 编译器插在你函数体前面的（wrap_original_function_body）
• ② co_await promise.yield_value(i) —— 你写的 co_yield i，编译器直接改写成这个
• ③ co_await promise.final_suspend() —— 编译器插在你函数体后面的

这三个 co_await 各按 3.2 的同一套模板展开成 await_ready / _Coro_resume_index = N / await_suspend / return / resume.N: / await_resume，展开完才是 dump 里的样子。所以下面的代码里没有 co_await 字样；每一段展开的起止位置用 ┌ … ┘ 标了出来，原来的那句 co_await 写在段首注释里。

跳转表为什么正好有 2/4/6 三项：编译器扫一遍函数体数出 co_await / co_yield 的个数（本例 1 个，源码里叫 await_count），加上自己插的两个，一共 3 个挂起点，编号 2、4、6；build_actor_fn 里两个 for (… < body_count …) 循环据此生成 resume 表的 case 0/2/4/6 和 destroy 表的 case 1/3/5/7。函数体里多写一个 co_await，两张表各多一项。

下面是 counter.actor 的全部代码。来源是 GCC 15.2 g++ -std=c++20 -O0 -fdump-tree-original 生成的 stackless.cpp.006t.original，只做了三件整理：去掉 dump 的 <<cleanup_point …>> 之类的包装标记；把 frame_ptr->_Coro_promise 简写成 promise、frame_ptr->Is_1_1 简写成 Is（其他帧字段同理）；把 dump 里的匿名标签 <D.12949> 改名为 loop_body。函数名、帧字段名、其余标签名、语句顺序都和 dump 一致。

void counter.actor(_Z7counterv.Frame* frame_ptr)
{
    // ═══════════ 3.1 分派器：每次进来先走这里 ═══════════
    if (frame_ptr->_Coro_resume_index & 1) {              // 奇数：destroy 路径
        switch (frame_ptr->_Coro_resume_index) {
            case 1:  goto coro.delete.promise;
            case 3:  goto destroy.2;
            case 5:  goto destroy.4;
            case 7:  goto destroy.6;
        }
    } else {                                              // 偶数：resume 路径
        switch (frame_ptr->_Coro_resume_index) {
            case 0:  goto actor.begin;                    // ramp 调进来，第一次
            case 2:  goto resume.2;
            case 4:  goto resume.4;
            case 6:  goto resume.6;
        }
    }
    __builtin_trap();                                     // 不该到这

actor.begin:
    frame_ptr->_Coro_initial_await_resume_called = false;
    frame_ptr->_Coro_frame_refcount += 1;                 // 函数体开始用这个帧
    try {
        // ┌─────── ① co_await promise.initial_suspend(); 展开后 ───────（3.3）
        Is = promise.initial_suspend();                   // suspend_always{}
        frame_ptr->_Coro_resume_index = 2;
        if (!Is.await_ready()) {                          // false → 挂起
            Is.await_suspend(coroutine_handle<promise_type>::from_address(frame_ptr));
            return;                                       // ★ 挂起：回到 ramp
        destroy.2:
            goto coro.delete.promise;
        }
    resume.2:
        frame_ptr->_Coro_initial_await_resume_called = true;
        Is.await_resume();
        // └──────────────────────────────────────────────────────────

        // ═══════════ 3.4 你写的函数体（for 已降成标签 + 跳转）═══════════
        i = 0;                                            // for 的初始化
        goto loop_cond;
    loop_body:
        printf("    counter: 准备让出 %d\n", i);
        // ┌─────── ② co_yield i;  即 co_await promise.yield_value(i); 展开后 ───────（3.4）
        Yd0 = promise.yield_value(i);                     // value = i 就发生在这个调用里
        frame_ptr->_Coro_resume_index = 4;
        if (!Yd0.await_ready()) {                         // false → 挂起
            Yd0.await_suspend(coroutine_handle<promise_type>::from_address(frame_ptr));
            return;                                       // ★ 挂起：回到 h.resume() 的调用者
        destroy.4:
            goto coro.delete.promise;
        }
    resume.4:
        Yd0.await_resume();
        // └──────────────────────────────────────────────────────────
        printf("    counter: 恢复了\n");
        ++i;                                              // for 的递增
    loop_cond:
        if (i <= 2) goto loop_body;                       // for 的条件

        promise.return_void();                            // 函数体自然结束 → 隐式 co_return
    } catch (...) {
        if (!frame_ptr->_Coro_initial_await_resume_called) throw;   // initial_suspend 之前的异常归 ramp 管
        frame_ptr->_Coro_resume_fn = nullptr;
        frame_ptr->_Coro_resume_index = 0;
        promise.unhandled_exception();
    }

final.suspend:
    frame_ptr->_Coro_resume_fn = nullptr;                 // ★ h.done() 查的就是这一格（编译器插的，3.5）
    // ┌─────── ③ co_await promise.final_suspend(); 展开后 ───────（3.5）
    Fs = promise.final_suspend();                         // suspend_always{}
    frame_ptr->_Coro_resume_index = 6;
    if (!Fs.await_ready()) {
        Fs.await_suspend(coroutine_handle<promise_type>::from_address(frame_ptr));
        return;                                           // ★ 挂起：协程结束，帧保留，等 h.destroy()
    destroy.6:
        goto coro.delete.promise;
    }
resume.6:
    Fs.await_resume();                                    // 到不了：_Coro_resume_fn 已清零
    // └──────────────────────────────────────────────────────────

    frame_ptr->_Coro_frame_refcount -= 1;                 // 函数体用完了
    if (frame_ptr->_Coro_frame_refcount != 0) return;     // ramp 还在用（initial_suspend 是 suspend_never 且一路跑完时）

    // ═══════════ 第 4 步：destroy 路径的汇合点 ═══════════
coro.delete.promise:
    /* 析构 promise —— 本例平凡，无代码 */
    if (frame_ptr->_Coro_frame_needs_free)
        operator delete(frame_ptr, 40);
    return;
}

dump 里每个挂起点的 return; 处原文是一个 switch，那是前端留给后面 pass 的占位，最终就是一句 return：

switch (.CO_YIELD (4, 0, &resume.4, &destroy.4, frame_ptr)) {
    case 0:  .CO_SUSPN (&actor.suspend.ret);    // → goto actor.suspend.ret，那里是 return
    case 1:  goto resume.4;
    default: goto destroy.4;
}

gcc/coroutine-passes.cc 把 .CO_YIELD 的结果折叠成 0（所以只剩 case 0），同时把分派器里的 .CO_ACTOR (4) 接到 resume.4、.CO_ACTOR (5) 接到 destroy.4。

这个函数就是状态机：开头的两个 switch 是跳转表；三个 ┌ … ┘ 段各是一个 co_await 展开后的"记位置、出去、回来"；你写的 for 循环就是 loop_body / loop_cond 之间那几行。下面 3.1～3.5 按块逐段解释。

---

3.1 分派器：先看最低位，再查跳转表

actor 的第一件事是读 frame_ptr->_Coro_resume_index，按「先建立坐标」里的编号规则跳转。和前面 A; B; C 那个小例子的区别只有一点：先用最低位区分 resume 还是 destroy，各查一张表：

if (frame_ptr->_Coro_resume_index & 1) {      // 奇数 → destroy 路径（第 4 步）
    switch (frame_ptr->_Coro_resume_index) {
        case 1: goto coro.delete.promise;     // unhandled_exception() 之后被 destroy
        case 3: goto destroy.2;               // 停在 initial_suspend 上被 destroy
        case 5: goto destroy.4;               // 停在 co_yield 上被 destroy
        case 7: goto destroy.6;               // 停在 final_suspend 上被 destroy（正常跑完后 ~Generator 走这里）
    }
} else {                                      // 偶数 → resume 路径
    switch (frame_ptr->_Coro_resume_index) {
        case 0: goto actor.begin;             // ramp 调进来，第一次
        case 2: goto resume.2;                // initial_suspend 之后
        case 4: goto resume.4;                // co_yield 之后
        case 6: goto resume.6;                // 到不了：final_suspend 前已把 _Coro_resume_fn 清零
    }
}
__builtin_trap();                             // 两张表都没命中：不该到这

dump 原文里 case 2: 的位置写的是 .CO_ACTOR (2); goto <默认出口>，coroutine-passes.cc 把它改接成 goto resume.2；上面直接写了改接后的结果。

恢复 = 一次普通函数调用 + 一个 switch。 h.resume() 取帧偏移 0 的函数指针调到这里，switch 里的 goto resume.N 跳到标签 resume.N:，而这个标签就贴在上次那句 return 的下一行（见 3.2 的展开），所以执行从上次离开的位置继续。

---

3.2 每个 co_await 的通用展开

任何 co_await expr 都展开成固定的这一套（expand_one_await_expression）：

{
    awaiter = expr;                                          // ★ awaiter 是帧字段，它要活过挂起点
    frame_ptr->_Coro_resume_index = N;                       //   先记下"下次从 resume.N 进来"
    if (!awaiter.await_ready()) {                            // ① 要不要挂起？
        awaiter.await_suspend(coroutine_handle<promise_type>::from_address(frame_ptr));
                                                             // ② 挂起前最后做点事（注册回调等），参数就是帧的 handle
        return;                                              // ★★★ ③ 挂起 = 从 actor return，回到调用 resume 的人
    destroy.N:
        goto coro.delete.promise;                            //   停在这个挂起点上被 destroy 时的落点
    }

resume.N:                                                    // ★★★ 恢复 = 分派器跳回这里
    /* 结果 = */ awaiter.await_resume();                     // ④ 取结果
}

传给 await_suspend 的 handle 每次都用 from_address(frame_ptr) 现算（就是把帧指针包一层，第 1 步讲过），再转成 coroutine_handle<>。N 从 2 开始，每个挂起点加 2。

awaiter 三件套的职责：

函数	什么时候调	干什么
await_ready()	挂起前	返回 true = 不用挂，直接往下走（快路径）
await_suspend(h)	正在挂起时	把 h 存到某处，将来好有人来 h.resume()（注册到 epoll、放进队列……）
await_resume()	恢复后	返回 co_await 表达式的值

std::suspend_always 的定义就三行，全是空的：

struct suspend_always {
    bool await_ready()   const noexcept { return false; }              // 我【一定】挂起
    void await_suspend(std::coroutine_handle<>) const noexcept { }     // 挂起时什么都不做
    void await_resume() const noexcept { }                             // 恢复时什么都不做
};

struct suspend_never {
    bool await_ready()   const noexcept { return true; }               // 我【绝不】挂起
    void await_suspend(std::coroutine_handle<>) const noexcept { }
    void await_resume() const noexcept { }
};

所以 suspend_always 的含义是："挂起，然后什么都不安排" —— 谁来 resume 我？外面的人自己拿着 handle 来调。generator 正是这样：g.next() 就是外面那个人。真正的异步库会在 await_suspend 里把 h 交给事件源，谁来调、要不要操作系统参与，见「补充 3」。

---

3.3 起点：co_await promise.initial_suspend()

ramp 把 _Coro_resume_index 设成 0 后调 actor，分派器跳到 actor.begin。这里执行的第一条语句不是你写的代码，而是编译器插入的（wrap_original_function_body）：

actor.begin:
    Is = promise.initial_suspend();                   // Is 是帧字段 Is_1_1，值为 suspend_always{}
    frame_ptr->_Coro_resume_index = 2;
    if (!Is.await_ready()) {                          // suspend_always → false → 挂起
        Is.await_suspend(coroutine_handle<promise_type>::from_address(frame_ptr));
        return;                                       // ★ 回到 ramp，ramp 接着 return
    }
resume.2:
    frame_ptr->_Coro_initial_await_resume_called = true;  // 之后再抛异常就归 unhandled_exception() 管
    Is.await_resume();
    // ↓ 从这里开始才是你写的函数体

这就是「counter() 返回时函数体一行没跑」的机制：ramp 调 actor，actor 在这里 return，ramp 再 return。第一次 h.resume() 时 _Coro_resume_index == 2，分派器跳到 resume.2，函数体才开始。

---

3.4 你写的函数体，以及 co_yield 展开成什么

for 循环先被降成标签 + 跳转（任何函数都这样，详见「补充 2」）：

    i = 0;                                        // ← for 的初始化（i 是帧字段 i_2_3，因为它要活过挂起点）
    goto loop_cond;
loop_body:
    printf("    counter: 准备让出 %d\n", i);
    /* co_yield i 的展开，见下 */
    printf("    counter: 恢复了\n");
    ++i;                                          // ← for 的递增
loop_cond:
    if (i <= 2) goto loop_body;                   // ← for 的条件（GCC 把 i < 3 写成 i <= 2）

co_yield i 分两层代入，最后落到 3.2 的通用展开：

第 1 层：co_yield 是 co_await 的语法糖

co_yield i;
// ↓ 编译器直接改写成
co_await promise.yield_value(i);

这就是为什么你的 promise_type 必须提供 yield_value —— 不提供，co_yield 就没法展开。

第 2 层：代入 yield_value 的定义

std::suspend_always yield_value(T v) { value = v; return {}; }

代进去：

promise.value = i;                    // ← yield_value 的函数体
co_await std::suspend_always{};       // ← 它的返回值，被 co_await

第 3 层：套用 3.2 的通用展开，N = 4

    Yd0 = promise.yield_value(i);                 // ← Yd0 是帧字段 Yd0_3_4；value = i 在 yield_value 里完成
    frame_ptr->_Coro_resume_index = 4;
    if (!Yd0.await_ready()) {                     // false → 挂起
        Yd0.await_suspend(coroutine_handle<promise_type>::from_address(frame_ptr));   // 空函数
        return;                                   // ★★★ 挂起 = return！
    }
resume.4:
    Yd0.await_resume();                           // 空函数

-O0 的 dump 里 yield_value 还是一次真实调用；开优化后它被内联，剩下的就是一句 frame_ptr->_Coro_promise.value = i。

---

3.5 结尾：return_void()、清零 _Coro_resume_fn、co_await promise.final_suspend()

循环退出后，编译器又插了一段（wrap_original_function_body 结尾）：

    promise.return_void();                        // ← 隐式的 co_return
final.suspend:
    frame_ptr->_Coro_resume_fn = nullptr;         // ★ 先清零 —— h.done() 查的就是这一格
    Fs = promise.final_suspend();                 // Fs 是帧字段 Fs_1_5
    frame_ptr->_Coro_resume_index = 6;
    if (!Fs.await_ready()) {
        Fs.await_suspend(coroutine_handle<promise_type>::from_address(frame_ptr));
        return;                                   // ★ 协程结束。帧还在，promise 还在，等 h.destroy()
    }
resume.6:                                         // 到不了
    Fs.await_resume();

源码注释："Before entering the final suspend point, we signal that this point has been reached by setting the resume function pointer to zero (this is what the 'done()' builtin tests)."

final_suspend 返回 suspend_always 的意义：协程跑完了但帧不自动释放，调用者还能读 promise.value，最后由 h.destroy() 收尾（第 4 步）。如果返回 suspend_never，协程跑完立刻走 destroy 路径释放帧，此后 h 就是悬垂指针。

---

第 4 步：counter.destroy 逐步拆解

第三个函数只有两行（build_destroy_fn）：

void counter.destroy(_Z7counterv.Frame* frame_ptr) {              // ← h.destroy() 调到这
    frame_ptr->_Coro_resume_index = frame_ptr->_Coro_resume_index | 1;   // 4.1 置最低位
    counter.actor(frame_ptr);                                      // 4.2 调 actor
}

这是 dump 原文的全部内容，一字未改。

销毁帧的代码全在 actor 里（destroy.N 和 coro.delete.promise），这个函数存在的唯一理由是 h.destroy() 和 h.resume() 的调用方式完全一样：取帧偏移 8（或 0）的函数指针，传帧指针，调用，别的什么都不做。actor 靠 _Coro_resume_index 的最低位决定走哪条路，而 h.destroy() 不会去改这个值，所以偏移 8 那格必须指向一个"先置位、再调 actor"的函数，这就是 counter.destroy。如果偏移 8 直接填 counter.actor，h.destroy() 就会变成一次 resume。

4.1 置最低位：协程只可能停在某个挂起点上，此时 _Coro_resume_index 是偶数 2/4/6。|= 1 变成 3/5/7，正好是同一个挂起点对应的 destroy 编号。

4.2 调 actor：3.1 的分派器看到奇数，走 destroy 那张表：

destroy.2:  goto coro.delete.promise;             // 停在 initial_suspend 上就被销毁
destroy.4:  goto coro.delete.promise;             // 停在 co_yield 上就被销毁（generator 没跑完）
destroy.6:  goto coro.delete.promise;             // 跑完了，停在 final_suspend 上 —— 本例 ~Generator() 走这里
coro.delete.promise:                              // case 1 也直接跳这里
    /* 析构 promise —— 本例平凡，无代码 */
    if (frame_ptr->_Coro_frame_needs_free) operator delete(frame_ptr, 40);
    return;

本例三个 destroy.N 长得一样，是因为 int 和 suspend_always 的析构都是平凡的。帧里有带析构函数的局部变量（比如一个 std::string）时，destroy.N 处会先析构"停在这个挂起点时还活着"的那些变量，再跳到 coro.delete.promise。

为什么 destroy 不是独立函数体而是调 actor：resume 和 destroy 两条路径共享同一套"哪个挂起点上哪些变量活着"的信息，放在一个函数里状态机只需优化一次（源码注释："the intent of keeping the resume and destroy paths together is that the conditionals controlling them are identical"）。

---

第 5 步：全部拼起来 —— 逐行对照

你写的：

Generator<int> counter() {
    for (int i = 0; i < 3; ++i) {
        printf("    counter: 准备让出 %d\n", i);
        co_yield i;
        printf("    counter: 恢复了\n");
    }
}

编译器生成的三个函数放在一起看（帧结构见「先建立坐标」，actor 的完整代码见第 3 步，这里只列骨架）：

// ================= ① ramp function（拆解见第 2 步）=================
//   ★ counter 这个【符号】编译后就是它，里面【没有】你写的 for 循环
Generator<int> counter() {
    auto* f = (_Z7counterv.Frame*)operator new(40);
    f->_Coro_frame_needs_free = true;
    f->_Coro_resume_fn  = &counter.actor;
    f->_Coro_destroy_fn = &counter.destroy;
    f->_Coro_resume_index = 0;
    auto _Coro_gro = f->_Coro_promise.get_return_object();   // ★ 必须在 initial_suspend 之前
    f->_Coro_frame_refcount = 1;
    counter.actor(f);              // ★ 跑到 initial_suspend 挂起就回来
    f->_Coro_frame_refcount -= 1;
    return _Coro_gro;              // ★ 函数体一行没跑就返回了
}

// ================= ② actor：真正的函数体 + 状态机（完整代码见第 3 步）=================
void counter.actor(_Z7counterv.Frame* frame_ptr) {
    /* 分派器：奇数查 destroy 表，偶数查 resume 表 */
actor.begin:
    /* 挂起点 ①  initial_suspend  → _Coro_resume_index = 2; return;   resume.2: */
    /* 你写的 for 循环 */
    /* 挂起点 ②  co_yield i       → _Coro_resume_index = 4; return;   resume.4: */
    /* promise.return_void() */
final.suspend:
    /* _Coro_resume_fn = nullptr */
    /* 挂起点 ③  final_suspend    → _Coro_resume_index = 6; return;   resume.6: */
coro.delete.promise:
    /* 析构 promise；if (_Coro_frame_needs_free) operator delete(frame_ptr) */
}

// ================= ③ destroy：一个 trivial shim（拆解见第 4 步）=================
void counter.destroy(_Z7counterv.Frame* frame_ptr) {   // ← h.destroy() 调到这
    frame_ptr->_Coro_resume_index |= 1;                  // ★ 置最低位 = "走 destroy 路径"
    counter.actor(frame_ptr);                            // ★ 跳进 actor
}

代码里的名字都能在 gcc/cp/coroutines.cc 找到：

代码里的写法	源码
resume.N / destroy.N	expand_one_await_expression 里 xasprintf ("resume.%d", …) / ("destroy.%d", …)
final.suspend / coro.delete.promise	create_named_label_with_ctx (loc, "final.suspend", …) / "coro.delete.promise"
from_address(frame_ptr)	data->hfa_m（handle-from-address），每个 await_suspend 调用前现算
f->_Coro_resume_fn = nullptr	final.suspend 标签后的 zero_resume，注释："this is what the 'done()' builtin tests"
_Coro_resume_index |= 1; actor(f)	build_destroy_fn

---

第 6 步：main 里的每一行对应什么

int main() {
    auto g = counter();
    while (g.next()) printf("main 收到 %d\n", g.value());
}

你写的	实际发生
counter()	调 ramp function → 分配帧、构造 promise、拿 Generator、调 counter.actor 跑到 initial_suspend 挂起、返回
g.next() 里的 h.resume()	f->_Coro_resume_fn(f) → counter.actor(f) → switch 跳到 resume.N
g.next() 里的 h.done()	读 f->_Coro_resume_fn == nullptr
g.value() 里的 h.promise()	*(promise_type*)((char*)f + 16) → 取 .value
~Generator() 里的 h.destroy()	f->_Coro_destroy_fn(f) → counter.destroy(f)（shim：置最低位后调 actor）→ actor 走 destroy 路径：析构 + operator delete

整条链路上没有一次虚函数调用、没有 RTTI。 全是"函数指针 + 编译期偏移 + 跳转表"。

---

第 7 步：三个关键结论

直接从上面那段展开代码读出来：

结论	代码里的依据
挂起 = return	co_yield 展开里那句 return; —— 所以协程必须借用调用者的栈，自己没有栈
恢复 = 一次普通函数调用 + 一次跳转表分派	f->_Coro_resume_fn(f) + switch (f->_Coro_resume_index) —— 所以切换只要几纳秒
帧大小编译期确定	sizeof(_Z7counterv.Frame) == 40 是常量 —— 只有 i 和三个 awaiter 进了帧

对比有栈协程：那边挂起是换 rsp，一整个 64 KB 的栈原封不动地留着。

---

拆解到此结束。下面补三个绕不开的问题：编译器凭什么知道要做这套变换、展开后 for 循环为什么不见了，以及挂起之后谁来调 h.resume()。

补充 1：编译器怎么知道"这是个协程"

上面拆的是"变换之后长什么样"。这里回头补一个更靠前的问题：编译器凭什么认定 counter() 是协程、要对它做这套变换？

判据只有一条：函数体里出现了那三个关键字之一。

co_await   /   co_yield   /   co_return

出现任意一个，这个函数就是协程。纯语法触发，没有别的条件（[dcl.fct.def.coroutine]）：

A function is a coroutine if its function-body encloses a coroutine-return-statement, an await-expression, or a yield-expression.

所以编译器扫到 co_yield i; 这一行，立刻就知道要走协程流程。

一个反直觉的推论：从声明完全看不出来。

Generator<int> counter();          // ← 是不是协程？看不出来

• 没有 async 关键字（不像 Rust 的 async fn、C# 的 async）
• 返回类型不是判据 —— Generator<int> 对编译器来说就是个普通类型
• 只由函数体决定

这意味着同一个声明，实现可以是协程也可以不是，调用方完全无感：

// 写法 ①：是协程
Generator<int> counter() { co_yield 1; }
// 写法 ②：不是协程，手工造一个 Generator 返回
Generator<int> counter() { return make_generator_somehow(); }

所以 C++ 的"函数染色"染的是返回类型（返回类型必须能提供 promise_type），而不是签名上的一个关键字。

确认是协程之后，编译器做四件事：

① 从【返回类型】找 promise_type
     std::coroutine_traits<Generator<int>>::promise_type
     默认就是 Generator<int>::promise_type
     找不到 → error: 'Generator<int>' has no member named 'promise_type'

② 检查 promise 提供了必需的成员
     get_return_object() / initial_suspend() / final_suspend() noexcept
     unhandled_exception() / return_void() 或 return_value()
     yield_value(x)        ★ 只有用了 co_yield 才要求

③ 检查禁止项
     不能有普通 return（只能 co_return）、不能是 constexpr/consteval、
     不能是构造/析构函数、不能是 main、不能有 C 风格变参、返回类型不能推导

④ 开始做状态机变换  ← 就是上面第 1～7 步拆的那一整套

你写的那些 promise_type 方法，本质是"编译器回调你的接口" —— 编译器负责切状态机，但"挂起后干什么、返回值长什么样"它不知道，只能问你。

---

补充 2：那 for 循环到底去哪了

回头看第 3 步那份 actor 代码 —— for 不见了，只剩 goto。很多人卡在这里 —— 其实：

for 循环没有消失，它变成了一条 goto。而且这一步和协程无关。

① for 本来就会被降级成跳转（任何编译器、任何函数都这样）

编译器内部从来不用"for 循环"这种结构表示代码，它先把源码转成控制流图（CFG）：基本块 + 跳转边。

for (int i = 0; i < 3; ++i) { BODY }

在编译器内部（不管是不是协程）本来就长这样：

    i = 0;                    ← 初始化
loop:
    if (!(i < 3)) goto done;  ← 条件
    BODY                      ← 循环体
    ++i;                      ← 递增
    goto loop;                ← ★ 这条边【就是】"循环"
done:

看普通函数的汇编就知道 —— 里面只有 cmp 和 jmp，根本没有"for"这个东西（实测：void loop() { for (int i = 0; i < 3; ++i) g += i; }，g++ -O0 -masm=intel）：

    mov  DWORD PTR -4[rbp], 0      ; i = 0
    jmp  .L2                       ; 先去判断条件
.L3:
    ...                            ; 循环体 g += i
    add  DWORD PTR -4[rbp], 1      ; ++i
.L2:
    cmp  DWORD PTR -4[rbp], 2      ; i <= 2 ?（GCC 把 i < 3 写成 i <= 2）
    jle  .L3                       ; ★ 跳回去 —— 这就是"循环"

这也正是第 3 步 actor 代码里 goto loop_cond; loop_body: … loop_cond: if (i <= 2) goto loop_body; 的形状。

所以问题该反过来问：不是"for 循环消失了"，而是 for 循环从来就只是一条跳转边，只不过平时你看的是源码，才觉得它是个"结构"。

② 协程特有的 —— 在挂起点把控制流"剪断"

现在函数体已经是一串带标签的直线代码。编译器扫一遍找到 co_yield，在那个位置做手术：

    i = 0;
loop:
    if (!(i < 3)) goto done;
    printf("准备让出 %d", i);

  ════════ ✂ 在挂起点剪断 ════════
    _Coro_resume_index = 4;
    return;                       ← 前半段：走人
  resume.4:                       ← 后半段：从这里回来
  ═══════════════════════════════

    printf("恢复了");
    ++i;
    goto loop;                    ← ★ for 循环在这儿，好好的
done:
    _Coro_resume_fn = nullptr;    ← 标记"已结束"
    return;

再在函数最开头加一个 switch，把所有恢复点串起来：

switch (_Coro_resume_index) {
    case 0: goto actor.begin;
    case 4: goto resume.4;
}

整个变换就是这样。 for 从头到尾没被动过 —— goto loop 那条边一直在。

③ 决定哪些变量要"跨过剪口"

做一次活跃性分析：i 在剪口前后都要用 → 它必须活过这次 return → 提升进堆上的协程帧。 其他没跨过剪口的变量留在栈上就行（反正 return 时它们已经死了）。

为什么必须"先摊平，才能切"

这是整件事的关键：

如果 for 还是一个不可分割的语法结构，就没法在中间剪断 —— 你不可能"从循环中间 return，下次直接跳回循环中间"。 但一旦它被降级成 label + goto，"跳回循环中间"就变成一件再普通不过的事。

一个手写的等价物最直白地证明了这一点（这里的 state 就是编译器的 _Coro_resume_index）：

switch (state) {
case 0:
    for (i = 0; i < 3; ++i) {
        printf("准备让出 %d\n", i);
        out = i; state = 1;
        return true;              // ← 从循环中间返回
case 1:                           // ★ case 标签【长在 for 循环体里面】
        printf("恢复了\n");
    }
}

这段代码能编译，正是因为 switch/case 在这个层面上也只是"标签 + 跳转"（C 标准允许 case 出现在 switch 体内任何位置，Duff's device 就是这么玩的）。

手写版和编译器做的，是同一件事。 编译器只是替你做了三件苦力活：找出哪些变量要进帧、给挂起点编号、把 switch 生成出来。

---

补充 3：谁来调 h.resume() —— 协程不依赖操作系统

谁来调 h.resume()，编译器不管，也不需要操作系统参与。 await_suspend 拿到的只是一个 coroutine_handle（8 字节的帧指针），把它存到哪、什么时候拿出来调 h.resume()，完全是库代码的事：

场景	await_suspend 把 h 放到哪	谁来调 h.resume()
本文的 generator	哪都不放（suspend_always 是空函数）	main 里的 g.next() —— 调用者自己就是驱动者
Linux 异步 I/O 库	和 fd 一起交给 epoll / io_uring	事件循环收到就绪事件后
Windows 异步 I/O 库	和 OVERLAPPED 一起交给 IOCP	GetQueuedCompletionStatus 返回后
macOS / BSD	kqueue	同上
线程池	塞进任务队列	某个工作线程
定时器	挂到定时器堆	到期时

表里的第一行就是本文的例子：没有事件循环，也没有任何 I/O，main 的 while (g.next()) 就是驱动协程的那个东西——每转一圈调一次 h.resume()。generator 是最简单的形态：驱动者和调用者是同一个人，同步地、按需地摇把手。其他几行只是把"摇把手的人"换成了事件循环或线程池。

所以协程本身不依赖任何平台特性：整套变换（帧、状态号、switch、三个函数）都是编译器在源码层面做的，MSVC / GCC / Clang 在 Windows、Linux、macOS 上都支持。epoll、IOCP 这些只是"等 I/O 的时候把 CPU 让出去"这个用途下的事件源，换一个平台换一个事件源，协程的代码一行不用改。「有栈协程为什么能工作 —— 三个前提条件」一节末尾有对比：有栈协程对平台有硬性要求，无栈协程一个条件都不需要。

---

真实编译器里怎么做的（GCC）

补充 2 里的"剪断"不是比喻，GCC 里真有这么几步（gcc/cp/coroutines.cc，入口 cp_coroutine_transform::apply_transforms）：

① 前端识别
     看到 co_yield → 确认是协程 → 从返回类型找 promise_type

② 切分（split_coroutine_body_from_ramp / wrap_original_function_body）
     把用户写的函数体从原函数里摘出来，前后包上
        co_await promise.initial_suspend();
        try { 用户函数体; promise.return_void(); } catch (...) { promise.unhandled_exception(); }
     final.suspend:
        _Coro_resume_fn = nullptr;
        co_await promise.final_suspend();

③ 建帧类型
     xref_tag (record_type, get_fn_local_identifier (orig_fn_decl, "Frame"))  → _Z7counterv.Frame
     字段：_Coro_resume_fn / _Coro_destroy_fn / _Coro_promise / _Coro_resume_index /
           _Coro_frame_refcount / _Coro_frame_needs_free / _Coro_initial_await_resume_called /
           参数拷贝 / 跨挂起点存活的局部变量和 awaiter（Is_1_1、i_2_3、Yd0_3_4、Fs_1_5）

④ 生成三个函数
     build_ramp_function     → counter          用户调的就是它，最后调一次 actor
     build_actor_fn          → counter.actor    ★ 分派器 + 真正的函数体
     build_destroy_fn        → counter.destroy  trivial shim：_Coro_resume_index |= 1; actor(f)
     ★ resume 和 destroy 两条路径都在 actor 里，靠 _Coro_resume_index 的最低位区分：
       偶数走 resume 的跳转表，奇数走 destroy 的
     ★ 每个 co_await 由 expand_one_await_expression 展开成
       "写 _Coro_resume_index = N；await_ready；await_suspend；return；resume.N: await_resume"，
       N 从 2 起每次加 2

⑤ 内建展开（gcc/coroutine-passes.cc）
     __builtin_coro_resume  / _destroy → 取帧偏移 0 / 8 的函数指针并调用
     __builtin_coro_done              → 帧偏移 0 是否为 null
     __builtin_coro_promise           → 帧地址 ± ROUND_UP(2*sizeof(void*), align)
     .CO_YIELD / .CO_ACTOR            → 把分派器里 case N 的跳转目标接到 resume.N / destroy.N

可以亲眼验证。 符号表里三个函数实打实存在（nm -C 输出见「先建立坐标」）。看前端变换刚做完时的中间表示，第 3 步那份完整代码就是从这里整理出来的：

g++ -std=c++20 -O0 -fdump-tree-original -c stackless.cpp
less stackless.cpp.*.original     # 搜 _Coro_resume_index，能看到分派器和三个挂起点

看帧的真实布局：

g++ -std=c++20 -g -c stackless.cpp && readelf --debug-dump=info stackless.o | grep -A40 '_Z7counterv.Frame'

（Compiler Explorer 上选 GCC + "Tree/RTL" 输出也能直接看到。）

---

新手最容易卡住的七个点

疑问	答案
coroutine_handle 是什么高级东西？	就是帧指针，sizeof == 8，方法全是"取偏移"
promise_type 为什么要我自己写？	编译器只负责切状态机，"挂起后干什么、返回值长什么样"必须你来定。它是编译器回调你的接口
co_yield 和 co_await 什么关系？	co_yield x ≡ co_await promise.yield_value(x)，co_await 才是唯一的核心
suspend_always 干了什么？	什么都没干。它只是 await_ready() 返回 false，意思是"挂起，然后等外面的人拿 handle 来 resume 我"
for 循环怎么不见了？	它一直都在 —— 就是展开代码里那句 goto loop。任何编译器都会先把 for 降级成"标签 + 跳转"，协程变换只是在这串已经摊平的代码上剪一刀
编译器怎么知道要做这个变换？	函数体里出现 co_await/co_yield/co_return 任意一个 —— 纯语法触发，从函数声明上完全看不出来
帧里那个 promise 到底装的是什么？	promise_type 的一个【对象】。大小只算数据成员，成员函数不占空间 —— 详见下面

详解：promise 那一格到底装的是什么？ 是 promise_type 这个类的一个对象（不是类型本身）—— 也就是你写在 Generator 里的那个嵌套 struct promise_type。 它的大小 只算数据成员：本例只有一个 T value，所以 sizeof(promise_type) == sizeof(int) == 4； 那 6 个成员函数是代码（在 .text 段），一个字节都不占。

static_assert(sizeof(Generator<int>::promise_type) == sizeof(int));   // 4

那这一格是不是就等于 T value？ 本例中【是】，但那是巧合 —— 因为你的 promise_type 只有这一个数据成员：

   0x5010  ┌──────────────────────────────────┐ ← promise 对象【起点】
           │ ╔══════════════════════════════╗ │
           │ ║  promise_type 的一个对象      ║ │
           │ ║      int value;              ║ │  ← ★ 恰好占满整格
           │ ╚══════════════════════════════╝ │
   0x5014  └──────────────────────────────────┘

   &h.promise()        == 0x5010     promise 对象的地址
   &h.promise().value  == 0x5010     value 的地址（在 promise 里偏移 0）
                                     ★ 本例两者重合

一旦往 promise_type 里多加成员，两者就分开了，协程帧也跟着变大：

struct promise_type {
    T                  value;         // 偏移 0
    std::exception_ptr exc;           // 偏移 8
    int                yield_count;   // 偏移 16
};   // sizeof == 24 → 帧里那一格从 4 字节涨到 24 字节

反过来，promise 也可以一个数据成员都没有（"发射后不管"型协程就不需要往外传值）， 这时 sizeof(promise_type) == 1（空类）。

value 这几个字节的作用，就是协程与调用者之间的传值通道： co_yield i → promise.yield_value(i) → value = v（写）； g.value() → h.promise().value（读）。 两边共享同一块内存，没有拷贝、没有队列、没有锁 —— 因为压根不并行。

它也是整个协程帧里唯一由你控制的部分：

帧里的东西	谁定的	你能改吗
_Coro_resume_fn / _Coro_destroy_fn	ABI 规定（偏移 0 / 8）	✗
promise	位置：ABI 规定（偏移 16） 内容：你（promise_type 的定义）	✓（内容）
_Coro_resume_index	编译器（按挂起点个数编号）	✗
i、awaiter	编译器（活跃性分析的结果）	✗

想让协程"随身携带"什么，就往 promise_type 里加数据成员 —— 帧会跟着变大。

（名字借自 future/promise 模型：协程内部用它"承诺"产出结果，外部用 Generator（相当于 future）取结果。 它和 std::promise 没有任何关系。）

机制拆完了，下面拿这些知识把 counter() 和 main 的完整执行流从头到尾走一遍。

执行顺序：counter() 调用时【什么都没发生】

有了上面的拆解，现在可以把谁先跑、谁后跑完整地串一遍了。

最常见的误解是："先执行到 co_yield，然后才回到 g.next()"。反过来 —— counter() 这一行什么都不执行，第一次 g.next() 才让函数体开始跑。

给代码标上号

Generator<int> counter() {
    for (int i = 0; i < 3; ++i) {
        printf("    counter: 准备让出 %d\n", i);   // ①
        co_yield i;                               // ②
        printf("    counter: 恢复了\n");           // ③
    }
}

int main() {
    auto g = counter();                           // ④
    while (g.next())                              // ⑤
        printf("main 收到 %d\n", g.value());       // ⑥
}

先展开一下 g.next() 到底调了什么

表里的 h.resume() 不是终点，它背后还有两层：

g.next()                                   ← 你自己写的成员函数
 │
 ├─ h.resume()                             ← std::coroutine_handle<>::resume()
 │   │                                        标准库里就一行：
 │   └─ (*(void(**)(void*))h)(h);          ← 取【协程帧偏移 0】的函数指针，
 │        │                                   调用它，参数就是帧自己
 │        └─ counter.actor(帧)             ← ★ 编译器生成的函数，
 │             │                              【你写的 for 循环在这里面】
 │             └─ switch (帧->_Coro_resume_index)   ← 跳到上次停下的地方
 │                  case 2: goto resume.2;     （第 1 次 next：从 initial_suspend 之后）
 │                  case 4: goto resume.4;     （之后每次：从 co_yield 之后）
 │
 └─ return !h.done();                      ← 读 帧->_Coro_resume_fn 是不是 nullptr

你写的	谁提供	实际做什么
g.next()	你	转发给 h.resume()，再查 h.done()
h.resume()	标准库	取帧头部第 1 个函数指针并调用（coroutine_handle 就是帧指针）
counter.actor(帧)	编译器	★ 真正的函数体：开头一个 switch，后面是你写的循环
g.value() → h.promise().value	你 + 标准库	从帧里按编译期偏移取出 promise，读它的 value

★ counter() 和 counter.actor() 是两个不同的函数。 counter() 只做开场（造帧、调一次 actor 到 initial_suspend 挂起、返回）；你写的 for 循环在 counter.actor() 里。 这也是为什么上面那张表第 1 步里"函数体一行都没跑" —— 那次调用虽然也进了 counter.actor()，但只执行了 initial_suspend 那一句就挂起返回了。

具体怎么做到的，见上一节「编译器做的状态机变换」的第 1～3 步。

完整执行顺序

#	谁在跑	执行什么	输出
1	main	④ counter()：造帧 → 构造 promise → get_return_object → 调 counter.actor → initial_suspend 挂起（_Coro_resume_index=2）→ return ★ 函数体一行都没跑，i 还没初始化	（无）
2	main	⑤ g.next() → h.resume() → counter.actor(帧)
3	协程	从 initial suspend 恢复，进入函数体；i = 0
4	协程	① printf	counter: 准备让出 0
5	协程	② co_yield 0：value=0，_Coro_resume_index=4，return
6	main	⑤ h.done() 为假 → next() 返回 true
7	main	⑥ printf	main 收到 0
8	main	⑤ 第 2 次 g.next() → h.resume() → counter.actor(帧)
9	协程	counter.actor 开头的 switch 看到 _Coro_resume_index==4，跳到 resume.4，即 ② 之后
10	协程	③ printf	counter: 恢复了
11	协程	++i → 1，循环条件成立
12	协程	① printf	counter: 准备让出 1
13	协程	② co_yield 1 → return
14	main	⑥ printf	main 收到 1
…		（第 3 轮同理）
N	main	⑤ 第 4 次 g.next() → h.resume() → counter.actor(帧)
N+1	协程	③ printf	counter: 恢复了
N+2	协程	++i → 3，循环条件失败，函数体结束
N+3	协程	return_void() → _Coro_resume_fn = nullptr → final_suspend 挂起 → return
N+4	main	⑤ h.done() 为真 → next() 返回 false → while 退出
N+5	main	g 析构 → h.destroy() → 释放帧

输出（标注谁打的）

    counter: 准备让出 0     ← 协程（第 1 次 next 触发）
main 收到 0                ← main
    counter: 恢复了         ← 协程（第 2 次 next 触发）
    counter: 准备让出 1     ← 协程（同一次 next 里连着打的）
main 收到 1                ← main
    counter: 恢复了         ← 协程（第 3 次 next）
    counter: 准备让出 2     ← 协程
main 收到 2                ← main
    counter: 恢复了         ← 协程（第 4 次 next）★ 后面没有"main 收到"了

注意最后一行："恢复了"出现了 3 次，但最后那次后面没有 main 收到 —— 因为那一次 resume 跑完了整个循环，协程结束，next() 返回 false。

两个容易搞反的点

① 不是"协程先跑到 co_yield"，而是"main 先按了按钮"

auto g = counter();   // ★ 只是"装好了一台机器"，没有启动它

心智模型：g.next() 是摇把手，摇一下协程往前走一段。 不摇它就一动不动。

这正是 initial_suspend() 返回 suspend_always 的含义 —— "造完帧先别开始"。

② 恢复了 和 准备让出 N 是同一次 next() 里连着打的

看第 8~13 步：一次 h.resume() 里，协程连续做了「打印恢复了 → ++i → 打印准备让出」三件事，然后才挂起。

所以输出里它们总是成对出现，中间不会插进 main 收到。

如果把 initial_suspend 改成 suspend_never

std::suspend_never initial_suspend() noexcept { return {}; }   // 只改这一行

counter() 就会立刻开始执行，一路跑到第一个 co_yield 才停：

    counter: 准备让出 0     ← ★ 在 auto g = counter(); 这一行就打印了！
    counter: 恢复了         ← 第 1 次 next()：从第一个 co_yield 之后继续
    counter: 准备让出 1
main 收到 1                ← ★★ 第一个值 0 【丢了】！
    counter: 恢复了
    counter: 准备让出 2
main 收到 2
    counter: 恢复了

第一个值 0 被跳过了。 因为 next() 的实现是"先 resume 再取值"：

bool next() { h.resume(); return !h.done(); }
//            ^^^^^^^^^^ 惰性启动时，这次 resume 才产出第一个值
//                       立即启动时，第一个值早就产出了，这次 resume 把它冲掉了

所以 generator 必须用 suspend_always 做 initial_suspend —— 这不是风格问题，选错会静默丢数据。 反过来，"创建即开跑"的异步任务（Task）常用 suspend_never，因为它要的就是这个效果。

小结

counter() 返回时函数体一行没跑 → 第 1 次 next() 才启动函数体，跑到第一个 co_yield 停 → main 取值打印 → 第 2 次 next() 从 co_yield 之后继续…… 两者严格交替、永不并行，而且总是 main 主动驱动，协程自己不会动。谁在跑，取决于最近一次是 resume（进协程）还是 co_yield / return（回 main）。

到这里，C++20 无栈协程的机制和执行流就完整了。下面换个语言看同一件事。

---

示例 2：Rust —— 同一个状态机，被显式暴露成一个 trait

C++20 把状态机藏在编译器里（_Z7counterv.Frame 是个你写不出名字的编译器内部类型）。Rust 把它暴露成语言层面的 Future trait —— 所以能直接对照第 3 步那份 counter.actor 代码看。

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};

// ---- 一个"让出一次"的 awaitable ----
struct Yield(bool);
impl Yield { fn new() -> Self { Yield(false) } }

impl Future for Yield {
    type Output = ();
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        if self.0 {
            Poll::Ready(())              // 第二次 poll：完成
        } else {
            self.0 = true;
            cx.waker().wake_by_ref();    // 告诉执行器"我还能继续，再来 poll 我"
            Poll::Pending                // ★ 第一次 poll：挂起
        }
    }
}

// ★ 函数体里出现 .await，这个函数就是协程（编译器把它变成状态机）
async fn counter() {
    for i in 0..3 {
        println!("    counter: 准备让出 {}", i);
        Yield::new().await;              // ← 挂起点【只能写在 async fn 里】
        println!("    counter: 恢复了");
    }
}

// ---- 一个最小的执行器 ----
fn block_on<F: Future>(fut: F) -> F::Output {
    let waker = noop_waker();
    let mut cx = Context::from_waker(&waker);
    let mut fut = Box::pin(fut);
    loop {
        match fut.as_mut().poll(&mut cx) {          // ★ "恢复" = 调一次 poll
            Poll::Ready(v) => return v,
            Poll::Pending  => println!("main: 协程让出了，我可以干别的"),
        }
    }
}

fn noop_waker() -> Waker {
    fn no_op(_: *const ()) {}
    fn clone(_: *const ()) -> RawWaker { RawWaker::new(std::ptr::null(), &VTABLE) }
    static VTABLE: RawWakerVTable = RawWakerVTable::new(clone, no_op, no_op, no_op);
    unsafe { Waker::from_raw(RawWaker::new(std::ptr::null(), &VTABLE)) }
}

fn main() { block_on(counter()); }

rustc main.rs -o counter && ./counter
# 无需任何依赖（tokio 之类的执行器在这里被 block_on 替代了）
# Rust 1.85+ 可以直接用 Waker::noop() 省掉 noop_waker

    counter: 准备让出 0
main: 协程让出了，我可以干别的
    counter: 恢复了
    counter: 准备让出 1
main: 协程让出了，我可以干别的
    counter: 恢复了
    counter: 准备让出 2
main: 协程让出了，我可以干别的
    counter: 恢复了

Rust 生成的状态机长什么样

第 3 步的 counter.actor，在 Rust 里几乎是字面对应的 —— 只不过 Rust 用 enum 而不是 _Coro_resume_index 那样的整数字段：

// 编译器把 counter() 大致变成这样一个匿名类型：
enum CounterFuture {
    Start,                                   // state = 0
    Suspended1 { i: i32, y: Yield },         // state = 1
                 // ^^^^^^^^^^^^^^ ★ 跨挂起点存活的变量被【提升】进 variant
    Done,                                    // state = -1
}

impl Future for CounterFuture {
    type Output = ();
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<()> {
        loop {
            match &mut *self {
                CounterFuture::Start => { /* i = 0，跳到循环 */ }
                CounterFuture::Suspended1 { i, y } => {
                    match Pin::new(y).poll(cx) {
                        Poll::Pending  => return Poll::Pending,   // ★ 挂起 = return
                        Poll::Ready(_) => { /* 打印"恢复了"，i += 1 */ }
                    }
                }
                CounterFuture::Done => panic!("polled after completion"),
            }
        }
    }
}

Poll::Pending 就是 counter.actor 里挂起点上的那句 return。 两者是同一个东西。

两种无栈协程的逐项对照

概念	C++20	Rust
协程帧	_Z7counterv.Frame（编译器生成、用户写不出名字的类型）	impl Future 匿名类型（通常是 enum）
状态号	unsigned short _Coro_resume_index 字段	enum 的判别式
恢复函数	_Coro_resume_fn(frame*)	Future::poll(&mut self, cx)
跨挂起点存活的变量	提升进 frame 结构体	提升进 enum variant 的字段
挂起关键字	co_await / co_yield	.await
挂起动作	return	return Poll::Pending
恢复由谁发起	handle.resume()	执行器调 poll()
恢复后如何通知	由 awaiter 的 await_suspend 自行安排	Waker（语言内建的统一约定）
是否惰性	取决于 initial_suspend()	总是惰性：不 poll 就什么都不发生
帧在哪	总是"概念上堆分配"（HALO 可优化掉）	默认在调用者栈上，Box::pin / spawn 才上堆
自引用怎么办	帧地址固定在堆上，天生没问题	需要 Pin
执行器	标准库不提供，自己写或用 libs	标准库只定义 trait，执行器交给 tokio / async-std

最值得注意的一条：Pin 与"帧放在哪"是同一枚硬币的两面

协程帧里的局部变量可能互相引用：

async fn f() {
    let x = 5;
    let r = &x;          // ★ 帧内部的自引用
    something().await;   // 挂起 —— r 和 x 都要存进帧，且 r 指向帧内部
    println!("{}", r);
}

一旦这个帧被移动（Rust 里 move 是逐字节 memcpy），r 就变成野指针。两种语言给出了相反的解法：

C++20：帧【固定分配在堆上】，地址永不改变
       → 不需要 Pin 这种东西
       → 代价：必须堆分配（HALO 优化能消除，但不保证）

Rust ：帧【默认放在调用者的栈上】，零分配
       → 但因此必须禁止"已经开始 poll 的 Future 被移动"
       → 这就是 Pin 的全部意义

Rust 用 Pin 的心智负担，换来了"async 默认零堆分配"。 这也是原文"心智负担重…Rust 要理解 Pin、Send、生命周期与 Future 的交互"那句话的来源。

Rust 的函数染色更"硬"

fn helper() {
    // Yield::new().await;   // ✗ error: `await` is only allowed inside `async` functions
}

async fn outer() {
    helper();                // helper 是普通函数，无法挂起
}

而且 Rust 的染色比 C++ 更明显地体现在类型上：

fn      f() -> i32          { 1 }              // 返回 i32
async fn g() -> i32          { 1 }             // 实际返回 impl Future<Output = i32>
//                                                调用方式、返回类型、生命周期全变了

再叠加 Send / Sync 约束（跨 .await 持有非 Send 的东西，整个 Future 就不是 Send，tokio::spawn 就拒收），以及 trait 里的 async（async_trait 宏 / RPITIT 的限制）—— 这些共同构成了 Rust 异步生态"上手陡峭"的名声。

相比之下，本节开头 Go 的示例里 level1 / level2 连自己在协程里都不知道。

函数染色：为什么"只能在协程函数体内挂起"

void helper() {
    // co_yield 1;      // ✗ 编译错误
    //                  //   一旦写了 co_yield，helper 就变成协程，
    //                  //   而它的返回类型 void 没有 promise_type
}

Generator<int> outer() {
    helper();           // ← helper 是普通函数：编译器【没有】给它做状态机变换，
                        //    它的局部变量在真实的机器栈上，无处可存 → 不可能挂起
    co_yield 1;         // ✓ 只有 outer 自己能挂起
}

想让 helper 也能挂起，只有一条路 —— 把它也变成协程，并让 outer 层层 co_await：

Task<void> helper() {                  // 返回类型必须变（染色第 1 步）
    co_await something();
}
Task<void> outer() {
    co_await helper();                 // 调用方式必须变（染色第 2 步）
}
Task<void> caller() {
    co_await outer();                  // 一路向上传染……
}

这就是"函数染色"（function coloring）：一个底层函数改成异步，会强制它的所有直接和间接调用者全部改成异步，一路传染到 main。

对比示例 1 的 Go 版本 —— level1 / level2 完全不知道自己在协程里，签名和调用方式都不用动。这是有栈与无栈最本质、也最影响工程实践的差别。

---

概念澄清 1："挂起点"到底指什么

挂起点（suspension point）= 执行流可以在此暂停、把控制权交出去、之后从这里继续的位置。

它是个概念，在不同体系里由完全不同的东西充当 —— 很多人以为它就等于 co_yield，其实不是。

无栈协程（C++20）：四类挂起点

挂起点	什么时候	说明
initial_suspend()	协程刚创建时	隐含的，你没写任何关键字它就在
co_await expr	你写的地方	最主要的一种
co_yield expr	你写的地方	是 co_await 的语法糖
final_suspend()	协程体结束 / co_return 后	隐含的

co_yield 其实就是 co_await：

co_yield x;
// 编译器展开成 ↓
co_await promise.yield_value(x);

这就是为什么前面那个 Generator 必须提供 yield_value：

std::suspend_always yield_value(T v) { value = v; return {}; }
//  ^^^^^^^^^^^^^^^ 返回的这个 awaiter 决定了"要不要真挂起"

所以 co_yield 不是独立机制，co_await 才是根本。C++20 协程真正的核心关键字只有 co_await 一个。

关键细节：写了 co_await 也未必真挂起

准确的标准术语是"潜在挂起点"（potential suspension point）：

co_await expr;
// 展开为：
{
    auto&& awaiter = /* 经 await_transform / operator co_await 得到 */;

    if (!awaiter.await_ready()) {        // ★ 返回 true 就【不挂起】，直接往下走
        awaiter.await_suspend(handle);   // 真挂起：保存状态、return 回调用者
        // ……被 resume 后从这里继续……
    }
    return awaiter.await_resume();       // 取结果
}

co_await std::suspend_never{};   // 写了 co_await，但 await_ready() 为 true → 【不挂起】
co_await std::suspend_always{};  // await_ready() 为 false → 真挂起

实践中这非常有用 —— 异步 I/O 的 awaiter 通常写成"数据已就绪就不挂、没就绪才挂"，避免不必要的状态保存：

bool await_ready() { return buffer_has_data(); }   // 快路径：零开销

但注意：编译器在做状态机变换时按"会挂起"处理（跨越它的局部变量都要提升进协程帧），只是运行时才由 await_ready() 决定要不要真的 return。

有栈协程：挂起点根本不是语法概念

有栈协程没有任何关键字。挂起点就是"调用了那个会换栈的函数"的地方：

static void co_yield_(int v) {       // ← 这个名字是随便起的
    g_value = v;
    swapcontext(&g_co, &g_main);     // ★ 真正的挂起点在这一行
}

澄清一个容易混的地方：前面 C++ 示例里那个函数写成 co_yield_（末尾带下划线），是因为 co_yield 在 C++20 里已经是关键字，不能当函数名。它和 C++20 的 co_yield 毫无关系 —— 只是个普通函数，内部调 swapcontext。

Go 里连"调了个特殊函数"都看不出来：

类别	挂起点长什么样	底层
显式	runtime.Gosched()	gopark
隐式（最常见）	ch <- v、<-ch、conn.Read()、mu.Lock()、time.Sleep()、wg.Wait()、select	gopark
编译器插入	每个函数调用的序言（栈增长检查顺带做抢占检查）	看不见
抢占	任意一条指令（SIGURG 打断）	完全无法预知

最后一行最能说明问题：Go 1.14+ 的异步抢占让"挂起点"可以是任意指令，压根不存在"点"这个概念了。

对照总结

	无栈（C++20）	有栈（ucontext / Go）
挂起点是什么	语法结构：co_await / co_yield / co_return + 隐含的 initial/final	一次函数调用：内部执行了换栈
源码里看得见吗	必须看得见 —— 编译器要靠它切分状态机	可以完全看不见（hook / 隐式 / 抢占）
能在第三方函数里吗	✗ 那个函数必须自己也是协程	✓ 任何普通函数里都行
一定会挂起吗	不一定，由 await_ready() 决定	调到了就一定挂
数量	有限、编译期确定（对应 switch 的 case）	Go 抢占下可以是任意指令

第二行是函数染色的根源：无栈协程的挂起点必须在源码里显式可见，编译器才知道在哪切状态机、哪些变量要提升进堆上的帧；有栈协程不需要编译器参与，所以挂起点可以藏在库里。

---

概念澄清 2："挂起点可以在任意调用深度"是什么意思

调用深度就是函数调用栈的层数。以协程入口为第 0 层往下数：

co_entry()              ← 第 0 层（协程函数本身）
  └─ level1()           ← 第 1 层
      └─ level2()       ← 第 2 层
          └─ level3()   ← 第 3 层

这句话的意思是：你可以在上面任何一层里让出，不管已经嵌套调用了多少层函数。

对比无栈协程 —— 它只能在第 0 层挂起。

为什么有栈可以

因为每一层的栈帧都在协程自己的栈上，挂起时原封不动地留着：

g_stack（协程自己的 64 KB）
   +-----------------------+  ← 栈底（高地址）
   |  co_entry 的帧        |   第 0 层
   +-----------------------+
   |  level1 的帧          |   第 1 层
   +-----------------------+
   |  level2 的帧          |   第 2 层
   +-----------------------+
   |  level3 的帧          |   第 3 层  ← 在这里调用 co_yield_
   |    局部变量 i         |
   +-----------------------+
   |  co_yield_ 的帧       |
   +-----------------------+  ← rsp

  挂起：mov [rdi], rsp     把栈顶记下来
        mov rsp, [rsi]     换到别的栈
        ★ 上面这 5 个帧一个字节都没动，就躺在 g_stack 里

  恢复：mov rsp, [rsi]     rsp 指回来
        ret                5 个帧全在，从 co_yield_ 的下一条继续

"挂起"对有栈协程来说只是换了个 rsp，栈本身完全没被触碰。 所以埋多深都无所谓。

为什么无栈不可以

因为无栈协程的"挂起"就是 return，而 return 必然销毁自己的栈帧：

真实机器栈（和调用者共用）
   +-----------------------+
   |  main 的帧            |
   +-----------------------+
   |  counter.actor 的帧  |  ← 协程只有这一层
   +-----------------------+  ← rsp

  挂起：return;
        ★ counter.actor 的帧【直接被销毁】
        只有事先被编译器挪进【堆上协程帧】的变量活下来

现在假设协程里调用了一个普通函数：

   +-----------------------+
   |  counter.actor 的帧  |   第 0 层
   +-----------------------+
   |  helper 的帧          |   第 1 层 ← 局部变量在【真实机器栈】上
   +-----------------------+  ← rsp

这时想在 helper 里挂起，做不到：

• 挂起 = counter.actor 要 return，但 helper 的帧压在它上面，物理上没法"只销毁自己、保留 helper"
• 就算能，helper 的局部变量也无处可存 —— 编译器没给 helper 做状态机变换，没为它分配堆上的帧

根源就是"没有自己的栈"：能保存的只有编译器提前分析出来、挪进堆上那个结构体的变量。而 helper 不是协程，编译器压根没分析过它。

工程上意味着什么

这不是理论细节，是每天都会撞上的事。

场景 1：第三方库的回调里想挂起

// 第三方库，你没有源码
void library_parse(void (*cb)(Item));

void my_handler(Item i) {
    int n = read(fd, buf, 100);     // ★ 想在这里挂起
    process(i, n);
}

// 有栈：完全正常
co_run([]{ library_parse(my_handler); });
// 调用链 co_entry → library_parse → my_handler → read → yield，四层，无所谓

// 无栈：做不到
// my_handler 要挂起就得变成协程 → 返回类型变成 Task<void>
// 但 library_parse 要的是 void(*)(Item) —— 类型对不上，也改不了它的签名

这就是"易迁移、可接入遗留代码"的具体含义。

场景 2：递归

void walk(const std::string& dir) {
    for (auto& e : list(dir)) {
        if (is_dir(e)) walk(e);          // 递归到第 N 层
        else auto data = read_file(e);   // ★ 第 N 层挂起
    }
}

有栈：递归多深都行，帧全在自己栈上。 无栈：每层递归都要是独立的协程、有独立的堆上帧，且帧大小编译期得确定 —— 所以 Rust 里递归 async fn 必须 Box<dyn Future> 装箱。这正是后面"局限性"一节说的递归困难。

场景 3：STL 算法、C 库回调

std::sort(v.begin(), v.end(), [](auto& a, auto& b) {
    return fetch_key(a) < fetch_key(b);   // ★ fetch_key 里想访问远程/磁盘
});
qsort(...); bsearch(...); nftw(...);      // 同理

有栈可以；无栈里这些全都不行 —— std::sort 的比较器必须是普通可调用对象。

一个容易混的点：无栈的 co_await 链不是同一回事

无栈协程也能做出"多层"的效果：

Task<void> deepest() { co_await io(); }
Task<void> middle()  { co_await deepest(); }
Task<void> outer()   { co_await middle(); }

但机制完全不同：

无栈的"链式"：
  堆上：[outer 帧] ──awaiting──> [middle 帧] ──awaiting──> [deepest 帧]
        三个【独立分配】的对象，靠 continuation 串起来
        恢复时从 deepest 开始，逐层往回 resume

有栈的"一整块"：
  g_stack: [帧][帧][帧][帧]
        一次 mov rsp 全部搞定，零分配

	有栈	无栈链式
分配	一次（创建协程时分配整个栈）	每层一次堆分配
挂起/恢复	一次换 rsp	逐层传递（对称转移可优化成尾调用）
中间层能否是别人的代码	可以（普通函数）	不可以（每层都必须是协程）

最后一行才是本质区别 —— 无栈的"深度"要求从上到下每一层都染色，而有栈的中间层可以是任何普通函数，包括你改不了的第三方代码。

一句话

有栈协程挂起时只是换了个 rsp，整条调用链的所有栈帧原封不动躺在自己的栈上，所以埋多少层深都能挂起；无栈协程挂起就是 return，return 必然销毁自己以及压在自己之上的所有帧，能活下来的只有编译器事先挪进堆上协程帧的那几个变量 —— 而编译器只对"写了 co_await 的那个函数"做了这个准备，所以它只能在第 0 层挂起。

---

有栈协程为什么能工作 —— 三个前提条件

一个常见的直觉是："因为 CPU 允许设置下一条指令的地址"。这说对了一半：改 PC 是必要条件，但不是最关键的那个。

而且严格说，x86 上根本不能直接写 RIP：

mov rip, rax        ; ✗ 非法指令，汇编器都不认

RIP 只能被控制转移指令间接修改（jmp / call / ret / iret）。"设置下一条指令地址"这个能力，在 x86 上从来不是直接给你的。

真正的关键：栈是普通内存，rsp 是普通寄存器

回头看前面那段 co_swap，把每条指令的角色标出来：

co_swap:
    push rbp                ; ③ 保存其余执行状态 —— 存到【当前这个栈】上
    push rbx
    push r12
    push r13
    push r14
    push r15

    mov  [rdi], rsp         ;    记住当前栈顶
    mov  rsp, [rsi]         ; ★★★ 换栈 —— 全部魔法就在这一条
                            ;     rsp 只是个通用寄存器，可以随便赋值

    pop  r15                ;    从【新栈】上恢复状态
    pop  r14
    pop  r13
    pop  r12
    pop  rbx
    pop  rbp

    ret                     ;    从【新栈】上弹出返回地址

注意最后那条 ret 完全是标准动作，它做的事和任何一个普通函数返回一模一样：pop rip。

它之所以能跳到另一个执行流的挂起点，不是因为我们"设置"了 PC，而是因为它读的是另一个栈。

把 mov rsp, [rsi] 这一条删掉，整段代码就退化成一个什么都不干的普通函数。

所以因果链是反的：

rsp 是可任意赋值的普通寄存器
        ↓
"换栈"成为可能
        ↓
ret 从新栈上弹出了另一个执行流的返回地址
        ↓
PC 自然而然地跳过去了       ← 这是【结果】，不是原因

改 PC 是任何有函数调用的架构都天然具备的能力（ret 本身就是在改 PC），它不构成门槛。真正的门槛是：你能不能自己搞一块内存当栈用，然后让 CPU 认它。

完整的必要条件清单

#	条件	是否稀缺
①	栈是可寻址的普通内存（能 malloc 一块当栈）	稀缺
②	栈指针是可任意赋值的通用寄存器	稀缺
③	能保存/恢复其余执行状态（callee-saved 寄存器）	push/pop 就够，不稀缺
④	能把控制流转到保存的地址	ret 就够，不稀缺
⑤	没有硬件强制的、程序不可控的影子栈	现代新增的约束（见反证 2）

①② 才是核心，③④ 任何架构都有。

反证 1：硬件栈的架构做不了

PIC16 单片机的调用栈是硬件实现的：固定 8 层深度，程序根本无法寻址，CALL/RETURN 直接操作硬件栈指针。

在这种架构上，即使你能改 PC，也永远做不出有栈协程 —— 因为你没有"另一个栈"可以换。这直接证明关键在 ①②，不在改 PC。

（x87 浮点栈也是这种"程序不可寻址的硬件栈"，只不过它不用于控制流。）

反证 2：Shadow Stack（Intel CET）—— 当代的真实麻烦

这是最好的现代例证。开启 CET 后，CPU 维护一份只有 call/ret 能修改的影子栈：

call  → 同时把返回地址压进普通栈【和】影子栈
ret   → 从两个栈各弹一个，【比对】，不一致就触发 #CP 异常

于是 co_swap 直接崩：换了普通栈，影子栈还是老的，ret 时两边对不上。

注意崩溃的原因恰恰是"换栈"，而不是"改 PC"。 解法是给每个协程再分配一个独立的影子栈，切换时一起换：

rdsspq   rax                 ; 读当前影子栈指针
mov      [rdi + SSP], rax    ; 保存
mov      rax, [rsi + SSP]
rstorssp [rax]               ; 恢复目标协程的影子栈

boost.context、glibc 的 makecontext 都为此做了专门适配。这件事再次说明：有栈协程的命门是"栈能不能被换"。

反证 3：setjmp / longjmp 不够

它俩能同时改 PC 和 rsp，但只能往"栈的更早处"跳 —— longjmp 假设目标 setjmp 点的栈帧仍然有效，跳过去等于把中间的帧全部丢弃。

它跳不到一个平行的、独立的栈上。所以标准的 setjmp/longjmp 做不了协程（硬要做得配合 sigaltstack + 信号处理程序骗出一个新栈，是个著名的 hack）。

这又一次说明：能改 PC ≠ 能做协程。

反过来看无栈协程 —— 它一个条件都不需要

挂起 = return;              // 普通函数返回
恢复 = counter.actor(f);   // 普通函数调用

无栈协程从头到尾不碰 rsp，也不"设置 PC" —— 它的状态保存在一个堆上的普通结构体里，跳转靠的是一个普通的 switch。

这就直接解释了后面对比表里的那两行：

	有栈	无栈
依赖	必须能换栈 → 每个架构一份汇编	只需要普通函数调用 → 纯语言层面，天然可移植
能跑在哪	需要 ①②⑤ 都满足的真实 CPU	任何地方，包括 JS 引擎、WASM、没有栈概念的虚拟机

WASM 是个绝好的例子：它的调用栈对程序不可见（和 PIC16 一样），所以 WASM 上无法实现传统的有栈协程 —— 但 C++20 无栈协程编译成 WASM 毫无问题。（这也正是 WASM 后来专门增加 stack-switching 提案的原因。）

小结

不是"CPU 允许设置下一条指令地址"，而是"栈只是一段普通内存，栈指针只是一个普通寄存器"。

改 PC 这件事 ret 天生就在做，任何架构都有；稀缺的是你能自己分配一块内存当栈、并让 rsp 指过去。有了这个，ret 会自动从新栈上弹出正确的返回地址 —— PC 的跳转是换栈的结果，不是原因。

反过来，凡是"栈对程序不可寻址"的环境（PIC16 单片机、WASM）就做不了有栈协程；而 Intel CET 的影子栈之所以会让 co_swap 崩溃，也恰恰是因为它把"栈"变成了一部分程序不可随意换的硬件结构。

---

二、逐项对比

维度	有栈协程	无栈协程
挂起位置	任意调用深度	仅协程函数体内
内存开销	每协程一个栈，几 KB 起（Go 2KB 起可增长；固定栈通常 64KB~1MB）	仅协程帧，几十到几百字节，编译器精确计算
切换开销	保存/恢复寄存器 + 换栈，~10-20ns（无系统调用）	一次函数调用 + switch，几 ns
函数染色	无，普通函数与协程函数没有区别	有，async 会向上传染整条调用链
编译器支持	不需要，库级实现即可	必须编译器/语言级支持
调试	栈完整，gdb 能看到调用链（需支持切栈）	栈是"扁平"的，回溯只能看到 resume 点，需要特殊调试工具
与现有代码集成	简单，阻塞式代码几乎不改（hook 系统调用即可）	困难，需要整个生态 async 化
生态兼容	可以直接调用 C 库、第三方阻塞函数	调用阻塞函数会卡住整个调度器
可移植性	依赖架构相关汇编	纯语言层面，天然可移植
迭代器/生成器	可以做但重	天然适合（generator 就是最简单的无栈协程）

---

三、优缺点详解

有栈协程

优点

• 透明性：写同步代码的方式写异步逻辑，read() 里面自动挂起，上层代码完全无感知（Go 的最大卖点）
• 无染色：任何函数都能挂起，库作者不用为异步单独写一套 API
• 易迁移：老的阻塞式代码用 hook 方式（如 libco hook read/write/connect）就能协程化
• 调试友好：完整调用栈

缺点

• 内存：栈大小是永恒的两难
  • 固定大栈：百万协程直接爆内存
  • 固定小栈：深递归或大局部数组会栈溢出，且很难提前发现
  • 可增长栈（Go 方案）：需要运行时能扫描栈、重写指针，这基本要求带 GC 的语言；C/C++ 做不到（指针可能藏在任何地方）
  • 分段栈（segmented stack）：Go 早期方案，有 "hot split" 问题（在段边界反复来回跨越，性能剧烈抖动），已被弃用
• 切换成本略高：要保存一整套寄存器，还可能有 cache/TLB 失效（不同栈的内存页不同）
• 与 TLS、C++ 异常、栈保护机制（如 Windows SEH、stack canary、ASAN）的交互容易踩坑
• 需要为每个 CPU 架构写汇编

无栈协程

优点

• 极致轻量：协程帧大小编译期确定，只包含跨挂起点存活的变量，百万级并发毫无压力
• 切换快：本质上是普通函数调用
• 零依赖架构：不需要汇编
• 可优化：编译器可以把协程帧直接放在栈上（HALO 优化）、内联、消除堆分配
• 表达力强：generator、async、异步流可以统一用同一套机制表达

缺点

• 函数染色（最被诟病）：async 传染性，整个生态分裂成同步/异步两套 API（Python 的 requests vs aiohttp，Rust 的 std::fs vs tokio::fs）。有栈协程的"一套 API"也不是白来的，见下面「同步/异步两套 API：有栈协程只是把代价挪到了运行时」
• 心智负担重：C++20 协程要理解 promise_type、awaiter、handle、生命周期，出了名的复杂；Rust 要理解 Pin、Send、生命周期与 Future 的交互
• 意外阻塞：不小心在协程里调了个阻塞函数（比如同步 DNS 解析、std::mutex），整个线程上的所有协程都停，且很难排查
• 调试困难：调用栈是碎片化的，异步回溯需要特殊支持（如 Rust 的 async-backtrace）
• 递归困难：递归协程需要装箱（Box<dyn Future>），因为帧大小需要编译期确定
• 对象生命周期陷阱：C++ 中引用参数在挂起后可能悬垂，lambda 捕获与协程帧的关系是经典 bug 源（有栈协程通过 void* 传参时同样会悬垂，只是无栈协程的写法和普通函数一样，坑更隐蔽）

---

四、局限性与常见误区

有栈协程的局限

1. 不适合无 GC 的语言做大规模并发：C/C++ 只能用固定栈，要在"安全"和"内存"之间选
2. 栈上的指针问题：如果把协程栈上的地址传出去（比如注册到 epoll 的 buffer），协程结束或栈被迁移后就是野指针
3. CPU 密集型任务无法被抢占（除非像 Go 一样做基于信号的异步抢占，实现极复杂）
4. 不能跨线程随意迁移：如果协程里用了线程局部存储或持有了线程绑定的锁，迁移到另一个线程执行就出错

无栈协程的局限

1. "只能在顶层挂起"是本质约束，不是实现问题——因为没有栈可以保存中间调用帧
2. 协程帧通常需要堆分配，高频创建小协程时分配开销可能反超收益
3. 无法协程化第三方阻塞库，只能用线程池包一层（这又回到了线程模型）
4. 协程帧可能"不小"：如果一个大数组跨越了挂起点，它就会进帧；有时编译器分析保守，帧比预期大很多

常见误区

• "无栈协程一定比有栈快"：单次切换更快，但真实场景的瓶颈通常在 I/O 和调度器，两者差距往往可以忽略
• "有栈协程内存一定大"：Go 通过可增长栈把起始栈压到 2KB，并不比一个复杂的 Rust Future 大多少
• "协程 = 并发"：协程只是控制流机制，真正的并发需要调度器 + 事件循环（epoll/io_uring）配合

同步/异步两套 API：有栈协程只是把代价挪到了运行时

两种协程面对的是同一个问题：一个普通的阻塞调用（read()、sleep()、std::mutex::lock()）会把整个线程卡住，协程机制对它完全不起作用。 区别只在解决手段落在哪一层：

	无栈（C++20 / Rust / Python asyncio）	有栈（Go / libco / Boost.Fiber）
协程里直接调 read()	阻塞整个线程	同样阻塞整个线程
怎么解决	改代码：换成 co_await async_read()，函数签名跟着变，一路向上传染	改运行时：hook read / connect / sleep（libco 用 dlsym 替换 libc 符号；Go 干脆不走 libc，自己实现 netpoller）
谁付代价	应用作者：调用链重写，同步/异步两套 API 并存	运行时作者：把每一个可能阻塞的系统调用都包一遍
漏掉一处会怎样	编译错误，逃不掉	静默阻塞：第三方 .so 里直接 syscall、Go 里的 cgo 调用、libco 没 hook 的 getaddrinfo……

所以有栈协程的"一套 API"是运行时替你 hook 出来的：Go 能做得干净，是因为它有自己的运行时和标准库，不依赖 libc；libco 这类寄生在 C 库上的方案，覆盖面永远不完整，而且漏网之鱼没有任何编译期提示。无栈协程把这个代价摊到每个调用点上，换来的是"漏了就编不过"。

为什么 C++ 标准选了无栈

"不需要平台支持"是原因之一，但不是主要原因。C++20 协程的提案（N4402、P0057，以及与有栈方案 P0099 的对比）给出的理由：

理由	有栈方案做不到的地方
可移植	换 rsp、分配栈、guard page、CET shadow stack、Windows 栈边界、ASan/TSan 的栈追踪，全是平台相关的；标准没法在语言层面规定"把栈指针换掉"。无栈是纯源码变换，能编 C++ 的地方都能做
内存和数量级	有栈每个协程至少几 KB 到几十 KB 栈，还得留余量防溢出；无栈帧只装跨挂起点的变量，本文的例子是 40 字节
可优化	无栈帧是编译器看得见的普通结构体：能把帧放到调用者栈上（HALO）、内联 await_*、常量折叠 await_ready。协程栈对编译器是黑盒
零开销原则	不用协程的代码一分钱不付；用了只付帧大小 + 一次间接调用。有栈方案需要运行时（调度器、栈池）
兼容现有 ABI 和工具	无栈生成的就是普通函数和普通结构体，调用约定、异常展开、调试器都不用改；栈切换会让 unwinder 和调试器看到断裂的调用链

代价就是上面那条函数染色，以及 promise_type / awaiter 那套复杂接口。委员会的判断是：接口复杂性可以由库包起来，平台相关性包不住。有栈方案后来以 Boost.Context / Boost.Coroutine2 的形式留在库层面，没有进标准。

---

五、混合与演化方向

• Java 虚拟线程（Loom）：有栈，但栈不是连续内存，挂起时把栈帧拷贝到堆上，恢复时再拷回来（"栈拷贝"方案），兼顾了内存和透明性，代价是挂起/恢复时的拷贝成本
• Kotlin：语言层无栈，但通过 suspend 关键字 + 编译器 CPS 变换，让写法接近有栈的透明度
• C++ 生态两条路并行：boost.context/libco 走有栈，C++20 走无栈，各自有一批拥趸
• io_uring + 无栈协程 是目前 Linux 高性能网络编程的热门组合

---

六、怎么选

场景	建议
需要接入大量遗留阻塞代码 / C 库	有栈
百万级连接、内存敏感	无栈（或 Go 式可增长栈）
语言有 GC 且能做栈扫描	有栈可增长栈是最舒服的方案
C/C++ 且追求极致轻量	无栈（C++20）
主要写业务逻辑、希望代码直观	有栈
需要 generator / 异步流语义	无栈天然合适
需要嵌入式 / 跨架构可移植	无栈

一句话总结：有栈协程用内存换透明性，无栈协程用生态复杂度换轻量。两者没有绝对优劣，Go 选前者、Rust 选后者都是各自语言约束（有无 GC、是否允许运行时）下的最优解。

---

七、附录：一个完整的精简版有栈协程库

前面讲清楚了原理，这里给一个可编译运行的完整实现，约 700 行。

它回答的问题是：如果自己要做一个有栈协程库，最少该提供哪些接口？

必须有 ───── co_go(fn)              创建协程
            co_sched_run()          事件循环
            co_yield_to_sched()     主动让出
            co_sleep(ms)            定时让出（★ 不是 ::sleep）
            co_in_coroutine()       判断当前上下文

强烈建议 ─── co_mutex               否则用户会用 std::mutex 然后炸
            co_chan<T>              协程间通信的主力
            co_wait_group           等待一组协程
            co_read/write/accept    带超时的 I/O

生产必须 ─── co_stack_used()        用来把栈调小
            co_foreach()            排查"谁卡住了"
            co_stats()              监控指标
            guard page              栈溢出立刻崩，而不是静默踩坏别人

设计原则：把 resume / yield 当内部原语，对外只暴露 go(fn) + channel + 同步原语 + 带超时的 I/O。 Go 之所以好用，正是因为它把前两个彻底藏了起来。

功能范围：单线程调度器（每线程一个）+ 定时器 + epoll + channel + mutex + waitgroup + guard page + 栈用量统计。 平台：Linux x86-64。

co_switch.S      20 行   上下文切换汇编
coroutine.h     190 行   公开接口
coroutine.cpp   400 行   调度器 / poller / I/O / 同步原语
example.cpp      90 行   演示

---

1. co_switch.S —— 上下文切换

# ============================================================
#  void co_switch(void** from_sp, void* to_sp);
#      rdi = &from->sp_   （把当前 rsp 存到这里）
#      rsi = to->sp_      （切到这个栈顶）
#  只需保存 callee-saved 寄存器：co_switch 本身是一次普通函数调用，
#  caller-saved 寄存器按 ABI 本来就允许被破坏。
# ============================================================
    .text
    .globl  co_switch
    .type   co_switch, @function
    .align  16
co_switch:
    pushq   %rbp
    pushq   %rbx
    pushq   %r12
    pushq   %r13
    pushq   %r14
    pushq   %r15

    movq    %rsp, (%rdi)        # ① 保存当前栈顶
    movq    %rsi, %rsp          # ② ★ 换栈 —— 全部魔法就这一条

    popq    %r15
    popq    %r14
    popq    %r13
    popq    %r12
    popq    %rbx
    popq    %rbp
    ret                         # ③ 从【新栈】弹出返回地址并跳过去

    .size   co_switch, .-co_switch
    .section .note.GNU-stack,"",@progbits

---

2. coroutine.h —— 公开接口

#pragma once
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <deque>
#include <exception>
#include <functional>
#include <string>
#include <sys/types.h>

struct sockaddr;

namespace co {

using Ms = std::chrono::milliseconds;

enum class State { Ready, Running, Suspended, Done };

// ============ 协程对象（用户一般不直接构造，用 co_go）============
class Coroutine {
public:
    std::uint64_t      id()         const noexcept { return id_; }
    const std::string& name()       const noexcept { return name_; }
    void               set_name(std::string s)     { name_ = std::move(s); }
    State              state()      const noexcept { return state_; }
    bool               done()       const noexcept { return state_ == State::Done; }
    std::size_t        stack_size() const noexcept { return size_; }

    Coroutine(const Coroutine&)            = delete;   // 栈不可拷贝
    Coroutine& operator=(const Coroutine&) = delete;

    // ---- 以下为内部字段，用户请勿直接访问 ----
    void*                 sp_      = nullptr;   // 保存的栈顶
    char*                 mmap_    = nullptr;   // mmap 起始（含 guard page）
    char*                 stack_   = nullptr;   // 可用栈起始
    std::size_t           size_    = 0;
    std::function<void()> fn_;
    State                 state_   = State::Ready;
    std::exception_ptr    exc_;
    int                   saved_errno_ = 0;
    std::uint64_t         id_      = 0;
    std::uint64_t         epoch_   = 0;         // 用于作废过期的定时器
    bool                  timed_out_ = false;
    bool                  guard_   = true;
    bool                  pooled_  = true;
    std::string           name_;

    Coroutine() = default;
    ~Coroutine() = default;      // 栈由调度器负责回收（要还给栈池）
};

// ============ 内部原语（同步原语要用）============
namespace detail {
Coroutine* current() noexcept;
void       block();                 // 挂起当前协程，【不】放入就绪队列
void       wake(Coroutine* co);     // 把协程放回就绪队列
}

// ============ 1. 核心 ============
struct StackConfig {
    std::size_t size       = 128 * 1024;
    bool        guard_page = true;
    bool        from_pool  = true;
};

void co_go(std::function<void()> fn);
void co_go(const StackConfig& cfg, std::function<void()> fn);

void co_sched_run();                // 事件循环，跑到没有可运行协程为止
void co_sched_stop();
void co_yield_to_sched();           // 主动让出（≈ runtime.Gosched）
void co_sleep(Ms d);                // 定时让出（★ 不是 ::sleep）
bool co_in_coroutine() noexcept;

// ============ 2. 同步原语（★ 绝不能用 std::mutex）============
class co_mutex {
public:
    void lock();
    bool try_lock();
    void unlock();
private:
    bool                    locked_ = false;
    std::deque<Coroutine*>  waiters_;
};

class co_lock_guard {
public:
    explicit co_lock_guard(co_mutex& m) : m_(m) { m_.lock(); }
    ~co_lock_guard() { m_.unlock(); }
    co_lock_guard(const co_lock_guard&) = delete;
private:
    co_mutex& m_;
};

class co_wait_group {
public:
    void add(int n);
    void done();
    void wait();
private:
    int                     count_ = 0;
    std::deque<Coroutine*>  waiters_;
};

template <class T>
class co_chan {
public:
    explicit co_chan(std::size_t cap = 0) : cap_(cap) {}

    void send(T v) {
        // 缓冲满（cap_==0 时永远"满"）且没有接收者在等 → 挂起
        while (!closed_ && buf_.size() >= cap_ && recvers_.empty()) {
            senders_.push_back(detail::current());
            detail::block();
        }
        if (closed_) return;                    // 已关闭，丢弃
        buf_.push_back(std::move(v));
        wake_one(recvers_);
    }

    // 返回 false 表示 channel 已关闭且无数据
    bool recv(T& out) {
        while (buf_.empty()) {
            if (closed_) return false;
            wake_one(senders_);                 // ★ 先唤醒可能在等接收者的发送方
            recvers_.push_back(detail::current());
            detail::block();
        }
        out = std::move(buf_.front());
        buf_.pop_front();
        wake_one(senders_);
        return true;
    }

    void close() {
        closed_ = true;
        while (!senders_.empty()) { detail::wake(senders_.front()); senders_.pop_front(); }
        while (!recvers_.empty()) { detail::wake(recvers_.front()); recvers_.pop_front(); }
    }

    std::size_t size() const noexcept { return buf_.size(); }

private:
    static void wake_one(std::deque<Coroutine*>& q) {
        if (!q.empty()) { detail::wake(q.front()); q.pop_front(); }
    }
    std::size_t             cap_;
    std::deque<T>           buf_;
    std::deque<Coroutine*>  senders_, recvers_;
    bool                    closed_ = false;
};

// ============ 3. I/O（全部带超时；timeout_ms < 0 表示永不超时）============
int     co_set_nonblock(int fd);
ssize_t co_read (int fd, void* buf, std::size_t n, int timeout_ms = -1);
ssize_t co_write(int fd, const void* buf, std::size_t n, int timeout_ms = -1);
int     co_accept(int fd, sockaddr* addr, socklen_t* len, int timeout_ms = -1);
int     co_connect(int fd, const sockaddr* addr, socklen_t len, int timeout_ms = -1);

// ============ 4. 可观测性（生产环境必须有）============
struct Stats {
    std::uint64_t alive         = 0;   // 存活协程数
    std::uint64_t total_created = 0;
    std::uint64_t switches      = 0;   // 切换次数
    std::uint64_t stack_bytes   = 0;   // 栈总占用
};
Stats       co_stats();
void        co_foreach(const std::function<void(const Coroutine&)>& fn);
std::size_t co_stack_used(const Coroutine& c);   // 实测用了多少字节

} // namespace co

---

3. coroutine.cpp —— 调度器 / poller / I/O / 同步原语

#include "coroutine.h"

#include <sys/epoll.h>
#include <sys/mman.h>
#include <sys/socket.h>
#include <unistd.h>
#include <fcntl.h>
#include <cerrno>
#include <cstring>
#include <cstdio>
#include <cassert>
#include <map>
#include <unordered_map>
#include <unordered_set>
#include <vector>

extern "C" void co_switch(void** from_sp, void* to_sp);   // 汇编实现

namespace co {
namespace {

using Clock     = std::chrono::steady_clock;
using TimePoint = Clock::time_point;

constexpr std::size_t kPage = 4096;

// ==================================================================
//  栈分配：mmap + guard page + 0xCD 填充 + 栈池复用
// ==================================================================
char* alloc_mmap(std::size_t size, bool guard) {
    std::size_t total = size + (guard ? kPage : 0);
    void* p = ::mmap(nullptr, total, PROT_READ | PROT_WRITE,
                     MAP_PRIVATE | MAP_ANONYMOUS | MAP_STACK, -1, 0);
    if (p == MAP_FAILED) return nullptr;
    // ★ 最低的一页设为不可访问：栈溢出会【立刻】SIGSEGV，
    //   而不是静默踩坏相邻协程的栈（后者是查三天都查不出来的 bug）
    if (guard) ::mprotect(p, kPage, PROT_NONE);
    return static_cast<char*>(p);
}

void free_mmap(char* p, std::size_t size, bool guard) {
    ::munmap(p, size + (guard ? kPage : 0));
}

class StackPool {
public:
    char* acquire(std::size_t size, bool guard) {
        auto& v = free_[key(size, guard)];
        if (!v.empty()) { char* p = v.back(); v.pop_back(); return p; }
        return alloc_mmap(size, guard);
    }
    void release(char* p, std::size_t size, bool guard) {
        auto& v = free_[key(size, guard)];
        if (v.size() >= kMaxCached) { free_mmap(p, size, guard); return; }
        v.push_back(p);
    }
    ~StackPool() {
        for (auto& [k, v] : free_)
            for (char* p : v) free_mmap(p, k >> 1, k & 1);
    }
private:
    static std::size_t key(std::size_t size, bool guard) { return (size << 1) | guard; }
    static constexpr std::size_t kMaxCached = 64;
    std::unordered_map<std::size_t, std::vector<char*>> free_;
};

// ==================================================================
//  epoll 封装
// ==================================================================
class Poller {
public:
    Poller() { epfd_ = ::epoll_create1(EPOLL_CLOEXEC); }
    ~Poller() { if (epfd_ >= 0) ::close(epfd_); }

    void add(int fd, std::uint32_t ev, Coroutine* c) {
        bool existed = fds_.count(fd) != 0;
        Ctx& x = fds_[fd];
        if (ev & EPOLLIN)  x.r = c;
        if (ev & EPOLLOUT) x.w = c;
        epoll_event e{};
        e.events  = (x.r ? EPOLLIN : 0u) | (x.w ? EPOLLOUT : 0u);
        e.data.fd = fd;
        ::epoll_ctl(epfd_, existed ? EPOLL_CTL_MOD : EPOLL_CTL_ADD, fd, &e);
    }

    void del(int fd, std::uint32_t ev) {
        auto it = fds_.find(fd);
        if (it == fds_.end()) return;
        Ctx& x = it->second;
        if (ev & EPOLLIN)  x.r = nullptr;
        if (ev & EPOLLOUT) x.w = nullptr;
        if (!x.r && !x.w) {
            ::epoll_ctl(epfd_, EPOLL_CTL_DEL, fd, nullptr);
            fds_.erase(it);
        } else {
            epoll_event e{};
            e.events  = (x.r ? EPOLLIN : 0u) | (x.w ? EPOLLOUT : 0u);
            e.data.fd = fd;
            ::epoll_ctl(epfd_, EPOLL_CTL_MOD, fd, &e);
        }
    }

    int poll(int timeout_ms);                       // 定义在下面（要用 detail::wake）
    std::size_t pending() const noexcept { return fds_.size(); }

private:
    struct Ctx { Coroutine* r = nullptr; Coroutine* w = nullptr; };
    int epfd_ = -1;
    std::unordered_map<int, Ctx> fds_;
};

// ==================================================================
//  调度器（每线程一个）
// ==================================================================
struct Timer {
    Coroutine*    co;
    std::uint64_t epoch;        // ★ 注册时的 epoch，用来作废已被 I/O 唤醒的定时器
};

struct Scheduler {
    Coroutine                       main_;      // 代表调度器自身（跑在线程原生栈上）
    Coroutine*                      cur_ = &main_;
    std::deque<Coroutine*>          ready_;
    std::multimap<TimePoint, Timer> timers_;
    std::unordered_set<Coroutine*>  all_;
    StackPool                       pool_;
    Poller                          poller_;
    Stats                           stats_;
    std::uint64_t                   next_id_ = 0;
    bool                            running_ = false;

    void resume(Coroutine* c);
    void destroy(Coroutine* c);
    void wake(Coroutine* c, bool timed_out);
    void add_timer(TimePoint tp, Coroutine* c) { timers_.emplace(tp, Timer{c, c->epoch_}); }
    void fire_timers();
    int  next_timeout_ms() const;
};

Scheduler& sched() {
    static thread_local Scheduler s;
    return s;
}

// ---------------- 协程入口蹦床 ----------------
void trampoline() {
    Scheduler& S = sched();
    Coroutine* self = S.cur_;
    try {
        self->fn_();
    } catch (...) {
        self->exc_ = std::current_exception();      // ★ 异常不能跨栈传播，先存下来
    }
    self->_Coro_resume_index_ = State::Done;
    S.stats_.switches++;
    co_switch(&self->sp_, S.main_.sp_);             // 切回调度器，【永不返回】
    __builtin_unreachable();
}

// ---------------- 创建协程 ----------------
Coroutine* make_co(const StackConfig& cfg, std::function<void()> fn) {
    Scheduler& S = sched();
    std::size_t size = (cfg.size + 15) & ~std::size_t(15);

    char* m = cfg.from_pool ? S.pool_.acquire(size, cfg.guard_page)
                            : alloc_mmap(size, cfg.guard_page);
    if (!m) return nullptr;

    auto* c    = new Coroutine();
    c->mmap_   = m;
    c->stack_  = m + (cfg.guard_page ? kPage : 0);
    c->size_   = size;
    c->guard_  = cfg.guard_page;
    c->pooled_ = cfg.from_pool;
    c->fn_     = std::move(fn);
    c->id_     = ++S.next_id_;

    std::memset(c->stack_, 0xCD, size);             // ★ 供 co_stack_used 反推用量

    // ---- 在栈顶布置"假的初始帧"，让 co_switch 的 ret 跳进 trampoline ----
    //   布局（高地址 → 低地址）：
    //     top-8   : 对齐占位
    //     top-16  : trampoline 地址   ← ret 从这里弹
    //     top-64  : 6 个 callee-saved 寄存器槽   ← sp_ 指向这里
    //   这样 ret 之后 rsp == top-8，满足 (rsp+8) % 16 == 0 的 ABI 要求
    char* top = c->stack_ + size;
    top = reinterpret_cast<char*>(reinterpret_cast<std::uintptr_t>(top) & ~std::uintptr_t(15));
    void** sp = reinterpret_cast<void**>(top);
    *--sp = nullptr;
    *--sp = reinterpret_cast<void*>(&trampoline);
    sp -= 6;
    for (int i = 0; i < 6; ++i) sp[i] = nullptr;
    c->sp_ = sp;

    S.all_.insert(c);
    S.stats_.total_created++;
    S.stats_.alive++;
    S.stats_.stack_bytes += size;
    return c;
}

} // namespace (匿名)

// ==================================================================
//  Scheduler 成员实现
// ==================================================================
namespace {

void Scheduler::resume(Coroutine* c) {
    cur_ = c;
    c->state_ = State::Running;

    int host_errno   = errno;           // ★ errno 是 thread_local 的，
    errno            = c->saved_errno_; //   两个协程共用一个线程，必须各存各的
    stats_.switches++;

    co_switch(&main_.sp_, c->sp_);      // ★★ 切进协程

    c->saved_errno_ = errno;
    errno           = host_errno;
    cur_            = &main_;

    if (c->state_ == State::Done) {
        if (c->exc_) {
            try { std::rethrow_exception(c->exc_); }
            catch (const std::exception& e) {
                std::fprintf(stderr, "[co] coroutine #%llu 抛出未捕获异常: %s\n",
                             (unsigned long long)c->id_, e.what());
            } catch (...) {
                std::fprintf(stderr, "[co] coroutine #%llu 抛出未知异常\n",
                             (unsigned long long)c->id_);
            }
        }
        destroy(c);
    }
}

void Scheduler::destroy(Coroutine* c) {
    stats_.alive--;
    stats_.stack_bytes -= c->size_;
    all_.erase(c);
    if (c->pooled_) pool_.release(c->mmap_, c->size_, c->guard_);
    else            free_mmap(c->mmap_, c->size_, c->guard_);
    delete c;
}

void Scheduler::wake(Coroutine* c, bool timed_out) {
    if (c->state_ != State::Suspended) return;      // 已经被唤醒过了
    c->timed_out_ = timed_out;
    c->epoch_++;                                    // ★ 作废所有为它注册的旧定时器
    c->state_     = State::Ready;
    ready_.push_back(c);
}

void Scheduler::fire_timers() {
    TimePoint now = Clock::now();
    while (!timers_.empty() && timers_.begin()->first <= now) {
        Timer t = timers_.begin()->second;
        timers_.erase(timers_.begin());
        if (t.co->epoch_ != t.epoch) continue;      // ★ 已被 I/O 唤醒，这个定时器过期作废
        wake(t.co, /*timed_out=*/true);
    }
}

int Scheduler::next_timeout_ms() const {
    if (timers_.empty()) return -1;
    auto d = std::chrono::duration_cast<Ms>(timers_.begin()->first - Clock::now()).count();
    return d <= 0 ? 0 : static_cast<int>(d);
}

int Poller::poll(int timeout_ms) {
    epoll_event evs[128];
    int n = ::epoll_wait(epfd_, evs, 128, timeout_ms);
    for (int i = 0; i < n; ++i) {
        auto it = fds_.find(evs[i].data.fd);
        if (it == fds_.end()) continue;
        std::uint32_t e = evs[i].events;
        Coroutine* r = it->second.r;
        Coroutine* w = it->second.w;
        if (r && (e & (EPOLLIN  | EPOLLERR | EPOLLHUP))) sched().wake(r, false);
        if (w && (e & (EPOLLOUT | EPOLLERR | EPOLLHUP))) sched().wake(w, false);
    }
    return n < 0 ? 0 : n;
}

} // namespace (匿名)

// ==================================================================
//  内部原语
// ==================================================================
namespace detail {

Coroutine* current() noexcept {
    Scheduler& S = sched();
    return S.cur_ == &S.main_ ? nullptr : S.cur_;
}

void block() {
    Scheduler& S = sched();
    Coroutine* self = S.cur_;
    assert(self != &S.main_ && "block() 只能在协程内调用");
    self->_Coro_resume_index_ = State::Suspended;
    S.stats_.switches++;
    co_switch(&self->sp_, S.main_.sp_);             // ★★ 切回调度器
    self->_Coro_resume_index_ = State::Running;                  // 被 resume 后从这里继续
}

void wake(Coroutine* c) { sched().wake(c, false); }

} // namespace detail

// ==================================================================
//  1. 核心接口
// ==================================================================
bool co_in_coroutine() noexcept { return detail::current() != nullptr; }

void co_go(const StackConfig& cfg, std::function<void()> fn) {
    Coroutine* c = make_co(cfg, std::move(fn));
    if (!c) { std::fprintf(stderr, "[co] 栈分配失败\n"); return; }
    c->state_ = State::Ready;
    sched().ready_.push_back(c);
}

void co_go(std::function<void()> fn) { co_go(StackConfig{}, std::move(fn)); }

void co_yield_to_sched() {
    Scheduler& S = sched();
    Coroutine* self = S.cur_;
    if (!self || self == &S.main_) return;          // 不在协程里，什么都不做
    self->_Coro_resume_index_ = State::Ready;
    S.ready_.push_back(self);                       // ★ 和 block 的唯一区别：自己排回队尾
    S.stats_.switches++;
    co_switch(&self->sp_, S.main_.sp_);
    self->_Coro_resume_index_ = State::Running;
}

void co_sleep(Ms d) {
    Scheduler& S = sched();
    Coroutine* self = detail::current();
    if (!self) { ::usleep(static_cast<useconds_t>(d.count()) * 1000); return; }
    S.add_timer(Clock::now() + d, self);
    detail::block();                                 // ★ 不进 ready_，等定时器把它放回来
}

void co_sched_stop() { sched().running_ = false; }

void co_sched_run() {
    Scheduler& S = sched();
    S.running_ = true;
    while (S.running_) {
        S.fire_timers();

        // 只跑本轮开始时就绪的那批，避免新产生的协程把本轮撑成无限循环
        std::size_t n = S.ready_.size();
        for (std::size_t i = 0; i < n && !S.ready_.empty(); ++i) {
            Coroutine* c = S.ready_.front();
            S.ready_.pop_front();
            S.resume(c);
        }

        if (!S.ready_.empty()) continue;                          // 还有活干
        if (S.timers_.empty() && S.poller_.pending() == 0) break;  // 彻底没事了
        S.poller_.poll(S.next_timeout_ms());                       // 等 I/O 或定时器
    }
    S.running_ = false;
}

// ==================================================================
//  2. 同步原语
// ==================================================================
void co_mutex::lock() {
    while (locked_) {
        waiters_.push_back(detail::current());
        detail::block();
    }
    locked_ = true;
}

bool co_mutex::try_lock() {
    if (locked_) return false;
    locked_ = true;
    return true;
}

void co_mutex::unlock() {
    locked_ = false;
    if (!waiters_.empty()) { detail::wake(waiters_.front()); waiters_.pop_front(); }
}

void co_wait_group::add(int n) { count_ += n; }

void co_wait_group::done() {
    if (--count_ <= 0)
        while (!waiters_.empty()) { detail::wake(waiters_.front()); waiters_.pop_front(); }
}

void co_wait_group::wait() {
    while (count_ > 0) {
        waiters_.push_back(detail::current());
        detail::block();
    }
}

// ==================================================================
//  3. I/O
// ==================================================================
int co_set_nonblock(int fd) {
    int fl = ::fcntl(fd, F_GETFL, 0);
    if (fl < 0) return -1;
    return ::fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}

namespace {
// 返回 false 表示超时
bool wait_io(int fd, std::uint32_t ev, int timeout_ms) {
    Scheduler& S = sched();
    Coroutine* self = S.cur_;
    S.poller_.add(fd, ev, self);
    if (timeout_ms >= 0) S.add_timer(Clock::now() + Ms(timeout_ms), self);
    self->timed_out_ = false;
    detail::block();
    S.poller_.del(fd, ev);
    return !self->timed_out_;
}
} // namespace

ssize_t co_read(int fd, void* buf, std::size_t n, int timeout_ms) {
    if (!co_in_coroutine()) return ::read(fd, buf, n);
    for (;;) {
        ssize_t r = ::read(fd, buf, n);
        if (r >= 0) return r;
        if (errno == EINTR) continue;
        if (errno != EAGAIN && errno != EWOULDBLOCK) return -1;
        if (!wait_io(fd, EPOLLIN, timeout_ms)) { errno = ETIMEDOUT; return -1; }
    }
}

ssize_t co_write(int fd, const void* buf, std::size_t n, int timeout_ms) {
    if (!co_in_coroutine()) return ::write(fd, buf, n);
    const char* p    = static_cast<const char*>(buf);
    std::size_t left = n;
    while (left > 0) {
        ssize_t r = ::write(fd, p, left);
        if (r > 0) { p += r; left -= static_cast<std::size_t>(r); continue; }
        if (r < 0 && errno == EINTR) continue;
        if (r < 0 && errno != EAGAIN && errno != EWOULDBLOCK)
            return (left == n) ? -1 : static_cast<ssize_t>(n - left);
        if (!wait_io(fd, EPOLLOUT, timeout_ms)) { errno = ETIMEDOUT; return -1; }
    }
    return static_cast<ssize_t>(n);
}

int co_accept(int fd, sockaddr* addr, socklen_t* len, int timeout_ms) {
    if (!co_in_coroutine()) return ::accept(fd, addr, len);
    for (;;) {
        int c = ::accept4(fd, addr, len, SOCK_NONBLOCK | SOCK_CLOEXEC);
        if (c >= 0) return c;
        if (errno == EINTR) continue;
        if (errno != EAGAIN && errno != EWOULDBLOCK) return -1;
        if (!wait_io(fd, EPOLLIN, timeout_ms)) { errno = ETIMEDOUT; return -1; }
    }
}

int co_connect(int fd, const sockaddr* addr, socklen_t len, int timeout_ms) {
    if (!co_in_coroutine()) return ::connect(fd, addr, len);
    if (::connect(fd, addr, len) == 0) return 0;
    if (errno != EINPROGRESS) return -1;
    if (!wait_io(fd, EPOLLOUT, timeout_ms)) { errno = ETIMEDOUT; return -1; }
    int err = 0; socklen_t l = sizeof err;
    if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &l) < 0) return -1;
    if (err) { errno = err; return -1; }
    return 0;
}

// ==================================================================
//  4. 可观测性
// ==================================================================
Stats co_stats() { return sched().stats_; }

void co_foreach(const std::function<void(const Coroutine&)>& fn) {
    for (const Coroutine* c : sched().all_) fn(*c);
}

std::size_t co_stack_used(const Coroutine& c) {
    if (!c.stack_) return 0;
    auto* p = reinterpret_cast<const unsigned char*>(c.stack_);
    std::size_t untouched = 0;                       // 从【低地址】开始数还有多少没被动过
    while (untouched < c.size_ && p[untouched] == 0xCD) ++untouched;
    return c.size_ - untouched;
}

} // namespace co

---

4. example.cpp —— 演示全部功能

#include "coroutine.h"
#include <cstdio>
#include <cstring>
#include <sys/socket.h>
#include <unistd.h>

using namespace co;

static co_chan<int>  ch(2);          // 容量 2 的 channel
static co_mutex      mu;
static co_wait_group wg;
static int           counter = 0;
static int           sv[2];

int main() {
    // ---- 1. 基本：创建 + 主动让出 ----
    co_go([] {
        for (int i = 0; i < 3; ++i) { printf("[A] step %d\n", i); co_yield_to_sched(); }
    });
    co_go([] {
        for (int i = 0; i < 3; ++i) { printf("[B] step %d\n", i); co_yield_to_sched(); }
    });

    // ---- 2. 定时器 ----
    co_go([] {
        printf("[timer] sleep 50ms ...\n");
        co_sleep(Ms(50));
        printf("[timer] woke up\n");
    });

    // ---- 3. channel：生产者 / 消费者 ----
    co_go([] {
        for (int i = 0; i < 5; ++i) { printf("[prod] send %d\n", i); ch.send(i); }
        ch.close();
    });
    co_go([] {
        int v;
        while (ch.recv(v)) printf("        [cons] recv %d\n", v);
        printf("        [cons] channel closed\n");
    });

    // ---- 4. mutex + waitgroup ----
    wg.add(3);
    for (int k = 0; k < 3; ++k)
        co_go([k] {
            for (int i = 0; i < 100; ++i) {
                co_lock_guard g(mu);
                ++counter;
                if (i % 40 == 0) co_yield_to_sched();   // 持锁时让出，考验互斥
            }
            printf("[worker %d] done\n", k);
            wg.done();
        });
    co_go([] {
        wg.wait();
        printf("[wg] all workers done, counter = %d (expect 300)\n", counter);
    });

    // ---- 5. I/O：socketpair 上的 echo（验证 epoll 集成）----
    socketpair(AF_UNIX, SOCK_STREAM, 0, sv);
    co_set_nonblock(sv[0]);
    co_set_nonblock(sv[1]);
    co_go([] {                                          // 服务端
        char buf[64];
        ssize_t n = co_read(sv[1], buf, sizeof buf);
        if (n > 0) { printf("[echo]   got %.*s\n", (int)n, buf); co_write(sv[1], buf, n); }
    });
    co_go([] {                                          // 客户端
        const char* msg = "hello coroutine";
        co_write(sv[0], msg, std::strlen(msg));
        char buf[64];
        ssize_t n = co_read(sv[0], buf, sizeof buf, 1000);   // ★ 1 秒超时
        if (n > 0) printf("[client] echoed back: %.*s\n", (int)n, buf);
        else       printf("[client] timeout\n");
    });

    // ---- 6. 可观测性 ----
    co_go([] {
        co_sleep(Ms(10));
        Stats s = co_stats();
        printf("\n--- stats: alive=%llu created=%llu switches=%llu stack=%lluKB ---\n",
               (unsigned long long)s.alive,     (unsigned long long)s.total_created,
               (unsigned long long)s.switches,  (unsigned long long)(s.stack_bytes / 1024));
        co_foreach([](const Coroutine& c) {
            printf("    co#%-2llu  stack_used = %6zu / %zu\n",
                   (unsigned long long)c.id(), co_stack_used(c), c.stack_size());
        });
        printf("\n");
    });

    co_sched_run();                                     // ★ 事件循环
    printf("all done\n");
}

5. 编译运行

g++ -std=c++17 -O2 -g -c coroutine.cpp -o coroutine.o
as  co_switch.S -o co_switch.o          # 或 g++ -c co_switch.S -o co_switch.o
g++ -std=c++17 -O2 -g example.cpp coroutine.o co_switch.o -o demo
./demo

或者一条命令：

g++ -std=c++17 -O2 -g example.cpp coroutine.cpp co_switch.S -o demo && ./demo

示意输出（具体交错顺序取决于调度）：

[A] step 0
[B] step 0
[timer] sleep 50ms ...
[prod] send 0
[prod] send 1
[prod] send 2
        [cons] recv 0
        [cons] recv 1
[A] step 1
[B] step 1
[prod] send 3
[prod] send 4
        [cons] recv 2
        [cons] recv 3
        [cons] recv 4
        [cons] channel closed
[A] step 2
[B] step 2
[worker 0] done
[worker 1] done
[worker 2] done
[wg] all workers done, counter = 300 (expect 300)
[echo]   got hello coroutine
[client] echoed back: hello coroutine

--- stats: alive=2 created=12 switches=47 stack=256KB ---
    co#3   stack_used =    312 / 131072
    co#12  stack_used =    408 / 131072

[timer] woke up
all done

注意最后 stack_used 只有 300 多字节 —— 而我们给了 128 KB 的栈。这就是有栈协程内存开销的来源：绝大部分栈空间从来没被用到，但必须预留（因为不知道调用链会有多深）。

拿这个数据把 StackConfig::size 调到 16 KB，内存占用立刻降到 1/8：

co_go(StackConfig{ .size = 16 * 1024 }, my_task);

---

6. 实现要点回顾

代码里标了 ★ 的地方，都是"不做就会出事"的细节：

位置	细节	不做会怎样
alloc_mmap	guard page（mprotect PROT_NONE）	栈溢出静默踩坏相邻协程的栈，查三天查不出来
make_co	栈顶假帧的 16 字节对齐	SSE 指令（movaps）崩溃
Scheduler::resume	errno 各存各的	协程 A 读到协程 B 的 errno
trampoline	try/catch 兜住异常	异常无法跨栈传播 → 直接 std::terminate
trampoline	末尾永不返回	返回到栈顶的垃圾地址 → 崩溃
Scheduler::wake	epoch_++ 作废旧定时器	I/O 先就绪后，超时定时器仍会把它误唤醒一次
Scheduler::destroy	只在 resume 返回之后销毁	销毁自己正在运行的栈 → 崩溃
co_sleep	用定时器 + block，不是 ::sleep	阻塞整个线程，同线程所有协程冻结
co_mutex	自己实现，不用 std::mutex	futex 挂起整个线程 → 冻结甚至死锁
co_chan::recv	先 wake_one(senders_) 再挂起	无缓冲 channel 上收发双方互等 → 死锁
co_yield_to_sched	自己放回 ready_	和 block() 的唯一区别；漏了就永远醒不来
co_sched_run	只跑本轮开始时就绪的那批	协程不断产生新协程时本轮永不结束

7. 已知限制（以及怎么补）

限制	怎么扩展
单线程调度（每线程一个独立调度器，协程不跨线程）	加全局队列 + work stealing → M:N。★ 同时必须提供 co_local<T> 替代 thread_local —— 协程被偷到别的线程后，thread_local 拿到的是另一个线程的副本
无抢占：纯计算的协程会饿死其他协程	参考 Go：sysmon 线程 + SIGURG 信号异步抢占
无缓冲 channel 是简化语义（发送方等到"有接收者在等"就交接并继续，而非等到对方真正取走）	引入 rendezvous 请求对象（SendReq{value, co, taken}）
未做 ASAN / Valgrind 集成	切换前后调 __sanitizer_start/finish_switch_fiber；VALGRIND_STACK_REGISTER
未 hook 系统调用	dlsym(RTLD_NEXT, "read") 劫持 libc 同名符号（libco 的做法）
异常只打印不传播	提供 Coroutine::exception() 让 co_go 的调用方查询，或走 channel 上报
仅 x86-64 SysV	ARM64 要另写 co_switch：保存 x19–x28 + x29/x30，返回地址在 X30 里
Windows 不可用	换 IOCP 替代 epoll；换栈时必须同步更新 TEB 的 StackBase/StackLimit，否则 SEH 和栈溢出检测全乱
未提供 co_local / co_context	前者是协程局部存储，后者是 Go 式的取消/超时传播（with_timeout / cancel）
开启 Intel CET 后会崩	每个协程再分配一个影子栈，切换时用 rdsspq / rstorssp 一起换

8. 这个实现有多"真"

	本实现	libco	libgo / coost
上下文切换	✅ 汇编	✅	✅
调度器 + 定时器	✅	✅	✅
epoll 集成	✅	✅	✅
协程版同步原语	✅ mutex / chan / waitgroup	部分	✅
guard page + 栈用量统计	✅	❌	部分
栈池复用	✅	✅	✅
多线程 M:N	❌	❌（1:N）	✅
系统调用 hook	❌	✅	✅
共享栈（省内存）	❌	✅	❌
抢占	❌	❌	❌

约 700 行就能覆盖 libco 的大部分核心能力 —— 这也说明：有栈协程的真正难点不在"切换"（60 行汇编 + 100 行调度器），而在栈管理、生态兼容和那一堆容易漏的细节。