async-io-model

异步 I/O 模型指南

Turso 使用协作式让出和显式状态机，而不使用 Rust async/await。

核心类型

pub enum IOCompletions {
    Single(Completion),
}

#[must_use]
pub enum IOResult<T> {
    Done(T),      // Operation complete, here's the result
    IO(IOCompletions),  // Need I/O, call me again after completions finish
}

返回 IOResult 的函数必须反复调用，直到返回 Done。

Completion 和 CompletionGroup

Completion 跟踪单个 I/O 操作：

pub struct Completion { /* ... */ }

impl Completion {
    pub fn finished(&self) -> bool;
    pub fn succeeded(&self) -> bool;
    pub fn get_error(&self) -> Option<CompletionError>;
}

要等待多个 I/O 操作，请使用 CompletionGroup：

let mut group = CompletionGroup::new(|_| {});

// Add individual completions
group.add(&completion1);
group.add(&completion2);

// Build into single completion that finishes when all complete
let combined = group.build();
io_yield_one!(combined);

CompletionGroup 的功能：

• 将多个 completion 聚合为一个
• 在所有操作完成时（或任一操作出错时）调用回调
• 可以嵌套组（将一个组的 completion 添加到另一个组）
• 可通过 group.cancel() 取消

辅助宏

return_if_io!

解包 IOResult，并将 IO 变体向上传播到调用栈：

let result = return_if_io!(some_io_operation());
// Only reaches here if operation returned Done

io_yield_one!

让出一个 completion：

io_yield_one!(completion);  // Returns Ok(IOResult::IO(Single(completion)))

状态机模式

可能让出执行的操作使用显式状态枚举：

enum MyOperationState {
    Start,
    WaitingForRead { page: PageRef },
    Processing { data: Vec<u8> },
    Done,
}

函数会循环，根据状态进行匹配并转换状态：

fn my_operation(&mut self) -> Result<IOResult<Output>> {
    loop {
        match &mut self.state {
            MyOperationState::Start => {
                let (page, completion) = start_read();
                self.state = MyOperationState::WaitingForRead { page };
                io_yield_one!(completion);
            }
            MyOperationState::WaitingForRead { page } => {
                let data = page.get_contents();
                self.state = MyOperationState::Processing { data: data.to_vec() };
                // No yield, continue loop
            }
            MyOperationState::Processing { data } => {
                let result = process(data);
                self.state = MyOperationState::Done;
                return Ok(IOResult::Done(result));
            }
            MyOperationState::Done => unreachable!(),
        }
    }
}

可重入性：关键陷阱

在让出点之前修改状态，会导致重新进入时出现错误。

错误示例

fn bad_example(&mut self) -> Result<IOResult<()>> {
    self.counter += 1;  // Mutates state
    return_if_io!(something_that_might_yield());  // If yields, re-entry will increment again!
    Ok(IOResult::Done(()))
}

如果 something_that_might_yield() 返回 IO，调用方会等待 completion，然后再次调用 bad_example()。counter 会被递增两次（或更多次）。

正确做法：在让出后修改状态

fn good_example(&mut self) -> Result<IOResult<()>> {
    return_if_io!(something_that_might_yield());
    self.counter += 1;  // Only reached once, after IO completes
    Ok(IOResult::Done(()))
}

正确做法：使用状态机

enum State { Start, AfterIO }

fn good_example(&mut self) -> Result<IOResult<()>> {
    loop {
        match self.state {
            State::Start => {
                // Don't mutate shared state here
                self.state = State::AfterIO;
                return_if_io!(something_that_might_yield());
            }
            State::AfterIO => {
                self.counter += 1;  // Safe: only entered once
                return Ok(IOResult::Done(()));
            }
        }
    }
}

常见的可重入性错误

模式	问题
vec.push(x); return_if_io!(...)	Vec 在每次重新进入时都会增长
idx += 1; return_if_io!(...)	索引会多次递增
map.insert(k,v); return_if_io!(...)	插入重复项或覆盖已有项
flag = true; return_if_io!(...)	通常没问题，但要检查逻辑

状态枚举设计

在状态变体中编码进度：

// Good: index is part of state, preserved across yields
enum ProcessState {
    Start,
    ProcessingItem { idx: usize, items: Vec<Item> },
    Done,
}

// Loop advances idx only when transitioning states
ProcessingItem { idx, items } => {
    return_if_io!(process_item(&items[idx]));
    if idx + 1 < items.len() {
        self.state = ProcessingItem { idx: idx + 1, items };
    } else {
        self.state = Done;
    }
}

Turso 实现

关键文件：

• core/types.rs - IOResult、IOCompletions、return_if_io!、return_and_restore_if_io!
• core/io/completions.rs - Completion、CompletionGroup
• core/util.rs - io_yield_one! 宏
• core/state_machine.rs - 通用 StateMachine 包装器
• core/storage/btree.rs - 多个状态机示例
• core/storage/pager.rs - CompletionGroup 用法示例

测试异步代码

可重入性错误通常只会在特定的 I/O 时序下显现。请使用：

• 确定性模拟（testing/simulator/）
• Whopper 并发 DST（testing/concurrent-simulator/）
• 故障注入，在不同位置强制让出执行

参考资料

• docs/manual.md 中关于 I/O 的章节