Skip to content

fix: make row publication concurrency-safe by default #45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
pathscale wants to merge 1 commit into from
Closed

fix: make row publication concurrency-safe by default #45

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
20 changes: 8 additions & 12 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,15 +69,11 @@ 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
Expand All @@ -87,15 +83,15 @@ 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. 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
Expand Down
1 change: 1 addition & 0 deletions codegen/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ repository = "https://github.com/pathscale/WorkTable"

[features]
s3-support = []
# Compatibility no-op retained for downstream manifests.
versioned-row-publication = []

[lib]
Expand Down
8 changes: 6 additions & 2 deletions codegen/src/generators/in_memory/queries/delete.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,11 @@ impl InMemoryGenerator {
return Err(e);
}
};
let row = self.0.select(pk.clone()).unwrap();
// A lock-free insert publishes index reachability before it
// clears the staged row's ghost bit. Treat that window as an
// absent row: this delete linearizes before the insert's
// publication instead of panicking on the hidden version.
let row = self.0.select(pk.clone()).ok_or(WorkTableError::NotFound)?;
#process
}
} else {
Expand All @@ -129,7 +133,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
}
}
Expand Down
88 changes: 31 additions & 57 deletions codegen/src/generators/in_memory/table/index_fns.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Link> = 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<Link> = 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! {
Expand Down Expand Up @@ -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 {
(
Expand All @@ -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,
Expand Down
8 changes: 6 additions & 2 deletions codegen/src/generators/persist/queries/delete.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,11 @@ impl PersistGenerator {
return Err(e);
}
};
let row = self.0.select(pk.clone()).unwrap();
// A lock-free insert publishes index reachability before it
// clears the staged row's ghost bit. Treat that window as an
// absent row: this delete linearizes before the insert's
// publication instead of panicking on the hidden version.
let row = self.0.select(pk.clone()).ok_or(WorkTableError::NotFound)?;
#process
}
} else {
Expand All @@ -122,7 +126,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
}
}
Expand Down
88 changes: 31 additions & 57 deletions codegen/src/generators/persist/table/index_fns.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Link> = 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<Link> = 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! {
Expand Down Expand Up @@ -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 {
(
Expand All @@ -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,
Expand Down
Loading
Loading