diff --git a/Cargo.toml b/Cargo.toml index 36903ce7..5005622d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,6 +22,8 @@ wti-hybrid-search = ["indexset/wt-slice-binary-search"] wti-predictable-search = ["indexset/custom-binary-search"] wti-std-search = ["indexset/std-binary-search"] wti-superslice-search = ["indexset/superslice-binary-search"] +# Compatibility no-op: immutable row publication is mandatory for the safe +# generated API, including `default-features = false` builds. versioned-row-publication = ["worktable_codegen/versioned-row-publication"] # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html @@ -32,14 +34,14 @@ arctic = { package = "arctic-wt", version = "0.1.4" } congee = { package = "congee-wt", version = "0.4.1" } convert_case = "0.6.0" crc32fast = "1.5.0" -data_bucket = "=0.5.1" +data_bucket = "=0.5.2" # data_bucket = { git = "https://github.com/pathscale/DataBucket", branch = "page_cdc_correction", version = "0.2.7" } # data_bucket = { path = "../DataBucket", version = "0.3.14" } derive_more = { version = "2.0.1", features = ["from", "error", "display", "debug", "into"] } eyre = "0.6.12" fastrand = "2.3.0" futures = "0.3.30" -indexset = { package = "WorkTablesIndex", version = "=0.0.4", default-features = false, features = ["concurrent", "cdc", "multimap"] } +indexset = { package = "WorkTablesIndex", version = "=0.0.5", default-features = false, features = ["concurrent", "cdc", "multimap"] } vanilla_indexset = { package = "indexset", version = "=0.15.0", features = ["concurrent", "cdc", "multimap"] } # indexset = { path = "../indexset", version = "0.15.0", features = ["concurrent", "cdc", "multimap"] } # indexset = { package = "wt-indexset", version = "=0.12.12", features = ["concurrent", "cdc", "multimap"] } diff --git a/README.md b/README.md index 85390100..c772af8f 100644 --- a/README.md +++ b/README.md @@ -69,33 +69,32 @@ WorkTablesIndex uses its predictable branch-based node search by default in Work ## Concurrent read/write publication -The default build preserves the existing lowest-latency page path and requires -applications to exclude reads that overlap page-byte mutation. Applications -that need generated reads to overlap updates, inserts, deletes, and vacuum can -opt into immutable row-version publication: - -```toml -[dependencies] -worktable = { version = "=1.0.0-beta.2", features = ["versioned-row-publication"] } -``` +Generated reads always use immutable row-version publication. This is also true +for `default-features = false` builds: disabling a Cargo feature must not expose +a safe API that can race deserialization against page-byte mutation. The former +`versioned-row-publication` feature name remains accepted as a compatibility +no-op for existing manifests. Generated point lookups use a strict backend-specific visibility contract by -default. WorkTablesIndex 0.0.4 keeps the structural mapping pinned until its +default. WorkTablesIndex 0.0.5 keeps the structural mapping pinned until its selected node is locked, making both hits and misses definitive; contended lookups release the structural guard before waiting and retry the mapping. Congee and Arctic use their native concurrent point lookups. The explicit vanilla `using indexset` backend remains experimental and is excluded from the stable concurrent-read contract because upstream IndexSet does not expose an equivalent validation primitive. This index-visibility contract is independent -of the optional row publication mode above. +of row publication. -In this mode, generated reads acquire an immutable owned row version instead +Generated reads acquire an immutable owned row version instead of borrowing the mutable archived page image. Writers replace a per-row version only after a complete page mutation, insert visibility is an atomic lifecycle transition after every index is installed, and deleted or relocated links are not reused until readers that could have captured them have drained. Page bytes remain the persistence image and are internally serialized; range queries are -still non-snapshot reads. The mode intentionally trades memory, an atomic +still non-snapshot reads. Archived-page mutations currently take one table-wide +writer barrier, so even disjoint writes can serialize and latency-sensitive +deployments should validate contention throughput and tail latency. The +protocol intentionally trades memory, an atomic read-side grace-period counter, and publication bookkeeping for this stronger concurrent-read contract. See [`docs/versioned-row-publication.md`](docs/versioned-row-publication.md) for the diff --git a/codegen/Cargo.toml b/codegen/Cargo.toml index 1add3e47..c32ce6f2 100644 --- a/codegen/Cargo.toml +++ b/codegen/Cargo.toml @@ -8,6 +8,7 @@ repository = "https://github.com/pathscale/WorkTable" [features] s3-support = [] +# Compatibility no-op retained for downstream manifests. versioned-row-publication = [] [lib] diff --git a/codegen/src/common/model/primary_key.rs b/codegen/src/common/model/primary_key.rs index 6664eea1..bbcb6441 100644 --- a/codegen/src/common/model/primary_key.rs +++ b/codegen/src/common/model/primary_key.rs @@ -1,10 +1,10 @@ +use indexmap::IndexMap; use proc_macro2::{Ident, TokenStream}; -use std::collections::HashMap; #[derive(Debug, Clone)] pub struct PrimaryKey { pub ident: Ident, - pub values: HashMap, + pub values: IndexMap, } #[derive(Debug, Clone, Copy, PartialEq)] diff --git a/codegen/src/generators/in_memory/primary_key.rs b/codegen/src/generators/in_memory/primary_key.rs index 31de463a..72ac82d5 100644 --- a/codegen/src/generators/in_memory/primary_key.rs +++ b/codegen/src/generators/in_memory/primary_key.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use indexmap::IndexMap; use crate::common::model::{GeneratorType, PrimaryKey}; use crate::common::name_generator::{WorktableNameGenerator, is_unsized_vec}; @@ -27,7 +27,7 @@ impl InMemoryGenerator { .clone(), ) }) - .collect::>(); + .collect::>(); let def = self.gen_primary_key_type()?; let impl_ = self.gen_table_primary_key_impl()?; diff --git a/codegen/src/generators/in_memory/queries/delete.rs b/codegen/src/generators/in_memory/queries/delete.rs index fad4375f..ee3fa92b 100644 --- a/codegen/src/generators/in_memory/queries/delete.rs +++ b/codegen/src/generators/in_memory/queries/delete.rs @@ -46,7 +46,7 @@ impl InMemoryGenerator { { let pk: #pk_ident = pk.into(); let op_lock = { #full_row_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -69,6 +69,7 @@ impl InMemoryGenerator { where #pk_ident: From { let pk: #pk_ident = pk.into(); + let _mutation_guard = self.0.lock_manager.mutation_guard(&pk); #delete_logic core::result::Result::Ok(()) } @@ -118,7 +119,12 @@ impl InMemoryGenerator { return Err(e); } }; - let row = self.0.select(pk.clone()).unwrap(); + // Low-level staged or hydrated state can publish index + // reachability before clearing the row's ghost bit. Ordinary + // insert shares this delete's per-key mutation gate, but keep + // this boundary defensive instead of panicking on a hidden + // version. + let row = self.0.select(pk.clone()).ok_or(WorkTableError::NotFound)?; #process } } else { @@ -129,7 +135,7 @@ impl InMemoryGenerator { .get_value(&pk) .map(Into::into) .ok_or(WorkTableError::NotFound)?; - let row = self.0.select(pk.clone()).unwrap(); + let row = self.0.select(pk.clone()).ok_or(WorkTableError::NotFound)?; #process } } diff --git a/codegen/src/generators/in_memory/queries/in_place.rs b/codegen/src/generators/in_memory/queries/in_place.rs index 0ed20b22..c0c38720 100644 --- a/codegen/src/generators/in_memory/queries/in_place.rs +++ b/codegen/src/generators/in_memory/queries/in_place.rs @@ -106,7 +106,7 @@ impl InMemoryGenerator { { let pk: #pk_type = by.into(); let op_lock = { #custom_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), diff --git a/codegen/src/generators/in_memory/queries/update.rs b/codegen/src/generators/in_memory/queries/update.rs index 59421a69..911b7147 100644 --- a/codegen/src/generators/in_memory/queries/update.rs +++ b/codegen/src/generators/in_memory/queries/update.rs @@ -62,7 +62,7 @@ impl InMemoryGenerator { if true { drop(_guard); let op_lock = { #full_row_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -85,7 +85,7 @@ impl InMemoryGenerator { pub async fn update(&self, row: #row_ident) -> core::result::Result<(), WorkTableError> { let pk = row.get_primary_key(); let op_lock = { #full_row_lock }; - let guard = LockGuard::new( + let guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -247,7 +247,7 @@ impl InMemoryGenerator { if need_to_reinsert { drop(_guard); let op_lock = { #full_row_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -465,7 +465,7 @@ impl InMemoryGenerator { { let pk: #pk_ident = pk.into(); let op_lock = { #custom_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -549,11 +549,12 @@ impl InMemoryGenerator { let mut need_to_reinsert = true; #(#fields_check)* if need_to_reinsert { + drop(_mutation_guard); let old_guard = guards.remove(&pk).expect("guard should exist for this pk"); drop(old_guard); let op_lock = { #full_row_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -638,6 +639,7 @@ impl InMemoryGenerator { if self.0.data.select_non_ghosted(link)?.#by_field != by { continue; } + let _mutation_guard = self.0.lock_manager.mutation_guard(&pk); let mut bytes = rkyv::to_bytes::(&row) .map_err(|_| WorkTableError::SerializeError)?; @@ -725,7 +727,7 @@ impl InMemoryGenerator { let pk = self.0.data.select_non_ghosted(link)?.get_primary_key().clone(); let op_lock = { #custom_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), diff --git a/codegen/src/generators/in_memory/table/impls.rs b/codegen/src/generators/in_memory/table/impls.rs index 60e2718e..4b577930 100644 --- a/codegen/src/generators/in_memory/table/impls.rs +++ b/codegen/src/generators/in_memory/table/impls.rs @@ -160,11 +160,11 @@ impl InMemoryGenerator { /// Inserts the row if its primary key is absent, updates it /// otherwise. /// - /// A definitely absent key takes the same optimistic lock-free - /// insert path as `insert`. Existing keys and insertion collisions - /// acquire one full-row lock across the repeated existence check - /// and selected mutation, so upserts, updates, and deletes on the - /// same key cannot invalidate that decision. + /// A definitely absent key takes the same optimistic synchronous + /// insert path as `insert`. Insert and generated locked mutations + /// share a per-key mutation gate; existing keys and insertion + /// collisions also acquire the full-row lock across the repeated + /// existence check and selected mutation. pub async fn upsert(&self, row: #row_type) -> core::result::Result<(), WorkTableError> { let pk = row.get_primary_key(); if !self.0.primary_index.pk_map.contains_key(&pk) { @@ -174,9 +174,14 @@ impl InMemoryGenerator { core::result::Result::Err(e) => return core::result::Result::Err(e), } } + // Retries only fire when an existence flip invalidated the + // optimistic decision (NotFound / row-absent). The FIFO + // per-key mutation gate guarantees forward progress; retain a + // bounded scheduler backoff for repeated decision races. + let mut backoff_spins: u32 = 0; loop { let op_lock = { #full_row_lock }; - let guard = LockGuard::new( + let guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -185,11 +190,15 @@ impl InMemoryGenerator { let result = if self.0.primary_index.pk_map.contains_key(&pk) { self.update_with_guard(row.clone(), guard).await } else { + // `insert` acquires the same per-key mutation gate as + // `guard`; release the row operation before entering + // the synchronous insertion protocol, then retry the + // locked decision if another writer won the race. + drop(guard); match self.insert(row.clone()) { core::result::Result::Ok(_) => core::result::Result::Ok(()), - core::result::Result::Err(WorkTableError::PrimaryAlreadyExists) => { - self.update_with_guard(row.clone(), guard).await - } + core::result::Result::Err(WorkTableError::PrimaryAlreadyExists) => + core::result::Result::Err(WorkTableError::NotFound), core::result::Result::Err(e) => core::result::Result::Err(e), } }; @@ -199,7 +208,19 @@ impl InMemoryGenerator { core::result::Result::Err(WorkTableError::PagesError(e)) if e.is_row_absent() => {} other => return other, } - tokio::task::yield_now().await; + if backoff_spins < 8 { + backoff_spins = backoff_spins.saturating_add(1); + tokio::task::yield_now().await; + } else { + // Cap the exponent BEFORE shifting: `1u64 << 64` panics + // (overflow) in debug/test builds. Clamp the shift to a + // 256µs ceiling and saturate the counter so a long + // starvation streak can never overflow. + let exponent = core::cmp::min(backoff_spins - 8, 8); + let micros = core::cmp::min(1u64 << exponent, 256); + backoff_spins = backoff_spins.saturating_add(1); + tokio::time::sleep(std::time::Duration::from_micros(micros)).await; + } } } } diff --git a/codegen/src/generators/in_memory/table/index_fns.rs b/codegen/src/generators/in_memory/table/index_fns.rs index 68670b69..9f30ad7f 100644 --- a/codegen/src/generators/in_memory/table/index_fns.rs +++ b/codegen/src/generators/in_memory/table/index_fns.rs @@ -83,36 +83,26 @@ impl InMemoryGenerator { row.#row_field_ident.eq(&by) } }; - let select = if cfg!(feature = "versioned-row-publication") { - quote! { - for _ in 0..64 { - let link: Link = self.0.indexes.#field_ident - .lookup_for_select(#by) - .map(Into::into)?; - if let Ok(row) = self.0.data.select_non_ghosted(link) { - if #predicate_matches { - return Some(row); - } - } - - let current_link: Option = self.0.indexes.#field_ident - .lookup_for_select(#by) - .map(Into::into); - if current_link == Some(link) { - return None; - } - std::hint::spin_loop(); - } - None - } - } else { - quote! { + let select = quote! { + for _ in 0..64 { let link: Link = self.0.indexes.#field_ident .lookup_for_select(#by) .map(Into::into)?; - let row = self.0.data.select_non_ghosted(link).ok()?; - #predicate_matches.then_some(row) + if let Ok(row) = self.0.data.select_non_ghosted(link) { + if #predicate_matches { + return Some(row); + } + } + + let current_link: Option = self.0.indexes.#field_ident + .lookup_for_select(#by) + .map(Into::into); + if current_link == Some(link) { + return None; + } + std::hint::spin_loop(); } + None }; Ok(quote! { @@ -182,33 +172,21 @@ impl InMemoryGenerator { let row_field_ident = &idx.field; let column_pascal = Ident::new(&i.to_string().to_case(Case::Pascal), Span::mixed_site()); - let revalidate = cfg!(feature = "versioned-row-publication"); let (range_bounds, range_arg) = if is_float(type_.to_string().as_str()) { ( quote! { std::ops::RangeBounds<#type_> }, - if revalidate { - quote! { - ( - predicate_range.0.as_ref().map(|v| OrderedFloat(*v)), - predicate_range.1.as_ref().map(|v| OrderedFloat(*v)), - ) - } - } else { - quote! { + quote! { ( - range.start_bound().map(|v| OrderedFloat(*v)), - range.end_bound().map(|v| OrderedFloat(*v)), + predicate_range.0.as_ref().map(|v| OrderedFloat(*v)), + predicate_range.1.as_ref().map(|v| OrderedFloat(*v)), ) - } }, ) - } else if revalidate { + } else { ( quote! { std::ops::RangeBounds<#type_> }, quote! { predicate_range.clone() }, ) - } else { - (quote! { std::ops::RangeBounds<#type_> }, quote! { range }) }; let (index_range, select_row) = if idx.is_unique { ( @@ -231,21 +209,17 @@ impl InMemoryGenerator { }, ) }; - let predicate_setup = revalidate.then(|| { - quote! { - let predicate_range = ( - range.start_bound().cloned(), - range.end_bound().cloned(), - ); - } - }); - let predicate_filter = revalidate.then(|| { - quote! { - .filter(move |row| { - std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident) - }) - } - }); + let predicate_setup = quote! { + let predicate_range = ( + range.start_bound().cloned(), + range.end_bound().cloned(), + ); + }; + let predicate_filter = quote! { + .filter(move |row| { + std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident) + }) + }; Ok(quote! { pub fn #fn_name<'a, R>(&'a self, range: R) -> SelectQueryBuilder<#row_ident, diff --git a/codegen/src/generators/persist/primary_key.rs b/codegen/src/generators/persist/primary_key.rs index 68622fcf..5929f0bf 100644 --- a/codegen/src/generators/persist/primary_key.rs +++ b/codegen/src/generators/persist/primary_key.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use indexmap::IndexMap; use crate::common::model::{GeneratorType, PrimaryKey}; use crate::common::name_generator::{WorktableNameGenerator, is_unsized_vec}; @@ -26,7 +26,7 @@ impl PersistGenerator { .clone(), ) }) - .collect::>(); + .collect::>(); let def = self.gen_primary_key_type()?; let impl_ = self.gen_table_primary_key_impl()?; diff --git a/codegen/src/generators/persist/queries/delete.rs b/codegen/src/generators/persist/queries/delete.rs index 0c3e2290..d0f39acf 100644 --- a/codegen/src/generators/persist/queries/delete.rs +++ b/codegen/src/generators/persist/queries/delete.rs @@ -46,7 +46,7 @@ impl PersistGenerator { { let pk: #pk_ident = pk.into(); let op_lock = { #full_row_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -69,6 +69,7 @@ impl PersistGenerator { where #pk_ident: From { let pk: #pk_ident = pk.into(); + let _mutation_guard = self.0.lock_manager.mutation_guard(&pk); #delete_logic core::result::Result::Ok(()) } @@ -111,7 +112,12 @@ impl PersistGenerator { return Err(e); } }; - let row = self.0.select(pk.clone()).unwrap(); + // Low-level staged or hydrated state can publish index + // reachability before clearing the row's ghost bit. Ordinary + // insert shares this delete's per-key mutation gate, but keep + // this boundary defensive instead of panicking on a hidden + // version. + let row = self.0.select(pk.clone()).ok_or(WorkTableError::NotFound)?; #process } } else { @@ -122,7 +128,7 @@ impl PersistGenerator { .get_value(&pk) .map(Into::into) .ok_or(WorkTableError::NotFound)?; - let row = self.0.select(pk.clone()).unwrap(); + let row = self.0.select(pk.clone()).ok_or(WorkTableError::NotFound)?; #process } } diff --git a/codegen/src/generators/persist/queries/in_place.rs b/codegen/src/generators/persist/queries/in_place.rs index 6de2d822..fcc066e8 100644 --- a/codegen/src/generators/persist/queries/in_place.rs +++ b/codegen/src/generators/persist/queries/in_place.rs @@ -108,7 +108,7 @@ impl PersistGenerator { { let pk: #pk_type = by.into(); let op_lock = { #custom_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), diff --git a/codegen/src/generators/persist/queries/update.rs b/codegen/src/generators/persist/queries/update.rs index b9108081..dbe0dfcb 100644 --- a/codegen/src/generators/persist/queries/update.rs +++ b/codegen/src/generators/persist/queries/update.rs @@ -62,7 +62,7 @@ impl PersistGenerator { if true { drop(_guard); let op_lock = { #full_row_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -85,7 +85,7 @@ impl PersistGenerator { pub async fn update(&self, row: #row_ident) -> core::result::Result<(), WorkTableError> { let pk = row.get_primary_key(); let op_lock = { #full_row_lock }; - let guard = LockGuard::new( + let guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -243,7 +243,7 @@ impl PersistGenerator { if need_to_reinsert { drop(_guard); let op_lock = { #full_row_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -422,7 +422,7 @@ impl PersistGenerator { { let pk: #pk_ident = pk.into(); let op_lock = { #custom_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -506,11 +506,12 @@ impl PersistGenerator { let mut need_to_reinsert = true; #(#fields_check)* if need_to_reinsert { + drop(_mutation_guard); let old_guard = guards.remove(&pk).expect("guard should exist for this pk"); drop(old_guard); let op_lock = { #full_row_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -595,6 +596,7 @@ impl PersistGenerator { if self.0.data.select_non_ghosted(link)?.#by_field != by { continue; } + let _mutation_guard = self.0.lock_manager.mutation_guard(&pk); let mut bytes = rkyv::to_bytes::(&row) .map_err(|_| WorkTableError::SerializeError)?; @@ -682,7 +684,7 @@ impl PersistGenerator { let pk = self.0.data.select_non_ghosted(link)?.get_primary_key().clone(); let op_lock = { #custom_lock }; - let _guard = LockGuard::new( + let _guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), diff --git a/codegen/src/generators/persist/table/impls.rs b/codegen/src/generators/persist/table/impls.rs index 4f8267c6..1187d5c4 100644 --- a/codegen/src/generators/persist/table/impls.rs +++ b/codegen/src/generators/persist/table/impls.rs @@ -279,11 +279,11 @@ impl PersistGenerator { /// Inserts the row if its primary key is absent, updates it /// otherwise. /// - /// A definitely absent key takes the same optimistic lock-free - /// insert path as `insert`. Existing keys and insertion collisions - /// acquire one full-row lock across the repeated existence check - /// and selected mutation, so upserts, updates, and deletes on the - /// same key cannot invalidate that decision. + /// A definitely absent key takes the same optimistic synchronous + /// insert path as `insert`. Insert and generated locked mutations + /// share a per-key mutation gate; existing keys and insertion + /// collisions also acquire the full-row lock across the repeated + /// existence check and selected mutation. pub async fn upsert(&self, row: #row_type) -> core::result::Result<(), WorkTableError> { let pk = row.get_primary_key(); if !self.0.primary_index.pk_map.contains_key(&pk) { @@ -293,9 +293,14 @@ impl PersistGenerator { core::result::Result::Err(e) => return core::result::Result::Err(e), } } + // Retries only fire when an existence flip invalidated the + // optimistic decision (NotFound / row-absent). The FIFO + // per-key mutation gate guarantees forward progress; retain a + // bounded scheduler backoff for repeated decision races. + let mut backoff_spins: u32 = 0; loop { let op_lock = { #full_row_lock }; - let guard = LockGuard::new( + let guard = LockGuard::new_with_mutation( op_lock, self.0.lock_manager.clone(), pk.clone(), @@ -304,11 +309,15 @@ impl PersistGenerator { let result = if self.0.primary_index.pk_map.contains_key(&pk) { self.update_with_guard(row.clone(), guard).await } else { + // `insert` acquires the same per-key mutation gate as + // `guard`; release the row operation before entering + // the synchronous insertion protocol, then retry the + // locked decision if another writer won the race. + drop(guard); match self.insert(row.clone()) { core::result::Result::Ok(_) => core::result::Result::Ok(()), - core::result::Result::Err(WorkTableError::PrimaryAlreadyExists) => { - self.update_with_guard(row.clone(), guard).await - } + core::result::Result::Err(WorkTableError::PrimaryAlreadyExists) => + core::result::Result::Err(WorkTableError::NotFound), core::result::Result::Err(e) => core::result::Result::Err(e), } }; @@ -318,7 +327,19 @@ impl PersistGenerator { core::result::Result::Err(WorkTableError::PagesError(e)) if e.is_row_absent() => {} other => return other, } - tokio::task::yield_now().await; + if backoff_spins < 8 { + backoff_spins = backoff_spins.saturating_add(1); + tokio::task::yield_now().await; + } else { + // Cap the exponent BEFORE shifting: `1u64 << 64` panics + // (overflow) in debug/test builds. Clamp the shift to a + // 256µs ceiling and saturate the counter so a long + // starvation streak can never overflow. + let exponent = core::cmp::min(backoff_spins - 8, 8); + let micros = core::cmp::min(1u64 << exponent, 256); + backoff_spins = backoff_spins.saturating_add(1); + tokio::time::sleep(std::time::Duration::from_micros(micros)).await; + } } } } diff --git a/codegen/src/generators/persist/table/index_fns.rs b/codegen/src/generators/persist/table/index_fns.rs index bf99fbe6..0259e010 100644 --- a/codegen/src/generators/persist/table/index_fns.rs +++ b/codegen/src/generators/persist/table/index_fns.rs @@ -83,36 +83,26 @@ impl PersistGenerator { row.#row_field_ident.eq(&by) } }; - let select = if cfg!(feature = "versioned-row-publication") { - quote! { - for _ in 0..64 { - let link: Link = self.0.indexes.#field_ident - .lookup_for_select(#by) - .map(Into::into)?; - if let Ok(row) = self.0.data.select_non_ghosted(link) { - if #predicate_matches { - return Some(row); - } - } - - let current_link: Option = self.0.indexes.#field_ident - .lookup_for_select(#by) - .map(Into::into); - if current_link == Some(link) { - return None; - } - std::hint::spin_loop(); - } - None - } - } else { - quote! { + let select = quote! { + for _ in 0..64 { let link: Link = self.0.indexes.#field_ident .lookup_for_select(#by) .map(Into::into)?; - let row = self.0.data.select_non_ghosted(link).ok()?; - #predicate_matches.then_some(row) + if let Ok(row) = self.0.data.select_non_ghosted(link) { + if #predicate_matches { + return Some(row); + } + } + + let current_link: Option = self.0.indexes.#field_ident + .lookup_for_select(#by) + .map(Into::into); + if current_link == Some(link) { + return None; + } + std::hint::spin_loop(); } + None }; Ok(quote! { @@ -182,33 +172,21 @@ impl PersistGenerator { let row_field_ident = &idx.field; let column_pascal = Ident::new(&i.to_string().to_case(Case::Pascal), Span::mixed_site()); - let revalidate = cfg!(feature = "versioned-row-publication"); let (range_bounds, range_arg) = if is_float(type_.to_string().as_str()) { ( quote! { std::ops::RangeBounds<#type_> }, - if revalidate { - quote! { - ( - predicate_range.0.as_ref().map(|v| OrderedFloat(*v)), - predicate_range.1.as_ref().map(|v| OrderedFloat(*v)), - ) - } - } else { - quote! { + quote! { ( - range.start_bound().map(|v| OrderedFloat(*v)), - range.end_bound().map(|v| OrderedFloat(*v)), + predicate_range.0.as_ref().map(|v| OrderedFloat(*v)), + predicate_range.1.as_ref().map(|v| OrderedFloat(*v)), ) - } }, ) - } else if revalidate { + } else { ( quote! { std::ops::RangeBounds<#type_> }, quote! { predicate_range.clone() }, ) - } else { - (quote! { std::ops::RangeBounds<#type_> }, quote! { range }) }; let (index_range, select_row) = if idx.is_unique { ( @@ -231,21 +209,17 @@ impl PersistGenerator { }, ) }; - let predicate_setup = revalidate.then(|| { - quote! { - let predicate_range = ( - range.start_bound().cloned(), - range.end_bound().cloned(), - ); - } - }); - let predicate_filter = revalidate.then(|| { - quote! { - .filter(move |row| { - std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident) - }) - } - }); + let predicate_setup = quote! { + let predicate_range = ( + range.start_bound().cloned(), + range.end_bound().cloned(), + ); + }; + let predicate_filter = quote! { + .filter(move |row| { + std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident) + }) + }; Ok(quote! { pub fn #fn_name<'a, R>(&'a self, range: R) -> SelectQueryBuilder<#row_ident, diff --git a/codegen/src/generators/read_only/primary_key.rs b/codegen/src/generators/read_only/primary_key.rs index 99a89aa6..2fe733ca 100644 --- a/codegen/src/generators/read_only/primary_key.rs +++ b/codegen/src/generators/read_only/primary_key.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use indexmap::IndexMap; use crate::common::model::{GeneratorType, PrimaryKey}; use crate::common::name_generator::{WorktableNameGenerator, is_unsized_vec}; @@ -26,7 +26,7 @@ impl ReadOnlyGenerator { .clone(), ) }) - .collect::>(); + .collect::>(); let def = self.gen_primary_key_type()?; let impl_ = self.gen_table_primary_key_impl()?; diff --git a/codegen/src/generators/read_only/table/index_fns.rs b/codegen/src/generators/read_only/table/index_fns.rs index 42f9ff92..23b33dfb 100644 --- a/codegen/src/generators/read_only/table/index_fns.rs +++ b/codegen/src/generators/read_only/table/index_fns.rs @@ -83,36 +83,26 @@ impl ReadOnlyGenerator { row.#row_field_ident.eq(&by) } }; - let select = if cfg!(feature = "versioned-row-publication") { - quote! { - for _ in 0..64 { - let link: Link = self.0.indexes.#field_ident - .lookup_for_select(#by) - .map(Into::into)?; - if let Ok(row) = self.0.data.select_non_ghosted(link) { - if #predicate_matches { - return Some(row); - } - } - - let current_link: Option = self.0.indexes.#field_ident - .lookup_for_select(#by) - .map(Into::into); - if current_link == Some(link) { - return None; - } - std::hint::spin_loop(); - } - None - } - } else { - quote! { + let select = quote! { + for _ in 0..64 { let link: Link = self.0.indexes.#field_ident .lookup_for_select(#by) .map(Into::into)?; - let row = self.0.data.select_non_ghosted(link).ok()?; - #predicate_matches.then_some(row) + if let Ok(row) = self.0.data.select_non_ghosted(link) { + if #predicate_matches { + return Some(row); + } + } + + let current_link: Option = self.0.indexes.#field_ident + .lookup_for_select(#by) + .map(Into::into); + if current_link == Some(link) { + return None; + } + std::hint::spin_loop(); } + None }; Ok(quote! { @@ -182,33 +172,21 @@ impl ReadOnlyGenerator { let row_field_ident = &idx.field; let column_pascal = Ident::new(&i.to_string().to_case(Case::Pascal), Span::mixed_site()); - let revalidate = cfg!(feature = "versioned-row-publication"); let (range_bounds, range_arg) = if is_float(type_.to_string().as_str()) { ( quote! { std::ops::RangeBounds<#type_> }, - if revalidate { - quote! { - ( - predicate_range.0.as_ref().map(|v| OrderedFloat(*v)), - predicate_range.1.as_ref().map(|v| OrderedFloat(*v)), - ) - } - } else { - quote! { + quote! { ( - range.start_bound().map(|v| OrderedFloat(*v)), - range.end_bound().map(|v| OrderedFloat(*v)), + predicate_range.0.as_ref().map(|v| OrderedFloat(*v)), + predicate_range.1.as_ref().map(|v| OrderedFloat(*v)), ) - } }, ) - } else if revalidate { + } else { ( quote! { std::ops::RangeBounds<#type_> }, quote! { predicate_range.clone() }, ) - } else { - (quote! { std::ops::RangeBounds<#type_> }, quote! { range }) }; let (index_range, select_row) = if idx.is_unique { ( @@ -231,21 +209,17 @@ impl ReadOnlyGenerator { }, ) }; - let predicate_setup = revalidate.then(|| { - quote! { - let predicate_range = ( - range.start_bound().cloned(), - range.end_bound().cloned(), - ); - } - }); - let predicate_filter = revalidate.then(|| { - quote! { - .filter(move |row| { - std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident) - }) - } - }); + let predicate_setup = quote! { + let predicate_range = ( + range.start_bound().cloned(), + range.end_bound().cloned(), + ); + }; + let predicate_filter = quote! { + .filter(move |row| { + std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident) + }) + }; Ok(quote! { pub fn #fn_name<'a, R>(&'a self, range: R) -> SelectQueryBuilder<#row_ident, diff --git a/codegen/src/worktable/mod.rs b/codegen/src/worktable/mod.rs index 3725345f..0097313f 100644 --- a/codegen/src/worktable/mod.rs +++ b/codegen/src/worktable/mod.rs @@ -137,6 +137,40 @@ mod tests { use super::expand; + fn assert_composite_primary_key_field_order(output: proc_macro2::TokenStream) { + let output = output.to_string(); + let get_primary_key = output + .split("fn get_primary_key") + .nth(1) + .expect("generated TableRow implementation"); + let tenant = get_primary_key + .find("self . tenant_id . clone") + .expect("first primary-key field"); + let record = get_primary_key + .find("self . record_id . clone") + .expect("second primary-key field"); + + assert!(tenant < record, "composite primary-key declaration order changed"); + } + + #[test] + fn composite_primary_key_codegen_preserves_declaration_order() { + for persist in [true, false] { + let output = expand(quote! { + name: CompositePrimaryKeyOrder, + persist: #persist, + columns: { + tenant_id: u64 primary_key, + record_id: u64 primary_key, + value: i64, + }, + }) + .unwrap(); + + assert_composite_primary_key_field_order(output); + } + } + #[test] fn absent_using_keeps_worktables_index_default() { let output = expand(quote! { diff --git a/docs/index-backend-dsl-proposal.md b/docs/index-backend-dsl-proposal.md index 72e3f94b..aac75ee5 100644 --- a/docs/index-backend-dsl-proposal.md +++ b/docs/index-backend-dsl-proposal.md @@ -158,22 +158,26 @@ Backend dispatch itself is static and should compile away. That does **not** mea - Both can emit persistence-compatible structural CDC. Direct dispatch preserves a strict generated point-read contract with a -provider-specific implementation. WorkTablesIndex 0.0.4 holds its structural +provider-specific implementation. WorkTablesIndex 0.0.5 holds its structural mapping stable until the selected node is locked, so both hits and misses are definitive; a contended lookup drops the structural guard before waiting and then retries the mapping. Vanilla IndexSet does not expose a comparable structural validation primitive; `using indexset` therefore remains experimental and is excluded from concurrent correctness and published -performance claims. This WorkTablesIndex visibility guarantee is independent of -`versioned-row-publication`, which addresses concurrent page bytes, ghost -publication, and reclamation rather than index routing. +performance claims. This WorkTablesIndex visibility guarantee composes with +the mandatory immutable row-publication protocol, which addresses concurrent +page bytes, ghost publication, and reclamation rather than index routing. ### Congee - Point lookup and mutation call Congee directly. - WorkTable links do not fit in Congee's one-word payload. The adapter stores an `Arc` pointer, so inserts allocate and reads clone the `Arc` before copying the link. - Ordered reads use Congee's native range scan and materialize only the requested key interval into a `Vec`; a full iteration is therefore O(n) with one result allocation, while a narrow range no longer dumps, re-probes, and sorts the whole tree. -- With `persist: true`, mutations additionally take a key-striped sequencing lock before producing one logical WAL event. Memory-only Congee does not pay that cost. +- Memory-only Congee point reads remain native and concurrent. Mutations use a + WorkTable adapter mutex because congee-wt 0.4.1 can otherwise lose disjoint + structural insert/remove updates. With `persist: true`, the persistence layer + additionally takes a key-striped sequencing lock before producing one logical + WAL event. ### Arctic @@ -205,7 +209,7 @@ Use allocator/RSS measurements for comparative memory results; do not treat the This implementation pins two narrow forks for typed topology import/export: -- `WorkTablesIndex 0.0.4` as the default `indexset` dependency alias already used by WorkTable; +- `WorkTablesIndex 0.0.5` as the default `indexset` dependency alias already used by WorkTable; - vanilla `indexset 0.15.0` under the `vanilla_indexset` Cargo name; - `congee-wt` at commit `005bfb1968e781800176f2d7e465e6a1af630e1a`; - `arctic-wt` at commit `e13fc7df3c040f14ae66c1cb56b1bd0a3f6da3fc`. diff --git a/docs/versioned-row-publication.md b/docs/versioned-row-publication.md index 9d79dd2b..4e22a047 100644 --- a/docs/versioned-row-publication.md +++ b/docs/versioned-row-publication.md @@ -1,6 +1,7 @@ # Versioned row publication -Status: feature-gated prototype behind `versioned-row-publication`. +Status: mandatory for generated safe APIs. The former +`versioned-row-publication` Cargo feature is retained as a compatibility no-op. ## Problem @@ -19,11 +20,12 @@ the interval from index lookup through acquisition of a stable row version. ## Protocol -With the feature enabled, `DataPages` maintains two representations: +`DataPages` maintains two representations: - Archived page bytes are the compact persistence and mutation image. All - accesses that can overlap a mutation are serialized by an internal page - barrier. + accesses that can overlap a mutation are serialized by one internal + table-wide page barrier. Its exclusive side currently serializes mutations + to disjoint rows and pages as well as mutations to the same page. - A concurrent link map holds an immutable application-visible row version. Each slot contains one `Arc` and its ghost, deleted, and vacuum lifecycle bits in a single version protected by a short per-slot lock. A reader cannot @@ -36,7 +38,7 @@ The generated API follows these publication rules: index predicates, clone the owned row, and release the guard. Unique and primary-key point reads retry when the mapping swings to a replacement link while it is being resolved. Point lookup itself uses each provider's strict - visibility path: WorkTablesIndex 0.0.4 holds its structural mapping stable + visibility path: WorkTablesIndex 0.0.5 holds its structural mapping stable until the selected node is locked, making hits and misses definitive, while the ART providers retain their native concurrent point algorithms. Vanilla `using indexset` is experimental and excluded from this concurrent-read @@ -63,7 +65,7 @@ The generated API follows these publication rules: admitting write traffic. Subsequent generated reads use the published version map. -The grace period is quiescent-state reclamation: a feature-only atomic counter +The grace period is quiescent-state reclamation: an atomic counter tracks generated reads, and retirement queues are drained when that counter is zero. `Arc` ownership independently keeps a version alive after a reader has acquired it. @@ -102,16 +104,22 @@ than spinning without a bound. The guarantee also does not cover callers that bypass generated table methods and directly invoke low-level `Data` page mutation APIs. -## Cost model and rollout - -The feature is off by default. Cargo features unify across a dependency graph, -so any dependency enabling it enables it for every WorkTable consumer in that -build. It adds one owned row copy plus slot/map -metadata per live physical link, an atomic increment/decrement per generated -read, a sharded publication-map lookup, and writer-side page serialization. -The index-visibility algorithm is always active and separate from row -publication: WorkTablesIndex acquires the selected node while its structural -mapping is pinned on the uncontended path, and may retry after node contention. -Those costs are inappropriate to impose silently on latency-sensitive users. -The default path remains unchanged; benchmark results for both modes must be -reported before this feature is proposed for default enablement. +## Cost model + +The protocol adds one owned row copy plus slot/map metadata per live physical +link, an atomic increment/decrement per generated read, a sharded publication +map lookup, and writer-side page serialization. `page_access` is currently one +`RwLock<()>` per table, not one lock per row or page: insert, update, delete, +hydration, reset/reuse, and vacuum operations that take its exclusive side form +a table-wide writer barrier. Immutable published reads do not take that +exclusive side after hydration, but disjoint writes can still serialize and +therefore require contention-throughput and tail-latency validation for +latency-sensitive deployments. + +These correctness costs are mandatory: the previous fast path allowed safe +generated reads to race mutation of archived bytes, and performance cannot +justify undefined behavior. Finer-grained or sharded page mutation barriers may +reduce write contention without weakening the publication protocol. The +index-visibility algorithm is separate: WorkTablesIndex acquires the selected +node while its structural mapping is pinned on the uncontended path, and may +retry after node contention. diff --git a/src/in_memory/mod.rs b/src/in_memory/mod.rs index 0da40382..fefa6add 100644 --- a/src/in_memory/mod.rs +++ b/src/in_memory/mod.rs @@ -1,7 +1,6 @@ mod data; mod empty_link_registry; mod pages; -#[cfg(feature = "versioned-row-publication")] mod publication; mod row; diff --git a/src/in_memory/pages.rs b/src/in_memory/pages.rs index c832d944..987639f9 100644 --- a/src/in_memory/pages.rs +++ b/src/in_memory/pages.rs @@ -1,6 +1,5 @@ use data_bucket::page::PageId; use derive_more::{Display, Error, From}; -#[cfg(feature = "versioned-row-publication")] use parking_lot::Mutex; use parking_lot::RwLock; #[cfg(feature = "perf_measurements")] @@ -12,13 +11,9 @@ use rkyv::{ ser::{Serializer, allocator::ArenaHandle, sharing::Share}, util::AlignedVec, }; -#[cfg(feature = "versioned-row-publication")] -use std::collections::HashMap; -use std::collections::VecDeque; -#[cfg(feature = "versioned-row-publication")] +use std::collections::{HashMap, HashSet, VecDeque}; use std::hash::{BuildHasherDefault, Hasher}; use std::marker::PhantomData; -#[cfg(feature = "versioned-row-publication")] use std::sync::atomic::AtomicUsize; use std::{ fmt::Debug, @@ -27,10 +22,8 @@ use std::{ }; use crate::in_memory::empty_link_registry::EmptyLinkRegistry; -#[cfg(feature = "versioned-row-publication")] use crate::in_memory::publication::{DELETED, GHOSTED, PublishedRow, VACUUMED}; use crate::prelude::ArchivedRowWrapper; -#[cfg(feature = "versioned-row-publication")] use crate::util::OffsetEqLink; use crate::{ in_memory::{ @@ -44,12 +37,9 @@ fn page_id_mapper(page_id: usize) -> usize { page_id - 1usize } -#[cfg(feature = "versioned-row-publication")] const PUBLICATION_SHARD_COUNT: usize = 64; -#[cfg(feature = "versioned-row-publication")] const RETIREMENT_BACKLOG_WARN_AT: usize = 1_024; -#[cfg(feature = "versioned-row-publication")] fn mix_publication_offset(mut value: u64) -> u64 { value ^= value >> 30; value = value.wrapping_mul(0xbf58_476d_1ce4_e5b9); @@ -62,17 +52,14 @@ fn mix_publication_offset(mut value: u64) -> u64 { /// storage offset. Avalanche that offset so both hash-table bucket bits and /// SIMD control bits remain distributed for aligned, monotonically allocated /// row positions. -#[cfg(feature = "versioned-row-publication")] struct PublicationHasher(u64); -#[cfg(feature = "versioned-row-publication")] impl Default for PublicationHasher { fn default() -> Self { Self(0xcbf2_9ce4_8422_2325) } } -#[cfg(feature = "versioned-row-publication")] impl Hasher for PublicationHasher { fn finish(&self) -> u64 { self.0 @@ -92,20 +79,16 @@ impl Hasher for PublicationHasher { } } -#[cfg(feature = "versioned-row-publication")] type PublicationMap = HashMap, Arc>, BuildHasherDefault>; -#[cfg(feature = "versioned-row-publication")] type PublicationShards = [RwLock>; PUBLICATION_SHARD_COUNT]; -#[cfg(feature = "versioned-row-publication")] fn publication_shard(key: &OffsetEqLink) -> usize { mix_publication_offset(key.absolute_index()) as usize & (PUBLICATION_SHARD_COUNT - 1) } -#[cfg(feature = "versioned-row-publication")] fn queue_retirement(queue: &Mutex>, pending_retirements: &AtomicUsize, queue_name: &'static str, value: T) { let mut queue = queue.lock(); queue.push(value); @@ -121,19 +104,17 @@ fn queue_retirement(queue: &Mutex>, pending_retirements: &AtomicUsize, } pub struct ReadGuard<'a> { - #[cfg(feature = "versioned-row-publication")] active_readers: &'a AtomicU64, marker: PhantomData<&'a ()>, } impl Drop for ReadGuard<'_> { fn drop(&mut self) { - #[cfg(feature = "versioned-row-publication")] self.active_readers.fetch_sub(1, Ordering::SeqCst); } } -/// Page storage and, when enabled, immutable row publication. +/// Page storage with immutable row publication. /// /// # Versioned-publication synchronization /// @@ -155,31 +136,24 @@ where { /// Immutable application-visible row versions. Published readers never /// borrow the mutable archived page image. - #[cfg(feature = "versioned-row-publication")] published_rows: PublicationShards, /// Protects the mutable page image used by writers, vacuum, and /// persistence. Application reads use `published_rows` after hydration. - #[cfg(feature = "versioned-row-publication")] page_access: RwLock<()>, /// Read-side grace period protecting the interval from index lookup until /// an immutable row version has been acquired. - #[cfg(feature = "versioned-row-publication")] active_readers: AtomicU64, - #[cfg(feature = "versioned-row-publication")] retired_links: Mutex>, - #[cfg(feature = "versioned-row-publication")] retired_pages: Mutex>, - #[cfg(feature = "versioned-row-publication")] retired_publications: Mutex>>, /// Avoids taking all retirement-queue mutexes on mutations when there is /// no reclamation work pending. - #[cfg(feature = "versioned-row-publication")] pending_retirements: AtomicUsize, /// Pages vector. Currently, not lock free. @@ -212,7 +186,6 @@ where Row: StorableRow, ::WrappedRow: RowWrapper, { - #[cfg(feature = "versioned-row-publication")] fn publication_flags(row: &::WrappedRow) -> u8 { let mut flags = 0; if row.is_ghosted() { @@ -227,7 +200,6 @@ where flags } - #[cfg(feature = "versioned-row-publication")] fn publish_wrapped_row(&self, link: Link, wrapped: ::WrappedRow) { let flags = Self::publication_flags(&wrapped); let row = wrapped.get_inner(); @@ -242,19 +214,16 @@ where } } - #[cfg(feature = "versioned-row-publication")] fn stage_published_row(&self, link: Link, row: Row) { let wrapped = ::WrappedRow::from_inner(row); self.publish_wrapped_row(link, wrapped); } - #[cfg(feature = "versioned-row-publication")] fn published_slot(&self, link: Link) -> Option>> { let key = OffsetEqLink(link); self.published_rows[publication_shard(&key)].read().get(&key).cloned() } - #[cfg(feature = "versioned-row-publication")] fn published_slot_or_hydrate(&self, link: Link) -> Result>, ExecutionError> where <::WrappedRow as Archive>::Archived: @@ -282,17 +251,14 @@ where } pub fn read_guard(&self) -> ReadGuard<'_> { - #[cfg(feature = "versioned-row-publication")] self.active_readers.fetch_add(1, Ordering::SeqCst); ReadGuard { - #[cfg(feature = "versioned-row-publication")] active_readers: &self.active_readers, marker: PhantomData, } } - #[cfg(feature = "versioned-row-publication")] fn reclaim_retired(&self) { if self.pending_retirements.load(Ordering::Acquire) == 0 { return; @@ -308,10 +274,16 @@ where return; } + // A whole-page retirement subsumes every free link within that page. + // Publishing both would let one allocator reset/reuse the page while + // another writes through an overlapping link from the same page. + let whole_pages: HashSet<_> = retired_pages.iter().copied().collect(); for link in retired_links.drain(..) { let key = OffsetEqLink(link); self.published_rows[publication_shard(&key)].write().remove(&key); - self.empty_links.push(link); + if !whole_pages.contains(&link.page_id) { + self.empty_links.push(link); + } } for key in retired_publications.drain(..) { self.published_rows[publication_shard(&key)].write().remove(&key); @@ -325,19 +297,12 @@ where pub fn new() -> Self { Self { - #[cfg(feature = "versioned-row-publication")] published_rows: std::array::from_fn(|_| RwLock::new(PublicationMap::default())), - #[cfg(feature = "versioned-row-publication")] page_access: RwLock::new(()), - #[cfg(feature = "versioned-row-publication")] active_readers: AtomicU64::new(0), - #[cfg(feature = "versioned-row-publication")] retired_links: Mutex::new(Vec::new()), - #[cfg(feature = "versioned-row-publication")] retired_pages: Mutex::new(Vec::new()), - #[cfg(feature = "versioned-row-publication")] retired_publications: Mutex::new(Vec::new()), - #[cfg(feature = "versioned-row-publication")] pending_retirements: AtomicUsize::new(0), // We are starting ID's from `1` because `0`'s page in file is info page. pages: RwLock::new(vec![Arc::new(Data::new(1.into()))]), @@ -356,19 +321,12 @@ where } else { let last_page_id = vec.len(); Self { - #[cfg(feature = "versioned-row-publication")] published_rows: std::array::from_fn(|_| RwLock::new(PublicationMap::default())), - #[cfg(feature = "versioned-row-publication")] page_access: RwLock::new(()), - #[cfg(feature = "versioned-row-publication")] active_readers: AtomicU64::new(0), - #[cfg(feature = "versioned-row-publication")] retired_links: Mutex::new(Vec::new()), - #[cfg(feature = "versioned-row-publication")] retired_pages: Mutex::new(Vec::new()), - #[cfg(feature = "versioned-row-publication")] retired_publications: Mutex::new(Vec::new()), - #[cfg(feature = "versioned-row-publication")] pending_retirements: AtomicUsize::new(0), pages: RwLock::new(vec), empty_links: EmptyLinkRegistry::default(), @@ -388,16 +346,11 @@ where ::WrappedRow: Archive + for<'a> Serialize, Share>, rkyv::rancor::Error>>, { - #[cfg(feature = "versioned-row-publication")] let general_row = ::WrappedRow::from_inner(row.clone()); - #[cfg(not(feature = "versioned-row-publication"))] - let general_row = ::WrappedRow::from_inner(row); - #[cfg(feature = "versioned-row-publication")] self.reclaim_retired(); if let Some(link) = self.empty_links.pop_max() { - #[cfg(feature = "versioned-row-publication")] let _page_access = self.page_access.write(); let pages = self.pages.read(); let current_page: usize = page_id_mapper(link.page_id.into()); @@ -408,7 +361,6 @@ where if let Some(l) = left_link { self.empty_links.push(l); } - #[cfg(feature = "versioned-row-publication")] self.stage_published_row(link, row); return Ok(link); } @@ -426,7 +378,6 @@ where loop { let (link, tried_page) = { - #[cfg(feature = "versioned-row-publication")] let _page_access = self.page_access.write(); let pages = self.pages.read(); let current_page = page_id_mapper(self.current_page_id.load(Ordering::Acquire) as usize); @@ -436,7 +387,6 @@ where }; match link { Ok(link) => { - #[cfg(feature = "versioned-row-publication")] self.stage_published_row(link, row); self.row_count.fetch_add(1, Ordering::Relaxed); return Ok(link); @@ -444,12 +394,17 @@ where Err(e) => match e { DataExecutionError::PageIsFull { .. } => { if tried_page == page_id_mapper(self.current_page_id.load(Ordering::Relaxed) as usize) { - let mut g = self.empty_pages.write(); - if let Some(page_id) = g.pop_front() { - let _pages = self.pages.write(); + let empty_page = self.empty_pages.write().pop_front(); + if let Some(page_id) = empty_page { + // Retired pages retain their old bytes until + // the read-side grace period completes. Reset + // only after reclamation made the page + // available for reuse. + let _page_access = self.page_access.write(); + let pages = self.pages.read(); + pages[page_id_mapper(page_id.into())].reset(); self.current_page_id.store(page_id.into(), Ordering::Release); } else { - drop(g); self.add_next_page(tried_page); } } @@ -492,7 +447,6 @@ where /// Allocates a new page or reuses a free page from `empty_pages`. /// Does **NOT** set the page as `current`. pub fn allocate_new_or_pop_free(&self) -> Arc::WrappedRow, DATA_LENGTH>> { - #[cfg(feature = "versioned-row-publication")] self.reclaim_retired(); let page_id = { @@ -501,7 +455,6 @@ where }; if let Some(page_id) = page_id { - #[cfg(feature = "versioned-row-publication")] let _page_access = self.page_access.write(); let pages = self.pages.read(); let index = page_id_mapper(page_id.into()); @@ -529,21 +482,8 @@ where Portable + Deserialize<::WrappedRow, HighDeserializer>, { let link = link.into(); - #[cfg(feature = "versioned-row-publication")] - { - let slot = self.published_slot_or_hydrate(link)?; - Ok(slot.snapshot().as_ref().clone()) - } - - #[cfg(not(feature = "versioned-row-publication"))] - { - let pages = self.pages.read(); - let page = pages - .get(page_id_mapper(link.page_id.into())) - .ok_or(ExecutionError::PageNotFound(link.page_id))?; - let gen_row = page.get_row(link).map_err(ExecutionError::DataPageError)?; - Ok(gen_row.get_inner()) - } + let slot = self.published_slot_or_hydrate(link)?; + Ok(slot.snapshot().as_ref().clone()) } pub fn select_non_ghosted(&self, link: Link) -> Result @@ -554,31 +494,15 @@ where <::WrappedRow as Archive>::Archived: Portable + Deserialize<::WrappedRow, HighDeserializer>, { - #[cfg(feature = "versioned-row-publication")] - { - let slot = self.published_slot_or_hydrate(link)?; - let (row, flags) = slot.load(); - if flags & GHOSTED != 0 { - return Err(ExecutionError::Ghosted); - } - if flags & DELETED != 0 { - return Err(ExecutionError::Deleted); - } - Ok(row.as_ref().clone()) + let slot = self.published_slot_or_hydrate(link)?; + let (row, flags) = slot.load(); + if flags & GHOSTED != 0 { + return Err(ExecutionError::Ghosted); } - - #[cfg(not(feature = "versioned-row-publication"))] - { - let pages = self.pages.read(); - let page = pages - .get(page_id_mapper(link.page_id.into())) - .ok_or(ExecutionError::PageNotFound(link.page_id))?; - let gen_row = page.get_row(link).map_err(ExecutionError::DataPageError)?; - if gen_row.is_ghosted() { - return Err(ExecutionError::Ghosted); - } - Ok(gen_row.get_inner()) + if flags & DELETED != 0 { + return Err(ExecutionError::Deleted); } + Ok(row.as_ref().clone()) } pub fn select_non_vacuumed(&self, link: Link) -> Result @@ -589,37 +513,18 @@ where <::WrappedRow as Archive>::Archived: Portable + Deserialize<::WrappedRow, HighDeserializer>, { - #[cfg(feature = "versioned-row-publication")] - { - let slot = self.published_slot_or_hydrate(link)?; - let (row, flags) = slot.load(); - if flags & GHOSTED != 0 { - return Err(ExecutionError::Ghosted); - } - if flags & VACUUMED != 0 { - return Err(ExecutionError::Vacuumed); - } - if flags & DELETED != 0 { - return Err(ExecutionError::Deleted); - } - Ok(row.as_ref().clone()) + let slot = self.published_slot_or_hydrate(link)?; + let (row, flags) = slot.load(); + if flags & GHOSTED != 0 { + return Err(ExecutionError::Ghosted); } - - #[cfg(not(feature = "versioned-row-publication"))] - { - let pages = self.pages.read(); - let page = pages - .get(page_id_mapper(link.page_id.into())) - .ok_or(ExecutionError::PageNotFound(link.page_id))?; - let gen_row = page.get_row(link).map_err(ExecutionError::DataPageError)?; - if gen_row.is_ghosted() { - return Err(ExecutionError::Ghosted); - } - if gen_row.is_vacuumed() { - return Err(ExecutionError::Vacuumed); - } - Ok(gen_row.get_inner()) + if flags & VACUUMED != 0 { + return Err(ExecutionError::Vacuumed); + } + if flags & DELETED != 0 { + return Err(ExecutionError::Deleted); } + Ok(row.as_ref().clone()) } #[cfg_attr(feature = "perf_measurements", performance_measurement(prefix_name = "DataPages"))] @@ -628,7 +533,6 @@ where Row: Archive + for<'a> Serialize, Share>, rkyv::rancor::Error>>, Op: Fn(&<::WrappedRow as Archive>::Archived) -> Res, { - #[cfg(feature = "versioned-row-publication")] let _page_access = self.page_access.read(); let pages = self.pages.read(); let page = pages @@ -649,7 +553,6 @@ where Deserialize<::WrappedRow, HighDeserializer>, Op: FnMut(&mut <::WrappedRow as Archive>::Archived) -> Res, { - #[cfg(feature = "versioned-row-publication")] let _page_access = self.page_access.write(); let pages = self.pages.read(); let page = pages @@ -664,7 +567,6 @@ where op(gen_row) }; - #[cfg(feature = "versioned-row-publication")] { let wrapped = page.get_row(link).map_err(ExecutionError::DataPageError)?; self.publish_wrapped_row(link, wrapped); @@ -685,21 +587,16 @@ where ::WrappedRow: Archive + for<'a> Serialize, Share>, rkyv::rancor::Error>>, { - #[cfg(feature = "versioned-row-publication")] let _page_access = self.page_access.write(); let pages = self.pages.read(); let page = pages .get(page_id_mapper(link.page_id.into())) .ok_or(ExecutionError::PageNotFound(link.page_id))?; - #[cfg(feature = "versioned-row-publication")] let gen_row = ::WrappedRow::from_inner(row.clone()); - #[cfg(not(feature = "versioned-row-publication"))] - let gen_row = ::WrappedRow::from_inner(row); let result = unsafe { page.save_row_by_link(&gen_row, link) .map_err(ExecutionError::DataPageError) }?; - #[cfg(feature = "versioned-row-publication")] self.stage_published_row(link, row); Ok(result) } @@ -715,19 +612,12 @@ where { unsafe { self.with_mut_ref(link, |r| r.delete())? } - #[cfg(feature = "versioned-row-publication")] - { - queue_retirement(&self.retired_links, &self.pending_retirements, "links", link); - self.reclaim_retired(); - } - - #[cfg(not(feature = "versioned-row-publication"))] - self.empty_links.push(link); + queue_retirement(&self.retired_links, &self.pending_retirements, "links", link); + self.reclaim_retired(); Ok(()) } pub fn select_raw(&self, link: Link) -> Result, ExecutionError> { - #[cfg(feature = "versioned-row-publication")] let _page_access = self.page_access.read(); let pages = self.pages.read(); let page = pages @@ -738,17 +628,8 @@ where pub fn mark_page_empty(&self, page_id: PageId) { if u32::from(page_id) != self.current_page_id.load(Ordering::Acquire) { - #[cfg(feature = "versioned-row-publication")] - { - queue_retirement(&self.retired_pages, &self.pending_retirements, "pages", page_id); - self.reclaim_retired(); - } - - #[cfg(not(feature = "versioned-row-publication"))] - { - let mut g = self.empty_pages.write(); - g.push_back(page_id); - } + queue_retirement(&self.retired_pages, &self.pending_retirements, "pages", page_id); + self.reclaim_retired(); } } @@ -794,7 +675,6 @@ where } pub fn get_bytes(&self) -> Vec<([u8; DATA_LENGTH], u32)> { - #[cfg(feature = "versioned-row-publication")] let _page_access = self.page_access.read(); let pages = self.pages.read(); pages @@ -803,18 +683,6 @@ where .collect() } - pub(crate) fn reset_page(&self, page_id: PageId) -> Result<(), ExecutionError> { - #[cfg(feature = "versioned-row-publication")] - let _page_access = self.page_access.write(); - let pages = self.pages.read(); - let page = pages - .get(page_id_mapper(page_id.into())) - .ok_or(ExecutionError::PageNotFound(page_id))?; - page.reset(); - - Ok(()) - } - /// Copies a row to another page without exposing either mutable byte /// image to application readers. /// @@ -837,7 +705,6 @@ where + Portable + Deserialize<::WrappedRow, HighDeserializer>, { - #[cfg(feature = "versioned-row-publication")] let _page_access = self.page_access.write(); let pages = self.pages.read(); let from_page = pages @@ -859,7 +726,6 @@ where archived.set_in_vacuum_process(); let new_link = to_page.save_raw_row(&raw_data).map_err(ExecutionError::DataPageError)?; - #[cfg(feature = "versioned-row-publication")] { let old_wrapped = from_page.get_row(from_link).map_err(ExecutionError::DataPageError)?; self.publish_wrapped_row(from_link, old_wrapped); @@ -871,19 +737,13 @@ where } pub(crate) fn retire_published_link(&self, link: Link) { - #[cfg(feature = "versioned-row-publication")] - { - queue_retirement( - &self.retired_publications, - &self.pending_retirements, - "publications", - OffsetEqLink(link), - ); - self.reclaim_retired(); - } - - #[cfg(not(feature = "versioned-row-publication"))] - let _ = link; + queue_retirement( + &self.retired_publications, + &self.pending_retirements, + "publications", + OffsetEqLink(link), + ); + self.reclaim_retired(); } pub fn get_page_count(&self) -> usize { @@ -950,10 +810,8 @@ mod tests { use std::collections::HashSet; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; - #[cfg(feature = "versioned-row-publication")] use std::sync::mpsc; use std::thread; - #[cfg(feature = "versioned-row-publication")] use std::time::Duration; use std::time::Instant; @@ -1091,7 +949,6 @@ mod tests { assert_eq!(res.err(), Some(PagesExecutionError::Ghosted)) } - #[cfg(feature = "versioned-row-publication")] #[test] fn versioned_insert_stays_hidden_until_unghost() { let pages = DataPages::::new(); @@ -1105,7 +962,6 @@ mod tests { assert_eq!(pages.select_non_ghosted(link), Ok(row)); } - #[cfg(feature = "versioned-row-publication")] #[test] fn versioned_reader_observes_old_row_while_page_update_is_incomplete() { let pages = Arc::new(DataPages::::new()); @@ -1147,7 +1003,6 @@ mod tests { assert_eq!(pages.select_non_ghosted(link), Ok(TestRow { a: 1, b: 1 })); } - #[cfg(feature = "versioned-row-publication")] #[test] fn retired_version_survives_link_reuse_for_in_flight_reader() { let pages = DataPages::::new(); @@ -1170,7 +1025,6 @@ mod tests { assert_eq!(pages.select_non_ghosted(reused_link), Ok(TestRow { a: 2, b: 2 })); } - #[cfg(feature = "versioned-row-publication")] #[test] fn read_grace_period_prevents_link_aba() { let pages = DataPages::::new(); @@ -1189,6 +1043,75 @@ mod tests { assert!(pages.get_empty_links().contains(&old_link)); } + #[test] + fn read_grace_period_prevents_vacuumed_page_reuse() { + let pages = DataPages::::from_data(vec![ + Arc::new(Data::new(1.into())), + Arc::new(Data::new(2.into())), + Arc::new(Data::new(3.into())), + ]); + pages.current_page_id.store(2, Ordering::Release); + let old_link = pages.insert(TestRow { a: 1, b: 1 }).unwrap(); + unsafe { + pages.with_mut_ref(old_link, |row| row.unghost()).unwrap(); + } + pages.current_page_id.store(3, Ordering::Release); + + let read_guard = pages.read_guard(); + pages.retire_published_link(old_link); + pages.mark_page_empty(old_link.page_id); + + let temporary_page = pages.allocate_new_or_pop_free(); + assert_ne!( + temporary_page.id, old_link.page_id, + "vacuumed source page was reused while an old index reader was active" + ); + assert_eq!( + pages.select_non_ghosted(old_link), + Ok(TestRow { a: 1, b: 1 }), + "the old publication must survive until the reader leaves" + ); + + drop(read_guard); + pages.reclaim_retired(); + assert!(pages.get_empty_pages().contains(&old_link.page_id)); + + let reused_page = pages.allocate_new_or_pop_free(); + assert_eq!(reused_page.id, old_link.page_id); + assert_eq!(reused_page.free_offset.load(Ordering::Acquire), 0); + assert!(pages.published_slot(old_link).is_none()); + } + + #[test] + fn whole_page_reclamation_does_not_publish_overlapping_empty_links() { + let pages = DataPages::::from_data(vec![ + Arc::new(Data::new(1.into())), + Arc::new(Data::new(2.into())), + Arc::new(Data::new(3.into())), + ]); + pages.current_page_id.store(2, Ordering::Release); + let old_link = pages.insert(TestRow { a: 1, b: 1 }).unwrap(); + unsafe { + pages.with_mut_ref(old_link, |row| row.unghost()).unwrap(); + } + pages.current_page_id.store(3, Ordering::Release); + + let read_guard = pages.read_guard(); + pages.delete(old_link).unwrap(); + pages.mark_page_empty(old_link.page_id); + drop(read_guard); + pages.reclaim_retired(); + + assert!(pages.get_empty_pages().contains(&old_link.page_id)); + assert!( + pages + .get_empty_links() + .iter() + .all(|link| link.page_id != old_link.page_id), + "whole-page and inner-link allocators must not receive overlapping storage" + ); + } + #[test] fn select_non_vacuumed_returns_row_when_valid() { let pages = DataPages::::new(); diff --git a/src/in_memory/row.rs b/src/in_memory/row.rs index 115b9b73..89eedaa1 100644 --- a/src/in_memory/row.rs +++ b/src/in_memory/row.rs @@ -2,18 +2,10 @@ use std::fmt::Debug; use rkyv::Archive; -#[cfg(feature = "versioned-row-publication")] pub trait PublicationSafe: Send + Sync + 'static {} -#[cfg(feature = "versioned-row-publication")] impl PublicationSafe for T {} -#[cfg(not(feature = "versioned-row-publication"))] -pub trait PublicationSafe {} - -#[cfg(not(feature = "versioned-row-publication"))] -impl PublicationSafe for T {} - /// Common trait for the `Row`s that can be stored on the [`Data`] page. /// /// [`Data`]: crate::in_memory::data::Data diff --git a/src/index/congee.rs b/src/index/congee.rs index f2e41846..06e8a4d5 100644 --- a/src/index/congee.rs +++ b/src/index/congee.rs @@ -6,6 +6,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use congee::{CongeeRaw, DefaultAllocator}; +use parking_lot::Mutex; use super::UniqueIndex; @@ -47,6 +48,10 @@ impl_congee_key!(u64); /// reclamation pattern used by Congee's own `CongeeArc` implementation. pub struct CongeeIndex { inner: CongeeRaw, + // congee-wt 0.4.1 can lose disjoint insert/remove mutations when their + // structural updates overlap. Keep point reads native and concurrent, but + // serialize mutations until the backend offers the required visibility. + mutation: Mutex<()>, len: AtomicUsize, marker: std::marker::PhantomData<(K, V)>, } @@ -72,6 +77,7 @@ where }; Self { inner: CongeeRaw::new_with_drainer(DefaultAllocator {}, drainer), + mutation: Mutex::new(()), len: AtomicUsize::new(0), marker: std::marker::PhantomData, } @@ -190,6 +196,7 @@ where let len = inner.keys().len(); Ok(Self { inner, + mutation: Mutex::new(()), len: AtomicUsize::new(len), marker: std::marker::PhantomData, }) @@ -224,6 +231,7 @@ where #[inline] fn insert_value(&self, key: K, value: V) -> Option { + let _mutation = self.mutation.lock(); let guard = self.inner.pin(); let pointer = Arc::into_raw(Arc::new(value)).expose_provenance(); match self.inner.insert(key.into_congee(), pointer, &guard) { @@ -242,6 +250,7 @@ where #[inline] fn insert_value_checked(&self, key: K, value: V) -> Option<()> { + let _mutation = self.mutation.lock(); let guard = self.inner.pin(); let pointer = Arc::into_raw(Arc::new(value)).expose_provenance(); let result = self @@ -269,6 +278,7 @@ where #[inline] fn remove_value(&self, key: &K) -> Option<(K, V)> { + let _mutation = self.mutation.lock(); let guard = self.inner.pin(); let pointer = self.inner.remove(&key.into_congee(), &guard)?; self.len.fetch_sub(1, Ordering::Relaxed); diff --git a/src/index/unique.rs b/src/index/unique.rs index d4ba8a01..55e559f3 100644 --- a/src/index/unique.rs +++ b/src/index/unique.rs @@ -292,9 +292,22 @@ mod tests { threads.push(std::thread::spawn(move || { for sequence in 0..1_000_u64 { let key = worker * 1_000 + sequence; - assert_eq!(index.insert_value_checked(key, key + 1), Some(())); - assert_eq!(index.get_value(&key), Some(key + 1)); - assert_eq!(index.remove_value(&key), Some((key, key + 1))); + let backend = std::any::type_name::(); + assert_eq!( + index.insert_value_checked(key, key + 1), + Some(()), + "backend={backend}, key={key}, operation=insert" + ); + assert_eq!( + index.get_value(&key), + Some(key + 1), + "backend={backend}, key={key}, operation=get" + ); + assert_eq!( + index.remove_value(&key), + Some((key, key + 1)), + "backend={backend}, key={key}, operation=remove" + ); } })); } diff --git a/src/index/unsized_node.rs b/src/index/unsized_node.rs index b569b5ab..019856d7 100644 --- a/src/index/unsized_node.rs +++ b/src/index/unsized_node.rs @@ -182,6 +182,16 @@ where } } + fn delete_at(&mut self, index: usize) -> Option { + let val = NodeLike::delete_at(&mut self.inner, index)?; + self.removed_length += val.aligned_size() + UnsizedIndexPageUtility::::slots_value_size(); + + if self.removed_length > self.length_capacity / 2 { + self.rebuild() + } + Some(val) + } + fn replace(&mut self, idx: usize, value: T) -> Option { let value_size = value.aligned_size(); if let Some(old) = self.inner.get_mut(idx) { @@ -271,6 +281,17 @@ mod test { assert_eq!(node.removed_length, 40); } + #[test] + fn test_delete_at_updates_removed_length() { + let mut node = UnsizedNode::::with_capacity(200); + node.insert(String::from_utf8(vec![b'1'; 16]).unwrap()); + node.insert(String::from_utf8(vec![b'2'; 24]).unwrap()); + + assert_eq!(node.delete_at(0), Some(String::from_utf8(vec![b'1'; 16]).unwrap())); + assert_eq!(node.removed_length, 32); + assert_eq!(node.inner.len(), 1); + } + #[test] fn test_get_works_as_expected_at_big_amounts() { let maximum_node_size = 1000; diff --git a/src/lock/map.rs b/src/lock/map.rs index 4e4b538f..2b563d55 100644 --- a/src/lock/map.rs +++ b/src/lock/map.rs @@ -1,17 +1,44 @@ use std::collections::HashMap; +use std::collections::hash_map::DefaultHasher; use std::fmt::Debug; -use std::hash::Hash; +use std::hash::{Hash, Hasher}; use std::sync::Arc; -use std::sync::atomic::{AtomicU16, Ordering}; +use std::sync::atomic::{AtomicU16, AtomicU64, Ordering}; use parking_lot::RwLock; use crate::lock::RowLock; +const MUTATION_STRIPE_COUNT: usize = 64; + +#[derive(Debug, Default)] +struct MutationStripe { + next_ticket: AtomicU64, + serving: AtomicU64, +} + +/// Synchronous, task-safe gate for one primary-key mutation stripe. +/// +/// Generated async row locks and synchronous inserts share these gates so a +/// synchronous API entry point cannot interleave its multi-structure +/// publication with an update or delete of the same key. +#[derive(Debug)] +pub struct MutationGuard { + stripes: Arc<[MutationStripe; MUTATION_STRIPE_COUNT]>, + stripe: usize, +} + +impl Drop for MutationGuard { + fn drop(&mut self) { + self.stripes[self.stripe].serving.fetch_add(1, Ordering::Release); + } +} + #[derive(Debug)] pub struct LockMap { map: RwLock>>>, next_id: AtomicU16, + mutation_stripes: Arc<[MutationStripe; MUTATION_STRIPE_COUNT]>, } impl Default for LockMap { @@ -19,6 +46,7 @@ impl Default for LockMap { Self { map: RwLock::new(HashMap::new()), next_id: AtomicU16::default(), + mutation_stripes: Arc::new(std::array::from_fn(|_| MutationStripe::default())), } } } @@ -104,4 +132,32 @@ where pub fn next_id(&self) -> u16 { self.next_id.fetch_add(1, Ordering::Relaxed) } + + /// Serializes the synchronous mutation phase for this key. + /// + /// The holder must not perform a suspending `.await`. Generated locked + /// operations acquire this only after their async predecessor wait has + /// completed, and the synchronous `insert` path never awaits. + pub fn mutation_guard(&self, key: &PrimaryKey) -> MutationGuard { + let mut hasher = DefaultHasher::new(); + key.hash(&mut hasher); + let stripe = (hasher.finish() as usize) % MUTATION_STRIPE_COUNT; + let gate = &self.mutation_stripes[stripe]; + let ticket = gate.next_ticket.fetch_add(1, Ordering::Relaxed); + let mut spins = 0u32; + + while gate.serving.load(Ordering::Acquire) != ticket { + if spins < 16 { + spins += 1; + std::hint::spin_loop(); + } else { + std::thread::yield_now(); + } + } + + MutationGuard { + stripes: self.mutation_stripes.clone(), + stripe, + } + } } diff --git a/src/lock/mod.rs b/src/lock/mod.rs index 51c1ec0d..81e194f8 100644 --- a/src/lock/mod.rs +++ b/src/lock/mod.rs @@ -14,7 +14,7 @@ use std::task::{Context, Poll}; use futures::task::AtomicWaker; use parking_lot::Mutex; -pub use map::LockMap; +pub use map::{LockMap, MutationGuard}; pub use row_lock::{FullRowLock, RowLock}; /// Maximum number of spin iterations before falling back to async waiting. @@ -32,6 +32,10 @@ pub struct LockGuard { lock: Arc, lock_map: Arc>, primary_key: PrimaryKey, + /// Present for single-row operations. Multi-row queries acquire one + /// mutation stripe only while processing each row, after all row locks are + /// held, so stripe collisions cannot invert their primary-key lock order. + _mutation_guard: Option, /// Marker to make this type ![`Sync`] (but still [`Send`]) _not_sync: PhantomData>, } @@ -48,6 +52,24 @@ where lock, lock_map, primary_key, + _mutation_guard: None, + _not_sync: PhantomData, + } + } + + /// Creates a row guard that also serializes the mutation phase with the + /// synchronous insert path for the same primary key. + pub fn new_with_mutation( + lock: Arc, + lock_map: Arc>, + primary_key: PrimaryKey, + ) -> Self { + let mutation_guard = lock_map.mutation_guard(&primary_key); + Self { + lock, + lock_map, + primary_key, + _mutation_guard: Some(mutation_guard), _not_sync: PhantomData, } } @@ -124,7 +146,7 @@ impl Lock { } pub fn unlock(&self) { - self.locked.store(false, Ordering::Relaxed); + self.locked.store(false, Ordering::Release); let guard = self.wakers.lock(); for w in guard.iter() { w.wake() @@ -136,7 +158,7 @@ impl Lock { } pub fn is_locked(&self) -> bool { - self.locked.load(Ordering::Relaxed) + self.locked.load(Ordering::Acquire) } pub fn wait(&self) -> LockWait { diff --git a/src/table/mod.rs b/src/table/mod.rs index aabf6d28..d01b1604 100644 --- a/src/table/mod.rs +++ b/src/table/mod.rs @@ -140,32 +140,19 @@ where Deserialize<::WrappedRow, HighDeserializer>, { let _read_guard = self.data.read_guard(); - #[cfg(feature = "versioned-row-publication")] - { - for _ in 0..64 { - let link = self.primary_index.pk_map.lookup_for_select(&pk).map(Into::into)?; - if let Ok(row) = self.data.select_non_ghosted(link) { - return Some(row); - } - - let current_link: Option = self.primary_index.pk_map.lookup_for_select(&pk).map(Into::into); - if current_link == Some(link) { - return None; - } - std::hint::spin_loop(); + for _ in 0..64 { + let link = self.primary_index.pk_map.lookup_for_select(&pk).map(Into::into)?; + if let Ok(row) = self.data.select_non_ghosted(link) { + return Some(row); } - None - } - #[cfg(not(feature = "versioned-row-publication"))] - { - let link = self.primary_index.pk_map.lookup_for_select(&pk).map(|value| value.0); - if let Some(link) = link { - self.data.select_non_ghosted(link).ok() - } else { - None + let current_link: Option = self.primary_index.pk_map.lookup_for_select(&pk).map(Into::into); + if current_link == Some(link) { + return None; } + std::hint::spin_loop(); } + None } #[cfg_attr(feature = "perf_measurements", performance_measurement(prefix_name = "WorkTable"))] @@ -186,6 +173,7 @@ where LockType: 'static, { let pk = row.get_primary_key().clone(); + let _mutation_guard = self.lock_manager.mutation_guard(&pk); let link = self.data.insert(row.clone()).map_err(WorkTableError::PagesError)?; if self.primary_index.insert_checked(pk.clone(), link).is_none() { self.data.delete(link).map_err(WorkTableError::PagesError)?; @@ -239,6 +227,7 @@ where PrimaryIndex: TableIndexCdc, { let pk = row.get_primary_key().clone(); + let _mutation_guard = self.lock_manager.mutation_guard(&pk); let (link, _) = match self.data.insert_cdc(row.clone()) { Ok(result) => result, diff --git a/src/table/vacuum/vacuum.rs b/src/table/vacuum/vacuum.rs index 5ff3f0d7..4343b458 100644 --- a/src/table/vacuum/vacuum.rs +++ b/src/table/vacuum/vacuum.rs @@ -146,6 +146,7 @@ where let mut free_pages = VecDeque::new(); let mut defragmented_pages = VecDeque::new(); free_pages.push_back(additional_allocated_page.id); + let mut pages_freed = 0; let pages_processed = per_page_info.len(); @@ -162,24 +163,27 @@ where } else if let Some(id) = free_pages.pop_front() { id } else { - unreachable!("I hope so") + // A source page cannot become a destination until every + // reader that could still hold one of its old links has + // left the grace period. This call reuses it immediately + // when reclamation is safe, or allocates a temporary page + // while a pre-existing reader is still active. + self.data_pages.allocate_new_or_pop_free().id }; match self.move_data_from(page_from, page_to).await? { (true, true) => { // from moved fully and on to no more space - free_pages.push_back(page_from); - self.free_page(page_from); + self.data_pages.mark_page_full(page_to); break; } (true, false) => { // from moved fully but to has space - free_pages.push_back(page_from); - self.free_page(page_from); defragmented_pages.push_back(page_to); break; } (false, true) => { // from was not moved but to have NO space + self.data_pages.mark_page_full(page_to); continue; } (false, false) => { @@ -187,10 +191,15 @@ where } } } + // Remove the page's empty-link fragments before reclamation can + // expose the whole page for reuse. Otherwise a concurrent insert + // could claim a stale fragment between retirement and cleanup. registry.remove_link_for_page(page_from); + self.data_pages.mark_page_empty(page_from); + pages_freed += 1; } - let pages_freed = free_pages.len(); + pages_freed += free_pages.len(); for id in free_pages { self.data_pages.mark_page_empty(id) } @@ -206,10 +215,6 @@ where }) } - fn free_page(&self, page_id: PageId) { - self.data_pages.reset_page(page_id).expect("should exist as called") - } - async fn move_data_from(&self, from: PageId, to: PageId) -> eyre::Result<(bool, bool)> { let to_page = self.data_pages.get_page(to).expect("should exist as link exists"); let to_free_space = to_page.free_space(); @@ -257,6 +262,7 @@ where for (from_link, pk) in links { let lock = self.full_row_lock(&pk).await; + let _mutation_guard = self.lock_manager.mutation_guard(&pk); if self .data_pages .with_ref(from_link.0, |r| r.is_deleted()) @@ -394,6 +400,8 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; + use data_bucket::Link; + use data_bucket::page::PageId; use worktable_codegen::{MemStat, worktable}; use crate::in_memory::{ArchivedRowWrapper, RowWrapper, StorableRow}; @@ -846,4 +854,74 @@ mod tests { assert_eq!(row, Some(expected)); } } + + #[tokio::test] + async fn vacuum_does_not_reuse_source_pages_during_a_read_grace_period() { + let table = TestWorkTable::default(); + let mut rows_by_page: HashMap> = HashMap::new(); + + // Two large rows fit on each page. Deleting one from many pages leaves + // enough fragmented source pages that the old vacuum implementation + // reset and recycled an earlier source as a later destination. + for i in 0..40u64 { + let row = TestRow { + id: table.get_next_pk().into(), + test: i as i64, + another: i, + exchange: format!("{i:02}-{}", "x".repeat(6_000)), + }; + let id = row.id; + table.insert(row.clone()).unwrap(); + let link = table + .0 + .primary_index + .pk_map + .get_value(&TestPrimaryKey::from(id)) + .unwrap() + .0; + rows_by_page.entry(link.page_id).or_default().push((id, row, link)); + } + + let current_page = table.0.data.current_page_id(); + let mut protected_rows = Vec::new(); + for (page_id, rows) in rows_by_page { + if page_id == current_page || rows.len() < 2 { + continue; + } + + protected_rows.push(rows[0].clone()); + for (id, _, _) in rows.into_iter().skip(1) { + table.delete(id).await.unwrap(); + } + } + assert!( + protected_rows.len() >= 3, + "test setup needs several fragmented source pages" + ); + + // Model a generated reader that already resolved each old physical + // link, then pauses while vacuum swings the indexes. + let read_guard = table.0.data.read_guard(); + create_vacuum(&table).defragment().await.unwrap(); + + let mut moved = 0; + for (id, expected, old_link) in &protected_rows { + let current_link = table + .0 + .primary_index + .pk_map + .get_value(&TestPrimaryKey::from(*id)) + .unwrap() + .0; + moved += usize::from(current_link != *old_link); + assert_eq!( + table.0.data.select_non_ghosted(*old_link), + Ok(expected.clone()), + "a retired source link was reset or republished before the reader left" + ); + } + assert!(moved >= 3, "test setup did not exercise enough vacuum moves"); + + drop(read_guard); + } } diff --git a/tests/persistence/tuple_primary_key.rs b/tests/persistence/tuple_primary_key.rs index 03e4b1bc..e7b07b85 100644 --- a/tests/persistence/tuple_primary_key.rs +++ b/tests/persistence/tuple_primary_key.rs @@ -53,6 +53,7 @@ async fn composite_primary_key_survives_mutations_and_reload() { for row in &rows { assert_eq!(table.select((row.tenant_id, row.record_id)), Some(row.clone())); } + table.close().await.unwrap(); } { @@ -84,5 +85,6 @@ async fn composite_primary_key_survives_mutations_and_reload() { assert!(table.select((7, 41)).is_none()); assert_eq!(table.select((7, 42)).unwrap().value, 99); assert_eq!(table.select((8, 1)), Some(rows[2].clone())); + table.close().await.unwrap(); } } diff --git a/tests/worktable/float.rs b/tests/worktable/float.rs index 576c707b..a924d343 100644 --- a/tests/worktable/float.rs +++ b/tests/worktable/float.rs @@ -54,7 +54,6 @@ fn unique_float_point_read_revalidates_the_returned_row() { assert_eq!(table.select_by_value(second.value), Some(second)); } -#[cfg(feature = "versioned-row-publication")] #[test] fn float_range_read_revalidates_each_resolved_row() { let table = TestFloatWorkTable::default(); diff --git a/tests/worktable/index/range.rs b/tests/worktable/index/range.rs index 19497bac..d9ac9d71 100644 --- a/tests/worktable/index/range.rs +++ b/tests/worktable/index/range.rs @@ -32,7 +32,6 @@ worktable!( } ); -#[cfg(feature = "versioned-row-publication")] #[tokio::test] async fn idle_select_builder_does_not_pin_retired_links() { let table = UniqueRangeTestWorkTable::default(); @@ -62,7 +61,6 @@ async fn idle_select_builder_does_not_pin_retired_links() { drop(idle_query); } -#[cfg(feature = "versioned-row-publication")] #[test] fn range_read_revalidates_each_resolved_row() { let table = RangeTestWorkTable::default(); diff --git a/tests/worktable/mod.rs b/tests/worktable/mod.rs index 4ce0fa42..ac946814 100644 --- a/tests/worktable/mod.rs +++ b/tests/worktable/mod.rs @@ -14,6 +14,7 @@ mod nid; mod option; mod tuple_primary_key; mod unsized_; +mod update_in_place_unsized; mod upsert; mod uuid; mod vacuum; diff --git a/tests/worktable/update_in_place_unsized.rs b/tests/worktable/update_in_place_unsized.rs new file mode 100644 index 00000000..453f287b --- /dev/null +++ b/tests/worktable/update_in_place_unsized.rs @@ -0,0 +1,125 @@ +//! Regression (KNOWN BUG, currently #[ignore]): updating a variable-length +//! (unsized / `String`) row to a value of the SAME serialized length still does +//! a full delete-and-reinsert instead of an in-place mutation. +//! +//! ## Why this matters +//! Reinsert allocates a fresh page slot (the row's `Link` changes), re-serializes +//! the whole row, and rebuilds every secondary index. On a `String`-bearing +//! table this makes UPDATE ~9x slower than an in-place field write — WorkTable +//! leads insert and point_read in the KV benchmark but is dead last on overwrite +//! purely because of this path. +//! +//! ## Root cause +//! `codegen/src/generators/in_memory/queries/update.rs`, custom-update size +//! check (`gen_size_check`): `let mut need_to_reinsert = true;` then +//! `need_to_reinsert |= `. Initialized to `true`, the `|=` can +//! never clear it, so EVERY update reinserts regardless of whether any unsized +//! field changed size. +//! +//! ## Why the obvious fix is not enough (do not just flip the initializer) +//! Setting the initializer to `false` correctly lets same-length updates skip +//! reinsert — but the in-place archived write in this custom-update path then +//! CORRUPTS variable-length rows in existing tests +//! (`worktable::unsized_::update_parallel_more_strings`, `update_many_times`, +//! `in_place::test_update_in_place_and_update_unsized_multithread`): reads come +//! back as raw archived bytes. So a real fix must make the in-place write of an +//! (even equal-length) archived `String` field safe in this path, not merely +//! change when the fast path is taken. That is a storage/codegen change beyond a +//! one-liner; tracked here so the fix has a proof. +//! +//! The observable is the row's physical `Link`: an in-place update keeps it, a +//! reinsert changes it. Remove `#[ignore]` when the in-place path is fixed. + +use worktable::prelude::*; +use worktable::worktable; + +worktable!( + name: UnsizedUpdate, + columns: { + id: u64 primary_key, + payload: String, + }, + queries: { + update: { + Payload(payload) by id, + } + } +); + +/// Read the current physical link for a primary key. +fn link_of(table: &UnsizedUpdateWorkTable, pk: u64) -> Link { + table + .0 + .primary_index + .pk_map + .get_value(&UnsizedUpdatePrimaryKey::from(pk)) + .map(Into::into) + .expect("row must exist") +} + +#[tokio::test] +#[ignore = "known bug: same-length unsized update reinserts; in-place write of a String field corrupts the row — needs a storage-path fix"] +async fn same_length_update_stays_in_place() { + let table = UnsizedUpdateWorkTable::default(); + table + .insert(UnsizedUpdateRow { + id: 1, + payload: "abcdefgh".to_string(), // 8 bytes + }) + .unwrap(); + + let before = link_of(&table, 1); + + table + .update_payload( + PayloadQuery { + payload: "12345678".to_string(), // 8 bytes — same length + }, + 1, + ) + .await + .unwrap(); + + let after = link_of(&table, 1); + + assert_eq!(table.select(1).unwrap().payload, "12345678"); + assert_eq!( + before, after, + "same-length unsized update must not reinsert (link changed: {before:?} -> {after:?})" + ); +} + +/// This one already holds on master and must keep holding through any fix: +/// a length change round-trips correctly (via reinsert). +#[tokio::test] +async fn different_length_update_is_correct() { + let table = UnsizedUpdateWorkTable::default(); + table + .insert(UnsizedUpdateRow { + id: 1, + payload: "abcdefghij".to_string(), + }) + .unwrap(); + + table + .update_payload( + PayloadQuery { + payload: "xy".to_string(), + }, + 1, + ) + .await + .unwrap(); + assert_eq!(table.select(1).unwrap().payload, "xy"); + + table + .update_payload( + PayloadQuery { + payload: "much longer payload".to_string(), + }, + 1, + ) + .await + .unwrap(); + assert_eq!(table.select(1).unwrap().payload, "much longer payload"); +} diff --git a/tests/worktable/upsert.rs b/tests/worktable/upsert.rs index 7ba61a3b..d297a8a4 100644 --- a/tests/worktable/upsert.rs +++ b/tests/worktable/upsert.rs @@ -27,12 +27,16 @@ async fn upsert_completes_under_same_key_churn() { /// Intense variant for the same-key upsert/delete linearization protocol. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn upsert_completes_under_extreme_same_key_churn() { - churn_run(5_000, 2_000).await; + // Keep this materially heavier than the normal case without making the + // assertion depend on runner speed: 2,000 churn flips perform both an + // upsert and a delete, alongside 4,000 competing upserts. + churn_run(2_000, 1_000).await; } -/// A synchronous insert does not participate in the generated async row lock. -/// Its collision and ghost-publication windows must still return typed errors, -/// never panic a concurrent locked delete or strand an upsert waiter. +/// A synchronous insert does not wait on the generated async row-lock chain, +/// but it shares the FIFO per-key mutation gate with delete and upsert. Its +/// collision and ghost-publication windows must return typed errors, never +/// panic a concurrent delete or strand an upsert waiter. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn raw_insert_delete_churn_never_panics_or_stalls() { let table = Arc::new(UpsertChurnWorkTable::default()); @@ -41,10 +45,24 @@ async fn raw_insert_delete_churn_never_panics_or_stalls() { let churn = { let table = table.clone(); tokio::spawn(async move { + let mut insert_successes = 0; + let mut insert_conflicts = 0; + let mut delete_successes = 0; + let mut delete_misses = 0; for i in 0..5_000u64 { - let _ = table.insert(UpsertChurnRow { id: KEY, val: i }); - let _ = table.delete(KEY).await; + match table.insert(UpsertChurnRow { id: KEY, val: i }) { + Ok(_) => insert_successes += 1, + Err(WorkTableError::PrimaryAlreadyExists) => insert_conflicts += 1, + Err(error) => panic!("raw insert returned an unexpected error: {error:?}"), + } + match table.delete(KEY).await { + Ok(()) => delete_successes += 1, + Err(WorkTableError::NotFound) => delete_misses += 1, + Err(error) => panic!("delete returned an unexpected error: {error:?}"), + } } + + (insert_successes, insert_conflicts, delete_successes, delete_misses) }) }; @@ -64,16 +82,77 @@ async fn raw_insert_delete_churn_never_panics_or_stalls() { })); } - timeout(Duration::from_secs(60), churn) + let (insert_successes, insert_conflicts, delete_successes, delete_misses) = timeout(Duration::from_secs(60), churn) .await .expect("raw insert/delete churn starved") .unwrap(); + assert_eq!(insert_successes + insert_conflicts, 5_000); + assert_eq!(delete_successes + delete_misses, 5_000); for handle in upserters { timeout(Duration::from_secs(60), handle) .await .expect("upserter starved during raw insert/delete churn") .unwrap(); } + + // Force a deterministic final state, then audit every layer used to reach + // the row. A liveness-only test would miss a stale reverse entry, a ghost + // publication, or a primary link that points at unrelated data. + let expected = UpsertChurnRow { id: KEY, val: 424_242 }; + table.upsert(expected.clone()).await.unwrap(); + + let pk = UpsertChurnPrimaryKey::from(KEY); + let link = table + .0 + .primary_index + .pk_map + .get_value(&pk) + .expect("final row must have one primary-index entry"); + assert_eq!(table.0.primary_index.pk_map.len(), 1); + assert_eq!(table.0.primary_index.reverse_pk_map.len(), 1); + assert_eq!( + table + .0 + .primary_index + .reverse_pk_map + .get(&link) + .map(|entry| entry.get().value.clone()), + Some(pk), + "reverse index must point back to the final primary key" + ); + assert_eq!( + table.0.data.select_non_ghosted(link.0), + Ok(expected.clone()), + "primary-index link must resolve to the final non-ghosted row" + ); + assert_eq!(table.select(KEY), Some(expected)); +} + +/// Pins the exact publication schedule that used to let delete unwrap a +/// ghosted row: data and primary-index reachability exist, but insert has not +/// yet cleared the lifecycle bit. Delete must linearize before publication and +/// leave the staged insert intact. +#[tokio::test] +async fn delete_during_insert_publication_window_returns_not_found() { + let table = UpsertChurnWorkTable::default(); + const KEY: u64 = 7; + let row = UpsertChurnRow { id: KEY, val: 11 }; + + let link = table.0.data.insert(row.clone()).unwrap(); + assert!( + table + .0 + .primary_index + .insert_checked(UpsertChurnPrimaryKey::from(KEY), link) + .is_some() + ); + + assert!(matches!(table.delete(KEY).await, Err(WorkTableError::NotFound))); + + unsafe { + table.0.data.with_mut_ref(link, |staged| staged.unghost()).unwrap(); + } + assert_eq!(table.select(KEY), Some(row)); } async fn churn_run(churn_flips: u64, upserts_per_task: u64) { @@ -88,12 +167,9 @@ async fn churn_run(churn_flips: u64, upserts_per_task: u64) { let table = table.clone(); tokio::spawn(async move { for i in 0..CHURN_FLIPS { - // Flip the key's existence as fast as possible through the - // locked operations. (Raw `insert` is deliberately not used - // here: it takes no row lock and publishes the pk entry - // before unghosting the data, which trips unrelated - // pre-existing races tracked separately in the issue on - // lock-free insert vs locked mutations.) + // Flip the key's existence through the async row-lock path. + // Raw synchronous insert has its own focused churn coverage + // above; keeping the paths separate makes failures diagnostic. table .upsert(UpsertChurnRow { id: KEY, val: i }) .await diff --git a/tests/worktable/vacuum.rs b/tests/worktable/vacuum.rs index 2e49cca6..85c18d25 100644 --- a/tests/worktable/vacuum.rs +++ b/tests/worktable/vacuum.rs @@ -217,8 +217,10 @@ async fn vacuum_parallel_with_upserts() { } #[tokio::test(flavor = "multi_thread", worker_threads = 3)] -#[ignore] +#[ignore = "bounded 10-second vacuum soak test"] async fn vacuum_loop_test() { + const SOAK_DURATION: Duration = Duration::from_secs(10); + let config = VacuumManagerConfig { check_interval: Duration::from_millis(1_000), ..Default::default() @@ -238,12 +240,13 @@ async fn vacuum_loop_test() { let vacuum = table.vacuum(); vacuum_manager.register(vacuum); - let _h = vacuum_manager.run_vacuum_task(); + let vacuum_task = vacuum_manager.run_vacuum_task(); let insert_table = table.clone(); - let _task = tokio::spawn(async move { + let stop_at = tokio::time::Instant::now() + SOAK_DURATION; + let task = tokio::spawn(async move { let mut i = 3001; - loop { + while tokio::time::Instant::now() < stop_at { let row = VacuumTestRow { id: insert_table.get_next_pk().into(), value: chrono::Utc::now().timestamp_nanos_opt().unwrap(), @@ -257,7 +260,7 @@ async fn vacuum_loop_test() { tokio::time::sleep(Duration::from_millis(1_000)).await; - loop { + while tokio::time::Instant::now() < stop_at { tokio::time::sleep(Duration::from_millis(1_000)).await; let outdated_ts = chrono::Utc::now() @@ -276,4 +279,7 @@ async fn vacuum_loop_test() { table.delete(row.id).await.unwrap(); } } + + task.await.unwrap(); + vacuum_task.abort(); }