mirror of
https://gitlab.com/veilid/veilid.git
synced 2025-08-19 19:58:45 -04:00
fix tabledb race condition
This commit is contained in:
parent
ab3cf25647
commit
0cf745a302
2 changed files with 33 additions and 54 deletions
|
@ -15,6 +15,7 @@ mod native;
|
||||||
use native::*;
|
use native::*;
|
||||||
|
|
||||||
use keyvaluedb::*;
|
use keyvaluedb::*;
|
||||||
|
use weak_table::WeakValueHashMap;
|
||||||
|
|
||||||
impl_veilid_log_facility!("tstore");
|
impl_veilid_log_facility!("tstore");
|
||||||
|
|
||||||
|
@ -72,7 +73,7 @@ pub struct TableInfo {
|
||||||
|
|
||||||
#[must_use]
|
#[must_use]
|
||||||
struct TableStoreInner {
|
struct TableStoreInner {
|
||||||
opened: BTreeMap<String, Weak<TableDBUnlockedInner>>,
|
opened: WeakValueHashMap<String, Weak<TableDBUnlockedInner>>,
|
||||||
encryption_key: Option<TypedSharedSecret>,
|
encryption_key: Option<TypedSharedSecret>,
|
||||||
all_table_names: HashMap<String, String>,
|
all_table_names: HashMap<String, String>,
|
||||||
all_tables_db: Option<Database>,
|
all_tables_db: Option<Database>,
|
||||||
|
@ -115,7 +116,7 @@ impl_veilid_component!(TableStore);
|
||||||
impl TableStore {
|
impl TableStore {
|
||||||
fn new_inner() -> TableStoreInner {
|
fn new_inner() -> TableStoreInner {
|
||||||
TableStoreInner {
|
TableStoreInner {
|
||||||
opened: BTreeMap::new(),
|
opened: WeakValueHashMap::new(),
|
||||||
encryption_key: None,
|
encryption_key: None,
|
||||||
all_table_names: HashMap::new(),
|
all_table_names: HashMap::new(),
|
||||||
all_tables_db: None,
|
all_tables_db: None,
|
||||||
|
@ -528,6 +529,7 @@ impl TableStore {
|
||||||
self.flush().await;
|
self.flush().await;
|
||||||
|
|
||||||
let mut inner = self.inner.lock();
|
let mut inner = self.inner.lock();
|
||||||
|
inner.opened.shrink_to_fit();
|
||||||
if !inner.opened.is_empty() {
|
if !inner.opened.is_empty() {
|
||||||
panic!(
|
panic!(
|
||||||
"all open databases should have been closed: {:?}",
|
"all open databases should have been closed: {:?}",
|
||||||
|
@ -539,15 +541,6 @@ impl TableStore {
|
||||||
inner.encryption_key = None;
|
inner.encryption_key = None;
|
||||||
}
|
}
|
||||||
|
|
||||||
#[instrument(level = "trace", target = "tstore", skip_all)]
|
|
||||||
pub(crate) fn on_table_db_drop(&self, table: String) {
|
|
||||||
veilid_log!(self trace "dropping table db: {}", table);
|
|
||||||
let mut inner = self.inner.lock();
|
|
||||||
if inner.opened.remove(&table).is_none() {
|
|
||||||
unreachable!("should have removed an item");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Get or create a TableDB database table. If the column count is greater than an
|
/// Get or create a TableDB database table. If the column count is greater than an
|
||||||
/// existing TableDB's column count, the database will be upgraded to add the missing columns.
|
/// existing TableDB's column count, the database will be upgraded to add the missing columns.
|
||||||
#[instrument(level = "trace", target = "tstore", skip_all)]
|
#[instrument(level = "trace", target = "tstore", skip_all)]
|
||||||
|
@ -566,10 +559,10 @@ impl TableStore {
|
||||||
|
|
||||||
// See if this table is already opened, if so the column count must be the same
|
// See if this table is already opened, if so the column count must be the same
|
||||||
{
|
{
|
||||||
let mut inner = self.inner.lock();
|
let inner = self.inner.lock();
|
||||||
if let Some(table_db_weak_inner) = inner.opened.get(&table_name) {
|
if let Some(table_db_unlocked_inner) = inner.opened.get(&table_name) {
|
||||||
match TableDB::try_new_from_weak_inner(table_db_weak_inner.clone(), column_count) {
|
let tdb = TableDB::new_from_unlocked_inner(table_db_unlocked_inner, column_count);
|
||||||
Some(tdb) => {
|
|
||||||
// Ensure column count isnt bigger
|
// Ensure column count isnt bigger
|
||||||
let existing_col_count = tdb.get_column_count()?;
|
let existing_col_count = tdb.get_column_count()?;
|
||||||
if column_count > existing_col_count {
|
if column_count > existing_col_count {
|
||||||
|
@ -581,11 +574,6 @@ impl TableStore {
|
||||||
|
|
||||||
return Ok(tdb);
|
return Ok(tdb);
|
||||||
}
|
}
|
||||||
None => {
|
|
||||||
inner.opened.remove(&table_name);
|
|
||||||
}
|
|
||||||
};
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Open table db using platform-specific driver
|
// Open table db using platform-specific driver
|
||||||
|
@ -637,7 +625,7 @@ impl TableStore {
|
||||||
// Keep track of opened DBs
|
// Keep track of opened DBs
|
||||||
inner
|
inner
|
||||||
.opened
|
.opened
|
||||||
.insert(table_name.clone(), table_db.weak_unlocked_inner());
|
.insert(table_name.clone(), table_db.unlocked_inner());
|
||||||
|
|
||||||
Ok(table_db)
|
Ok(table_db)
|
||||||
}
|
}
|
||||||
|
|
|
@ -38,13 +38,6 @@ impl fmt::Debug for TableDBUnlockedInner {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Drop for TableDBUnlockedInner {
|
|
||||||
fn drop(&mut self) {
|
|
||||||
let table_store = self.registry.table_store();
|
|
||||||
table_store.on_table_db_drop(self.table.clone());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub struct TableDB {
|
pub struct TableDB {
|
||||||
|
@ -87,12 +80,11 @@ impl TableDB {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(super) fn try_new_from_weak_inner(
|
pub(super) fn new_from_unlocked_inner(
|
||||||
weak_inner: Weak<TableDBUnlockedInner>,
|
unlocked_inner: Arc<TableDBUnlockedInner>,
|
||||||
opened_column_count: u32,
|
opened_column_count: u32,
|
||||||
) -> Option<Self> {
|
) -> Self {
|
||||||
weak_inner.upgrade().map(|table_db_unlocked_inner| {
|
let db = &unlocked_inner.database;
|
||||||
let db = &table_db_unlocked_inner.database;
|
|
||||||
let total_columns = db.num_columns().unwrap();
|
let total_columns = db.num_columns().unwrap();
|
||||||
Self {
|
Self {
|
||||||
opened_column_count: if opened_column_count == 0 {
|
opened_column_count: if opened_column_count == 0 {
|
||||||
|
@ -100,13 +92,12 @@ impl TableDB {
|
||||||
} else {
|
} else {
|
||||||
opened_column_count
|
opened_column_count
|
||||||
},
|
},
|
||||||
unlocked_inner: table_db_unlocked_inner,
|
unlocked_inner,
|
||||||
}
|
}
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(super) fn weak_unlocked_inner(&self) -> Weak<TableDBUnlockedInner> {
|
pub(super) fn unlocked_inner(&self) -> Arc<TableDBUnlockedInner> {
|
||||||
Arc::downgrade(&self.unlocked_inner)
|
self.unlocked_inner.clone()
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get the internal name of the table
|
/// Get the internal name of the table
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue