异步
从
0.16开始,Zig 重新拥有了异步能力。它不再依靠async/await关键字,而是由标准库中的std.Io接口提供。
历史回顾
在 0.10 之前,Zig 通过 async、await、suspend、resume 等关键字实现了一套基于无栈协程的异步机制。这套机制在 self-hosted 编译器中一直没有完成移植,最终相关关键字在 0.11 以后被移除(详见 issue #6025)。
新的设计换了一种思路:异步不再是语言特性,而是 I/O 实现提供的能力。
- 语言层面没有
async关键字,也没有“异步函数”与“普通函数”之分,从而彻底避开了“函数染色(function coloring)”问题 - 所有可能阻塞的操作(文件、网络、睡眠、锁、进程……)都通过
std.Io接口完成 - 异步任务通过
io.async、io.concurrent等普通函数发起,具体如何调度由Io的实现决定
如果你熟悉 Zig 的分配器,就很容易理解这种设计:需要分配内存的函数接收一个 std.mem.Allocator,需要做 I/O 的函数接收一个 std.Io。由调用者(通常是 main)决定用哪一种实现,库代码只依赖接口。
std.Io 接口
为了行文简洁,下文正文中的 Io 均指 std.Io,例如 Io.Threaded 即 std.Io.Threaded;示例代码中则一律写出完整的 std.Io。
std.Io 本身非常小,只有两个字段:类型为 ?*anyopaque 的 userdata,以及类型为 *const std.Io.VTable 的 vtable。和 std.mem.Allocator 一样,它是一个“胖指针”:userdata 指向具体实现的状态,vtable 是一张函数指针表。这张表涵盖了:
- 异步与并发:
async、concurrent、await、cancel,以及任务组Group - 取消:
checkCancel、swapCancelProtection、recancel - 同步原语的底层支持:
futexWait、futexWake - 文件系统:
Dir、File的各种操作 - 网络:监听、连接、收发数据、DNS 查询
- 进程:创建子进程、等待、获取当前路径等
- 时间与随机数:
now、sleep、random
标准库目前提供了以下实现:
| 实现 | 说明 |
|---|---|
Io.Threaded | 基于线程池,功能最完整,也是目前的默认实现 |
Io.Threaded.init_single_threaded | 单线程、阻塞式,不提供并发能力 |
Io.Evented | 基于 fiber(见下方说明)的事件驱动实现,仍处于实验阶段 |
Io.Uring / Io.Kqueue / Io.Dispatch | 分别基于 io_uring、kqueue、Grand Central Dispatch,Io.Evented 会按平台选用 |
Io.failing | 模拟一个“什么都不支持”的系统,常用于测试 |
🅿️ 什么是 fiber?
fiber(纤程)可以理解为“由程序自己调度的轻量级线程”,也常被称为有栈协程(stackful coroutine)或用户态线程。
和操作系统线程一样,每个 fiber 都拥有自己独立的栈,所以它可以在任意深度的函数调用中途暂停,之后再从暂停的地方继续执行。不同之处在于,fiber 之间的切换完全发生在用户态:只需保存当前 fiber 的栈指针、帧指针和指令指针,再恢复另一个 fiber 的这三个值即可,不需要陷入内核,也不需要内核参与调度。标准库中的 std.Io.fiber.Context 保存的正是这三个寄存器。
Io.Evented 的工作方式大致如下:
- 每个
io.async/io.concurrent任务都运行在一个 fiber 上 - 任务发起 I/O(读文件、收网络数据、
sleep等)时,把请求提交给 io_uring / kqueue / GCD,然后主动切换到另一个就绪的 fiber - 等内核通知 I/O 完成,再切换回原来的 fiber,让它从暂停处继续执行
这样少量的系统线程就能同时推进大量任务,而任务代码本身看起来和普通的阻塞式代码完全一样。
| 对比项 | 系统线程 | fiber | Zig 旧版 async(无栈协程) |
|---|---|---|---|
| 调度者 | 操作系统内核 | Io 实现(用户态) | 用户代码(suspend/resume) |
| 独立的栈 | 有 | 有 | 没有,状态保存在 @Frame 中 |
| 切换开销 | 较大,需要进入内核 | 很小,几条指令 | 很小 |
| 函数染色 | 无 | 无 | 隐式存在:会挂起的函数由编译器推断为 async,并沿调用链向上传播 |
正因为 fiber 有独立的栈,普通函数无需任何标记就能在其中被挂起,新的 std.Io 才得以避开函数染色问题。代价是栈切换需要针对每种 CPU 架构编写汇编,目前 std.Io.fiber 只支持 x86_64、aarch64 和 riscv64,在其他架构上 Io.Evented 不可用。
获取 Io
最简单的方式是让 main 接收 std.process.Init,运行时会为你准备好默认的 Io 实现(同时还有 gpa、arena 等)。你也可以自己创建一个 Io.Threaded:
// 方式一:使用 `std.process.Init`,由运行时选好默认的 std.Io 实现
pub fn mainWithInit(init: std.process.Init) !void {
const io = init.io;
try std.Io.File.stdout().writeStreamingAll(io, "hello from init.io\n");
}
// 方式二:自己创建一个 std.Io 实现(这里是基于线程池的 `std.Io.Threaded`)
pub fn mainWithThreaded() !void {
var threaded: std.Io.Threaded = .init(std.heap.page_allocator, .{});
defer threaded.deinit();
const io = threaded.io();
try std.Io.File.stdout().writeStreamingAll(io, "hello from std.Io.Threaded\n");
}在单元测试中,可以直接使用 std.testing.io。
🅿️ 提示
Io 和分配器一样,应该由应用程序的入口来选择,然后一路作为参数传下去。编写库时,请让需要做 I/O 的函数接收一个 io: std.Io 参数,而不是自己创建实现。
Io 是如何工作的
std.Io 本身不包含任何 I/O 逻辑。标准库中 std.Io 的 sleep、async,以及 File、Dir、net 中的各种操作,都只是薄薄的一层包装:整理一下参数,然后调用 io.vtable.xxx(io.userdata, ...)。真正干活的,是把自己的函数填进 vtable 的那个实现。
下面两行代码的效果完全相同,第二行就是第一行在标准库内部展开后的样子:
fn sleepTwice(io: std.Io) !void {
// 平时的写法
try io.sleep(.fromMilliseconds(10), .awake);
// 与上一行完全等价:std.Io.sleep 所做的,只是把参数打包成 Timeout,
// 再把 userdata 连同参数一起交给 vtable 中的函数
try io.vtable.sleep(io.userdata, .{ .duration = .{
.raw = .fromMilliseconds(10),
.clock = .awake,
} });
}以 io.sleep(...) 在 Io.Threaded 上的执行过程为例,整条调用链是:
- 接口层:
std.Io.sleep把参数打包成Timeout,调用io.vtable.sleep(io.userdata, timeout)。它不知道、也不关心背后是线程、fiber 还是一个模拟系统 - 装配:
Io.Threaded的io()方法在生成std.Io时,把自身的指针放进userdata,把自己的sleep、async、now等函数填进 vtable - 实现层:调用落到
Io.Threaded的sleep函数,它先把userdata转回*Threaded,再按平台调用clock_nanosleep等系统调用阻塞当前线程
同样是 sleep,换成 Io.Evented,就会变成向 io_uring 提交一个定时器,然后切换到别的 fiber。
这和 std.mem.Allocator 的设计如出一辙:allocator.alloc 最终调用的也是 allocator.vtable.alloc(allocator.ptr, ...)。正因为所有 I/O 都要经过这张表,替换一个 Io 实现就能改变程序全部的 I/O 行为,而调用方的代码一个字都不用改。本章最后我们会利用这一点,亲手实现一个 Io。
🅿️ 提示
想进一步了解细节,可以直接阅读标准库源码:接口层位于 lib/std/Io.zig,Io.Threaded 的实现位于 lib/std/Io/Threaded.zig。
async 与 await
io.async(function, args) 会“异步地”调用 function,并立即返回一个 Io.Future(T),其中 T 是 function 的返回类型。之后可以通过 future.await(io) 拿到结果。
本节以及后面的大部分示例中,io 都来自 main 的参数 init: std.process.Init,也就是运行时默认选择的 Io.Threaded。它内部维护一个线程池,io.async 发起的任务会交给池中的线程执行:
/// 模拟一个耗时的 I/O 操作,例如网络请求
fn fetch(io: std.Io, id: u32, delay_ms: i64) !u32 {
try io.sleep(.fromMilliseconds(delay_ms), .awake);
return id * 10;
}
pub fn main(init: std.process.Init) !void {
// init.io 是运行时为 main 准备的默认 std.Io 实现,即基于线程池的 std.Io.Threaded,
// 因此下面的两个任务会被分配到不同的线程上同时执行
const io = init.io;
const start = std.Io.Clock.awake.now(io);
// 发起两个异步任务,它们返回 `std.Io.Future(T)`
var a = io.async(fetch, .{ io, 1, 100 });
// 无论从哪条路径离开作用域,都要确保任务被取消、资源被回收;
// 这里有意丢弃 cancel 返回的结果,原因见正文
defer _ = a.cancel(io) catch {};
var b = io.async(fetch, .{ io, 2, 100 });
defer _ = b.cancel(io) catch {};
// 等待结果;await 是幂等的,之后再调用 cancel 只会直接返回结果
const ra = try a.await(io);
const rb = try b.await(io);
const elapsed = start.untilNow(io, .awake).toMilliseconds();
std.debug.print("async_await: a = {d}, b = {d}, 耗时约 {d} ms\n", .{ ra, rb, elapsed });
}两个 fetch 各自睡眠 100ms,但总耗时只有约 100ms,因为它们被分配到了不同的线程上同时执行。如果换成别的 Io 实现,同样的代码可能会有不同的表现,下一小节就会看到这一点。
用 defer 取消任务
上面的代码在每次 io.async 之后都紧跟了一句 defer _ = a.cancel(io) catch {};,它的作用是:无论从哪条路径离开作用域,任务都会被结束,任务占用的资源都会被回收。如果没有这一句,一旦两次 await 之间的代码出错提前返回,任务 b 就再也不会被等待,它占用的资源也就泄漏了。
你可能会注意到,catch {} 把 cancel 返回的错误直接丢掉了,什么也没有输出。这是有意为之,标准库自己也是这样写的(例如 std.Io.net.HostName 中 DNS 查询的实现)。分两种情况来看:
- 正常路径:
await已经拿到了结果,错误也已经通过try处理过了。Future的await和cancel都是幂等的,此时cancel只是把同一个结果再返回一遍,丢弃它并不会漏掉任何东西 - 提前返回的路径:函数已经带着另一个错误在返回了,而
defer中的表达式本身也无法再向上返回错误。此时任务刚刚被要求取消,它的结果多半就是error.Canceled,已经没有人关心了;cancel真正重要的作用是等任务结束并回收资源
但下面两种情况不能简单地丢弃结果:
- 任务返回了需要释放的资源(例如打开的文件、分配的内存):要在
defer中把它释放掉,例如defer if (f.cancel(io)) |file| file.close(io) else |_| {}; - 提前返回时也想知道任务出了什么错:可以在
catch中把错误记录下来,例如catch |err| std.log.warn("task failed: {t}", .{err})
async 不等于并发
async 表达的是 异步(asynchrony):这次调用与后续代码相互独立,结果可以晚一点再要。它并不保证任务会与调用者同时运行。
Io 实现完全可以在 io.async 内部就地把函数执行完,再把结果交给 Future。下面的代码和上一节完全相同,只是 io 不再来自 init.io,而是我们自己创建的单线程实现 Io.Threaded.init_single_threaded:
fn fetch(io: std.Io, id: u32, delay_ms: i64) !u32 {
try io.sleep(.fromMilliseconds(delay_ms), .awake);
return id * 10;
}
pub fn main() !void {
// 单线程、阻塞式的 std.Io 实现:不提供任何并发能力
var threaded: std.Io.Threaded = .init_single_threaded;
const io = threaded.io();
const start = std.Io.Clock.awake.now(io);
// 代码和上一个例子完全一样,但 `async` 会直接就地执行函数
var a = io.async(fetch, .{ io, 1, 50 });
defer _ = a.cancel(io) catch {};
var b = io.async(fetch, .{ io, 2, 50 });
defer _ = b.cancel(io) catch {};
_ = try a.await(io);
_ = try b.await(io);
// 结果依然正确,只不过耗时是两次 sleep 之和
const elapsed = start.untilNow(io, .awake).toMilliseconds();
std.debug.print("async_not_concurrent: 耗时约 {d} ms\n", .{elapsed});
}程序依然正确,只是总耗时变成了两次睡眠之和。这正是 async 的设计意图:用 async 编写的代码,在任何 Io 实现上都能正确运行,并发只是锦上添花的优化。因此 io.async 永远不会失败。
io.async 内部发生了什么
为什么 async 有时并发、有时又是就地执行的?答案同样在接口层和实现层的分工里。
接口层:类型擦除。vtable 中的函数指针必须有固定的签名,没法直接接收任意的 function 和 args。因此 std.Io.async 在编译期为每一种被调用的函数生成一个签名固定的“跳板”函数 start(context, result):参数元组以字节的形式交给实现(由实现负责复制一份),跳板再把字节还原成参数元组,调用真正的函数,并把返回值写入 result 指向的位置,也就是 Future 中的 result 字段。
实现层:两种结局。拿到跳板之后,实现有两种选择:
- 在
vtable.async里就地调用start,结果直接写进Future,然后返回null。此后future.await发现没有待完成的任务,会直接返回结果,不再调用实现 - 把任务交给别人执行(例如线程池中的线程或另一个 fiber),返回一个任务句柄。等到
await时,再通过vtable.await等待任务完成、取回结果
Io.Threaded 会优先把任务交给线程池,但在以下情况会退化为就地执行,而不是报错:
- 以单线程模式编译(
builtin.single_threaded) - 为任务分配内存失败
- 正在执行任务的线程数已经达到上限
async_limit(默认是 CPU 核数减一,init_single_threaded则是 0) - 线程池中没有空闲线程,并且创建新线程失败
这就是 io.async 永远不会失败、却也不保证并发的原因。io.concurrent 走的是另一条路:vtable.concurrent 在无法分配并发单元时直接返回 error.ConcurrencyUnavailable,而不会退化为就地执行。这就是下一节的主题。
concurrent:需要真正的并发
有些代码只有在多个任务同时运行时才能正确工作。典型例子是生产者与消费者通过一个容量有限的队列通信:如果生产者在队列满时阻塞,而消费者又必须等生产者执行完才开始运行,程序就会死锁。
这种场景应该使用 io.concurrent。它和 io.async 的用法一样,但要求 Io 实现必须为任务分配一个独立的“并发单元”(例如线程或 fiber),否则返回 error.ConcurrencyUnavailable:
fn producer(io: std.Io, queue: *std.Io.Queue(u32)) !void {
// 函数返回时关闭队列,通知消费者不会再有新数据
defer queue.close(io);
for (1..11) |i| {
// 队列满时,putOne 会阻塞,直到消费者取走数据
try queue.putOne(io, @intCast(i));
}
}
fn consumer(io: std.Io, queue: *std.Io.Queue(u32)) !u32 {
var sum: u32 = 0;
while (true) {
const value = queue.getOne(io) catch |err| switch (err) {
// 队列已关闭并且数据已经取完
error.Closed => return sum,
error.Canceled => |e| return e,
};
sum += value;
}
}
pub fn main(init: std.process.Init) !void {
const io = init.io; // 默认的 std.Io.Threaded,支持真正的并发
// 队列容量只有 2,生产者和消费者必须“同时”运行才能完成任务
var buffer: [2]u32 = undefined;
var queue: std.Io.Queue(u32) = .init(&buffer);
// concurrent 要求 std.Io 实现必须真正并发地执行任务,否则返回 error.ConcurrencyUnavailable
var p = try io.concurrent(producer, .{ io, &queue });
defer p.cancel(io) catch {};
var c = try io.concurrent(consumer, .{ io, &queue });
defer _ = c.cancel(io) catch {};
try p.await(io);
const sum = try c.await(io);
std.debug.print("concurrent: sum = {d}\n", .{sum});
}这里用到的 Io.Queue(T) 是标准库提供的多生产者多消费者队列,它的 putOne / getOne 在队列满或空时会阻塞当前任务,close 后再取数据会在取完剩余元素后返回 error.Closed。
在不支持并发的实现上,concurrent 会直接报错,而不是让程序悄悄死锁:
pub fn unavailable() !void {
var threaded: std.Io.Threaded = .init_single_threaded;
const io = threaded.io();
var buffer: [2]u32 = undefined;
var queue: std.Io.Queue(u32) = .init(&buffer);
// 单线程实现无法提供并发能力,这里会直接得到错误,而不是死锁
const result = io.concurrent(producer, .{ io, &queue });
try std.testing.expectError(error.ConcurrencyUnavailable, result);
}如何选择
- 只是想“让这几件事尽量重叠执行”:用
async,它在任何实现上都正确 - 程序的正确性依赖于多个任务同时运行:用
concurrent,并处理error.ConcurrencyUnavailable
取消
future.cancel(io) 与 await 类似,但会先向任务发出取消请求。任务在下一次到达取消点(cancelation point)时会收到 error.Canceled。所谓取消点,就是任何错误集合中包含 error.Canceled 的 Io 函数,例如 sleep、文件读写、网络收发、Mutex.lock 等。
fn slowTask(io: std.Io) !void {
// sleep 是一个“取消点”(cancelation point):
// 当任务被取消时,它会返回 error.Canceled
try io.sleep(.fromSeconds(10), .awake);
std.debug.print("这一行不会被执行\n", .{});
}
pub fn main(init: std.process.Init) !void {
const io = init.io; // 默认的 std.Io.Threaded,支持真正的并发
var task = try io.concurrent(slowTask, .{io});
try io.sleep(.fromMilliseconds(10), .awake);
// cancel 会请求取消并等待任务结束,返回任务的结果
const result = task.cancel(io);
if (result) |_| {
std.debug.print("cancel: 任务正常完成\n", .{});
} else |err| {
std.debug.print("cancel: 任务结束,结果为 {t}\n", .{err});
}
}关于取消,还有几条规则需要了解:
- 取消请求只会让下一个取消点返回
error.Canceled,之后的取消点不会重复报告。因此忽略error.Canceled通常是一个 bug,正确的做法是把它向上传播 - 如果确实需要先处理完手头的事情再响应取消,可以在捕获
error.Canceled后调用io.recancel(),让下一个取消点再次返回它 - 对于长时间运行的纯计算任务,可以定期调用
io.checkCancel()主动检查是否被取消
如果某段代码必须完整执行、不能被打断,可以通过 swapCancelProtection 临时屏蔽取消:
fn criticalSection(io: std.Io) !void {
// 在这段代码中屏蔽取消请求,保证它能完整地执行完毕
const old = io.swapCancelProtection(.blocked);
defer _ = io.swapCancelProtection(old);
io.sleep(.fromMilliseconds(1), .awake) catch |err| switch (err) {
// 取消已被屏蔽,不可能收到 error.Canceled
error.Canceled => unreachable,
};
}Group:任务组
当需要启动数量不定的任务、并且只关心“全部完成”时,可以使用 Io.Group。组内的任务不返回值(返回类型必须能转换为 Io.Cancelable!void),只能作为一个整体被等待或取消:
fn worker(io: std.Io, id: usize, results: []u64) std.Io.Cancelable!void {
try io.sleep(.fromMilliseconds(10), .awake);
// 每个任务只写自己的那一格,因此不需要加锁
results[id] = id * id;
}
pub fn main(init: std.process.Init) !void {
const io = init.io; // 默认的 std.Io.Threaded,支持真正的并发
var results: [8]u64 = undefined;
var group: std.Io.Group = .init;
// 如果中途出错,取消组内所有任务
defer group.cancel(io);
for (0..results.len) |i| {
group.async(io, worker, .{ io, i, &results });
}
// 等待组内全部任务完成
try group.await(io);
var sum: u64 = 0;
for (results) |r| sum += r;
std.debug.print("group: sum = {d}\n", .{sum});
}Group 也提供了 concurrent 方法,语义与 io.concurrent 一致。它的一个重要特点是:每个任务结束时就会释放自己占用的资源,而不是等整个组完成,因此一个长期存在、不断加入新任务的组(例如服务器为每个连接创建一个任务)也不会造成资源泄漏。
Select:等待最先完成的任务
Io.Select(U) 可以同时启动多个不同类型的任务,并等待其中最先完成的那一个。U 是一个带标签的联合类型,每个字段对应一种任务的返回类型。
一个常见用法是为操作加上超时:
fn slowQuery(io: std.Io) std.Io.Cancelable!u32 {
try io.sleep(.fromSeconds(5), .awake);
return 42;
}
pub fn main(init: std.process.Init) !void {
const io = init.io; // 默认的 std.Io.Threaded,支持真正的并发
// 每个字段对应一种任务,字段类型就是对应函数的返回类型
const Result = union(enum) {
query: std.Io.Cancelable!u32,
timeout: std.Io.Cancelable!void,
};
var buffer: [2]Result = undefined;
var select: std.Io.Select(Result) = .init(io, &buffer);
// 取消剩余的任务,并丢弃它们的结果
defer select.cancelDiscard();
try select.concurrent(.query, slowQuery, .{io});
try select.concurrent(.timeout, std.Io.sleep, .{ io, .fromMilliseconds(20), .awake });
// 等待第一个完成的任务
switch (try select.await()) {
.query => |r| std.debug.print("select: 查询结果 {d}\n", .{try r}),
.timeout => std.debug.print("select: 查询超时\n", .{}),
}
}await 返回第一个完成的结果后,cancelDiscard 会取消剩下的任务并丢弃它们的结果。如果剩下的任务可能返回需要释放的资源,应该改用 cancel,循环调用直到返回 null,逐个释放结果。
同步原语
由于任务可能运行在线程上,也可能运行在 fiber 上,同步原语必须由 Io 实现来配合阻塞与唤醒。因此原先的 std.Thread.Mutex 等已经被移除,取而代之的是 std.Io 中一套与实现无关的版本,它们都需要传入 io:
Io.Mutex:互斥锁Io.RwLock:读写锁Io.Condition:条件变量Io.Semaphore:信号量Io.Event:一次性事件,可以set并让其他任务waitIo.Queue(T):前面用到的阻塞队列
const Counter = struct {
mutex: std.Io.Mutex = .init,
value: u64 = 0,
fn add(counter: *Counter, io: std.Io, n: u64) std.Io.Cancelable!void {
// std.Io.Mutex 的加锁同样需要传入 io,并且是一个取消点
try counter.mutex.lock(io);
defer counter.mutex.unlock(io);
counter.value += n;
}
};
fn worker(io: std.Io, counter: *Counter) std.Io.Cancelable!void {
for (0..1000) |_| try counter.add(io, 1);
}
pub fn main(init: std.process.Init) !void {
const io = init.io; // 默认的 std.Io.Threaded,支持真正的并发
var counter: Counter = .{};
var group: std.Io.Group = .init;
defer group.cancel(io);
for (0..4) |_| group.async(io, worker, .{ io, &counter });
try group.await(io);
std.debug.print("mutex: counter = {d}\n", .{counter.value});
}这些原语都建立在 Io 的 futexWait / futexWake 之上。在 Io.Threaded 中它们对应真正的线程阻塞,而在事件驱动的实现中则对应挂起当前 fiber、切换到其他任务。
实现一个自己的 Io
理解 std.Io 最好的办法就是亲手实现一个。下面我们实现一个极简的“模拟系统” SimIo:
- 标准输入来自一段预设的字符串,标准输出写入内存
- 拥有一个虚拟时钟,
sleep不会真正等待,而是立刻把时钟向前拨 async就地执行,不提供并发能力- 其余操作全部沿用
Io.failing,即表现为一个“什么都没有”的系统
/// 一个极简的 `std.Io` 实现:
/// * 标准输入来自一段预设的字符串,标准输出与标准错误写入内存
/// * 拥有一个“虚拟时钟”,sleep 会立刻返回并把时钟向前拨
/// * async 直接就地执行(不提供并发),concurrent 返回 error.ConcurrencyUnavailable
/// * 随机数来自固定种子的伪随机数生成器
/// 其余所有操作都沿用 `std.Io.failing` 的行为(模拟一个“什么都没有”的系统)。
const SimIo = struct {
stdin: []const u8,
stdout: std.Io.Writer,
now_ns: i96 = 0,
prng: std.Random.DefaultPrng = .init(0),
cancel_protection: std.Io.CancelProtection = .unblocked,
pub fn init(stdin: []const u8, stdout_buffer: []u8) SimIo {
return .{ .stdin = stdin, .stdout = .fixed(stdout_buffer) };
}
/// 已经写入“虚拟终端”的内容
pub fn output(sim: *const SimIo) []const u8 {
return sim.stdout.buffered();
}
pub fn io(sim: *SimIo) std.Io {
return .{ .userdata = sim, .vtable = &vtable };
}
// 以 `std.Io.failing` 为模板,只覆盖我们关心的函数
const vtable: std.Io.VTable = vt: {
var v = std.Io.failing.vtable.*;
v.async = std.Io.noAsync; // 就地执行,返回 null 表示结果已经就绪
v.groupAsync = std.Io.noGroupAsync;
v.recancel = recancel;
v.swapCancelProtection = swapCancelProtection;
v.checkCancel = checkCancel;
v.operate = operate;
v.fileWritePositional = fileWritePositional;
v.fileReadPositional = fileReadPositional;
v.now = now;
v.sleep = sleep;
v.random = random;
break :vt v;
};
fn cast(userdata: ?*anyopaque) *SimIo {
return @ptrCast(@alignCast(userdata.?));
}
// 只有一个任务在运行,也就不会有人来取消它
fn recancel(_: ?*anyopaque) void {}
fn swapCancelProtection(userdata: ?*anyopaque, new: std.Io.CancelProtection) std.Io.CancelProtection {
const sim = cast(userdata);
defer sim.cancel_protection = new;
return sim.cancel_protection;
}
fn checkCancel(_: ?*anyopaque) std.Io.Cancelable!void {}
/// 所有“流式”读写最终都会走到这里
fn operate(userdata: ?*anyopaque, operation: std.Io.Operation) std.Io.Cancelable!std.Io.Operation.Result {
const sim = cast(userdata);
switch (operation) {
.file_write_streaming => |op| {
if (op.file.handle != std.Io.File.stdout().handle and
op.file.handle != std.Io.File.stderr().handle)
return .{ .file_write_streaming = error.NotOpenForWriting };
const n = sim.stdout.writeSplatHeader(op.header, op.data, op.splat) catch
return .{ .file_write_streaming = error.NoSpaceLeft };
return .{ .file_write_streaming = n };
},
.file_read_streaming => |op| {
if (op.file.handle != std.Io.File.stdin().handle)
return .{ .file_read_streaming = error.NotOpenForReading };
if (sim.stdin.len == 0)
return .{ .file_read_streaming = error.EndOfStream };
var n: usize = 0;
for (op.data) |buf| {
const len = @min(buf.len, sim.stdin.len - n);
@memcpy(buf[0..len], sim.stdin[n..][0..len]);
n += len;
}
sim.stdin = sim.stdin[n..];
return .{ .file_read_streaming = n };
},
else => return std.Io.failingOperate(userdata, operation),
}
}
// 终端不支持定位读写,返回 Unseekable 后,File.Reader/Writer 会自动退回流式模式
fn fileWritePositional(
_: ?*anyopaque,
_: std.Io.File,
_: []const u8,
_: []const []const u8,
_: usize,
_: u64,
) std.Io.File.WritePositionalError!usize {
return error.Unseekable;
}
fn fileReadPositional(
_: ?*anyopaque,
_: std.Io.File,
_: []const []u8,
_: u64,
) std.Io.File.ReadPositionalError!usize {
return error.Unseekable;
}
// 无论哪种时钟,都返回同一个虚拟时间
fn now(userdata: ?*anyopaque, _: std.Io.Clock) std.Io.Timestamp {
return .fromNanoseconds(cast(userdata).now_ns);
}
// 不真正等待,只是把虚拟时钟拨到超时时刻
fn sleep(userdata: ?*anyopaque, timeout: std.Io.Timeout) std.Io.Cancelable!void {
const sim = cast(userdata);
switch (timeout) {
// 永远等待:单任务的系统中没有人能唤醒我们
.none => unreachable,
.duration => |d| sim.now_ns += d.raw.nanoseconds,
.deadline => |d| sim.now_ns = @max(sim.now_ns, d.raw.nanoseconds),
}
}
fn random(userdata: ?*anyopaque, buffer: []u8) void {
cast(userdata).prng.fill(buffer);
}
};这里有几个值得注意的地方:
- 以
Io.failing为模板:VTable有近百个函数,我们不必全部实现。在comptime中复制一份Io.failing.vtable.*,再覆盖我们关心的几个即可 async可以就地执行:标准库提供了Io.noAsync,它直接调用函数并返回null,表示结果已经写好,此后的await不会再调用实现。Io.noGroupAsync同理- 流式读写统一经过
operate:File.Reader/File.Writer的流式读写最终都会变成Io.Operation,交给vtable.operate处理 - 定位读写返回
error.Unseekable:File.Writer默认先尝试定位写入(pwritev),遇到error.Unseekable后会自动退回流式模式,就像写入一个真实的终端或管道一样
下面这段“业务代码”只依赖 Io 接口:它逐行读取标准输入,每行“处理”一秒钟,再把结果写到标准输出:
/// 一段普通的“业务代码”:它只依赖 `std.Io` 接口,并不知道底层是哪种实现
fn app(io: std.Io) !void {
var in_buffer: [64]u8 = undefined;
var stdin_reader = std.Io.File.stdin().reader(io, &in_buffer);
const stdin = &stdin_reader.interface;
var out_buffer: [64]u8 = undefined;
var stdout_writer = std.Io.File.stdout().writer(io, &out_buffer);
const stdout = &stdout_writer.interface;
const start = std.Io.Clock.awake.now(io);
// 逐行读取输入,处理后写回输出
while (try stdin.takeDelimiter('\n')) |line| {
try io.sleep(.fromSeconds(1), .awake); // 模拟一个耗时的处理过程
var task = io.async(countVowels, .{line});
try stdout.print("{s}: {d} vowels\n", .{ line, task.await(io) });
}
const elapsed = start.untilNow(io, .awake);
try stdout.print("elapsed: {d}s\n", .{elapsed.toSeconds()});
try stdout.flush();
}
fn countVowels(text: []const u8) usize {
var n: usize = 0;
for (text) |c| {
if (std.mem.indexOfScalar(u8, "aeiouAEIOU", c) != null) n += 1;
}
return n;
}
pub fn main(init: std.process.Init) !void {
var buffer: [256]u8 = undefined;
var sim: SimIo = .init("hello\nzig\nasync io\n", &buffer);
// 在模拟系统中运行业务代码:瞬间完成,输出被捕获到内存里
try app(sim.io());
// 再用真正的 std.Io 把捕获的结果打印出来
try std.Io.File.stdout().writeStreamingAll(init.io, sim.output());
}app 中用到的 File.stdin().reader、Writer.print、io.sleep、Clock.now、io.async 全是标准库的普通 API,它完全不知道自己运行在模拟系统中。传入 SimIo 时,整个程序瞬间完成,输出为:
hello: 2 vowels
zig: 1 vowels
async io: 3 vowels
elapsed: 3s虽然程序“睡”了 3 秒,但实际上一点时间也没有花。这正是把 I/O 抽象成接口所带来的好处之一:可以把真实世界替换成一个完全确定、可控的环境,从而编写出快速、稳定、可复现的测试,或者模拟各种难以复现的故障(磁盘已满、网络断开、时钟回拨……)。
🅿️ 提示
如果想更进一步,实现一个真正支持并发的 Io,就需要让 async / concurrent 把任务交给其他执行单元,并在 await、futexWait、sleep 等阻塞点上把当前任务挂起、切换到其他任务。Io.Threaded 使用线程来做到这一点,而 Io.Evented 则借助 std.Io.fiber 在用户态切换栈。阅读它们的源码是深入理解 Zig 异步模型的好途径。
更多关于 std.Io 引入时的变更细节,可以参考 0.16.0 版本说明。