企业官网建设流程全解析

• 数据存储
• 后端

本文基于 OpenDAL 仓库中的 RFC 文档 0191_async_streaming_io.md，完整解读这份 2022 年提出、主张用Stream/Sink替代AsyncRead的异步流式 IO 提案：其动机、API 设计、基准数据与被否决的方案，并结合当前仓库源码，说明该提案的核心思想如何演化为今天Service/oio的读写抽象与into_bytes_stream等用户侧 API。读完后你可以理解 OpenDAL 读写路径从“包装 Reader”到“返回流式句柄”的架构演进逻辑。

一、提案背景：Accessor的 read/write 为何成为瓶颈

RFC 0191（Proposal Name:async_streaming_io，开始于 2022-03-28，对应 PR #191 / Tracking Issue #190）针对的痛点是：Accessor定位为“所有后端实现者要面对的底层 trait”，但它返回的BoxedAsyncReader包装层次过多，且读写行为不对称：

pub type BoxedAsyncReader = Box<dyn AsyncRead + Unpin + Send>; pub trait Accessor { async fn read(&self, args: &OpRead) -> Result<BoxedAsyncReader> { let _ = args; unimplemented!() } }

1.1 过度包装：一次 poll_read 要穿四层

提案给出的核心论据是调用栈深度。当时 OpenDAL 对最终用户暴露实现AsyncRead与AsyncSeek的Reader，而每一次Reader::poll_read()调用都要经历：

• Reader::poll_read()
• BoxedAsyncReader::poll_read()
• IntoAsyncRead<ByteStream>::poll_read()
• ByteStream::poll_next()

也就是说，AsyncRead与底层Stream之间夹着IntoAsyncRead这样的适配器。如果Accessor::read直接返回Stream，调用栈可以压缩为：

• Reader::poll_read()
• ByteStream::poll_next()

代价是：直接操作底层 IO 流后，调用者必须自己跟踪读取状态。但换来的是更短、更可控的路径。

1.2 读写行为不对称

提案指出的第二个问题是read与write语义不一致：

pub type BoxedAsyncReader = Box<dyn AsyncRead + Unpin + Send>; pub trait Accessor: Send + Sync + Debug { async fn read(&self, args: &OpRead) -> Result<BoxedAsyncReader> { let _ = args; unimplemented!() } async fn write(&self, r: BoxedAsyncReader, args: &OpWrite) -> Result<usize> { let (_, _) = (r, args); unimplemented!() } }

read返回一个BoxedAsyncReader，用户可以自行决定何时、以何种方式消费数据；而write却要求用户传入一个BoxedAsyncReader，写入逻辑完全由 OpenDAL 接管——写缓冲多大？何时flush？用户都无从控制。这种不对称意味着Accessor作为“底层 trait”不够底层。

1.3 服务侧是流式的，API 却无法表达

第三个动机来自服务本身的特性。OpenDAL 对存储服务的细节（传输协议、缓冲策略）比用户更了解，但返回BoxedAsyncReader使这些优势无法释放。以对象存储为例，大多数服务基于 HTTP 传输数据，而 HTTP 底层是 TCP 流。最高效的方式是整块消费 TCP 缓冲，但用户并不知道这一点，可能会发起一连串小读；为了弥补短板，用户只能自己套一层BufReader增加缓冲，却不知道“最佳”缓冲尺寸是多少。

提案认为直接返回Stream可以两头讨好：

• 想要底层控制的用户可以直接操作Stream；
• 不关心细节的用户则继续使用 OpenDAL 提供的Reader，由它始终采用最优优化。

二、提案设计：Stream/Sink 作为 IO 的基座

2.1 指南级 API：Object::stream与Object::sink

RFC 计划在async_streaming_iofeature 下为Object增加两个新 API：

impl Object { pub async fn stream(&self, offset: Option<u64>, size: Option<u64>) -> Result<BytesStream> {} pub async fn sink(&self, size: u64) -> Result<BytesSink> {} }

这样用户就能直接控制字节的底层流转：

