尧图网页设计 
Serenity OS 异步资源与异步输入流设计指南AK::AsyncResource 与 AsyncInputStream 深度解析【免费下载链接】serenityThe Serenity Operating System 项目地址: https://gitcode.com/GitHub_Trending/se/serenityDocumentation/AsynchronousDesign.md是 Serenity OS 异步 I/O 体系的权威设计文档它定义了AK::AsyncResource带可失败/异步析构的通用资源抽象与AK::AsyncInputStream全缓冲异步输入流的完整语义、状态机与调用约定。本文以该文档为骨架结合 AK/AsyncStream.h、AK/AsyncStreamHelpers.h、AK/AsyncStreamTransform.h 等源码实现以及 LibHTTP 的实际应用与测试用例系统讲解异步资源关闭/重置纪律、peek/read 缓冲工作流、长度条件约束、EOF 检测语义与错误码约定帮助你写出正确、健壮且不违反协议约束的异步流代码。一、异步资源AsyncResource可失败/异步析构的资源抽象AK::AsyncResource表示一类析构过程本身可能失败、可能异步的通用资源例如 POSIX 文件描述符、AsyncStream、HTTP 响应体等。这类资源的典型场景是一个异步 socket 在关闭前可能希望等待所有未完成的传输结束并通知调用方服务器是否已确认所有缓冲写入。该抽象的核心矛盾在于资源的释放不再是一个同步且无错误可言的操作因此必须显式地定义关闭Close与重置Reset两个抽象操作Abstract OperationAO并以close()/reset()/is_open()三个接口暴露给用户。接口定义见 AK/AsyncStream.hvirtual void reset() 0断言资源处于打开状态后执行 Reset AOvirtual CoroutineErrorOrvoid close() 0断言资源已完全构造且处于打开状态然后执行 Close AO 并等待其结果virtual bool is_open() const 0查询资源是否仍处于打开状态。类本身通过AK_MAKE_NONCOPYABLE/AK_MAKE_NONMOVABLE禁止拷贝与移动确保生命周期完全由所有者掌控。1.1 Close AO优雅关闭的五步协议源码注释AK/AsyncStream.h为 Close AO 定义了严格顺序断言当前没有任何人在等待await该资源确保后续对该资源的任何等待操作都会触发断言可能异步地关闭底层资源且必须保证若资源状态是干净的该状态将无限期保持。何为干净由资源类型自行定义——对流而言通常指没有未完成的写、没有未读的数据检查资源状态是否干净若不干净则调用 Reset AO 并返回错误优先返回EBUSY可能异步地释放底层资源返回成功。也就是说close()是有体检功能的它会把数据没读完/没写完这类不干净状态检测出来并以错误上报而不是默默丢弃。1.2 Reset AO同步暴力回收Reset AOAK/AsyncStream.h则完全不同向当前所有等待者调度返回错误优先返回ECANCELED确保后续等待操作会断言同步释放底层资源最好以能清晰向事件生产者指示错误的方式进行同步返回。可见 Reset 是即时、同步、粗暴的错误路径而 Close 是优雅、可等待、可失败的正常路径。AsyncResource的析构函数语义正是二者的组合AK/AsyncStream.h先断言无人等待若资源仍打开则执行 Reset AO。这就是文档反复强调的自动 reset-on-destruction机制。1.3 关闭/重置纪律不要带着打开的异步资源退出文档强调使用AsyncResource时唯一需要注意的就是不要在用完资源后仍让它处于打开状态——这既是为了行为一致性也是为了避免向对端误发虚假的数据结束信号。好消息是这种卫生习惯在实践中不难做到每个AsyncResource在析构时都会自动 reset且部分资源在任意接口返回错误时也会自动 reset即reset-on-error行为。因此下面这段代码是正确的文档原例CoroutineErrorOrvoid do_very_meaningful_work() { Core::AsyncTCPSocket socket make_me_a_socket(); // Core::AsyncTCPSocket 是 AK::AsyncStream它本身当然是 AsyncResource // 表现出 reset-on-error 行为。 auto object CO_TRY(co_await socket.read_objectVeryImportantObject()); auto response CO_TRY(process_object(object)); CO_TRY(co_await socket.write(response)); CO_TRY(co_await socket.close()); }其行为保证是如果任何一步包括process_object失败socket 会被 reset表现为发送 TCP RST 中断连接如果一切顺利则优雅关闭完成 TCP 四次挥手。这正是 Close/Reset 两条路径在错误传播上的体现。1.4 非局部资源必须手动 reset自动析构重置并不能取代所有显式 reset。当异步资源不是某个可失败函数的局部变量、而是外部传入的引用时错误路径上必须手动 reset否则会污染调用方的流状态文档原例CoroutineErrorOrvoid do_very_meaningful_work_second_time(AsyncStream stream) { auto object CO_TRY(co_await stream.read_objectVeryImportantObject()); auto response_or_error process_object(object); if (response_or_error.is_error()) { stream.reset(); co_return response_or_error.release_error(); } CO_TRY(co_await stream.write(response_or_error.release_value())); }有人会想干脆什么都不做但这会让流在函数失败时处于未知状态迫使调用方事后逐个检查stream-is_open()——因为对已关闭/已重置的流做任何操作都会触发断言。与其让错误蔓延到调用方不如在错误点就地 reset。1.5 is_open不会自动变坏的状态除了close和reset资源还提供is_open()用于判断资源是否已被关闭或重置无论显式还是由失败的接口隐式触发。文档特别指出一个重要事实is_open状态不会随时间自行变化——比如服务端在无人监听时发送了 RSTsocket 不会因此自动变为 closed只有对资源调用某个方法open 状态才可能改变。这是懒状态更新的设计状态的转换只发生在调用边界上这为上层协议实现提供了可预测性。二、输入流AsyncInputStream全缓冲的 peek/read 工作流AK::AsyncInputStream是所有异步输入流的基类AK/AsyncStream.h。其设计有一个根本性特点所有输入流在结构上都是带缓冲的——每个输入流都维护一个内部读缓冲区数据先被读入该缓冲区peek和read返回的都是指向缓冲区内容的ReadonlyBytes视图而非拷贝。2.1 典型工作流多次 peek 一次 read典型的AsyncInputStream使用流程是反复peek直到在流中找到想要的东西然后执行一次read把刚 peek 到的对象从缓冲区中移除。下面这个例子从流中读取一个大小未知的对象文档原例CoroutineErrorOrVeryImportantObject read_some_very_important_object(AsyncInputStream stream) { size_t byte_size; while (true) { auto bytes CO_TRY(co_await stream.peek()); Optionalsize_t maybe_size figure_out_size_from_prefix(bytes); if (maybe_size.has_value()) { // 太好了我们已经读了足够的数据来推断对象的长度。 byte_size maybe_size.value(); break; } } auto bytes CO_TRY(co_await stream.read(byte_size)); auto object_or_error parse_object_from_data(bytes); if (object_or_error.is_error()) { stream.reset(); co_return object_or_error.release_error(); } co_return object_or_error.release_value(); }如果大小事先已知读取就更简单stream-read(size)即可。但文档给出了一个必须牢记的告诫[!IMPORTANT] 永远不要做超出绝对必要的 peek。2.2 长度条件Length Conditionread 的形式化约束为什么不能过度 peek文档给出了形式化定义。设在某次read之前(s_1, s_2, ..., s_n)是自上次read或自流创建以来peek/peek_or_eof返回的各ReadonlyBytes视图的长度序列若n 1read的bytes参数可以取任意值若n 1则s_{n-1}必须不大于bytes参数此外若流数据没有子字节结构且尚未到达 EOFbytes应当大于s_{n-1}。违反该条件几乎总意味着调用方代码有 bug。虽然文档没有保证违反条件时会发生什么但明确说明异步流框架在违反长度条件时不保证线性渐进时间复杂度——也就是说乱 peek 可能让整个流的复杂度从线性退化为更差同时还可能意外触发 EOF 读取。在实践中该条件并不难满足。例如在read_some_very_important_object例子中条件只要求当对象已经完全包含在bytes参数内时figure_out_size_from_prefix必须能解出大小。2.3 read 的精确语义read(bytes)总是精确返回bytes字节。如果bytes大于最后一次 peek 到的数据量s_n或n为 0即从未 peek 过read会返回比已 peek 数据更多的内容显然是通过继续读流实现的。实现见 AK/AsyncStream.h当缓冲区数据不足时read循环调用enqueue_some补充数据直到缓冲区达到bytes大小然后dequeue并返回buffer.slice(0, bytes)视图bytes 0时则直接返回空视图。另一个重要保证是peek 后必读原则[!IMPORTANT] 如果你从流中 peek 了某些数据就应该把它们 read 掉。2.4 错误码约定EIO 与 EBUSY如果你的用例不需要知道 EOF 的位置即事先就知道要读多少不依赖 EOF 判断那么只需在最后使用peek、read和close若输入因 EOF 而过早结束peek/read会reset 流并报错close时若流按约定应被完整读完却仍有剩余数据close会报错错误码约定意外的流结束返回EIO未读完整个流就关闭返回EBUSY后者的错误码选择与 AsyncResource 的 Close AO 第 4 步优先返回 EBUSY完全一致。在 AK/AsyncStream.h 中可以看到peek的 EIO 路径peek内部调用peek_or_eof若 EOF 标志置位则先reset()再返回Error::from_errno(EIO)。read在缓冲不足且enqueue_some返回 falseEOF时同样执行 reset 并返回 EIO。三、EOF 感知peek_or_eof 与精确的 is_eof 实现对于需要感知 EOF 位置的用例AsyncInputStream提供peek_or_eof()它在peek的基础上额外返回一个is_eof标志PeekOrEofResult { ReadonlyBytes data; bool is_eof; }见 AK/AsyncStream.h。EOF 检测语义与 POSIX 流类似第一次peek_or_eof返回到达 EOF 为止的数据且is_eof为 false下一次调用才返回同样的数据且is_eof置为 true。也就是说EOF 的宣判滞后一拍。3.1 一个可靠的 is_eof为什么需要 read(0)文档用一个例子滥用peek_or_eof来实现绝对可靠的is_eofCoroutineErrorOrbool accurate_is_eof(AsyncInputStream stream) { auto [_, is_eof] CO_TRY(co_await stream.peek_or_eof()); must_sync(stream.read(0)); co_return is_eof; }这里那个看似无用的read(0)是干什么的考虑一个场景流缓冲区为空且整条流只剩 1 个字节。第一次peek_or_eof会返回该字节但is_eof为 false于是accurate_is_eof返回 false。之后若有人用普通peek再次 peek 该字节peek内部会再走一次peek_or_eof——这次is_eof就会被置位peek检查到标志后报 EIO 并 reset 流把一次无辜的查询变成了致命错误。read(0)的作用就是消费掉这次 peek 产生的前瞻peek-ahead状态把peek_or_eof的读取 peek 性质降级为非读取 peek从而避免下一次peek被强制升级为读取 peek。这正是文档peek 了就要 read原则的极端体现——哪怕只 read 0 个字节。四、peek / peek_or_eof / read 的形式化行为规范文档指出这三个方法的实现看似简单却蕴含了大量概念复杂性。其形式化规范如下每次peek_or_eof调用被分为读取型 peekreading peek与非读取型 peeknon-reading peek读取型 peek总是读取新数据非读取型 peek仅在缓冲区无数据时才读取缓冲区已有数据时的非读取型 peek 称为no-op peek无操作 peek缓冲区无数据、不得不读取时的非读取型 peek 称为promoted peek升级型 peek即被晋升为读取。分类规则紧随另一个 peek 之后的 peek 永远是读取型 peek紧随read之后的 peek 是非读取型 peek。这就是前述长度条件的由来——peek 在某种意义上总是对新数据的请求一次不必要的 peek 可能无意中读到 EOF 而报错。流程细节与 AK/AsyncStream.h 的实现一致若是非读取型 peek 且缓冲区非空直接返回缓冲视图is_eof为 false这是 no-op peek否则读取型或 promoted peek调用enqueue_some从底层流读取数据检查是否遇到 EOF即底层 read 返回 0 个新字节若未到 EOF把新数据追加进缓冲区无论哪种类型最终都返回缓冲区的视图。协议违规与逻辑错误处理文档明确列举均有源码对应在已到达 EOF 后调用peek、或read试图越过 EOF 读取协议违规返回EIO任何错误包括 EIO都会导致流 reset因此若继续对该流操作会触发断言并发调用读操作断言因为这是逻辑错误enqueue_some的实现也要求并发调用必须断言见 AK/AsyncStream.h对未打开的流调用peek/peek_or_eof/read逻辑错误同样断言三个方法开头均有VERIFY(is_open())。五、源码纵深AsyncInputStream 的三个钩子与派生组件5.1 三个纯虚钩子如何实现一个自定义输入流文档对应的 AK/AsyncStream.h 定义了实现新输入流必须重写的三个钩子理解了它们才能真正理解缓冲流的内部enqueue_some(BadgeAsyncInputStream)若未到 EOF从底层流至少读 1 字节进缓冲区并返回 true若已到 EOF不得改动缓冲区并返回 false。若读取失败返回 Error必须执行 Reset AO或等价操作——因此对AsyncInputStream而言一切读取错误都是致命的。这是唯一可被reset中断的方法buffered_data_unchecked(BadgeAsyncInputStream)仅返回缓冲区的视图且不得使先前返回的视图失效dequeue(BadgeAsyncInputStream, size_t bytes)从缓冲区移除bytes字节调用时保证缓冲区中确有这么多数据同样不得使先前返回的视图失效。注释还提到若使用AsyncStreamBuffer作为流缓冲区dequeue与enqueue_some将获得均摊 O(stream_length) 的复杂度。此外还有一个便捷方法read_objectT()AK/AsyncStream.h直接read(sizeof(T))后通过 union memcpy把字节重解释为类型T的对象返回——文档第一个示例中的socket.read_objectVeryImportantObject()用的正是它。5.2 输出流与全双工流AsyncOutputStreamAK/AsyncStream.h是输出侧基类核心抽象是write_some(ReadonlyBytes)默认的write会循环调用write_some直到所有缓冲写尽天然支持分片写入AsyncStream同时继承AsyncInputStream与AsyncOutputStream是全双工流的公共基类StreamWrapperTAK/AsyncStream.h把任意实现了 AsyncResource 接口的流包装成 AsyncResource 子类转发 reset/close/is_open。5.3 复合组件AsyncStreamTransform、consume_until、Slice 与 StreamPairAK/AsyncStreamTransform.h 的AsyncStreamTransformT是一个变换流它持有一个底层流和一个AK::GeneratorEmpty, ErrorOrvoid生成器把生成器每次co_yield的内容作为变换后的数据供上层读取。其close()语义很能体现 Close AO 的干净状态检查若生成器尚未结束还有数据未消费则 reset 并返回 EBUSY。AK/AsyncStreamHelpers.h 提供两个实用工具AsyncStreamHelpers::consume_until(stream, delimiter, max_size)循环 peek 直到在缓冲区中找到分隔符然后read消费到分隔符末尾支持可选的max_size上限——这是按行/按定界符解析的通用原语AsyncInputStreamSliceAK/AsyncStreamHelpers.h在流上切出一个固定长度的切片读满length字节即视为切片 EOF未消费完就 close 会返回 EBUSYAsyncStreamPairAK/AsyncStreamHelpers.h把独立的输入流与输出流组合成一个全双工AsyncStream并保证一侧出错时另一侧也被 reset如enqueue_some失败时 reset 输出流、write_some失败时 reset 输入流是错误传播联动的现成范本。六、真实案例LibHTTP 中的异步流实战异步流并非纸上谈兵Serenity OS 的 LibHTTP 客户端正是其大规模使用者是研读本文档语义的最佳配套代码。6.1 按行解析响应头consume_until 的典型应用Userland/Libraries/LibHTTP/Http11Connection.cpp 的receive_response_headers展示了peek 找定界符 → read 消费的标准姿势CoroutineErrorOrStatusCodeAndHeaders receive_response_headers(AsyncStream stream) { auto status_line CO_TRY(co_await AsyncStreamHelpers::consume_until(stream, \r\nsv)); // ... 用 GenericLexer 解析状态行失败时 stream.reset() 并返回错误 ... VectorHeader headers; while (true) { auto header StringView { CO_TRY(co_await AsyncStreamHelpers::consume_until(stream, \r\nsv)) }; if (header \r\nsv) break; // ... 解析 Header 行格式错误时 stream.reset() ... } co_return StatusCodeAndHeaders { ... }; }注意其中的错误处理模式任何解析失败都先stream.reset()再返回错误——这正是文档错误路径上必须显式 reset 非局部资源原则的忠实执行因为stream是从外部传入的引用。6.2 分块传输编码AsyncStreamTransform 的实战Userland/Libraries/LibHTTP/Http11Connection.cpp 的ChunkedBodyStream继承自AsyncStreamTransformAsyncInputStream用一个生成器实现 HTTP chunked 解码循环consume_until读块长度行、peek/read分片搬运块数据并co_yield最后校验块尾\r\n读到0长度块即结束。整个解码逻辑被封装成看起来像普通输入流的组件上层只需无脑read——这就是变换流的威力。6.3 测试验证AsyncTestStreams 与随机分片Userland/Libraries/LibTest/AsyncTestStreams.cpp 提供了两个用于测试的异步内存流AsyncMemoryInputStreamAsyncTestStreams.cpp用Vectorsize_t控制每次enqueue_some吐出的字节数其dequeue里VERIFY(m_last_enqueue m_read_head m_read_head m_peek_head)直接对长度条件做了运行时校验enqueue_some在 reset 后返回ECANCELED验证 Reset AO 对等待者的错误调度AsyncMemoryOutputStreamAsyncTestStreams.cpp在析构时通过StreamCloseExpectation断言流的最终状态到底是 Reset 还是 Close——把文档的关闭/重置卫生变成了可自动检查的测试契约。配套的 Tests/LibHTTP/TestHttp11Connection.cpp 则用Test::randomly_partition_inputAsyncTestStreams.cpp把 HTTP 响应随机切成大小不一的片再通过AsyncStreamPair喂给Http11Connection校验 chunked 解码后的 body 是否与期望一致——这相当于把任意分片下 AsyncInputStream 语义仍正确变成了自动化回归测试。七、实践要点速查不要带着打开的异步资源退出局部资源靠析构自动 reset外部传入的资源在错误路径上必须手动reset()。Close 是优雅路径Reset 是错误路径close()会检查干净状态数据未消费完会返回EBUSYreset()同步粗暴释放并向等待者返回ECANCELED。peek 不要过量s_{n-1}不得大于随后read的bytes否则违反长度条件、破坏渐进复杂度并可能误读 EOF。peek 了就要 read哪怕read(0)也行——accurate_is_eof的实现证明了 0 字节 read 也能重置 peek 状态避免后续peek升级为读取型而报 EIO。EOF 检测滞后一拍peek_or_eof第一次返回数据不带 EOF 标志下一次调用才置位用peek在 EOF 后读取是协议违规报 EIO 并 reset。错误即终点任何读错误对AsyncInputStream都是致命的enqueue_some失败必须执行 Reset AO对已关闭/已重置的流操作会断言对流的并发读操作也会断言。错误码速记意外 EOF →EIO未读完整流就 close →EBUSYreset 打断等待者 →ECANCELED。【免费下载链接】serenityThe Serenity Operating System 项目地址: https://gitcode.com/GitHub_Trending/se/serenity创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Serenity OS 异步资源与异步输入流设计指南：AK::AsyncResource 与 AsyncInputStream 深度解析
Serenity OS 异步资源与异步输入流设计指南AK::AsyncResource 与 AsyncInputStream 深度解析【免费下载链接】serenityThe Serenity Operating System 项目地址: https://gitcode.com/GitHub_Trending/se/serenityDocumentation/AsynchronousDesign.md是 Serenity OS 异步 I/O 体系的权威设计文档它定义了AK::AsyncResource带可失败/异步析构的通用资源抽象与AK::AsyncInputStream全缓冲异步输入流的完整语义、状态机与调用约定。本文以该文档为骨架结合 AK/AsyncStream.h、AK/AsyncStreamHelpers.h、AK/AsyncStreamTransform.h 等源码实现以及 LibHTTP 的实际应用与测试用例系统讲解异步资源关闭/重置纪律、peek/read 缓冲工作流、长度条件约束、EOF 检测语义与错误码约定帮助你写出正确、健壮且不违反协议约束的异步流代码。一、异步资源AsyncResource可失败/异步析构的资源抽象AK::AsyncResource表示一类析构过程本身可能失败、可能异步的通用资源例如 POSIX 文件描述符、AsyncStream、HTTP 响应体等。这类资源的典型场景是一个异步 socket 在关闭前可能希望等待所有未完成的传输结束并通知调用方服务器是否已确认所有缓冲写入。该抽象的核心矛盾在于资源的释放不再是一个同步且无错误可言的操作因此必须显式地定义关闭Close与重置Reset两个抽象操作Abstract OperationAO并以close()/reset()/is_open()三个接口暴露给用户。接口定义见 AK/AsyncStream.hvirtual void reset() 0断言资源处于打开状态后执行 Reset AOvirtual CoroutineErrorOrvoid close() 0断言资源已完全构造且处于打开状态然后执行 Close AO 并等待其结果virtual bool is_open() const 0查询资源是否仍处于打开状态。类本身通过AK_MAKE_NONCOPYABLE/AK_MAKE_NONMOVABLE禁止拷贝与移动确保生命周期完全由所有者掌控。1.1 Close AO优雅关闭的五步协议源码注释AK/AsyncStream.h为 Close AO 定义了严格顺序断言当前没有任何人在等待await该资源确保后续对该资源的任何等待操作都会触发断言可能异步地关闭底层资源且必须保证若资源状态是干净的该状态将无限期保持。何为干净由资源类型自行定义——对流而言通常指没有未完成的写、没有未读的数据检查资源状态是否干净若不干净则调用 Reset AO 并返回错误优先返回EBUSY可能异步地释放底层资源返回成功。也就是说close()是有体检功能的它会把数据没读完/没写完这类不干净状态检测出来并以错误上报而不是默默丢弃。1.2 Reset AO同步暴力回收Reset AOAK/AsyncStream.h则完全不同向当前所有等待者调度返回错误优先返回ECANCELED确保后续等待操作会断言同步释放底层资源最好以能清晰向事件生产者指示错误的方式进行同步返回。可见 Reset 是即时、同步、粗暴的错误路径而 Close 是优雅、可等待、可失败的正常路径。AsyncResource的析构函数语义正是二者的组合AK/AsyncStream.h先断言无人等待若资源仍打开则执行 Reset AO。这就是文档反复强调的自动 reset-on-destruction机制。1.3 关闭/重置纪律不要带着打开的异步资源退出文档强调使用AsyncResource时唯一需要注意的就是不要在用完资源后仍让它处于打开状态——这既是为了行为一致性也是为了避免向对端误发虚假的数据结束信号。好消息是这种卫生习惯在实践中不难做到每个AsyncResource在析构时都会自动 reset且部分资源在任意接口返回错误时也会自动 reset即reset-on-error行为。因此下面这段代码是正确的文档原例CoroutineErrorOrvoid do_very_meaningful_work() { Core::AsyncTCPSocket socket make_me_a_socket(); // Core::AsyncTCPSocket 是 AK::AsyncStream它本身当然是 AsyncResource // 表现出 reset-on-error 行为。 auto object CO_TRY(co_await socket.read_objectVeryImportantObject()); auto response CO_TRY(process_object(object)); CO_TRY(co_await socket.write(response)); CO_TRY(co_await socket.close()); }其行为保证是如果任何一步包括process_object失败socket 会被 reset表现为发送 TCP RST 中断连接如果一切顺利则优雅关闭完成 TCP 四次挥手。这正是 Close/Reset 两条路径在错误传播上的体现。1.4 非局部资源必须手动 reset自动析构重置并不能取代所有显式 reset。当异步资源不是某个可失败函数的局部变量、而是外部传入的引用时错误路径上必须手动 reset否则会污染调用方的流状态文档原例CoroutineErrorOrvoid do_very_meaningful_work_second_time(AsyncStream stream) { auto object CO_TRY(co_await stream.read_objectVeryImportantObject()); auto response_or_error process_object(object); if (response_or_error.is_error()) { stream.reset(); co_return response_or_error.release_error(); } CO_TRY(co_await stream.write(response_or_error.release_value())); }有人会想干脆什么都不做但这会让流在函数失败时处于未知状态迫使调用方事后逐个检查stream-is_open()——因为对已关闭/已重置的流做任何操作都会触发断言。与其让错误蔓延到调用方不如在错误点就地 reset。1.5 is_open不会自动变坏的状态除了close和reset资源还提供is_open()用于判断资源是否已被关闭或重置无论显式还是由失败的接口隐式触发。文档特别指出一个重要事实is_open状态不会随时间自行变化——比如服务端在无人监听时发送了 RSTsocket 不会因此自动变为 closed只有对资源调用某个方法open 状态才可能改变。这是懒状态更新的设计状态的转换只发生在调用边界上这为上层协议实现提供了可预测性。二、输入流AsyncInputStream全缓冲的 peek/read 工作流AK::AsyncInputStream是所有异步输入流的基类AK/AsyncStream.h。其设计有一个根本性特点所有输入流在结构上都是带缓冲的——每个输入流都维护一个内部读缓冲区数据先被读入该缓冲区peek和read返回的都是指向缓冲区内容的ReadonlyBytes视图而非拷贝。2.1 典型工作流多次 peek 一次 read典型的AsyncInputStream使用流程是反复peek直到在流中找到想要的东西然后执行一次read把刚 peek 到的对象从缓冲区中移除。下面这个例子从流中读取一个大小未知的对象文档原例CoroutineErrorOrVeryImportantObject read_some_very_important_object(AsyncInputStream stream) { size_t byte_size; while (true) { auto bytes CO_TRY(co_await stream.peek()); Optionalsize_t maybe_size figure_out_size_from_prefix(bytes); if (maybe_size.has_value()) { // 太好了我们已经读了足够的数据来推断对象的长度。 byte_size maybe_size.value(); break; } } auto bytes CO_TRY(co_await stream.read(byte_size)); auto object_or_error parse_object_from_data(bytes); if (object_or_error.is_error()) { stream.reset(); co_return object_or_error.release_error(); } co_return object_or_error.release_value(); }如果大小事先已知读取就更简单stream-read(size)即可。但文档给出了一个必须牢记的告诫[!IMPORTANT] 永远不要做超出绝对必要的 peek。2.2 长度条件Length Conditionread 的形式化约束为什么不能过度 peek文档给出了形式化定义。设在某次read之前(s_1, s_2, ..., s_n)是自上次read或自流创建以来peek/peek_or_eof返回的各ReadonlyBytes视图的长度序列若n 1read的bytes参数可以取任意值若n 1则s_{n-1}必须不大于bytes参数此外若流数据没有子字节结构且尚未到达 EOFbytes应当大于s_{n-1}。违反该条件几乎总意味着调用方代码有 bug。虽然文档没有保证违反条件时会发生什么但明确说明异步流框架在违反长度条件时不保证线性渐进时间复杂度——也就是说乱 peek 可能让整个流的复杂度从线性退化为更差同时还可能意外触发 EOF 读取。在实践中该条件并不难满足。例如在read_some_very_important_object例子中条件只要求当对象已经完全包含在bytes参数内时figure_out_size_from_prefix必须能解出大小。2.3 read 的精确语义read(bytes)总是精确返回bytes字节。如果bytes大于最后一次 peek 到的数据量s_n或n为 0即从未 peek 过read会返回比已 peek 数据更多的内容显然是通过继续读流实现的。实现见 AK/AsyncStream.h当缓冲区数据不足时read循环调用enqueue_some补充数据直到缓冲区达到bytes大小然后dequeue并返回buffer.slice(0, bytes)视图bytes 0时则直接返回空视图。另一个重要保证是peek 后必读原则[!IMPORTANT] 如果你从流中 peek 了某些数据就应该把它们 read 掉。2.4 错误码约定EIO 与 EBUSY如果你的用例不需要知道 EOF 的位置即事先就知道要读多少不依赖 EOF 判断那么只需在最后使用peek、read和close若输入因 EOF 而过早结束peek/read会reset 流并报错close时若流按约定应被完整读完却仍有剩余数据close会报错错误码约定意外的流结束返回EIO未读完整个流就关闭返回EBUSY后者的错误码选择与 AsyncResource 的 Close AO 第 4 步优先返回 EBUSY完全一致。在 AK/AsyncStream.h 中可以看到peek的 EIO 路径peek内部调用peek_or_eof若 EOF 标志置位则先reset()再返回Error::from_errno(EIO)。read在缓冲不足且enqueue_some返回 falseEOF时同样执行 reset 并返回 EIO。三、EOF 感知peek_or_eof 与精确的 is_eof 实现对于需要感知 EOF 位置的用例AsyncInputStream提供peek_or_eof()它在peek的基础上额外返回一个is_eof标志PeekOrEofResult { ReadonlyBytes data; bool is_eof; }见 AK/AsyncStream.h。EOF 检测语义与 POSIX 流类似第一次peek_or_eof返回到达 EOF 为止的数据且is_eof为 false下一次调用才返回同样的数据且is_eof置为 true。也就是说EOF 的宣判滞后一拍。3.1 一个可靠的 is_eof为什么需要 read(0)文档用一个例子滥用peek_or_eof来实现绝对可靠的is_eofCoroutineErrorOrbool accurate_is_eof(AsyncInputStream stream) { auto [_, is_eof] CO_TRY(co_await stream.peek_or_eof()); must_sync(stream.read(0)); co_return is_eof; }这里那个看似无用的read(0)是干什么的考虑一个场景流缓冲区为空且整条流只剩 1 个字节。第一次peek_or_eof会返回该字节但is_eof为 false于是accurate_is_eof返回 false。之后若有人用普通peek再次 peek 该字节peek内部会再走一次peek_or_eof——这次is_eof就会被置位peek检查到标志后报 EIO 并 reset 流把一次无辜的查询变成了致命错误。read(0)的作用就是消费掉这次 peek 产生的前瞻peek-ahead状态把peek_or_eof的读取 peek 性质降级为非读取 peek从而避免下一次peek被强制升级为读取 peek。这正是文档peek 了就要 read原则的极端体现——哪怕只 read 0 个字节。四、peek / peek_or_eof / read 的形式化行为规范文档指出这三个方法的实现看似简单却蕴含了大量概念复杂性。其形式化规范如下每次peek_or_eof调用被分为读取型 peekreading peek与非读取型 peeknon-reading peek读取型 peek总是读取新数据非读取型 peek仅在缓冲区无数据时才读取缓冲区已有数据时的非读取型 peek 称为no-op peek无操作 peek缓冲区无数据、不得不读取时的非读取型 peek 称为promoted peek升级型 peek即被晋升为读取。分类规则紧随另一个 peek 之后的 peek 永远是读取型 peek紧随read之后的 peek 是非读取型 peek。这就是前述长度条件的由来——peek 在某种意义上总是对新数据的请求一次不必要的 peek 可能无意中读到 EOF 而报错。流程细节与 AK/AsyncStream.h 的实现一致若是非读取型 peek 且缓冲区非空直接返回缓冲视图is_eof为 false这是 no-op peek否则读取型或 promoted peek调用enqueue_some从底层流读取数据检查是否遇到 EOF即底层 read 返回 0 个新字节若未到 EOF把新数据追加进缓冲区无论哪种类型最终都返回缓冲区的视图。协议违规与逻辑错误处理文档明确列举均有源码对应在已到达 EOF 后调用peek、或read试图越过 EOF 读取协议违规返回EIO任何错误包括 EIO都会导致流 reset因此若继续对该流操作会触发断言并发调用读操作断言因为这是逻辑错误enqueue_some的实现也要求并发调用必须断言见 AK/AsyncStream.h对未打开的流调用peek/peek_or_eof/read逻辑错误同样断言三个方法开头均有VERIFY(is_open())。五、源码纵深AsyncInputStream 的三个钩子与派生组件5.1 三个纯虚钩子如何实现一个自定义输入流文档对应的 AK/AsyncStream.h 定义了实现新输入流必须重写的三个钩子理解了它们才能真正理解缓冲流的内部enqueue_some(BadgeAsyncInputStream)若未到 EOF从底层流至少读 1 字节进缓冲区并返回 true若已到 EOF不得改动缓冲区并返回 false。若读取失败返回 Error必须执行 Reset AO或等价操作——因此对AsyncInputStream而言一切读取错误都是致命的。这是唯一可被reset中断的方法buffered_data_unchecked(BadgeAsyncInputStream)仅返回缓冲区的视图且不得使先前返回的视图失效dequeue(BadgeAsyncInputStream, size_t bytes)从缓冲区移除bytes字节调用时保证缓冲区中确有这么多数据同样不得使先前返回的视图失效。注释还提到若使用AsyncStreamBuffer作为流缓冲区dequeue与enqueue_some将获得均摊 O(stream_length) 的复杂度。此外还有一个便捷方法read_objectT()AK/AsyncStream.h直接read(sizeof(T))后通过 union memcpy把字节重解释为类型T的对象返回——文档第一个示例中的socket.read_objectVeryImportantObject()用的正是它。5.2 输出流与全双工流AsyncOutputStreamAK/AsyncStream.h是输出侧基类核心抽象是write_some(ReadonlyBytes)默认的write会循环调用write_some直到所有缓冲写尽天然支持分片写入AsyncStream同时继承AsyncInputStream与AsyncOutputStream是全双工流的公共基类StreamWrapperTAK/AsyncStream.h把任意实现了 AsyncResource 接口的流包装成 AsyncResource 子类转发 reset/close/is_open。5.3 复合组件AsyncStreamTransform、consume_until、Slice 与 StreamPairAK/AsyncStreamTransform.h 的AsyncStreamTransformT是一个变换流它持有一个底层流和一个AK::GeneratorEmpty, ErrorOrvoid生成器把生成器每次co_yield的内容作为变换后的数据供上层读取。其close()语义很能体现 Close AO 的干净状态检查若生成器尚未结束还有数据未消费则 reset 并返回 EBUSY。AK/AsyncStreamHelpers.h 提供两个实用工具AsyncStreamHelpers::consume_until(stream, delimiter, max_size)循环 peek 直到在缓冲区中找到分隔符然后read消费到分隔符末尾支持可选的max_size上限——这是按行/按定界符解析的通用原语AsyncInputStreamSliceAK/AsyncStreamHelpers.h在流上切出一个固定长度的切片读满length字节即视为切片 EOF未消费完就 close 会返回 EBUSYAsyncStreamPairAK/AsyncStreamHelpers.h把独立的输入流与输出流组合成一个全双工AsyncStream并保证一侧出错时另一侧也被 reset如enqueue_some失败时 reset 输出流、write_some失败时 reset 输入流是错误传播联动的现成范本。六、真实案例LibHTTP 中的异步流实战异步流并非纸上谈兵Serenity OS 的 LibHTTP 客户端正是其大规模使用者是研读本文档语义的最佳配套代码。6.1 按行解析响应头consume_until 的典型应用Userland/Libraries/LibHTTP/Http11Connection.cpp 的receive_response_headers展示了peek 找定界符 → read 消费的标准姿势CoroutineErrorOrStatusCodeAndHeaders receive_response_headers(AsyncStream stream) { auto status_line CO_TRY(co_await AsyncStreamHelpers::consume_until(stream, \r\nsv)); // ... 用 GenericLexer 解析状态行失败时 stream.reset() 并返回错误 ... VectorHeader headers; while (true) { auto header StringView { CO_TRY(co_await AsyncStreamHelpers::consume_until(stream, \r\nsv)) }; if (header \r\nsv) break; // ... 解析 Header 行格式错误时 stream.reset() ... } co_return StatusCodeAndHeaders { ... }; }注意其中的错误处理模式任何解析失败都先stream.reset()再返回错误——这正是文档错误路径上必须显式 reset 非局部资源原则的忠实执行因为stream是从外部传入的引用。6.2 分块传输编码AsyncStreamTransform 的实战Userland/Libraries/LibHTTP/Http11Connection.cpp 的ChunkedBodyStream继承自AsyncStreamTransformAsyncInputStream用一个生成器实现 HTTP chunked 解码循环consume_until读块长度行、peek/read分片搬运块数据并co_yield最后校验块尾\r\n读到0长度块即结束。整个解码逻辑被封装成看起来像普通输入流的组件上层只需无脑read——这就是变换流的威力。6.3 测试验证AsyncTestStreams 与随机分片Userland/Libraries/LibTest/AsyncTestStreams.cpp 提供了两个用于测试的异步内存流AsyncMemoryInputStreamAsyncTestStreams.cpp用Vectorsize_t控制每次enqueue_some吐出的字节数其dequeue里VERIFY(m_last_enqueue m_read_head m_read_head m_peek_head)直接对长度条件做了运行时校验enqueue_some在 reset 后返回ECANCELED验证 Reset AO 对等待者的错误调度AsyncMemoryOutputStreamAsyncTestStreams.cpp在析构时通过StreamCloseExpectation断言流的最终状态到底是 Reset 还是 Close——把文档的关闭/重置卫生变成了可自动检查的测试契约。配套的 Tests/LibHTTP/TestHttp11Connection.cpp 则用Test::randomly_partition_inputAsyncTestStreams.cpp把 HTTP 响应随机切成大小不一的片再通过AsyncStreamPair喂给Http11Connection校验 chunked 解码后的 body 是否与期望一致——这相当于把任意分片下 AsyncInputStream 语义仍正确变成了自动化回归测试。七、实践要点速查不要带着打开的异步资源退出局部资源靠析构自动 reset外部传入的资源在错误路径上必须手动reset()。Close 是优雅路径Reset 是错误路径close()会检查干净状态数据未消费完会返回EBUSYreset()同步粗暴释放并向等待者返回ECANCELED。peek 不要过量s_{n-1}不得大于随后read的bytes否则违反长度条件、破坏渐进复杂度并可能误读 EOF。peek 了就要 read哪怕read(0)也行——accurate_is_eof的实现证明了 0 字节 read 也能重置 peek 状态避免后续peek升级为读取型而报 EIO。EOF 检测滞后一拍peek_or_eof第一次返回数据不带 EOF 标志下一次调用才置位用peek在 EOF 后读取是协议违规报 EIO 并 reset。错误即终点任何读错误对AsyncInputStream都是致命的enqueue_some失败必须执行 Reset AO对已关闭/已重置的流操作会断言对流的并发读操作也会断言。错误码速记意外 EOF →EIO未读完整流就 close →EBUSYreset 打断等待者 →ECANCELED。【免费下载链接】serenityThe Serenity Operating System 项目地址: https://gitcode.com/GitHub_Trending/se/serenity创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考