Loading...
Loading...
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
npx skill4agent add tursodatabase/turso async-io-modelpub 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
}IOResultDoneCompletionpub struct Completion { /* ... */ }
impl Completion {
pub fn finished(&self) -> bool;
pub fn succeeded(&self) -> bool;
pub fn get_error(&self) -> Option<CompletionError>;
}CompletionGrouplet 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);CompletionGroupgroup.cancel()return_if_io!IOResultlet result = return_if_io!(some_io_operation());
// Only reaches here if operation returned Doneio_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()IObad_example()counterfn 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(()));
}
}
}
}| Pattern | Problem |
|---|---|
| Vec grows on each re-entry |
| Index advances multiple times |
| Duplicate inserts or overwrites |
| Usually ok, but check logic |
// 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;
}
}core/types.rsIOResultIOCompletionsreturn_if_io!return_and_restore_if_io!core/io/completions.rsCompletionCompletionGroupcore/util.rsio_yield_one!core/state_machine.rsStateMachinecore/storage/btree.rscore/storage/pager.rsCompletionGrouptesting/simulator/testing/concurrent-simulator/docs/manual.md