• 按需读：stream.next().await
• 按需写：sink.feed(bs).await; sink.close().await;

在此基座之上，Object再提供优化后的便捷函数，例如：

• async read(offset: Option<u64>, size: Option<u64>) -> Result<bytes::Bytes>
• async write(bs: bytes::Bytes) -> Result<()>

这正是 RFC 中“服务侧简单、辅助逻辑上移到 Object 层”的分层思想（后文第三、五节可见这一思路在现行代码中的体现）。

2.2 参考级设计：Accessor 的流式化与 BodySinker

参考级设计将Accessor的read/write重构为基于流的形态：

pub type BytesStream = Box<dyn Stream + Unpin + Send>; pub type BytesSink = Box<dyn Sink + Unpin + Send>; pub trait Accessor: Send + Sync + Debug { async fn read(&self, args: &OpRead) -> Result<BytesStream> { let _ = args; unimplemented!() } async fn write(&self, args: &OpWrite) -> Result<BytesSink> { let _ = args; unimplemented!() } }

其余 IO 函数则随这一变化适配。各后端的实现策略在 RFC 中分两类：

• fs 这类本地后端：直接为tokio::fs::File实现Stream和Sink，实现上是简单直接的；
• HTTP 系后端：统一返回BodySinker，其内部持有一个由hyper构造的put_objectResponseFuture，以及一个 channel 的sender端。用户发送的所有数据都通过无缓冲（unbuffered）channel转发给ResponseFuture：

struct BodySinker { fut: ResponseFuture, sender: Sender<bytes::Bytes> }

值得注意的是，这里写路径返回的是一个“已建立、可喂数据的 sink 句柄”，用户掌控喂数据的节奏——这恰好修复了 1.2 节指出的读写不对称问题。

三、代价与验证：RFC 附带的基准数据

3.1 缺点：fs 上约 70% 的纯内存开销

RFC 明确承认fs不是流式后端，把Reader转成Stream并非零成本。基于IntoStream的纯内存基准，性能损耗接近 70%：

into_stream/into_stream time: [1.3046 ms 1.3056 ms 1.3068 ms] thrpt: [2.9891 GiB/s 2.9919 GiB/s 2.9942 GiB/s] into_stream/raw_reader time: [382.10 us 383.52 us 385.16 us] thrpt: [10.142 GiB/s 10.185 GiB/s 10.223 GiB/s]

提案的判断是：真实磁盘远慢于内存，主要开销落在磁盘侧，因此这一回归在当时可以接受。这是理解 OpenDAL“先追求架构一致性，再用基准证明热路径不劣化”这种工程取舍的一个典型样本。

3.2 读路径：Reader → Stream 的整体基准

read_full在改造前后的对比（注意统计显著性：p < 0.05 才算有显著变化）：

小文件区间基本持平，中大文件区间有统计意义上约 8%–19% 的提升，说明“更薄的读路径”在数据量增大时开始兑现收益。

3.3 写路径：额外 channel 反而略有收益

BodySinker引入的无缓冲 channel 曾被担心成为额外开销，但write_once基准显示它在多个档位反而有提升：

3.4 “服务侧复杂度增加”的回应

RFC 预判了“返回 Stream/Sink 会让服务实现更复杂”的质疑，并给出回应：乍看如此，实则不然。关键前提是 HTTP（尤其是 hyper）本身就是流式的：

• 返回stream比返回reader更直接；
• 返回Sink由全局共享的BodySinker统一覆盖；
• 其余辅助函数由 Object 层承担，服务实现者无需关心。

四、被否决的替代方案（Prior Art）

RFC 系统性地梳理并否决了三种“看起来更自然”的 API 形态，其论证过程对理解流式 IO 抽象的取舍很有价值。

4.1 返回BoxedAsyncWriter

最直接的扩展是让write返回BoxedAsyncWriter：

