aljaz/turso
1
0
Fork 0
mirror of https://github.com/aljazceru/turso.git synced 2026-02-10 18:54:22 +01:00
Code Issues Packages Projects Releases Wiki Activity

core/storage/pager: Replace io_yield_many with completion groups

Browse Source
This commit is contained in:
Pekka Enberg
2025-10-13 10:47:41 +03:00
parent 07ba7276b2
commit 986faa42da
1 changed files with 18 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
core/storage/pager.rs
View File

@@ -10,11 +10,11 @@ use crate::storage::{
};
use crate::types::{IOCompletions, WalState};
use crate::util::IOExt as _;
use crate::{io_yield_many, io_yield_one, IOContext};
use crate::{
return_if_io, turso_assert, types::WalFrameInfo, Completion, Connection, IOResult, LimboError,
Result, TransactionState,
io::CompletionGroup, return_if_io, turso_assert, types::WalFrameInfo, Completion, Connection,
IOResult, LimboError, Result, TransactionState,
};
use crate::{io_yield_one, IOContext};
use parking_lot::RwLock;
use std::cell::{RefCell, UnsafeCell};
use std::collections::HashSet;
@@ -1627,7 +1627,7 @@ impl Pager {
CommitState::Checkpoint => {
match self.checkpoint()? {
IOResult::IO(cmp) => {
let completions = {
let completion = {
let mut commit_info = self.commit_info.write();
match cmp {
IOCompletions::Single(c) => {
@@ -1637,10 +1637,14 @@ impl Pager {
commit_info.completions.extend(c);
}
}
std::mem::take(&mut commit_info.completions)
let mut group = CompletionGroup::new(|_| {});
for c in commit_info.completions.drain(..) {
group.add(&c);
}
group.build()
};
// TODO: remove serialization of checkpoint path
io_yield_many!(completions);
io_yield_one!(completion);
}
IOResult::Done(res) => {
let mut commit_info = self.commit_info.write();
@@ -1679,21 +1683,25 @@ impl Pager {
.unwrap()
.as_millis()
);
let (should_finish, result, completions) = {
let (should_finish, result, completion) = {
let mut commit_info = self.commit_info.write();
if commit_info.completions.iter().all(|c| c.succeeded()) {
commit_info.completions.clear();
commit_info.state = CommitState::PrepareWal;
(true, commit_info.result.take(), Vec::new())
(true, commit_info.result.take(), Completion::new_yield())
} else {
(false, None, std::mem::take(&mut commit_info.completions))
let mut group = CompletionGroup::new(|_| {});
for c in commit_info.completions.drain(..) {
group.add(&c);
}
(false, None, group.build())
}
};
if should_finish {
wal.borrow_mut().finish_append_frames_commit()?;
return Ok(IOResult::Done(result.expect("commit result should be set")));
}
io_yield_many!(completions);
io_yield_one!(completion);
}
}
}