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 の機能：

• 複数の完了をひとつにまとめます
• すべて完了したとき（またはエラーが発生したとき）にコールバックを呼び出します
• グループをネストできます（あるグループの完了を別のグループに追加できます）
• 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!

単一の完了を中断して待ちます。

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 を返すと、呼び出し元は完了を待ってから bad_example() を再度呼び出します。counter は 2 回以上インクリメントされます。

正しい例：中断後に変更する

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/）
• 障害注入を使って、さまざまなポイントで中断を強制する

参考資料

• I/O に関する docs/manual.md のセクション