use crate::db::TransactionGuard;
use crate::error::CommitError;
use crate::multimap_table::ReadOnlyUntypedMultimapTable;
use crate::sealed::Sealed;
use crate::sync::Mutex;
use crate::table::ReadOnlyUntypedTable;
use crate::transaction_tracker::{SavepointId, TransactionId, TransactionTracker};
#[cfg(all(debug_assertions, not(redb_no_std)))]
use crate::tree_store::PageNumberHashSet;
use crate::tree_store::{
AllocationPolicy, Btree, BtreeHeader, BtreeMut, InternalTableDefinition, MAX_PAIR_LENGTH,
MAX_VALUE_LENGTH, Page, PageAllocator, PageHint, PageListMut, PageNumber, PageNumberHashMap,
PageResolver, PageTracker, SerializedSavepoint, ShrinkPolicy, TableTree, TableTreeMut,
TableType, TransactionalMemory,
};
use crate::types::{Key, Value};
use crate::{
AccessGuard, AccessGuardMutInPlace, ExtractIf, MultimapTable, MultimapTableDefinition,
MultimapTableHandle, MutInPlaceValue, Range, ReadOnlyMultimapTable, ReadOnlyTable, Result,
Savepoint, SavepointError, SetDurabilityError, StorageError, Table, TableDefinition,
TableError, TableHandle, TransactionError, TypeName, UntypedMultimapTableHandle,
UntypedTableHandle,
};
use alloc::boxed::Box;
use alloc::collections::{BTreeMap, BTreeSet};
use alloc::string::String;
use alloc::string::ToString;
use alloc::sync::Arc;
use alloc::vec;
use alloc::vec::Vec;
use core::borrow::Borrow;
use core::cmp::min;
use core::fmt::{Debug, Display, Formatter};
use core::marker::PhantomData;
use core::mem;
use core::mem::size_of;
use core::ops::{RangeBounds, RangeFull};
use core::panic;
use core::sync::atomic::{AtomicBool, Ordering};
#[cfg(feature = "logging")]
use log::{debug, warn};
const MAX_PAGES_PER_COMPACTION: usize = 1_000_000;
const NEXT_SAVEPOINT_TABLE: SystemTableDefinition<(), SavepointId> =
SystemTableDefinition::new("next_savepoint_id");
pub(crate) const SAVEPOINT_TABLE: SystemTableDefinition<SavepointId, SerializedSavepoint> =
SystemTableDefinition::new("persistent_savepoints");
pub(crate) const DATA_ALLOCATED_TABLE: SystemTableDefinition<
TransactionIdWithPagination,
PageList,
> = SystemTableDefinition::new("data_pages_allocated");
pub(crate) const DATA_FREED_TABLE: SystemTableDefinition<TransactionIdWithPagination, PageList> =
SystemTableDefinition::new("data_pages_unreachable");
pub(crate) const SYSTEM_FREED_TABLE: SystemTableDefinition<TransactionIdWithPagination, PageList> =
SystemTableDefinition::new("system_pages_unreachable");
pub(crate) const ALLOCATOR_STATE_TABLE_NAME: &str = "allocator_state";
pub(crate) type AllocatorStateTree = Btree<AllocatorStateKey, &'static [u8]>;
pub(crate) type AllocatorStateTreeMut = BtreeMut<AllocatorStateKey, &'static [u8]>;
pub(crate) type SystemFreedTree = BtreeMut<TransactionIdWithPagination, PageList<'static>>;
#[derive(Debug)]
pub(crate) struct PageList<'a> {
data: &'a [u8],
}
impl PageList<'_> {
fn required_bytes(len: usize) -> usize {
2 + PageNumber::serialized_size() * len
}
pub(crate) fn len(&self) -> usize {
u16::from_le_bytes(self.data[..size_of::<u16>()].try_into().unwrap()).into()
}
pub(crate) fn get(&self, index: usize) -> PageNumber {
let start = size_of::<u16>() + PageNumber::serialized_size() * index;
PageNumber::from_le_bytes(
self.data[start..(start + PageNumber::serialized_size())]
.try_into()
.unwrap(),
)
}
}
impl Value for PageList<'_> {
type SelfType<'a>
= PageList<'a>
where
Self: 'a;
type AsBytes<'a>
= &'a [u8]
where
Self: 'a;
fn fixed_width() -> Option<usize> {
None
}
fn from_bytes<'a>(data: &'a [u8]) -> Self::SelfType<'a>
where
Self: 'a,
{
PageList { data }
}
fn as_bytes<'a, 'b: 'a>(value: &'a Self::SelfType<'b>) -> &'b [u8]
where
Self: 'b,
{
value.data
}
fn type_name() -> TypeName {
TypeName::internal("redb::PageList")
}
}
impl MutInPlaceValue for PageList<'_> {
type BaseRefType = PageListMut;
fn initialize(data: &mut [u8]) {
assert!(data.len() >= 8);
data[..8].fill(0);
}
fn from_bytes_mut(data: &mut [u8]) -> &mut Self::BaseRefType {
unsafe { &mut *(core::ptr::from_mut::<[u8]>(data) as *mut PageListMut) }
}
}
#[derive(Debug)]
pub(crate) struct TransactionIdWithPagination {
pub(crate) transaction_id: u64,
pub(crate) pagination_id: u64,
}
impl Value for TransactionIdWithPagination {
type SelfType<'a>
= TransactionIdWithPagination
where
Self: 'a;
type AsBytes<'a>
= [u8; 2 * size_of::<u64>()]
where
Self: 'a;
fn fixed_width() -> Option<usize> {
Some(2 * size_of::<u64>())
}
fn from_bytes<'a>(data: &'a [u8]) -> Self
where
Self: 'a,
{
let transaction_id = u64::from_le_bytes(data[..size_of::<u64>()].try_into().unwrap());
let pagination_id = u64::from_le_bytes(data[size_of::<u64>()..].try_into().unwrap());
Self {
transaction_id,
pagination_id,
}
}
fn as_bytes<'a, 'b: 'a>(value: &'a Self::SelfType<'b>) -> [u8; 2 * size_of::<u64>()]
where
Self: 'b,
{
let mut result = [0u8; 2 * size_of::<u64>()];
result[..size_of::<u64>()].copy_from_slice(&value.transaction_id.to_le_bytes());
result[size_of::<u64>()..].copy_from_slice(&value.pagination_id.to_le_bytes());
result
}
fn type_name() -> TypeName {
TypeName::internal("redb::TransactionIdWithPagination")
}
}
impl Key for TransactionIdWithPagination {
fn compare(data1: &[u8], data2: &[u8]) -> core::cmp::Ordering {
let value1 = Self::from_bytes(data1);
let value2 = Self::from_bytes(data2);
match value1.transaction_id.cmp(&value2.transaction_id) {
core::cmp::Ordering::Greater => core::cmp::Ordering::Greater,
core::cmp::Ordering::Equal => value1.pagination_id.cmp(&value2.pagination_id),
core::cmp::Ordering::Less => core::cmp::Ordering::Less,
}
}
}
#[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Debug)]
pub(crate) enum AllocatorStateKey {
Deprecated,
Region(u32),
RegionTracker,
TransactionId,
}
impl Value for AllocatorStateKey {
type SelfType<'a> = Self;
type AsBytes<'a> = [u8; 1 + size_of::<u32>()];
fn fixed_width() -> Option<usize> {
Some(1 + size_of::<u32>())
}
fn from_bytes<'a>(data: &'a [u8]) -> Self::SelfType<'a>
where
Self: 'a,
{
match data[0] {
0..=2 => Self::Deprecated,
3 => Self::Region(u32::from_le_bytes(data[1..].try_into().unwrap())),
4 => Self::RegionTracker,
5 => Self::TransactionId,
_ => unreachable!(),
}
}
fn as_bytes<'a, 'b: 'a>(value: &'a Self::SelfType<'b>) -> Self::AsBytes<'a>
where
Self: 'a,
Self: 'b,
{
let mut result = Self::AsBytes::default();
match value {
Self::Region(region) => {
result[0] = 3;
result[1..].copy_from_slice(&u32::to_le_bytes(*region));
}
Self::RegionTracker => {
result[0] = 4;
}
Self::TransactionId => {
result[0] = 5;
}
AllocatorStateKey::Deprecated => {
result[0] = 0;
}
}
result
}
fn type_name() -> TypeName {
TypeName::internal("redb::AllocatorStateKey")
}
}
impl Key for AllocatorStateKey {
fn compare(data1: &[u8], data2: &[u8]) -> core::cmp::Ordering {
Self::from_bytes(data1).cmp(&Self::from_bytes(data2))
}
}
pub struct SystemTableDefinition<'a, K: Key + 'static, V: Value + 'static> {
name: &'a str,
_key_type: PhantomData<K>,
_value_type: PhantomData<V>,
}
impl<'a, K: Key + 'static, V: Value + 'static> SystemTableDefinition<'a, K, V> {
pub const fn new(name: &'a str) -> Self {
assert!(!name.is_empty());
Self {
name,
_key_type: PhantomData,
_value_type: PhantomData,
}
}
}
impl<K: Key + 'static, V: Value + 'static> TableHandle for SystemTableDefinition<'_, K, V> {
fn name(&self) -> &str {
self.name
}
}
impl<K: Key, V: Value> Sealed for SystemTableDefinition<'_, K, V> {}
impl<K: Key + 'static, V: Value + 'static> Clone for SystemTableDefinition<'_, K, V> {
fn clone(&self) -> Self {
*self
}
}
impl<K: Key + 'static, V: Value + 'static> Copy for SystemTableDefinition<'_, K, V> {}
impl<K: Key + 'static, V: Value + 'static> Display for SystemTableDefinition<'_, K, V> {
fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
write!(
f,
"{}<{}, {}>",
self.name,
K::type_name().name(),
V::type_name().name()
)
}
}
#[derive(Debug)]
pub struct DatabaseStats {
pub(crate) tree_height: u32,
pub(crate) allocated_pages: u64,
pub(crate) leaf_pages: u64,
pub(crate) branch_pages: u64,
pub(crate) stored_leaf_bytes: u64,
pub(crate) metadata_bytes: u64,
pub(crate) fragmented_bytes: u64,
pub(crate) page_size: usize,
}
impl DatabaseStats {
pub fn tree_height(&self) -> u32 {
self.tree_height
}
pub fn allocated_pages(&self) -> u64 {
self.allocated_pages
}
pub fn leaf_pages(&self) -> u64 {
self.leaf_pages
}
pub fn branch_pages(&self) -> u64 {
self.branch_pages
}
pub fn stored_bytes(&self) -> u64 {
self.stored_leaf_bytes
}
pub fn metadata_bytes(&self) -> u64 {
self.metadata_bytes
}
pub fn fragmented_bytes(&self) -> u64 {
self.fragmented_bytes
}
pub fn page_size(&self) -> usize {
self.page_size
}
}
#[derive(Copy, Clone, Debug)]
#[non_exhaustive]
pub enum Durability {
None,
Immediate,
}
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
enum InternalDurability {
None,
Immediate,
}
#[derive(Copy, Clone, Eq, PartialEq)]
enum PostCommitFree {
Enabled,
Disabled,
}
pub struct SystemTable<'s, K: Key + 'static, V: Value + 'static> {
name: String,
namespace: &'s mut SystemNamespace,
tree: BtreeMut<K, V>,
transaction_guard: Arc<TransactionGuard>,
}
impl<'s, K: Key + 'static, V: Value + 'static> SystemTable<'s, K, V> {
fn new(
name: &str,
table_root: Option<BtreeHeader>,
freed_pages: Arc<Mutex<Vec<PageNumber>>>,
guard: Arc<TransactionGuard>,
page_allocator: PageAllocator,
namespace: &'s mut SystemNamespace,
) -> SystemTable<'s, K, V> {
let ignore = Arc::new(PageTracker::ignore());
SystemTable {
name: name.to_string(),
namespace,
tree: BtreeMut::new(
table_root,
guard.clone(),
page_allocator,
freed_pages,
ignore,
),
transaction_guard: guard,
}
}
fn get<'a>(&self, key: impl Borrow<K::SelfType<'a>>) -> Result<Option<AccessGuard<'_, V>>>
where
K: 'a,
{
self.tree.get(key.borrow())
}
fn range<'a, KR>(&self, range: impl RangeBounds<KR> + 'a) -> Result<Range<'_, K, V>>
where
K: 'a,
KR: Borrow<K::SelfType<'a>> + 'a,
{
self.tree
.range(&range)
.map(|x| Range::new(x, self.transaction_guard.clone()))
}
pub fn extract_from_if<'a, KR, F: for<'f> FnMut(K::SelfType<'f>, V::SelfType<'f>) -> bool>(
&mut self,
range: impl RangeBounds<KR> + 'a,
predicate: F,
) -> Result<ExtractIf<'_, K, V, F>>
where
KR: Borrow<K::SelfType<'a>> + 'a,
{
self.tree
.extract_from_if(&range, predicate)
.map(|inner| ExtractIf::new(inner, None))
}
pub fn insert<'k, 'v>(
&mut self,
key: impl Borrow<K::SelfType<'k>>,
value: impl Borrow<V::SelfType<'v>>,
) -> Result<Option<AccessGuard<'_, V>>> {
let value_len = V::as_bytes(value.borrow()).as_ref().len();
if value_len > MAX_VALUE_LENGTH {
return Err(StorageError::ValueTooLarge(value_len));
}
let key_len = K::as_bytes(key.borrow()).as_ref().len();
if key_len > MAX_VALUE_LENGTH {
return Err(StorageError::ValueTooLarge(key_len));
}
if value_len + key_len > MAX_PAIR_LENGTH {
return Err(StorageError::ValueTooLarge(value_len + key_len));
}
self.tree.insert(key.borrow(), value.borrow())
}
pub fn remove<'a>(
&mut self,
key: impl Borrow<K::SelfType<'a>>,
) -> Result<Option<AccessGuard<'_, V>>>
where
K: 'a,
{
self.tree.remove(key.borrow())
}
}
impl<K: Key + 'static, V: MutInPlaceValue + 'static> SystemTable<'_, K, V> {
pub fn insert_reserve<'a>(
&mut self,
key: impl Borrow<K::SelfType<'a>>,
value_length: usize,
) -> Result<AccessGuardMutInPlace<'_, V>> {
if value_length > MAX_VALUE_LENGTH {
return Err(StorageError::ValueTooLarge(value_length));
}
let key_len = K::as_bytes(key.borrow()).as_ref().len();
if key_len > MAX_VALUE_LENGTH {
return Err(StorageError::ValueTooLarge(key_len));
}
if value_length + key_len > MAX_PAIR_LENGTH {
return Err(StorageError::ValueTooLarge(value_length + key_len));
}
self.tree.insert_reserve(key.borrow(), value_length)
}
}
impl<K: Key + 'static, V: Value + 'static> Drop for SystemTable<'_, K, V> {
fn drop(&mut self) {
self.namespace.close_table(
&self.name,
&self.tree,
self.tree.get_root().map(|x| x.length).unwrap_or_default(),
);
}
}
struct SystemNamespace {
table_tree: TableTreeMut,
freed_pages: Arc<Mutex<Vec<PageNumber>>>,
transaction_guard: Arc<TransactionGuard>,
}
impl SystemNamespace {
fn new(
root_page: Option<BtreeHeader>,
guard: Arc<TransactionGuard>,
page_allocator: PageAllocator,
) -> Self {
let ignore = Arc::new(PageTracker::ignore());
let freed_pages = Arc::new(Mutex::new(vec![]));
Self {
table_tree: TableTreeMut::new(
root_page,
guard.clone(),
page_allocator,
freed_pages.clone(),
ignore,
),
freed_pages,
transaction_guard: guard.clone(),
}
}
fn system_freed_pages(&self) -> Arc<Mutex<Vec<PageNumber>>> {
self.freed_pages.clone()
}
fn open_system_table<'s, K: Key + 'static, V: Value + 'static>(
&'s mut self,
transaction: &WriteTransaction,
definition: SystemTableDefinition<K, V>,
) -> Result<SystemTable<'s, K, V>> {
let (root, _) = self
.table_tree
.get_or_create_table::<K, V>(definition.name(), TableType::Normal)
.map_err(|e| {
e.into_storage_error_or_corrupted("Internal error. System table is corrupted")
})?;
self.table_tree
.clear_pending_table_update(definition.name());
transaction.dirty.store(true, Ordering::Release);
let page_allocator = self.table_tree.page_allocator().clone();
Ok(SystemTable::new(
definition.name(),
root,
self.freed_pages.clone(),
self.transaction_guard.clone(),
page_allocator,
self,
))
}
fn get_system_table_root<K: Key + 'static, V: Value + 'static>(
&self,
definition: SystemTableDefinition<K, V>,
) -> Result<Option<BtreeHeader>> {
let table = self
.table_tree
.get_table::<K, V>(definition.name(), TableType::Normal)
.map_err(|e| {
e.into_storage_error_or_corrupted("Internal error. System table is corrupted")
})?;
Ok(table.and_then(|definition| match definition {
InternalTableDefinition::Normal { table_root, .. } => table_root,
InternalTableDefinition::Multimap { .. } => unreachable!(),
}))
}
fn close_table<K: Key + 'static, V: Value + 'static>(
&mut self,
name: &str,
table: &BtreeMut<K, V>,
length: u64,
) {
self.table_tree
.stage_update_table_root(name, table.get_root(), length);
}
}
struct TableNamespace {
open_tables: BTreeMap<String, &'static panic::Location<'static>>,
allocated_pages: Arc<PageTracker>,
freed_pages: Arc<Mutex<Vec<PageNumber>>>,
table_tree: TableTreeMut,
}
impl TableNamespace {
fn new(
root_page: Option<BtreeHeader>,
guard: Arc<TransactionGuard>,
page_allocator: PageAllocator,
) -> Self {
let allocated = Arc::new(PageTracker::new_tracking());
let freed_pages = Arc::new(Mutex::new(vec![]));
let table_tree = TableTreeMut::new(
root_page,
guard,
page_allocator,
freed_pages.clone(),
allocated.clone(),
);
Self {
open_tables: BTreeMap::default(),
table_tree,
freed_pages,
allocated_pages: allocated,
}
}
fn set_dirty(&mut self, transaction: &WriteTransaction) {
transaction.dirty.store(true, Ordering::Release);
if !transaction.transaction_tracker.any_savepoint_exists() {
self.allocated_pages.disable();
}
}
fn set_root(&mut self, root: Option<BtreeHeader>) {
assert!(self.open_tables.is_empty());
self.table_tree.set_root(root);
}
#[track_caller]
fn inner_open<K: Key + 'static, V: Value + 'static>(
&mut self,
name: &str,
table_type: TableType,
) -> Result<(Option<BtreeHeader>, u64), TableError> {
if let Some(location) = self.open_tables.get(name) {
return Err(TableError::TableAlreadyOpen(name.to_string(), location));
}
let root = self
.table_tree
.get_or_create_table::<K, V>(name, table_type)?;
self.table_tree.clear_pending_table_update(name);
self.open_tables
.insert(name.to_string(), panic::Location::caller());
Ok(root)
}
#[track_caller]
pub fn open_multimap_table<'txn, K: Key + 'static, V: Key + 'static>(
&mut self,
transaction: &'txn WriteTransaction,
definition: MultimapTableDefinition<K, V>,
) -> Result<MultimapTable<'txn, K, V>, TableError> {
#[cfg(feature = "logging")]
debug!("Opening multimap table: {definition}");
let (root, length) = self.inner_open::<K, V>(definition.name(), TableType::Multimap)?;
self.set_dirty(transaction);
Ok(MultimapTable::new(
definition.name(),
root,
length,
self.freed_pages.clone(),
self.allocated_pages.clone(),
self.table_tree.page_allocator().clone(),
transaction,
))
}
#[track_caller]
pub fn open_table<'txn, K: Key + 'static, V: Value + 'static>(
&mut self,
transaction: &'txn WriteTransaction,
definition: TableDefinition<K, V>,
) -> Result<Table<'txn, K, V>, TableError> {
#[cfg(feature = "logging")]
debug!("Opening table: {definition}");
let (root, _) = self.inner_open::<K, V>(definition.name(), TableType::Normal)?;
self.set_dirty(transaction);
Ok(Table::new(
definition.name(),
root,
self.freed_pages.clone(),
self.allocated_pages.clone(),
self.table_tree.page_allocator().clone(),
transaction,
))
}
#[track_caller]
fn inner_rename(
&mut self,
name: &str,
new_name: &str,
table_type: TableType,
) -> Result<(), TableError> {
if let Some(location) = self.open_tables.get(name) {
return Err(TableError::TableAlreadyOpen(name.to_string(), location));
}
self.table_tree.rename_table(name, new_name, table_type)
}
#[track_caller]
fn rename_table(
&mut self,
transaction: &WriteTransaction,
name: &str,
new_name: &str,
) -> Result<(), TableError> {
#[cfg(feature = "logging")]
debug!("Renaming table: {name} to {new_name}");
self.set_dirty(transaction);
self.inner_rename(name, new_name, TableType::Normal)
}
#[track_caller]
fn rename_multimap_table(
&mut self,
transaction: &WriteTransaction,
name: &str,
new_name: &str,
) -> Result<(), TableError> {
#[cfg(feature = "logging")]
debug!("Renaming multimap table: {name} to {new_name}");
self.set_dirty(transaction);
self.inner_rename(name, new_name, TableType::Multimap)
}
#[track_caller]
fn inner_delete(&mut self, name: &str, table_type: TableType) -> Result<bool, TableError> {
if let Some(location) = self.open_tables.get(name) {
return Err(TableError::TableAlreadyOpen(name.to_string(), location));
}
self.table_tree.delete_table(name, table_type)
}
#[track_caller]
fn delete_table(
&mut self,
transaction: &WriteTransaction,
name: &str,
) -> Result<bool, TableError> {
#[cfg(feature = "logging")]
debug!("Deleting table: {name}");
self.set_dirty(transaction);
self.inner_delete(name, TableType::Normal)
}
#[track_caller]
fn delete_multimap_table(
&mut self,
transaction: &WriteTransaction,
name: &str,
) -> Result<bool, TableError> {
#[cfg(feature = "logging")]
debug!("Deleting multimap table: {name}");
self.set_dirty(transaction);
self.inner_delete(name, TableType::Multimap)
}
pub(crate) fn close_table<K: Key + 'static, V: Value + 'static>(
&mut self,
name: &str,
table: &BtreeMut<K, V>,
length: u64,
) {
self.open_tables.remove(name).unwrap();
self.table_tree
.stage_update_table_root(name, table.get_root(), length);
}
pub(crate) fn close_table_without_update(&mut self, name: &str) {
self.open_tables.remove(name).unwrap();
}
}
#[derive(Default)]
struct SavepointTransactionState {
created_persistent: BTreeSet<(SavepointId, TransactionId)>,
deleted_persistent: Vec<(SavepointId, TransactionId)>,
invalidated: BTreeSet<SavepointId>,
}
impl SavepointTransactionState {
fn record_created(&mut self, id: SavepointId, transaction_id: TransactionId) {
self.created_persistent.insert((id, transaction_id));
}
fn record_deleted(&mut self, id: SavepointId, transaction_id: TransactionId) {
self.deleted_persistent.push((id, transaction_id));
}
fn record_invalidated(&mut self, ids: impl IntoIterator<Item = SavepointId>) {
self.invalidated.extend(ids);
}
fn is_invalidated(&self, id: SavepointId) -> bool {
self.invalidated.contains(&id)
}
fn pending_deleted_ids(&self) -> BTreeSet<SavepointId> {
self.deleted_persistent.iter().map(|(id, _)| *id).collect()
}
fn has_created_or_deleted(&self) -> bool {
!self.created_persistent.is_empty() || !self.deleted_persistent.is_empty()
}
fn apply_on_commit(&mut self, tracker: &TransactionTracker) {
for (savepoint, transaction) in self.deleted_persistent.drain(..) {
tracker.deallocate_savepoint(savepoint, transaction);
}
tracker.invalidate_savepoints(core::mem::take(&mut self.invalidated));
self.created_persistent.clear();
}
fn apply_on_abort(&mut self, tracker: &TransactionTracker) {
for (savepoint, transaction) in mem::take(&mut self.created_persistent) {
tracker.deallocate_savepoint(savepoint, transaction);
}
self.deleted_persistent.clear();
self.invalidated.clear();
}
}
struct AllocatorStateLatch {
mem: Option<Arc<TransactionalMemory>>,
}
impl AllocatorStateLatch {
fn arm(mem: Arc<TransactionalMemory>) -> Self {
Self { mem: Some(mem) }
}
fn disarm(mut self) {
self.mem = None;
}
}
impl Drop for AllocatorStateLatch {
fn drop(&mut self) {
if let Some(mem) = self.mem.take() {
mem.invalidate_allocator_state();
}
}
}
pub struct WriteTransaction {
transaction_tracker: Arc<TransactionTracker>,
mem: Arc<TransactionalMemory>,
transaction_guard: Arc<TransactionGuard>,
transaction_id: TransactionId,
tables: Mutex<TableNamespace>,
system_tables: Mutex<SystemNamespace>,
completed: bool,
dirty: AtomicBool,
poisoned: AtomicBool,
durability: InternalDurability,
two_phase_commit: bool,
shrink_policy: ShrinkPolicy,
quick_repair: bool,
post_commit_free: PostCommitFree,
restored_transaction: Option<TransactionId>,
savepoint_state: Mutex<SavepointTransactionState>,
}
impl WriteTransaction {
pub(crate) fn new(
guard: TransactionGuard,
transaction_tracker: Arc<TransactionTracker>,
mem: Arc<TransactionalMemory>,
allocation_policy: AllocationPolicy,
) -> Result<Self> {
let transaction_id = guard.id();
let guard = Arc::new(guard);
let root_page = mem.get_data_root();
let system_page = mem.get_system_root();
let page_allocator = PageAllocator::new(mem.clone(), allocation_policy);
let tables = TableNamespace::new(root_page, guard.clone(), page_allocator.clone());
let system_tables = SystemNamespace::new(system_page, guard.clone(), page_allocator);
Ok(Self {
transaction_tracker,
mem: mem.clone(),
transaction_guard: guard.clone(),
transaction_id,
tables: Mutex::new(tables),
system_tables: Mutex::new(system_tables),
completed: false,
dirty: AtomicBool::new(false),
poisoned: AtomicBool::new(false),
durability: InternalDurability::Immediate,
two_phase_commit: false,
quick_repair: false,
post_commit_free: PostCommitFree::Enabled,
restored_transaction: None,
shrink_policy: ShrinkPolicy::Default,
savepoint_state: Mutex::new(SavepointTransactionState::default()),
})
}
pub(crate) fn set_shrink_policy(&mut self, shrink_policy: ShrinkPolicy) {
self.shrink_policy = shrink_policy;
}
pub(crate) fn poison(&self) {
self.poisoned.store(true, Ordering::Release);
}
fn is_poisoned(&self) -> bool {
self.poisoned.load(Ordering::Acquire)
}
fn page_allocator(&self) -> PageAllocator {
self.tables
.lock()
.unwrap()
.table_tree
.page_allocator()
.clone()
}
fn read_existing_system_table<K: Key + 'static, V: Value + 'static, T>(
&self,
definition: SystemTableDefinition<K, V>,
read: impl FnOnce(&Btree<K, V>) -> Result<T>,
) -> Result<Option<T>> {
let system_tables = self.system_tables.lock().unwrap();
let Some(root) = system_tables.get_system_table_root(definition)? else {
return Ok(None);
};
let table = Btree::new(
Some(root),
PageHint::None,
self.transaction_guard.clone(),
PageResolver::new(self.mem.clone()),
)?;
read(&table).map(Some)
}
pub(crate) fn pending_free_pages(&self) -> Result<bool> {
let system_tables = self.system_tables.lock().unwrap();
Ok(system_tables
.get_system_table_root(DATA_FREED_TABLE)?
.is_some()
|| system_tables
.get_system_table_root(SYSTEM_FREED_TABLE)?
.is_some())
}
#[cfg(all(debug_assertions, not(redb_no_std)))]
pub fn print_allocated_page_debug(&self) {
let mut all_allocated = PageNumberHashSet::from_iter(self.mem.all_allocated_pages());
self.mem.debug_check_allocator_consistency();
let mut table_pages = vec![];
self.tables
.lock()
.unwrap()
.table_tree
.visit_all_pages(|path| {
table_pages.push(path.page_number());
Ok(())
})
.unwrap();
println!("Tables");
for p in table_pages {
assert!(all_allocated.remove(&p));
println!("{p:?}");
}
let mut system_table_pages = vec![];
self.system_tables
.lock()
.unwrap()
.table_tree
.visit_all_pages(|path| {
system_table_pages.push(path.page_number());
Ok(())
})
.unwrap();
println!("System tables");
for p in system_table_pages {
assert!(all_allocated.remove(&p));
println!("{p:?}");
}
{
println!("Pending free (in data freed table)");
let mut system_tables = self.system_tables.lock().unwrap();
let data_freed = system_tables
.open_system_table(self, DATA_FREED_TABLE)
.unwrap();
for entry in data_freed.range::<TransactionIdWithPagination>(..).unwrap() {
let (_, entry) = entry.unwrap();
let value = entry.value();
for i in 0..value.len() {
let p = value.get(i);
assert!(all_allocated.remove(&p));
println!("{p:?}");
}
}
}
{
println!("Pending free (in system freed table)");
let mut system_tables = self.system_tables.lock().unwrap();
let system_freed = system_tables
.open_system_table(self, SYSTEM_FREED_TABLE)
.unwrap();
for entry in system_freed
.range::<TransactionIdWithPagination>(..)
.unwrap()
{
let (_, entry) = entry.unwrap();
let value = entry.value();
for i in 0..value.len() {
let p = value.get(i);
assert!(all_allocated.remove(&p));
println!("{p:?}");
}
}
}
{
let tables = self.tables.lock().unwrap();
let pages = tables.freed_pages.lock().unwrap();
if !pages.is_empty() {
println!("Pages in in-memory data freed_pages");
for p in pages.iter() {
println!("{p:?}");
assert!(all_allocated.remove(p));
}
}
}
{
let system_tables = self.system_tables.lock().unwrap();
let pages = system_tables.freed_pages.lock().unwrap();
if !pages.is_empty() {
println!("Pages in in-memory system freed_pages");
for p in pages.iter() {
println!("{p:?}");
assert!(all_allocated.remove(p));
}
}
}
if !all_allocated.is_empty() {
println!("Leaked pages");
for p in all_allocated {
println!("{p:?}");
}
}
}
pub fn persistent_savepoint(&self) -> Result<u64, SavepointError> {
if self.durability != InternalDurability::Immediate {
return Err(SavepointError::ImmediateDurabilityRequired);
}
let mut savepoint = self.ephemeral_savepoint()?;
let mut system_tables = self.system_tables.lock().unwrap();
let mut next_table = system_tables.open_system_table(self, NEXT_SAVEPOINT_TABLE)?;
next_table.insert((), savepoint.get_id().next())?;
drop(next_table);
let mut savepoint_table = system_tables.open_system_table(self, SAVEPOINT_TABLE)?;
savepoint_table.insert(
savepoint.get_id(),
SerializedSavepoint::from_savepoint(&savepoint),
)?;
savepoint.set_persistent();
self.transaction_tracker
.mark_savepoint_persistent(savepoint.get_id());
self.savepoint_state
.lock()
.unwrap()
.record_created(savepoint.get_id(), savepoint.get_transaction_id());
Ok(savepoint.get_id().0)
}
pub(crate) fn transaction_guard(&self) -> Arc<TransactionGuard> {
self.transaction_guard.clone()
}
pub(crate) fn next_persistent_savepoint_id(&self) -> Result<Option<SavepointId>> {
let Some(value) = self.read_existing_system_table(NEXT_SAVEPOINT_TABLE, |next_table| {
let value = next_table.get(&())?;
Ok(value.map(|next_id| next_id.value()))
})?
else {
return Ok(None);
};
Ok(value)
}
pub fn get_persistent_savepoint(&self, id: u64) -> Result<Savepoint, SavepointError> {
let Some(value) = self.read_existing_system_table(SAVEPOINT_TABLE, |table| {
let value = table.get(&SavepointId(id))?;
value
.map(|x| x.value().to_savepoint(self.transaction_tracker.clone()))
.transpose()
})?
else {
return Err(SavepointError::InvalidSavepoint);
};
value.ok_or(SavepointError::InvalidSavepoint)
}
pub fn delete_persistent_savepoint(&self, id: u64) -> Result<bool, SavepointError> {
if self.durability != InternalDurability::Immediate {
return Err(SavepointError::ImmediateDurabilityRequired);
}
let mut system_tables = self.system_tables.lock().unwrap();
if system_tables
.get_system_table_root(SAVEPOINT_TABLE)?
.is_none()
{
return Ok(false);
}
let mut table = system_tables.open_system_table(self, SAVEPOINT_TABLE)?;
let savepoint = if let Some(serialized) = table.get(SavepointId(id))? {
serialized
.value()
.to_savepoint(self.transaction_tracker.clone())?
} else {
return Ok(false);
};
table.remove(SavepointId(id))?;
self.savepoint_state
.lock()
.unwrap()
.record_deleted(savepoint.get_id(), savepoint.get_transaction_id());
Ok(true)
}
pub fn list_persistent_savepoints(&self) -> Result<impl Iterator<Item = u64>> {
let Some(savepoints) = self.read_existing_system_table(SAVEPOINT_TABLE, |table| {
let mut savepoints = vec![];
for savepoint in table.range::<RangeFull, SavepointId>(&..)? {
savepoints.push(savepoint?.key().0);
}
Ok(savepoints)
})?
else {
return Ok(vec![].into_iter());
};
Ok(savepoints.into_iter())
}
fn allocate_savepoint(&self) -> Result<(SavepointId, TransactionId)> {
let transaction_id = self
.transaction_tracker
.register_read_transaction(&self.mem)?;
let id = self.transaction_tracker.allocate_savepoint(transaction_id);
Ok((id, transaction_id))
}
pub fn ephemeral_savepoint(&self) -> Result<Savepoint, SavepointError> {
let (id, transaction_id) = {
let _tables = self.tables.lock().unwrap();
if self.dirty.load(Ordering::Acquire) {
return Err(SavepointError::InvalidSavepoint);
}
self.allocate_savepoint()?
};
#[cfg(feature = "logging")]
debug!("Creating savepoint id={id:?}, txn_id={transaction_id:?}");
let root = self.mem.get_data_root();
let savepoint = Savepoint::new_ephemeral(
&self.mem,
self.transaction_tracker.clone(),
id,
transaction_id,
root,
);
Ok(savepoint)
}
pub fn restore_savepoint(&mut self, savepoint: &Savepoint) -> Result<(), SavepointError> {
if core::ptr::from_ref(self.transaction_tracker.as_ref()) != savepoint.db_address() {
return Err(SavepointError::InvalidSavepoint);
}
if !self
.transaction_tracker
.is_valid_savepoint(savepoint.get_id())
|| self
.savepoint_state
.lock()
.unwrap()
.is_invalidated(savepoint.get_id())
{
return Err(SavepointError::InvalidSavepoint);
}
if self.durability != InternalDurability::Immediate
&& self
.list_persistent_savepoints()?
.any(|id| id > savepoint.get_id().0)
{
return Err(SavepointError::ImmediateDurabilityRequired);
}
#[cfg(feature = "logging")]
debug!(
"Beginning savepoint restore (id={:?}) in transaction id={:?}",
savepoint.get_id(),
self.transaction_id
);
assert_eq!(self.mem.get_version(), savepoint.get_version());
self.dirty.store(true, Ordering::Release);
{
self.tables
.lock()
.unwrap()
.set_root(savepoint.get_user_root());
}
let txn_id = savepoint.get_transaction_id().next().raw_id();
{
let lower = TransactionIdWithPagination {
transaction_id: txn_id,
pagination_id: 0,
};
let mut system_tables = self.system_tables.lock().unwrap();
let mut data_freed = system_tables.open_system_table(self, DATA_FREED_TABLE)?;
let drain = || -> Result<(), StorageError> {
let mut iter = data_freed.extract_from_if(lower.., |_, _| true)?;
for entry in &mut iter {
entry?;
}
iter.close()
};
let result = drain();
if result.is_err() {
self.poison();
}
result?;
}
{
let tables = self.tables.lock().unwrap();
let page_allocator = tables.table_tree.page_allocator();
for page in tables.allocated_pages.reset() {
debug_assert!(page_allocator.uncommitted(page));
debug_assert!(self.mem.is_allocated(page));
page_allocator.free(page, &PageTracker::ignore());
}
let mut data_freed_pages = tables.freed_pages.lock().unwrap();
data_freed_pages.clear();
let mut system_tables = self.system_tables.lock().unwrap();
let data_allocated = system_tables.open_system_table(self, DATA_ALLOCATED_TABLE)?;
let lower = TransactionIdWithPagination {
transaction_id: txn_id,
pagination_id: 0,
};
for entry in data_allocated.range(lower..)? {
let (_, value) = entry?;
for i in 0..value.value().len() {
data_freed_pages.push(value.value().get(i));
}
}
for page in self
.mem
.unpersisted_allocations_after(savepoint.get_transaction_id())
{
data_freed_pages.push(page);
}
}
let invalidated = self
.transaction_tracker
.list_savepoints_after(savepoint.get_id());
self.savepoint_state
.lock()
.unwrap()
.record_invalidated(invalidated);
for persistent_savepoint in self.list_persistent_savepoints()? {
if persistent_savepoint > savepoint.get_id().0 {
self.delete_persistent_savepoint(persistent_savepoint)?;
}
}
self.restored_transaction = Some(savepoint.get_transaction_id());
Ok(())
}
pub fn set_durability(&mut self, durability: Durability) -> Result<(), SetDurabilityError> {
let persistent_modified = self
.savepoint_state
.lock()
.unwrap()
.has_created_or_deleted();
if persistent_modified && !matches!(durability, Durability::Immediate) {
return Err(SetDurabilityError::PersistentSavepointModified);
}
self.durability = match durability {
Durability::None => InternalDurability::None,
Durability::Immediate => InternalDurability::Immediate,
};
Ok(())
}
pub fn set_two_phase_commit(&mut self, enabled: bool) {
self.two_phase_commit = enabled;
}
pub fn set_quick_repair(&mut self, enabled: bool) {
self.quick_repair = enabled;
}
pub(crate) fn disable_post_commit_free(&mut self) {
self.post_commit_free = PostCommitFree::Disabled;
}
#[track_caller]
pub fn open_table<'txn, K: Key + 'static, V: Value + 'static>(
&'txn self,
definition: TableDefinition<K, V>,
) -> Result<Table<'txn, K, V>, TableError> {
self.tables.lock().unwrap().open_table(self, definition)
}
#[track_caller]
pub fn open_multimap_table<'txn, K: Key + 'static, V: Key + 'static>(
&'txn self,
definition: MultimapTableDefinition<K, V>,
) -> Result<MultimapTable<'txn, K, V>, TableError> {
self.tables
.lock()
.unwrap()
.open_multimap_table(self, definition)
}
pub(crate) fn close_table<K: Key + 'static, V: Value + 'static>(
&self,
name: &str,
table: &BtreeMut<K, V>,
length: u64,
) {
let mut tables = self.tables.lock().unwrap();
if self.is_poisoned() {
tables.close_table_without_update(name);
} else {
tables.close_table(name, table, length);
}
}
pub fn rename_table(
&self,
definition: impl TableHandle,
new_name: impl TableHandle,
) -> Result<(), TableError> {
let name = definition.name().to_string();
drop(definition);
self.tables
.lock()
.unwrap()
.rename_table(self, &name, new_name.name())
}
pub fn rename_multimap_table(
&self,
definition: impl MultimapTableHandle,
new_name: impl MultimapTableHandle,
) -> Result<(), TableError> {
let name = definition.name().to_string();
drop(definition);
self.tables
.lock()
.unwrap()
.rename_multimap_table(self, &name, new_name.name())
}
pub fn delete_table(&self, definition: impl TableHandle) -> Result<bool, TableError> {
let name = definition.name().to_string();
drop(definition);
self.tables.lock().unwrap().delete_table(self, &name)
}
pub fn delete_multimap_table(
&self,
definition: impl MultimapTableHandle,
) -> Result<bool, TableError> {
let name = definition.name().to_string();
drop(definition);
self.tables
.lock()
.unwrap()
.delete_multimap_table(self, &name)
}
pub fn list_tables(&self) -> Result<impl Iterator<Item = UntypedTableHandle> + '_> {
self.tables
.lock()
.unwrap()
.table_tree
.list_tables(TableType::Normal)
.map(|x| x.into_iter().map(UntypedTableHandle::new))
}
pub fn list_multimap_tables(
&self,
) -> Result<impl Iterator<Item = UntypedMultimapTableHandle> + '_> {
self.tables
.lock()
.unwrap()
.table_tree
.list_tables(TableType::Multimap)
.map(|x| x.into_iter().map(UntypedMultimapTableHandle::new))
}
pub fn commit(mut self) -> Result<(), CommitError> {
self.completed = true;
if self.is_poisoned() {
self.abort_inner()?;
return Err(CommitError::TransactionPoisoned);
}
self.commit_inner()
}
fn commit_inner(&mut self) -> Result<(), CommitError> {
let latch = AllocatorStateLatch::arm(self.mem.clone());
let result = self.commit_inner_helper();
if result.is_ok() {
latch.disarm();
}
result
}
fn commit_inner_helper(&mut self) -> Result<(), CommitError> {
if self.quick_repair {
self.two_phase_commit = true;
}
if let Some(transaction_id) = self.restored_transaction {
self.mem.drop_unpersisted_data_freed_after(transaction_id);
}
let (user_root, allocated_pages, data_freed) =
self.tables.lock().unwrap().table_tree.flush_and_close()?;
if self.durability == InternalDurability::Immediate {
let pages = self.mem.take_post_commit_allocations();
self.page_allocator().adopt_unpersisted(pages);
}
let stored_data_freed_pages = match self.durability {
InternalDurability::None => {
let stored_pages = !data_freed.is_empty();
self.mem
.record_unpersisted_data_freed(self.transaction_id, data_freed);
stored_pages
}
InternalDurability::Immediate => self.store_data_freed_pages(data_freed)?,
};
#[cfg(feature = "logging")]
debug!(
"Committing transaction id={:?} with durability={:?} two_phase={} quick_repair={}",
self.transaction_id, self.durability, self.two_phase_commit, self.quick_repair
);
let allocated_pages: Vec<PageNumber> = allocated_pages.into_iter().collect();
match self.durability {
InternalDurability::None => {
self.non_durable_commit(user_root, allocated_pages, stored_data_freed_pages)?;
self.apply_savepoint_state_on_commit();
}
InternalDurability::Immediate => self.durable_commit(user_root, allocated_pages)?,
}
assert!(
self.system_tables
.lock()
.unwrap()
.system_freed_pages()
.lock()
.unwrap()
.is_empty()
);
assert!(
self.tables
.lock()
.unwrap()
.freed_pages
.lock()
.unwrap()
.is_empty()
);
#[cfg(feature = "logging")]
debug!(
"Finished commit of transaction id={:?}",
self.transaction_id
);
Ok(())
}
fn apply_savepoint_state_on_commit(&self) {
self.savepoint_state
.lock()
.unwrap()
.apply_on_commit(&self.transaction_tracker);
}
fn store_data_freed_pages(&self, freed_pages: Vec<PageNumber>) -> Result<bool> {
let stored_pages = !freed_pages.is_empty();
self.store_data_freed_pages_for(self.transaction_id, freed_pages)?;
Ok(stored_pages)
}
fn store_data_freed_pages_for(
&self,
transaction_id: TransactionId,
mut freed_pages: Vec<PageNumber>,
) -> Result {
let mut system_tables = self.system_tables.lock().unwrap();
let mut freed_table = system_tables.open_system_table(self, DATA_FREED_TABLE)?;
let mut pagination_counter = 0;
#[cfg(debug_assertions)]
let page_allocator = self.page_allocator();
while !freed_pages.is_empty() {
let chunk_size = 400;
let buffer_size = PageList::required_bytes(chunk_size);
let key = TransactionIdWithPagination {
transaction_id: transaction_id.raw_id(),
pagination_id: pagination_counter,
};
let mut access_guard = freed_table.insert_reserve(&key, buffer_size)?;
let len = freed_pages.len();
access_guard.as_mut().clear();
for page in freed_pages.drain(len - min(len, chunk_size)..) {
debug_assert!(
self.mem.is_allocated(page),
"Page is not allocated: {page:?}"
);
#[cfg(debug_assertions)]
debug_assert!(
!page_allocator.uncommitted(page),
"Page is uncommitted: {page:?}"
);
access_guard.as_mut().push_back(page);
}
pagination_counter += 1;
}
Ok(())
}
fn flush_data_allocated_pages(&self, data_allocated_pages: Vec<PageNumber>) -> Result<u64> {
#[cfg(debug_assertions)]
{
let page_allocator = self.page_allocator();
for page in &data_allocated_pages {
debug_assert!(
self.mem.is_allocated(*page),
"Page is not allocated: {page:?}"
);
debug_assert!(
page_allocator.uncommitted(*page),
"Page is committed: {page:?}"
);
}
}
let unpersisted = self.mem.take_unpersisted_allocations();
let mut system_tables = self.system_tables.lock().unwrap();
let mut allocated_table = system_tables.open_system_table(self, DATA_ALLOCATED_TABLE)?;
for (txn_id, pages) in unpersisted {
Self::write_allocated_pages_entry(
&mut allocated_table,
txn_id,
pages.into_iter().collect(),
)?;
}
Self::write_allocated_pages_entry(
&mut allocated_table,
self.transaction_id,
data_allocated_pages,
)?;
let deleted_savepoints = self.savepoint_state.lock().unwrap().pending_deleted_ids();
let oldest = self
.transaction_tracker
.oldest_savepoint_excluding(&deleted_savepoints)
.map_or(u64::MAX, |(_, x)| x.raw_id());
let key = TransactionIdWithPagination {
transaction_id: oldest,
pagination_id: 0,
};
for entry in allocated_table.extract_from_if(..key, |_, _| true)? {
entry?;
}
Ok(oldest)
}
fn write_allocated_pages_entry(
allocated_table: &mut SystemTable<'_, TransactionIdWithPagination, PageList<'static>>,
transaction_id: TransactionId,
mut pages: Vec<PageNumber>,
) -> Result {
let mut pagination_counter = 0;
while !pages.is_empty() {
let chunk_size = 400;
let buffer_size = PageList::required_bytes(chunk_size);
let key = TransactionIdWithPagination {
transaction_id: transaction_id.raw_id(),
pagination_id: pagination_counter,
};
let mut access_guard = allocated_table.insert_reserve(&key, buffer_size)?;
let len = pages.len();
access_guard.as_mut().clear();
for page in pages.drain(len - min(len, chunk_size)..) {
access_guard.as_mut().push_back(page);
}
pagination_counter += 1;
}
Ok(())
}
pub fn abort(mut self) -> Result {
self.completed = true;
self.abort_inner()
}
fn abort_inner(&mut self) -> Result {
#[cfg(feature = "logging")]
debug!("Aborting transaction id={:?}", self.transaction_id);
self.tables
.lock()
.unwrap()
.table_tree
.clear_root_updates_and_close();
self.savepoint_state
.lock()
.unwrap()
.apply_on_abort(&self.transaction_tracker);
self.mem.check_io_errors()?;
self.page_allocator().rollback_all();
#[cfg(feature = "logging")]
debug!("Finished abort of transaction id={:?}", self.transaction_id);
Ok(())
}
pub(crate) fn durable_commit(
&mut self,
user_root: Option<BtreeHeader>,
allocated_pages: Vec<PageNumber>,
) -> Result {
for (transaction_id, pages) in self.mem.take_unpersisted_data_freed() {
self.store_data_freed_pages_for(transaction_id, pages)?;
}
let free_until_transaction = self
.transaction_tracker
.oldest_live_read_transaction()
.map_or(self.transaction_id, |x| x.next());
self.process_freed_pages(free_until_transaction)?;
let savepoint_horizon = self.flush_data_allocated_pages(allocated_pages)?;
let mut system_tables = self.system_tables.lock().unwrap();
let system_freed_pages = system_tables.system_freed_pages();
let system_root = {
let system_tree = system_tables.table_tree.flush_table_root_updates()?;
system_tree
.delete_table(ALLOCATOR_STATE_TABLE_NAME, TableType::Normal)
.map_err(|e| e.into_storage_error_or_corrupted("Unexpected TableError"))?;
if self.quick_repair {
system_tree.create_table_and_flush_table_root(
ALLOCATOR_STATE_TABLE_NAME,
|system_tree_ref, tree: &mut AllocatorStateTreeMut| {
loop {
let num_regions = self
.mem
.reserve_allocator_state(tree, self.transaction_id)?;
self.store_system_freed_pages(
system_tree_ref,
self.transaction_id,
system_freed_pages.clone(),
None,
)?;
if self.mem.try_save_allocator_state(tree, num_regions)? {
return Ok(());
}
while let Some(guards) = tree.last()? {
let key = guards.0.value();
drop(guards);
tree.remove(&key)?;
}
}
},
)?;
}
system_tree.finalize_dirty_checksums()?
};
let page_allocator = self.page_allocator();
self.mem.commit(
user_root,
system_root,
self.transaction_id,
self.two_phase_commit,
self.shrink_policy,
)?;
let _ = page_allocator.take_allocated_since_commit();
self.transaction_tracker.clear_pending_non_durable_commits();
for page in system_freed_pages.lock().unwrap().drain(..) {
page_allocator.free(page, &PageTracker::ignore());
}
drop(system_tables);
self.apply_savepoint_state_on_commit();
if self.post_commit_free == PostCommitFree::Enabled {
self.process_data_freed_pages_after_commit(
user_root,
&page_allocator,
savepoint_horizon,
)?;
}
Ok(())
}
fn process_data_freed_pages_after_commit(
&self,
user_root: Option<BtreeHeader>,
page_allocator: &PageAllocator,
savepoint_horizon: u64,
) -> Result {
let epilogue_transaction = self.transaction_id.next();
let mut free_until = self
.transaction_tracker
.oldest_live_read_transaction()
.map_or(epilogue_transaction, |x| x.next());
if savepoint_horizon != u64::MAX {
free_until = free_until.min(TransactionId::new(savepoint_horizon).next());
}
let mut freed_any = false;
let (system_root, stored_system_freed_pages, extracted_data_transactions) = {
let mut system_tables = self.system_tables.lock().unwrap();
let system_freed_pages = system_tables.system_freed_pages();
let extracted_data_transactions = self.extract_freed_pages(
&mut system_tables,
DATA_FREED_TABLE,
free_until,
|page| {
freed_any = true;
debug_assert!(!self.mem.unpersisted(page));
page_allocator.free(page, &PageTracker::ignore());
},
)?;
if !freed_any {
return Ok(());
}
let system_tree = system_tables.table_tree.flush_table_root_updates()?;
let stored_system_freed_pages = self.store_system_freed_pages(
system_tree,
epilogue_transaction,
system_freed_pages,
None,
)?;
(
system_tree.finalize_dirty_checksums()?,
stored_system_freed_pages,
extracted_data_transactions,
)
};
let epilogue_allocations = page_allocator.take_allocated_since_commit();
self.mem
.record_post_commit_allocations(epilogue_allocations.iter().copied());
self.mem.non_durable_commit(
user_root,
system_root,
epilogue_transaction,
epilogue_allocations,
)?;
self.transaction_tracker
.reserve_transaction_id(epilogue_transaction, self.transaction_id);
self.transaction_tracker.register_non_durable_commit(
epilogue_transaction,
self.transaction_id,
stored_system_freed_pages,
);
self.transaction_tracker
.mark_non_durable_freed_pages_processed(extracted_data_transactions);
Ok(())
}
pub(crate) fn non_durable_commit(
&mut self,
user_root: Option<BtreeHeader>,
allocated_pages: Vec<PageNumber>,
stored_data_freed_pages: bool,
) -> Result {
let free_until_transaction = self
.transaction_tracker
.oldest_live_read_nondurable_transaction()
.map_or(self.transaction_id, |x| x.next());
self.process_freed_pages_nondurable(free_until_transaction)?;
let mut post_commit_frees = vec![];
let (system_root, stored_system_freed_pages) = {
let mut system_tables = self.system_tables.lock().unwrap();
let system_freed_pages = system_tables.system_freed_pages();
system_tables.table_tree.flush_table_root_updates()?;
for page in system_freed_pages
.lock()
.unwrap()
.extract_if(.., |p| self.mem.unpersisted(*p))
{
post_commit_frees.push(page);
}
let stored_system_freed_pages = self.store_system_freed_pages(
&mut system_tables.table_tree,
self.transaction_id,
system_freed_pages,
Some(&mut post_commit_frees),
)?;
let system_root = system_tables
.table_tree
.flush_table_root_updates()?
.finalize_dirty_checksums()?;
(system_root, stored_system_freed_pages)
};
let newly_unpersisted = self.page_allocator().take_allocated_since_commit();
self.mem.non_durable_commit(
user_root,
system_root,
self.transaction_id,
newly_unpersisted,
)?;
self.mem
.record_unpersisted_allocations(self.transaction_id, allocated_pages);
let stored_freed_pages = stored_data_freed_pages || stored_system_freed_pages;
self.transaction_tracker.register_non_durable_commit(
self.transaction_id,
self.mem.get_last_durable_transaction_id()?,
stored_freed_pages,
);
for page in post_commit_frees {
let removed = self.mem.free_if_unpersisted(page, &PageTracker::ignore());
assert!(removed);
}
Ok(())
}
pub(crate) fn compact_pages(&mut self) -> Result<bool> {
let mut progress = false;
let mut highest_pages = BTreeMap::new();
let mut tables = self.tables.lock().unwrap();
let table_tree = &mut tables.table_tree;
table_tree.highest_index_pages(MAX_PAGES_PER_COMPACTION, &mut highest_pages)?;
let mut system_tables = self.system_tables.lock().unwrap();
let system_table_tree = &mut system_tables.table_tree;
system_table_tree.highest_index_pages(MAX_PAGES_PER_COMPACTION, &mut highest_pages)?;
let page_allocator = table_tree.page_allocator().clone();
let mut relocation_map = PageNumberHashMap::default();
for path in highest_pages.into_values().rev() {
if relocation_map.contains_key(&path.page_number()) {
continue;
}
let old_page = page_allocator.get_page(path.page_number(), PageHint::None)?;
let mut new_page =
page_allocator.allocate_lowest(old_page.memory().len(), &PageTracker::ignore())?;
let new_page_number = new_page.get_page_number();
new_page.memory_mut()[0] = old_page.memory()[0];
drop(new_page);
if new_page_number < path.page_number() {
relocation_map.insert(path.page_number(), new_page_number);
for parent in path.parents() {
if relocation_map.contains_key(parent) {
continue;
}
let old_parent = page_allocator.get_page(*parent, PageHint::None)?;
let mut new_page = page_allocator
.allocate_lowest(old_parent.memory().len(), &PageTracker::ignore())?;
let new_page_number = new_page.get_page_number();
new_page.memory_mut()[0] = old_parent.memory()[0];
drop(new_page);
relocation_map.insert(*parent, new_page_number);
}
} else {
page_allocator.free(new_page_number, &PageTracker::ignore());
break;
}
}
if !relocation_map.is_empty() {
progress = true;
}
table_tree.relocate_tables(&relocation_map)?;
system_table_tree.relocate_tables(&relocation_map)?;
Ok(progress)
}
fn process_freed_pages(&mut self, free_until: TransactionId) -> Result {
assert_eq!(PageNumber::serialized_size(), 8);
let page_allocator = self.page_allocator();
let mut free_page = |page| {
debug_assert!(!self.mem.unpersisted(page));
page_allocator.free(page, &PageTracker::ignore());
};
let extracted_transactions = {
let mut system_tables = self.system_tables.lock().unwrap();
let mut extracted_transactions = self.extract_freed_pages(
&mut system_tables,
DATA_FREED_TABLE,
free_until,
&mut free_page,
)?;
extracted_transactions.extend(self.extract_freed_pages(
&mut system_tables,
SYSTEM_FREED_TABLE,
free_until,
&mut free_page,
)?);
extracted_transactions
};
self.transaction_tracker
.mark_non_durable_freed_pages_processed(extracted_transactions);
Ok(())
}
fn extract_freed_pages(
&self,
system_tables: &mut SystemNamespace,
definition: SystemTableDefinition<TransactionIdWithPagination, PageList>,
free_until: TransactionId,
mut process_page: impl FnMut(PageNumber),
) -> Result<Vec<TransactionId>> {
if system_tables.get_system_table_root(definition)?.is_none() {
return Ok(vec![]);
}
let mut freed = system_tables.open_system_table(self, definition)?;
let key = TransactionIdWithPagination {
transaction_id: free_until.raw_id(),
pagination_id: 0,
};
let mut extracted_transactions = vec![];
for entry in freed.extract_from_if(..key, |_, _| true)? {
let (key, page_list) = entry?;
let transaction_id = TransactionId::new(key.value().transaction_id);
if extracted_transactions.last().copied() != Some(transaction_id) {
extracted_transactions.push(transaction_id);
}
let page_list = page_list.value();
for i in 0..page_list.len() {
process_page(page_list.get(i));
}
}
Ok(extracted_transactions)
}
fn process_freed_pages_nondurable_helper(
&mut self,
free_until: TransactionId,
definition: SystemTableDefinition<TransactionIdWithPagination, PageList>,
) -> Result<Vec<TransactionId>> {
let mut processed = vec![];
let mut system_tables = self.system_tables.lock().unwrap();
let last_key = TransactionIdWithPagination {
transaction_id: free_until.raw_id(),
pagination_id: 0,
};
let oldest_unprocessed = self
.transaction_tracker
.oldest_unprocessed_non_durable_commit()
.map_or(free_until.raw_id(), |x| x.raw_id());
let first_key = TransactionIdWithPagination {
transaction_id: oldest_unprocessed,
pagination_id: 0,
};
let mut data_freed = system_tables.open_system_table(self, definition)?;
let mut candidate_transactions = vec![];
for entry in data_freed.range(first_key..last_key)? {
let (key, _) = entry?;
let transaction_id = TransactionId::new(key.value().transaction_id);
if self
.transaction_tracker
.is_unprocessed_non_durable_commit(transaction_id)
&& candidate_transactions.last().copied() != Some(transaction_id)
{
candidate_transactions.push(transaction_id);
}
}
for transaction_id in candidate_transactions {
let mut key = TransactionIdWithPagination {
transaction_id: transaction_id.raw_id(),
pagination_id: 0,
};
loop {
let Some(entry) = data_freed.get(&key)? else {
break;
};
let pages = entry.value();
let mut new_pages = vec![];
for i in 0..pages.len() {
let page = pages.get(i);
if !self.mem.free_if_unpersisted(page, &PageTracker::ignore()) {
new_pages.push(page);
}
}
if new_pages.len() != pages.len() {
drop(entry);
if new_pages.is_empty() {
data_freed.remove(&key)?;
} else {
let required = PageList::required_bytes(new_pages.len());
let mut page_list_mut = data_freed.insert_reserve(&key, required)?;
for page in new_pages {
page_list_mut.as_mut().push_back(page);
}
}
}
key.pagination_id += 1;
}
processed.push(transaction_id);
}
Ok(processed)
}
fn process_freed_pages_nondurable(&mut self, free_until: TransactionId) -> Result {
assert_eq!(PageNumber::serialized_size(), 8);
let oldest_unprocessed = self
.transaction_tracker
.oldest_unprocessed_non_durable_commit()
.unwrap_or(free_until);
let mut processed =
self.mem
.process_unpersisted_data_freed(oldest_unprocessed, free_until, |page| {
self.mem.free_if_unpersisted(page, &PageTracker::ignore())
});
processed
.extend(self.process_freed_pages_nondurable_helper(free_until, SYSTEM_FREED_TABLE)?);
for transaction_id in processed {
self.transaction_tracker
.mark_non_durable_freed_pages_processed([transaction_id]);
}
Ok(())
}
fn store_system_freed_pages(
&self,
system_tree: &mut TableTreeMut,
transaction_id: TransactionId,
system_freed_pages: Arc<Mutex<Vec<PageNumber>>>,
mut unpersisted_pages: Option<&mut Vec<PageNumber>>,
) -> Result<bool> {
assert_eq!(PageNumber::serialized_size(), 8); if system_freed_pages.lock().unwrap().is_empty() {
return Ok(false);
}
let mut stored_pages = false;
system_tree.open_table_and_flush_table_root(
SYSTEM_FREED_TABLE.name(),
|system_freed_tree: &mut SystemFreedTree| {
let mut pagination_id =
Self::next_system_freed_pagination_id(system_freed_tree, transaction_id)?;
while !system_freed_pages.lock().unwrap().is_empty() {
let chunk_size = 200;
let buffer_size = PageList::required_bytes(chunk_size);
let key = TransactionIdWithPagination {
transaction_id: transaction_id.raw_id(),
pagination_id,
};
let mut access_guard = system_freed_tree.insert_reserve(&key, buffer_size)?;
let mut freed_pages = system_freed_pages.lock().unwrap();
let len = freed_pages.len();
access_guard.as_mut().clear();
for page in freed_pages.drain(len - min(len, chunk_size)..) {
if let Some(ref mut unpersisted_pages) = unpersisted_pages
&& self.mem.unpersisted(page)
{
unpersisted_pages.push(page);
} else {
access_guard.as_mut().push_back(page);
stored_pages = true;
}
}
drop(access_guard);
pagination_id += 1;
}
Ok(())
},
)?;
Ok(stored_pages)
}
fn next_system_freed_pagination_id(
system_freed_tree: &SystemFreedTree,
transaction_id: TransactionId,
) -> Result<u64> {
let first_key = TransactionIdWithPagination {
transaction_id: transaction_id.raw_id(),
pagination_id: 0,
};
let next_transaction_key = TransactionIdWithPagination {
transaction_id: transaction_id.next().raw_id(),
pagination_id: 0,
};
let transaction_range = first_key..next_transaction_key;
let mut existing_entries = system_freed_tree.range(&transaction_range)?;
Ok(existing_entries
.next_back()
.transpose()?
.map_or(0, |entry| entry.key().pagination_id + 1))
}
pub fn stats(&self) -> Result<DatabaseStats> {
let tables = self.tables.lock().unwrap();
let table_tree = &tables.table_tree;
let data_tree_stats = table_tree.stats()?;
let system_tables = self.system_tables.lock().unwrap();
let system_table_tree = &system_tables.table_tree;
let system_tree_stats = system_table_tree.stats()?;
let total_metadata_bytes = data_tree_stats.metadata_bytes()
+ system_tree_stats.metadata_bytes
+ system_tree_stats.stored_leaf_bytes;
let total_fragmented = data_tree_stats.fragmented_bytes()
+ system_tree_stats.fragmented_bytes
+ self.mem.count_free_pages()? * (self.mem.get_page_size() as u64);
Ok(DatabaseStats {
tree_height: data_tree_stats.tree_height(),
allocated_pages: self.mem.count_allocated_pages()?,
leaf_pages: data_tree_stats.leaf_pages(),
branch_pages: data_tree_stats.branch_pages(),
stored_leaf_bytes: data_tree_stats.stored_bytes(),
metadata_bytes: total_metadata_bytes,
fragmented_bytes: total_fragmented,
page_size: self.mem.get_page_size(),
})
}
#[allow(dead_code)]
#[cfg(not(redb_no_std))]
pub(crate) fn print_debug(&self) -> Result {
let mut tables = self.tables.lock().unwrap();
if let Some(page) = tables
.table_tree
.flush_table_root_updates()
.unwrap()
.finalize_dirty_checksums()
.unwrap()
{
eprintln!("Master tree:");
let master_tree: Btree<&str, InternalTableDefinition> = Btree::new(
Some(page),
PageHint::None,
self.transaction_guard.clone(),
PageResolver::new(self.mem.clone()),
)?;
master_tree.print_debug(true)?;
}
let mut system_tables = self.system_tables.lock().unwrap();
if let Some(page) = system_tables
.table_tree
.flush_table_root_updates()
.unwrap()
.finalize_dirty_checksums()
.unwrap()
{
eprintln!("System tree:");
let master_tree: Btree<&str, InternalTableDefinition> = Btree::new(
Some(page),
PageHint::None,
self.transaction_guard.clone(),
PageResolver::new(self.mem.clone()),
)?;
master_tree.print_debug(true)?;
}
Ok(())
}
}
impl Drop for WriteTransaction {
fn drop(&mut self) {
if !self.completed && !crate::panicking() && !self.mem.storage_failure() {
#[allow(unused_variables)]
if let Err(error) = self.abort_inner() {
#[cfg(feature = "logging")]
warn!("Failure automatically aborting transaction: {error}");
}
} else if !self.completed && self.mem.storage_failure() {
self.tables
.lock()
.unwrap()
.table_tree
.clear_root_updates_and_close();
}
}
}
pub struct ReadTransaction {
mem: Arc<TransactionalMemory>,
tree: TableTree,
}
impl ReadTransaction {
pub(crate) fn new(
mem: Arc<TransactionalMemory>,
guard: TransactionGuard,
) -> Result<Self, TransactionError> {
let root_page = mem.get_data_root();
let guard = Arc::new(guard);
let resolver = PageResolver::new(mem.clone());
Ok(Self {
mem,
tree: TableTree::new(root_page, PageHint::Clean, guard, resolver)
.map_err(TransactionError::Storage)?,
})
}
pub fn open_table<K: Key + 'static, V: Value + 'static>(
&self,
definition: TableDefinition<K, V>,
) -> Result<ReadOnlyTable<K, V>, TableError> {
let header = self
.tree
.get_table::<K, V>(definition.name(), TableType::Normal)?
.ok_or_else(|| TableError::TableDoesNotExist(definition.name().to_string()))?;
match header {
InternalTableDefinition::Normal { table_root, .. } => Ok(ReadOnlyTable::new(
definition.name().to_string(),
table_root,
PageHint::Clean,
self.tree.transaction_guard().clone(),
PageResolver::new(self.mem.clone()),
)?),
InternalTableDefinition::Multimap { .. } => unreachable!(),
}
}
pub fn open_untyped_table(
&self,
handle: impl TableHandle,
) -> Result<ReadOnlyUntypedTable, TableError> {
let name = handle.name();
let header = self
.tree
.get_table_untyped(name, TableType::Normal)?
.ok_or_else(|| TableError::TableDoesNotExist(name.to_string()))?;
match header {
InternalTableDefinition::Normal {
table_root,
fixed_key_size,
fixed_value_size,
..
} => Ok(ReadOnlyUntypedTable::new(
name,
table_root,
PageHint::Clean,
fixed_key_size,
fixed_value_size,
PageResolver::new(self.mem.clone()),
)),
InternalTableDefinition::Multimap { .. } => unreachable!(),
}
}
pub fn open_multimap_table<K: Key + 'static, V: Key + 'static>(
&self,
definition: MultimapTableDefinition<K, V>,
) -> Result<ReadOnlyMultimapTable<K, V>, TableError> {
let header = self
.tree
.get_table::<K, V>(definition.name(), TableType::Multimap)?
.ok_or_else(|| TableError::TableDoesNotExist(definition.name().to_string()))?;
match header {
InternalTableDefinition::Normal { .. } => unreachable!(),
InternalTableDefinition::Multimap {
table_root,
table_length,
..
} => Ok(ReadOnlyMultimapTable::new(
definition.name(),
table_root,
table_length,
PageHint::Clean,
self.tree.transaction_guard().clone(),
PageResolver::new(self.mem.clone()),
)?),
}
}
pub fn open_untyped_multimap_table(
&self,
handle: impl MultimapTableHandle,
) -> Result<ReadOnlyUntypedMultimapTable, TableError> {
let name = handle.name();
let header = self
.tree
.get_table_untyped(name, TableType::Multimap)?
.ok_or_else(|| TableError::TableDoesNotExist(name.to_string()))?;
match header {
InternalTableDefinition::Normal { .. } => unreachable!(),
InternalTableDefinition::Multimap {
table_root,
table_length,
fixed_key_size,
fixed_value_size,
..
} => Ok(ReadOnlyUntypedMultimapTable::new(
name,
table_root,
table_length,
PageHint::Clean,
fixed_key_size,
fixed_value_size,
PageResolver::new(self.mem.clone()),
)),
}
}
pub fn list_tables(&self) -> Result<impl Iterator<Item = UntypedTableHandle>> {
self.tree
.list_tables(TableType::Normal)
.map(|x| x.into_iter().map(UntypedTableHandle::new))
}
pub fn list_multimap_tables(&self) -> Result<impl Iterator<Item = UntypedMultimapTableHandle>> {
self.tree
.list_tables(TableType::Multimap)
.map(|x| x.into_iter().map(UntypedMultimapTableHandle::new))
}
pub fn close(self) -> Result<(), TransactionError> {
if Arc::strong_count(self.tree.transaction_guard()) > 1 {
return Err(TransactionError::ReadTransactionStillInUse(Box::new(self)));
}
Ok(())
}
}
impl Debug for ReadTransaction {
fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
f.write_str("ReadTransaction")
}
}
#[cfg(test)]
mod test {
#[cfg(feature = "experimental-api-5")]
use crate::ReadableTable;
use crate::{Database, ReadableDatabase, StorageError, TableDefinition, TransactionError};
const X: TableDefinition<&str, &str> = TableDefinition::new("x");
const BIG_VALUE: TableDefinition<u64, &[u8]> = TableDefinition::new("big_value");
#[test]
fn discarded_allocator_state_poisons_database() {
let tmpfile = crate::create_tempfile();
let db = Database::create(tmpfile.path()).unwrap();
let txn = db.begin_write().unwrap();
{
let mut table = txn.open_table(X).unwrap();
for i in 0..100u32 {
table.insert(format!("key{i}").as_str(), "value").unwrap();
}
}
txn.commit().unwrap();
let mut txn = db.begin_write().unwrap();
{
let mut table = txn.open_table(X).unwrap();
for i in 0..50u32 {
table.remove(format!("key{i}").as_str()).unwrap();
}
}
txn.disable_post_commit_free();
txn.commit().unwrap();
db.get_memory().invalidate_allocator_state();
for _ in 0..2 {
match db.begin_write() {
Err(TransactionError::Storage(StorageError::Corrupted(_))) => {}
Err(err) => panic!("unexpected error: {err}"),
Ok(_) => panic!("begin_write() must fail"),
}
}
{
let read_txn = db.begin_read().unwrap();
let table = read_txn.open_table(X).unwrap();
assert!(table.get("key0").unwrap().is_none());
assert_eq!(table.get("key99").unwrap().unwrap().value(), "value");
}
drop(db);
let mut db = Database::open(tmpfile.path()).unwrap();
assert!(db.check_integrity().unwrap());
{
let read_txn = db.begin_read().unwrap();
let table = read_txn.open_table(X).unwrap();
assert_eq!(table.get("key99").unwrap().unwrap().value(), "value");
}
let txn = db.begin_write().unwrap();
{
let mut table = txn.open_table(X).unwrap();
table.insert("after-repair", "value").unwrap();
}
txn.commit().unwrap();
}
#[test]
fn transaction_id_persistence() {
let tmpfile = crate::create_tempfile();
let db = Database::create(tmpfile.path()).unwrap();
let write_txn = db.begin_write().unwrap();
{
let mut table = write_txn.open_table(X).unwrap();
table.insert("hello", "world").unwrap();
}
let first_txn_id = write_txn.transaction_id;
write_txn.commit().unwrap();
drop(db);
let db2 = Database::create(tmpfile.path()).unwrap();
let write_txn = db2.begin_write().unwrap();
assert!(write_txn.transaction_id > first_txn_id);
}
#[test]
fn post_commit_epilogue_reserves_transaction_id() {
let tmpfile = crate::create_tempfile();
let db = Database::create(tmpfile.path()).unwrap();
let value = vec![0; 512 * 1024];
let write_txn = db.begin_write().unwrap();
{
let mut table = write_txn.open_table(BIG_VALUE).unwrap();
table.insert(0, value.as_slice()).unwrap();
}
write_txn.commit().unwrap();
let write_txn = db.begin_write().unwrap();
{
let mut table = write_txn.open_table(BIG_VALUE).unwrap();
table.remove(0).unwrap();
}
let remove_txn_id = write_txn.transaction_id;
write_txn.commit().unwrap();
let write_txn = db.begin_write().unwrap();
assert!(write_txn.transaction_id > remove_txn_id.next());
}
}