--- name: async-io-model description: Explanations of common asynchronous patterns used in tursodb. Involves IOResult, state machines, re-entrancy pitfalls, CompletionGroup. Always use these patterns in `core` when doing anything IO --- # Async I/O Model Guide Turso uses cooperative yielding with explicit state machines instead of Rust async/await. ## Core Types ```rust pub enum IOCompletions { Single(Completion), } #[must_use] pub enum IOResult { Done(T), // Operation complete, here's the result IO(IOCompletions), // Need I/O, call me again after completions finish } ``` Functions returning `IOResult` must be called repeatedly until `Done`. ## Completion and CompletionGroup A `Completion` tracks a single I/O operation: ```rust pub struct Completion { /* ... */ } impl Completion { pub fn finished(&self) -> bool; pub fn succeeded(&self) -> bool; pub fn get_error(&self) -> Option; } ``` To wait for multiple I/O operations, use `CompletionGroup`: ```rust 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` features: - Aggregates multiple completions into one - Calls callback when all complete (or any errors) - Can nest groups (add a group's completion to another group) - Cancellable via `group.cancel()` ## Helper Macros ### `return_if_io!` Unwraps `IOResult`, propagates IO variant up the call stack: ```rust let result = return_if_io!(some_io_operation()); // Only reaches here if operation returned Done ``` ### `io_yield_one!` Yields a single completion: ```rust io_yield_one!(completion); // Returns Ok(IOResult::IO(Single(completion))) ``` ## State Machine Pattern Operations that may yield use explicit state enums: ```rust enum MyOperationState { Start, WaitingForRead { page: PageRef }, Processing { data: Vec }, Done, } ``` The function loops, matching on state and transitioning: ```rust fn my_operation(&mut self) -> Result> { 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!(), } } } ``` ## Re-Entrancy: The Critical Pitfall **State mutations before yield points cause bugs on re-entry.** ### Wrong ```rust fn bad_example(&mut self) -> Result> { self.counter += 1; // Mutates state return_if_io!(something_that_might_yield()); // If yields, re-entry will increment again! Ok(IOResult::Done(())) } ``` If `something_that_might_yield()` returns `IO`, caller waits for completion, then calls `bad_example()` again. `counter` gets incremented twice (or more). ### Correct: Mutate After Yield ```rust fn good_example(&mut self) -> Result> { return_if_io!(something_that_might_yield()); self.counter += 1; // Only reached once, after IO completes Ok(IOResult::Done(())) } ``` ### Correct: Use State Machine ```rust enum State { Start, AfterIO } fn good_example(&mut self) -> Result> { 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(())); } } } } ``` ## Common Re-Entrancy Bugs | Pattern | Problem | |---------|---------| | `vec.push(x); return_if_io!(...)` | Vec grows on each re-entry | | `idx += 1; return_if_io!(...)` | Index advances multiple times | | `map.insert(k,v); return_if_io!(...)` | Duplicate inserts or overwrites | | `flag = true; return_if_io!(...)` | Usually ok, but check logic | ## State Enum Design Encode progress in state variants: ```rust // Good: index is part of state, preserved across yields enum ProcessState { Start, ProcessingItem { idx: usize, items: Vec }, 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 Implementation Key files: - `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!` macro - `core/state_machine.rs` - Generic `StateMachine` wrapper - `core/storage/btree.rs` - Many state machine examples - `core/storage/pager.rs` - `CompletionGroup` usage examples ## Testing Async Code Re-entrancy bugs often only manifest under specific IO timing. Use: - Deterministic simulation (`testing/simulator/`) - Whopper concurrent DST (`testing/concurrent-simulator/`) - Fault injection to force yields at different points ## References - `docs/manual.md` section on I/O

async-io-model

Scan your own AI components