aljaz/turso
1
0
Fork 0
mirror of https://github.com/aljazceru/turso.git synced 2026-02-07 09:14:26 +01:00
Code Issues Packages Projects Releases Wiki Activity

allow delete of dirty page in cacheflush

Browse Source 
Dirty pages can be deleted in `cacheflush`. Furthermore, there could be
multiple live references in the stack of a cursor so let's allow them to
exist while deleting.
This commit is contained in:
Pere Diaz Bou
2025-05-19 11:07:30 +02:00
parent e2f99a1ad2
commit 67e260ff71
4 changed files with 53 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
core/storage/btree.rs
View File

@@ -2815,7 +2815,7 @@ impl BTreeCursor {
if *new_id != page.get().id {
page.get().id = *new_id;
// FIXME: why would there be another page at this id/frame?
self.pager.put_loaded_page(*new_id, page.clone());
self.pager.put_loaded_page(*new_id, page.clone())?;
}
}

23
core/storage/page_cache.rs
View File

@@ -26,12 +26,6 @@ struct PageCacheEntry {
next: Option<NonNull<PageCacheEntry>>,
}
impl PageCacheEntry {
fn as_non_null(&mut self) -> NonNull<PageCacheEntry> {
NonNull::new(&mut *self).unwrap()
}
}
pub struct DumbLruPageCache {
capacity: usize,
map: RefCell<HashMap<PageCacheKey, NonNull<PageCacheEntry>>>,
@@ -166,18 +160,9 @@ impl DumbLruPageCache {
}
if clean_page {
if let Some(page_mut) = Arc::get_mut(&mut entry_mut.page) {
page_mut.clear_loaded();
debug!("cleaning up page {}", page_mut.get().id);
let _ = page_mut.get().contents.take();
} else {
let page_id = unsafe { &entry.as_mut().page.get().id };
debug!(
"detach page {}: can't clean, there are other references",
page_id
);
return Err(CacheError::ActiveRefs);
}
entry_mut.page.clear_loaded();
debug!("cleaning up page {}", entry_mut.page.get().id);
let _ = entry_mut.page.get().contents.take();
}
self.unlink(entry);
Ok(())
@@ -950,6 +935,7 @@ mod tests {
}
#[test]
#[should_panic(expected = "Attempted to insert different page with same key")]
fn test_resize_larger() {
let mut cache = DumbLruPageCache::new(2);
let _ = insert_page(&mut cache, 1);
@@ -965,6 +951,7 @@ mod tests {
let _ = insert_page(&mut cache, i);
}
assert_eq!(cache.len(), 5);
// FIXME: For now this will assert because we cannot insert a page with same id but different contents of page.
assert!(cache.insert(create_key(4), page_with_content(4)).is_err());
cache.verify_list_integrity();
}

59
core/storage/pager.rs
View File

@@ -13,7 +13,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tracing::trace;
use super::page_cache::{CacheError, DumbLruPageCache, PageCacheKey, CacheResizeResult};
use super::page_cache::{CacheError, CacheResizeResult, DumbLruPageCache, PageCacheKey};
use super::wal::{CheckpointMode, CheckpointStatus};
pub struct PageInner {
@@ -333,8 +333,15 @@ impl Pager {
match page_cache.insert(page_key, page.clone()) {
Ok(_) => {}
Err(CacheError::Full) => return Err(LimboError::CacheFull),
Err(CacheError::KeyExists) => unreachable!("Page should not exist in cache after get() miss"),
Err(e) => return Err(LimboError::InternalError(format!("Failed to insert page into cache: {:?}", e))),
Err(CacheError::KeyExists) => {
unreachable!("Page should not exist in cache after get() miss")
}
Err(e) => {
return Err(LimboError::InternalError(format!(
"Failed to insert page into cache: {:?}",
e
)))
}
}
return Ok(page);
}
@@ -348,8 +355,15 @@ impl Pager {
match page_cache.insert(page_key, page.clone()) {
Ok(_) => {}
Err(CacheError::Full) => return Err(LimboError::CacheFull),
Err(CacheError::KeyExists) => unreachable!("Page should not exist in cache after get() miss"),
Err(e) => return Err(LimboError::InternalError(format!("Failed to insert page into cache: {:?}", e))),
Err(CacheError::KeyExists) => {
unreachable!("Page should not exist in cache after get() miss")
}
Err(e) => {
return Err(LimboError::InternalError(format!(
"Failed to insert page into cache: {:?}",
e
)))
}
}
Ok(page)
}
@@ -373,9 +387,14 @@ impl Pager {
page.set_uptodate();
}
match page_cache.insert(page_key, page.clone()) {
Err(CacheError::KeyExists) => {}, // Exists but same page, not error
Err(CacheError::KeyExists) => {} // Exists but same page, not error
Err(CacheError::Full) => return Err(LimboError::CacheFull),
Err(e) => return Err(LimboError::InternalError(format!("Failed to insert page into cache during load: {:?}", e))),
Err(e) => {
return Err(LimboError::InternalError(format!(
"Failed to insert page into cache during load: {:?}",
e
)))
}
Ok(_) => {}
}
return Ok(());
@@ -383,10 +402,15 @@ impl Pager {
}
match page_cache.insert(page_key, page.clone()) {
Err(CacheError::KeyExists) => {}, // Ensures same page
Err(CacheError::Full) => return Err(LimboError::CacheFull),
Err(e) => return Err(LimboError::InternalError(format!("Failed to insert page into cache during load: {:?}", e))),
Ok(_) => {}
Err(CacheError::KeyExists) => {} // Ensures same page
Err(CacheError::Full) => return Err(LimboError::CacheFull),
Err(e) => {
return Err(LimboError::InternalError(format!(
"Failed to insert page into cache during load: {:?}",
e
)))
}
Ok(_) => {}
};
sqlite3_ondisk::begin_read_page(
@@ -440,8 +464,8 @@ impl Pager {
for page_id in self.dirty_pages.borrow().iter() {
let mut cache = self.page_cache.write();
let page_key = PageCacheKey::new(*page_id, Some(max_frame));
let page = cache.get(&page_key).expect("we somehow added a page to dirty list but we didn't mark it as dirty, causing cache to drop it.");
if let Some(wal) = &self.wal {
let page = cache.get(&page_key).expect("we somehow added a page to dirty list but we didn't mark it as dirty, causing cache to drop it.");
let page_type = page.get().contents.as_ref().unwrap().maybe_page_type();
trace!("cacheflush(page={}, page_type={:?}", page_id, page_type);
wal.borrow_mut().append_frame(
@@ -450,6 +474,7 @@ impl Pager {
self.flush_info.borrow().in_flight_writes.clone(),
)?;
}
page.clear_dirty();
// This page is no longer valid.
// For example:
// We took page with key (page_num, max_frame) -- this page is no longer valid for that max_frame
@@ -579,7 +604,10 @@ impl Pager {
}
}
// TODO: only clear cache of things that are really invalidated
self.page_cache.write().clear().expect("Failed to clear page cache");
self.page_cache
.write()
.clear()
.expect("Failed to clear page cache");
checkpoint_result
}
@@ -731,7 +759,8 @@ impl Pager {
cache.insert(page_key, page.clone()).map_err(|e| {
LimboError::InternalError(format!(
"Failed to insert loaded page {} into cache: {:?}", id, e
"Failed to insert loaded page {} into cache: {:?}",
id, e
))
})?;
@@ -807,7 +836,7 @@ mod tests {
std::thread::spawn(move || {
let mut cache = cache.write();
let page_key = PageCacheKey::new(1, None);
cache.insert(page_key, Arc::new(Page::new(1)));
cache.insert(page_key, Arc::new(Page::new(1))).unwrap();
})
};
let _ = thread.join();

4
core/translate/pragma.rs
View File

@@ -305,5 +305,7 @@ fn update_cache_size(value: i64, header: Arc<SpinLock<DatabaseHeader>>, pager: R
pager.write_database_header(&header_copy);
// update cache size
pager.change_page_cache_size(cache_size);
pager
.change_page_cache_size(cache_size)
.expect("couldn't update page cache size");
}