start rewrite of transaction api
This commit is contained in:
parent
897af34e37
commit
137f6cd372
178
src/lib.rs
178
src/lib.rs
@ -1,6 +1,5 @@
|
||||
use std::{
|
||||
borrow::BorrowMut,
|
||||
collections::{BTreeMap, HashMap},
|
||||
collections::{BTreeMap, VecDeque},
|
||||
fmt::Debug,
|
||||
fs::File,
|
||||
mem::size_of,
|
||||
@ -10,10 +9,14 @@ use std::{
|
||||
|
||||
mod allocator;
|
||||
mod atomic_arc;
|
||||
mod mapped;
|
||||
mod transaction;
|
||||
|
||||
use allocator::{AllocatorState, FreeList, GeneralPurposeAllocator, SlabListPointer, SlabPointer};
|
||||
use allocator::{AllocatorState, GeneralPurposeAllocator, SlabListPointer, SlabPointer};
|
||||
use atomic_arc::AtomicArc;
|
||||
use mapped::ReaderTrait;
|
||||
use memmap::{Mmap, MmapMut};
|
||||
use transaction::TransactionHandle;
|
||||
use zerocopy::{AsBytes, FromBytes, LayoutVerified, Unaligned, LE};
|
||||
|
||||
const PAGE_SIZE: u64 = 4096;
|
||||
@ -176,7 +179,7 @@ pub struct Db {
|
||||
map: MmapMut,
|
||||
slabs: BTreeMap<u32, SlabPointer>,
|
||||
state: Arc<AtomicArc<Snapshot>>,
|
||||
snapshots: Vec<SnapshotAndFreeList>,
|
||||
snapshots: VecDeque<SnapshotAndFreeList>,
|
||||
}
|
||||
|
||||
struct SnapshotAndFreeList {
|
||||
@ -184,126 +187,6 @@ struct SnapshotAndFreeList {
|
||||
to_free: Vec<FileRange>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
struct Replaced {
|
||||
from: FileRange,
|
||||
to: FileRange,
|
||||
}
|
||||
|
||||
pub struct TransactionHandle<'t> {
|
||||
db: &'t mut Db,
|
||||
replaced: HashMap<FilePointer, Replaced>,
|
||||
new: HashMap<FilePointer, FileRange>,
|
||||
}
|
||||
|
||||
impl<'t> TransactionHandle<'t> {
|
||||
pub unsafe fn modify_raw(&mut self, range: FileRange) -> (FileRange, &mut [u8]) {
|
||||
let range = if let Some(&replaced) = self.replaced.get(&range.start) {
|
||||
assert_eq!(replaced.from, range);
|
||||
replaced.to
|
||||
} else if let Some(&new) = self.new.get(&range.start) {
|
||||
assert_eq!(new, range);
|
||||
new
|
||||
} else {
|
||||
let (new, _) = self.allocate_raw(range.len());
|
||||
|
||||
self.db.copy(range, new);
|
||||
|
||||
let res = self.replaced.insert(
|
||||
new.start,
|
||||
Replaced {
|
||||
from: range,
|
||||
to: new,
|
||||
},
|
||||
);
|
||||
debug_assert!(res.is_none());
|
||||
|
||||
new
|
||||
};
|
||||
|
||||
(range, &mut self.db.map[range.as_range()])
|
||||
}
|
||||
|
||||
pub fn read_raw(&mut self, range: FileRange) -> (FileRange, &[u8]) {
|
||||
let range = if let Some(&replaced) = self.replaced.get(&range.start) {
|
||||
assert_eq!(replaced.from, range);
|
||||
replaced.to
|
||||
} else if let Some(&new) = self.new.get(&range.start) {
|
||||
assert_eq!(new, range);
|
||||
new
|
||||
} else {
|
||||
range
|
||||
};
|
||||
|
||||
(range, &self.db.map[range.as_range()])
|
||||
}
|
||||
|
||||
pub fn allocate_raw(&mut self, length: u64) -> (FileRange, &mut [u8]) {
|
||||
unsafe {
|
||||
let range = self.allocate_range(length);
|
||||
|
||||
let res = self.new.insert(range.start, range);
|
||||
debug_assert!(res.is_none());
|
||||
|
||||
(range, &mut self.db.map[range.as_range()])
|
||||
}
|
||||
}
|
||||
|
||||
pub fn modify_range<T: FromBytes + AsBytes>(
|
||||
&mut self,
|
||||
range: FileRange,
|
||||
) -> (FileRange, &mut T) {
|
||||
unsafe {
|
||||
let (ptr, _) = self.modify_raw(range);
|
||||
(ptr, self.db.modify_range(ptr))
|
||||
}
|
||||
}
|
||||
|
||||
pub fn modify<T: FromBytes + AsBytes>(&mut self, at: FilePointer) -> (FileRange, &mut T) {
|
||||
self.modify_range(at.range(size_of::<T>() as u64))
|
||||
}
|
||||
|
||||
pub fn read_range<T: FromBytes>(&mut self, range: FileRange) -> (FileRange, &T) {
|
||||
unsafe {
|
||||
let (ptr, _) = self.read_raw(range);
|
||||
(ptr, self.db.reference_range(ptr))
|
||||
}
|
||||
}
|
||||
|
||||
pub fn read<T: FromBytes>(&mut self, at: FilePointer) -> (FileRange, &T) {
|
||||
unsafe {
|
||||
let (ptr, _) = self.read_raw(at.range(size_of::<T>() as u64));
|
||||
(ptr, self.db.reference_range(ptr))
|
||||
}
|
||||
}
|
||||
|
||||
pub fn allocate_size<T: FromBytes + AsBytes>(&mut self, length: u64) -> (FileRange, &mut T) {
|
||||
unsafe {
|
||||
let (ptr, _) = self.allocate_raw(length);
|
||||
(ptr, self.db.modify_range(ptr))
|
||||
}
|
||||
}
|
||||
|
||||
pub fn allocate<T: FromBytes + AsBytes>(&mut self) -> (FileRange, &mut T) {
|
||||
unsafe {
|
||||
let (ptr, _) = self.allocate_raw(size_of::<T>() as u64);
|
||||
(ptr, self.db.modify_range(ptr))
|
||||
}
|
||||
}
|
||||
|
||||
pub fn free(&mut self, range: FileRange) {
|
||||
self.db.free(range)
|
||||
}
|
||||
|
||||
fn allocate_range(&mut self, size: u64) -> FileRange {
|
||||
self.db.allocate(size)
|
||||
}
|
||||
|
||||
fn root(&self) -> FilePointer {
|
||||
self.db.header().root
|
||||
}
|
||||
}
|
||||
|
||||
impl Db {
|
||||
fn header(&self) -> &Header {
|
||||
unsafe {
|
||||
@ -332,42 +215,35 @@ impl Db {
|
||||
}
|
||||
|
||||
fn transaction(&mut self, f: impl FnOnce(&mut TransactionHandle) -> FilePointer) {
|
||||
let mut handle = TransactionHandle {
|
||||
db: self,
|
||||
replaced: HashMap::new(),
|
||||
new: HashMap::new(),
|
||||
};
|
||||
let mut handle = TransactionHandle::new(self);
|
||||
|
||||
let root = f(&mut handle);
|
||||
|
||||
let to_free = handle
|
||||
.replaced
|
||||
.values()
|
||||
.map(|replaced| replaced.from)
|
||||
.collect();
|
||||
let to_free = handle.to_free();
|
||||
|
||||
let snapshot = self.update_root(root);
|
||||
|
||||
self.snapshots
|
||||
.push(SnapshotAndFreeList { snapshot, to_free });
|
||||
.push_back(SnapshotAndFreeList { snapshot, to_free });
|
||||
}
|
||||
|
||||
fn free_old_epochs(&mut self) {
|
||||
let mut snapshots = std::mem::take(&mut self.snapshots);
|
||||
|
||||
snapshots.retain_mut(|snapshot| {
|
||||
if Arc::get_mut(&mut snapshot.snapshot).is_some() {
|
||||
println!("freeing epoch");
|
||||
// if the snapshot is uniqe we are the only owner and can
|
||||
// free the epoch.
|
||||
for allocation in std::mem::take(&mut snapshot.to_free) {
|
||||
self.free(allocation);
|
||||
}
|
||||
false
|
||||
} else {
|
||||
true
|
||||
while snapshots
|
||||
.front_mut()
|
||||
.is_some_and(|snapshot| Arc::get_mut(&mut snapshot.snapshot).is_some())
|
||||
{
|
||||
// if the snapshot is uniqe we are the only owner and can free the epoch.
|
||||
|
||||
println!("freeing epoch");
|
||||
|
||||
let snapshot = snapshots.pop_front().unwrap();
|
||||
|
||||
for allocation in snapshot.to_free {
|
||||
self.free(allocation);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
self.snapshots = snapshots;
|
||||
}
|
||||
@ -530,7 +406,7 @@ impl Db {
|
||||
file,
|
||||
map,
|
||||
slabs: BTreeMap::new(),
|
||||
snapshots: Vec::new(),
|
||||
snapshots: VecDeque::new(),
|
||||
};
|
||||
|
||||
unsafe {
|
||||
@ -557,7 +433,7 @@ impl Db {
|
||||
file,
|
||||
map,
|
||||
slabs: BTreeMap::new(),
|
||||
snapshots: Vec::new(),
|
||||
snapshots: VecDeque::new(),
|
||||
};
|
||||
|
||||
let _ = db.state.swap(Arc::new(Snapshot {
|
||||
@ -839,7 +715,7 @@ mod tests {
|
||||
root
|
||||
};
|
||||
|
||||
let (root, &data) = transaction.read::<DataHeader>(root);
|
||||
let &data = transaction.read::<DataHeader>(root);
|
||||
|
||||
assert_eq!(data.generation.get(), i);
|
||||
|
||||
@ -848,7 +724,7 @@ mod tests {
|
||||
element.next = data.list;
|
||||
element.data = i.into();
|
||||
|
||||
let (root, data) = transaction.modify::<DataHeader>(root.start);
|
||||
let (root, data) = transaction.modify::<DataHeader>(root);
|
||||
data.list = elem_ptr.start;
|
||||
data.generation = (i + 1).into();
|
||||
root.start
|
||||
|
||||
14
src/mapped.rs
Normal file
14
src/mapped.rs
Normal file
@ -0,0 +1,14 @@
|
||||
use std::mem::size_of;
|
||||
|
||||
use zerocopy::{FromBytes, LayoutVerified};
|
||||
|
||||
use crate::{FilePointer, FileRange};
|
||||
|
||||
pub trait ReaderTrait {
|
||||
fn read_raw(&self, ptr: FileRange) -> &[u8];
|
||||
fn read<T: FromBytes>(&self, ptr: FilePointer) -> &T {
|
||||
LayoutVerified::<_, T>::new(self.read_raw(ptr.range(size_of::<T>() as u64)))
|
||||
.unwrap()
|
||||
.into_ref()
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user