pub trait Accessor: Send + Sync + Debug { /// Read data from the underlying storage into input writer. async fn read(&self, args: &OpRead) -> Result<BoxedAsyncReader> { let _ = args; unimplemented!() } /// Write data from input reader to the underlying storage. async fn write(&self, args: &OpWrite) -> Result<BoxedAsyncWriter> { let _ = args; unimplemented!() } }

它只解决了“读写不对称”这一个痛点，对过度包装和服务侧优化空间两个问题没有帮助。

4.2 基于切片（slice）的 API

Rust 标准 IO API 大多基于切片，因此也可以设想：

pub trait Accessor: Send + Sync + Debug { async fn read(&self, args: &OpRead, bs: &mut [u8]) -> Result<usize> { let _ = args; unimplemented!() } async fn write(&self, args: &OpWrite, bs: &[u8]) -> Result<usize> { let _ = args; unimplemented!() } }

问题在于Accessor是无状态的：若要求数据一次性传入，无法支持大文件读写；若允许多次调用，则等于要在Accessor外再实现一套Reader/Writer式的状态机——绕回原点。

4.3 接受Reader/Writer作为参数

pub trait Accessor: Send + Sync + Debug { async fn read(&self, args: &OpRead, w: BoxedAsyncWriter) -> Result<usize> { let _ = args; unimplemented!() } async fn write(&self, args: &OpWrite, r: BoxedAsyncReader) -> Result<usize> { let _ = args; unimplemented!() } }

这一设计覆盖了前述全部担忧，但实际使用困难：首先，futures::AsyncRead与tokio::AsyncRead无法同时兼容；其次，比如无法接受一个Box::new(Vec::new())，用户也拿不回这个 Vec。数据的所有权与生命周期问题无法回避。

五、提案的结局与思想在现行代码中的延续

必须如实指出：RFC 文档头部标注了Reverted——即该提案所规划的stream/sink特性并未以原样保留。但“让后端返回流式/下沉句柄，而不是返回一个包装了 AsyncRead 的 Box”这一核心方向，在 OpenDAL 后续架构中得到了更彻底的实现。从源码结构看，当前的证据链如下：

5.1Servicetrait：read/write 返回句柄而非数据

现行核心中，Accessor已被Service取代（见 accessor.rs）。其read/write的形态与 RFC 的参考级设计同构——同步返回一个可后续喂取/写入的句柄，并返回Result<Self::Reader>/Result<Self::Writer>：

fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader>; fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer>;

其中Reader: oio::Read、Writer: oio::Write由各服务通过关联类型自行定义，层与 Operator 则通过ServiceDyn（同文件 L283 起）以类型擦除方式组合。RFC 中“后端负责流式基座、上层负责辅助语义”的分层在Service上完整继承。

5.2oio::Read：open(range) 返回流

raw/oio/read/api.rs 中，内部读抽象Read的核心方法正是 RFC 想要的那条流式路径：

pub trait Read: Unpin + Send + Sync { /// Open a range stream for the given range. fn open(&self, range: BytesRange) -> impl Future<Output = Result<(RpRead, Box<dyn ReadStreamDyn>)>> + MaybeSend; /// Read an exact bounded range into [`Buffer`]. fn read(&self, range: BytesRange) -> impl Future<Output = Result<(RpRead, Buffer)>> + MaybeSend; }

open返回的ReadStreamDyn提供read() -> Result<Buffer>（逐块取数）与默认的read_all()（循环取到空 Buffer 为止，L109-L127）。这正是当年BytesStream的对应物：数据以“块（Buffer）”为单位沿流前进，读侧状态由句柄自持。

5.3StreamRead/StreamReader：只暴露原生 range 流即可

stream_read.rs 提供了一个适配器，允许服务只实现“按 range 打开原生流”这一件事：

pub trait StreamRead: Send + Sync + Unpin + 'static { fn open(&self, range: BytesRange) -> impl Future<Output = Result<(RpRead, Box<dyn oio::ReadStreamDyn>)>> + MaybeSend; }

StreamReader<R>会自动补全oio::Read的完整契约——read(range)由open(range)加read_all实现，并严格校验实际读到的长度与请求范围一致（L59-L75）。这对应了 RFC 第 3.4 节的思路：把“流式读取”的复杂性收敛到一处，服务实现只保留最有信息量的open(range)。仓库中 services/memory/backend.rs 等服务即基于该模式接入；而 HTTP 系服务（如 services/s3、services/azblob 等，各自在src/reader.rs中）则是 RFC 中“HTTP 后端统一流式处理”思想的落地版本——每个后端提供自己的ReadStream实现。

5.4oio::Write：feed + close 的 Sink 语义

raw/oio/write/api.rs 中的Writetrait 在语义上就是当年BytesSink的继承：

pub trait Write: Unpin + Send + Sync { fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<()>> + MaybeSend; fn close(&mut self) -> impl Future<Output = Result<Metadata>> + MaybeSend; fn abort(&mut self) -> impl Future<Output = Result<()>> + MaybeSend; }

注意write(bs)的契约：Ok(())意味着全部字节被接受，实现必须对部分写入返回错误（L29-L33）——这正是 RFC 中sink.feed(bs).await; sink.close().await的形态：用户掌握喂数节奏，close完成提交（并回传Metadata），abort放弃未提交数据。Service::write返回Self::Writer句柄（accessor.rs L181）而非直接消费数据，读写两侧的对称性问题由此在架构层面消除。

5.5 用户侧：into_bytes_stream兑现“Stream 给控制，Reader 给便利”

RFC 给最终用户开出的两张牌——“想控制就用 Stream，不想控制就用 Reader”——在现行 API 中完整保留。Reader同时提供两条出口（见 types/read/reader.rs）：

/// 转换为实现 futures::Stream 的 FuturesBytesStream pub async fn into_bytes_stream( self, range: impl Into<BytesRange>, ) -> Result<FuturesBytesStream>; /// 转换为 futures::io::AsyncRead pub async fn into_futures_async_read( self, range: impl Into<BytesRange>, ) -> Result<FuturesAsyncReader>;

典型用法（取自源码中的文档示例）：

use bytes::Bytes; use futures::TryStreamExt; use opendal_core::Operator; async fn test(op: Operator) -> std::io::Result<()> { let mut s = op .reader("hello.txt") .await? .into_bytes_stream(1024..2048) .await?; let bs: Vec<Bytes> = s.try_collect().await?; Ok(()) }

其中FuturesBytesStream（futures_bytes_stream.rs）是对内部块流的futures::Stream适配器，底层由 buffer_stream.rs 中的BufferStream负责按块产出bytes::Bytes。也就是说，RFC 时代的BytesStream概念如今以FuturesBytesStream/BufferStream的形式存在于用户可见层，而服务侧则以oio::ReadStream/ReadStreamDyn的形式存在于内部层——提案的“双层结构”（服务返回流，Object 层包装出便捷与流式两种消费方式）在代码中得到了比原提案更清晰的实现。

六、结语：从一份被 Reverted 的 RFC 看 API 演进的判定标准

回顾 0191_async_streaming_io.md 的价值，不在于它的stream/sink特性原样上线，而在于它确立了 OpenDAL 读写 API 的三条判定标准，这些标准在此后的架构中一以贯之：

1. 底层 trait 必须比 AsyncRead 更底层：Service::read返回oio::Read句柄、Service::write返回oio::Write句柄，避免中间夹一层适配器；
2. 读写必须对称：读是“open 一个流、逐块 read”，写是“write 一块、close 提交”，状态都由句柄持有；
3. 复杂度分层收敛：服务侧只暴露最有信息量的原生操作（open(range)、write(bs)/close()），通用语义由StreamReader之类的适配器和上层Reader/Operator补齐。

对阅读 OpenDAL 源码的开发者，这份 RFC 与当前 raw/oio 目录对照阅读，是理解其 IO 抽象从提案到落地最短的一条路径；文档中的基准数据（fs 70% 纯内存损耗、读路径 16 MiB 提升 8%–19%、写路径 channel 非负收益）则是评估“架构一致性 vs. 热路径开销”时可直接引用的实证材料。

• 数据存储
• 后端