# 多线程开发深度教程：无栈状态机与并发编程
> **对标版本**：XTC v1.0-rc
---
## 架构概述
玄铁的并发模型是**双层架构**：OS 线程池 + 用户态调度器。
```
┌──────────────────────────────────────┐
│ XTC 编译器 │
│ 编译异步 → 状态机 IR + entry wrapper │
└──────────────┬───────────────────────┘
│
┌──────────────▼───────────────────────┐
│ 用户态调度器（主线程） │
│ ┌──────────┐ ┌────────────────┐ │
│ │ 就绪队列 │ │ 定时器堆 │ │
│ │ (FIFO) │ │ (小顶堆) │ │
│ └──────────┘ └────────────────┘ │
│ │ │ │
│ fiber.run() sleep/wakeup │
└──────────────┬───────────────────────┘
│
┌──────────────▼───────────────────────┐
│ OS 线程池（CPU 密集型） │
│ 异步{纯计算} → xt_async_spawn │
└──────────────────────────────────────┘
```
**核心原则**：CPU 密集型任务跑在线程池，I/O 等待型任务由调度器管理。状态机 fiber 不阻塞 OS 线程——需要等待时挂起并让出 CPU，等待条件满足后由调度器恢复。
---
## 无栈状态机
### 什么是无栈状态机
传统有栈协程（如 Go 的 goroutine）为每个协程分配独立栈（~2KB+），切换时需要保存/恢复寄存器。无栈状态机将协程编译为**状态枚举 + switch-case**，状态存储在堆分配的紧凑结构体中，切换成本为零。
```xuantie
// 用户代码
异步 {
示("步骤1")
等待(taskA)
示("步骤2")
等待(taskB)
示("完成")
}
```
编译器将其转换为：
```llvm
; 状态结构体（堆分配，16 字节）
%AsyncState = type { i32, i64, i64 } ; state, taskA_handle, taskB_handle
; poll 函数（无栈，由调度器调用）
define i32 @poll(%AsyncState* %self) {
entry:
switch i32 %state, label %done [i32 0, label %s0 i32 1, label %s1 i32 2, label %s2]
s0: ; 初始状态
store state=1
call @xt_print(..."步骤1")
store taskA_handle → field[1]
call @xt_scheduler_wait_task(taskA)
ret i32 0 ; PENDING
s1: ; taskA 完成后恢复
call @xt_async_wait(taskA) ; 验证完成
store state=2
call @xt_print(..."步骤2")
store taskB_handle → field[2]
call @xt_scheduler_wait_task(taskB)
ret i32 0 ; PENDING
s2: ; taskB 完成后恢复
call @xt_async_wait(taskB)
call @xt_print(..."完成")
ret i32 1 ; DONE
}
```
### 编译器如何决定
编译器在编译 `异步 { ... }` 时调用 `含等标记` 扫描块体：
| 检测到 | 策略 |
|--------|------|
| 语句级 `等待` / `.收()` / `.发()` / `选` / `连` / `流.读` / `流.写` / `时.睡` | fiber 状态机（挂起点让出，不占 OS 线程） |
| 上述挂起点出现在**循环体内** | **整块降级**为线程池任务：循环体局部状态无法跨 poll 存活，降级后挂起操作按阻塞执行；编译期打印提示 |
| 含 `求` / `执` / `输`（C 侧全程阻塞，无非阻塞路径） | **整块降级**为线程池任务；编译期打印提示 |
| 无上述 | 独立函数 → 线程池 |
注：`文件.读` / `文件.写` 被显式排除在切分点之外（按阻塞调用执行）；网络侧的 `连` / `流.读` / `流.写` 是 fiber 就绪切分点。
### 状态机 vs 线程池
| 特性 | 状态机 | 线程池 |
|------|--------|--------|
| 内存开销 | ~(8 + N×8) 字节 | ~4KB 栈 |
| 切换成本 | 0（调度器循环） | 上下文切换 |
| 适用场景 | I/O 等待、通道操作 | CPU 计算 |
| 并行度 | 调度器单线程 | N 核并行 |
---
## 调度器
### 事件循环
```
xt_scheduler_run():
while running:
now = 当前微秒时间
tick_timers(now) // 唤醒到期的 SLEEPING fiber
fiber = dequeue() // 从 READY 队列取
if no fiber:
if any WAITING: sleep(1ms) // 等待任务完成通知
else: break // 全部完成
result = fiber.poll(fiber.state)
if result == 0: // PENDING（fiber 已自行挂起）
else: // DONE
```
### Fiber 生命周期
```
[spawn] → READY → dequeue → RUNNING
↑ │
│ ┌─────────┼──────────┐
│ │ │ │
enqueue wait_task sleep ret 1
│ │ │ │
│ ▼ ▼ ▼
│ WAITING SLEEPING DONE
│ │ │
│ │ timer_tick
└────┴─────────┘
```
### 调度器 API（内部，用户不可直接调用）
| C 函数 | 说明 |
|--------|------|
| `xt_scheduler_init()` | 初始化调度器（`xt_init` 自动调用） |
| `xt_scheduler_run()` | 启动事件循环（main 函数末尾自动调用） |
| `xt_scheduler_spawn(state, poll)` | 创建 fiber 并入队 |
| `xt_scheduler_wait_task(task)` | 当前 fiber 挂起等待任务 |
| `xt_scheduler_wake_task(task)` | 任务完成时唤醒等待 fiber |
| `xt_scheduler_sleep_us(us)` | 当前 fiber 定时睡眠 |
---
## 通道深度用法
### 非阻塞 vs 阻塞
| 上下文 | `.发()` | `.收()` |
|--------|---------|---------|
| 普通代码 | 非阻塞（满则丢弃） | 非阻塞（空则返回 `空`） |
| `异步` 块内 | 阻塞（等待空间，条件变量） | 阻塞（等待数据，条件变量） |
阻塞版本使用条件变量，不占 OS 线程——等待的 fiber 被挂起，线程池线程释放。
### 生产者-消费者
```xuantie
设 ch = 道(4) // 缓冲 4，平滑速率差
设 producer = 异步 {
遍历 i 于 [1, 2, 3, 4, 5] {
时.睡(50) // 模拟生产耗时
ch.发(i)
示("生产: " & i)
}
}
设 consumer = 异步 {
设 sum = 0
设 count = 0
当 count < 5 {
设 val = ch.收() // 阻塞等待
sum = sum + val
count = count + 1
示("消费: " & val)
}
示("合计: " & sum)
}
等待(producer)
等待(consumer)
```
### 扇出（Fan-Out）
多个 worker 从同一通道取任务：
```xuantie
设 tasks = 道(10)
设 results = 道(10)
// 填充任务
遍历 i 于 [1, 2, 3, 4, 5, 6] {
tasks.发(i)
}
// 3 个 worker 竞争消费
设 worker = 函(id) {
异步 {
循 {
设 t = tasks.收()
若 t == 空 { 断 }
时.睡(t * 20) // 模拟处理
results.发(t * 2)
示("Worker " & id & " 完成: " & t)
}
}
}
设 w1 = worker(1)
设 w2 = worker(2)
设 w3 = worker(3)
等待(w1); 等待(w2); 等待(w3)
```
### 选择（Select）
`选(通道数组, 超时毫秒)` — 等待多个通道中第一个就绪的：
```xuantie
设 data = 道(2)
设 ctrl = 道(1)
设 sender = 异步 {
时.睡(200)
data.发("payload")
}
设 controller = 异步 {
ctrl.发("stop")
}
设 selector = 异步 {
设 idx = 选([data, ctrl], 1000)
匹配 idx {
0 -> { 示("数据通道就绪") }
1 -> { 示("控制通道就绪") }
_ -> { 示("超时") }
}
}
```
**超时模式**：用 `选([ch], timeout)` 实现带超时的通道等待。
### 通道 vs 共享内存
| 维度 | 通道 | 共享内存 + 锁 |
|------|------|--------------|
| 数据所有权 | 转移（发送后不再使用） | 共享 |
| 竞态风险 | 无（编译器/运行时保证） | 用户负责 |
| 性能 | 条件变量等待 | 锁竞争 |
| 推荐场景 | 首选 | 性能关键路径 |
---
## 实战模式
### 流水线（Pipeline）
多个阶段通过通道串联：
```xuantie
设 raw = 道(5)
设 cooked = 道(5)
设 stage1 = 异步 { raw.发(1); raw.发(2); raw.发(3) }
设 stage2 = 异步 { 当 (设 v = raw.收()) != 空 { cooked.发(v * 10) } }
设 stage3 = 异步 { 当 (设 v = cooked.收()) != 空 { 示("结果: " & v) } }
等待(stage1); 等待(stage2); 等待(stage3)
```
### 超时重试
```xuantie
设 ch = 道(1)
设 op = 异步 {
设 attempt = 0
当 attempt < 3 {
attempt = attempt + 1
示("尝试 " & attempt)
设 idx = 选([ch], 1000)
若 idx >= 0 { 示("成功"); 断 }
示("超时，重试...")
}
}
```
### 并发等待多个任务
```xuantie
设 t1 = 异步 { 时.睡(200); 返 "A" }
设 t2 = 异步 { 时.睡(100); 返 "B" }
设 t3 = 异步 { 时.睡(300); 返 "C" }
设 collector = 异步 {
设 ch = 道(3)
设 runner = 函(t, label) {
异步 { 等待(t); ch.发(label) }
}
runner(t1, "A")
runner(t2, "B")
runner(t3, "C")
设 results = []
设 i = 0; 当 i < 3 { results.追加(ch.收()); i = i + 1 }
示("全部完成: " & results.连接(","))
}
等待(collector)
```
### 计数信号量（限制并发数）
```xuantie
设 sem = 道(3) // 最多 3 并发
设 limitedTask = 函(id) {
异步 {
sem.发(1) // 获取许可
示("任务 " & id & " 开始")
时.睡(id * 100)
示("任务 " & id & " 完成")
设 _ = sem.收() // 释放许可
}
}
遍历 i 于 [1, 2, 3, 4, 5] {
limitedTask(i)
}
```
---
## 性能与调试
### Fiber 开销
| 指标 | 数值 |
|------|------|
| 每 fiber 内存 | ~(8 + N×8) 字节（N=等待点数） |
| 创建/销毁 | malloc/free 一次 |
| 切换（poll） | 函数调用 + switch 跳转 |
| 最大 fiber 数 | 65535（`XT_MAX_FIBERS`，编译期常量） |
| 最大定时器 | 8192（`XT_TIMER_HEAP_SIZE`，编译期常量；耗尽显式报错，不静默丢 fiber） |
### 常见问题
**Q: 为什么我的异步块没有并发执行？**
A: 检查是否调用了 `等待(task)` —— 未等待的任务仍在执行。程序末尾的 `xt_scheduler_run()` 会自动调度所有 fiber。
**Q: 状态机中能使用 `时.睡` 吗？**
A: 能，且不阻塞线程：状态机内的 `时.睡` 是挂起点，挂到调度器定时器堆（`xt_scheduler_sleep_us`）后让出。实测 64 个并发 `时.睡(300)` 总耗时 300ms——若每个都占住 worker 线程则需约 1800ms。但若该异步块因"循环体内有挂起点"或"含 `求`/`执`/`输`"被整块降级为线程池任务，块内执行的是阻塞版 `xt_time_sleep`，会占住一个 worker 线程。
**Q: 如何避免死锁？**
A: (1) 通道容量 > 0 作为缓冲；(2) 避免循环等待（A 等 B，B 等 A）；(3) 使用 `选` 的超时机制。
### 调试技巧
```xuantie
// 打印 fiber 信息（开发中）
// 当前可用：通过输出来追踪状态转换
异步 {
示("[DEBUG] s0 开始")
等待(task)
示("[DEBUG] s1 恢复")
}
```
---
## 已知限制与路线图
### 当前限制
| 限制 | 影响 | 预计修复 |
|------|------|---------|
| 循环体内的挂起点会降级 | `等待`/`收`/`发`/`选`/`时.睡`/`连`/`读`/`写` 出现在循环体里时，整个异步块降级为线程池任务（运行期独占一个 worker 线程），失去 fiber 轻量调度；编译期会打印提示 | 设计取舍：循环体局部状态无法跨 poll 恢复 |
| 同一条语句内的重放 | `等待` 所在的那条语句中、`等待` 之前的部分在恢复后会重复执行（`设 x = 副作用() + 等待(t)` 会让 `副作用()` 跑两次）；跨语句不重放 | P6b |
| 外层可变变量按"任务开始执行时"读取 | 块内读到的不是创建时刻的值，而是任务真正开始执行时的值——循环里批量起任务会全部读到终值 | 语义规则，见 [异步](关键字/异步.md) 的捕获说明；`遍历` 循环变量本身逐迭代正确 |
| 阻塞原语不可调度 | `求` / `执` / `输`，以及 `文件.读` / `文件.写`，走阻塞实现；含它们的异步块整块降级为线程池任务 | P6b（网络侧已是 fiber 就绪切分点，文件侧未接入） |
| 无 fiber 取消 | 无法终止等待中的 fiber，也没有超时取消接口 | P7 |
| 容量为编译期常量 | fiber 池 65535、定时器堆 8192；耗尽会显式报错退出 | P7（改 `runtime/xt_scheduler.h` 重编即可调） |
> 调度器状态由互斥锁保护（`g_sched_mutex`），线程池自带互斥锁与条件变量；worker 数按 CPU 核数自动确定（上限 64）。尚未做过专门的竞态压力审计。
### 路线图
| 阶段 | 内容 | 状态 |
|------|------|------|
| P1 | `含等标记` 扫描器 | ✅ |
| P2 | 单等待状态机 | ✅ |
| P3 | 多等待状态机 | ✅ |
| P4 | 通道 + `选` + 阻塞收发 | ✅ |
| P5 | 调度器原型（就绪队列 + 定时器堆 + 事件驱动唤醒） | ✅ |
| P6 | 编译路径自动路由：线性挂起走 fiber 状态机，循环内挂起与阻塞原语整块降级线程池（均带编译期提示） | ✅ |
| P6b | 系统 IO 集成（文件/网络 → 状态机） | 部分完成：网络侧 `连`/`流.读`/`流.写` 已是 fiber 就绪切分点；文件 IO 仍为阻塞路径 |
| P7 | 性能调优（work-stealing、NUMA、CPU affinity） | 规划中 |
---
## 完整示例：并发爬虫
```xuantie
// 简易并发爬虫——展示异步+通道+选的实际应用
设 urls = ["http://example.com", "http://example.org"]
设 results = 道(5)
设 done = 道(1)
设 crawl = 函(url) {
异步 {
设 resp = 求(url)
若 resp.成功 {
results.发(url & " → " & resp.值.截取(0, 100))
} 否 {
results.发(url & " → 失败")
}
}
}
// 并发发起
遍历 u 于 urls {
crawl(u)
}
// 收集结果
设 collector = 异步 {
设 count = 0
当 count < urls.长度 {
设 idx = 选([results], 5000)
若 idx >= 0 {
设 r = results.收()
示(r)
count = count + 1
} 否 {
示("超时，停止等待")
断
}
}
done.发("完成")
}
等待(collector)
```