Fence feature operations during teardown
This commit is contained in:
@@ -8,6 +8,7 @@
|
|||||||
use std::collections::{HashMap, HashSet};
|
use std::collections::{HashMap, HashSet};
|
||||||
use std::fs;
|
use std::fs;
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
|
use std::sync::atomic::{AtomicBool, Ordering};
|
||||||
use std::sync::{Arc, Mutex, Weak};
|
use std::sync::{Arc, Mutex, Weak};
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
@@ -221,6 +222,9 @@ impl FeatureStorage {
|
|||||||
.values()
|
.values()
|
||||||
.filter_map(Weak::upgrade)
|
.filter_map(Weak::upgrade)
|
||||||
.collect::<Vec<_>>();
|
.collect::<Vec<_>>();
|
||||||
|
for connection in &connections {
|
||||||
|
connection.fence();
|
||||||
|
}
|
||||||
state.connections.clear();
|
state.connections.clear();
|
||||||
connections
|
connections
|
||||||
};
|
};
|
||||||
@@ -270,6 +274,7 @@ impl FeatureStorage {
|
|||||||
let connection = Arc::new(FeatureDatabaseInner {
|
let connection = Arc::new(FeatureDatabaseInner {
|
||||||
workspace_id: workspace_id.to_string(),
|
workspace_id: workspace_id.to_string(),
|
||||||
feature_id: registration.feature_id.clone(),
|
feature_id: registration.feature_id.clone(),
|
||||||
|
accepting_operations: AtomicBool::new(true),
|
||||||
connection: Mutex::new(Some(connection)),
|
connection: Mutex::new(Some(connection)),
|
||||||
});
|
});
|
||||||
state.connections.insert(key, Arc::downgrade(&connection));
|
state.connections.insert(key, Arc::downgrade(&connection));
|
||||||
@@ -381,6 +386,9 @@ impl FeatureStorage {
|
|||||||
.filter(|((candidate, _), _)| candidate == workspace_id)
|
.filter(|((candidate, _), _)| candidate == workspace_id)
|
||||||
.filter_map(|(_, connection)| connection.upgrade())
|
.filter_map(|(_, connection)| connection.upgrade())
|
||||||
.collect::<Vec<_>>();
|
.collect::<Vec<_>>();
|
||||||
|
for connection in &connections {
|
||||||
|
connection.fence();
|
||||||
|
}
|
||||||
state
|
state
|
||||||
.connections
|
.connections
|
||||||
.retain(|(candidate, _), _| candidate != workspace_id);
|
.retain(|(candidate, _), _| candidate != workspace_id);
|
||||||
@@ -513,6 +521,7 @@ pub struct FeatureDatabase {
|
|||||||
struct FeatureDatabaseInner {
|
struct FeatureDatabaseInner {
|
||||||
workspace_id: String,
|
workspace_id: String,
|
||||||
feature_id: String,
|
feature_id: String,
|
||||||
|
accepting_operations: AtomicBool,
|
||||||
connection: Mutex<Option<Connection>>,
|
connection: Mutex<Option<Connection>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -531,9 +540,11 @@ impl FeatureDatabase {
|
|||||||
&self,
|
&self,
|
||||||
operation: impl FnOnce(&Connection) -> Result<T>,
|
operation: impl FnOnce(&Connection) -> Result<T>,
|
||||||
) -> Result<T> {
|
) -> Result<T> {
|
||||||
|
self.inner.ensure_accepting()?;
|
||||||
let connection = self.inner.connection.lock().map_err(|_| {
|
let connection = self.inner.connection.lock().map_err(|_| {
|
||||||
FeatureStorageError::Operation("Feature database lock was poisoned".to_string())
|
FeatureStorageError::Operation("Feature database lock was poisoned".to_string())
|
||||||
})?;
|
})?;
|
||||||
|
self.inner.ensure_accepting()?;
|
||||||
let connection = connection
|
let connection = connection
|
||||||
.as_ref()
|
.as_ref()
|
||||||
.ok_or(FeatureStorageError::DatabaseClosed)?;
|
.ok_or(FeatureStorageError::DatabaseClosed)?;
|
||||||
@@ -546,9 +557,11 @@ impl FeatureDatabase {
|
|||||||
&self,
|
&self,
|
||||||
operation: impl FnOnce(&Transaction<'_>) -> Result<T>,
|
operation: impl FnOnce(&Transaction<'_>) -> Result<T>,
|
||||||
) -> Result<T> {
|
) -> Result<T> {
|
||||||
|
self.inner.ensure_accepting()?;
|
||||||
let mut connection = self.inner.connection.lock().map_err(|_| {
|
let mut connection = self.inner.connection.lock().map_err(|_| {
|
||||||
FeatureStorageError::Operation("Feature database lock was poisoned".to_string())
|
FeatureStorageError::Operation("Feature database lock was poisoned".to_string())
|
||||||
})?;
|
})?;
|
||||||
|
self.inner.ensure_accepting()?;
|
||||||
let connection = connection
|
let connection = connection
|
||||||
.as_mut()
|
.as_mut()
|
||||||
.ok_or(FeatureStorageError::DatabaseClosed)?;
|
.ok_or(FeatureStorageError::DatabaseClosed)?;
|
||||||
@@ -572,7 +585,20 @@ impl FeatureDatabase {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl FeatureDatabaseInner {
|
impl FeatureDatabaseInner {
|
||||||
|
fn ensure_accepting(&self) -> Result<()> {
|
||||||
|
if self.accepting_operations.load(Ordering::Acquire) {
|
||||||
|
Ok(())
|
||||||
|
} else {
|
||||||
|
Err(FeatureStorageError::DatabaseClosed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn fence(&self) {
|
||||||
|
self.accepting_operations.store(false, Ordering::Release);
|
||||||
|
}
|
||||||
|
|
||||||
fn close(&self) -> Result<()> {
|
fn close(&self) -> Result<()> {
|
||||||
|
self.fence();
|
||||||
let mut connection = self.connection.lock().map_err(|_| {
|
let mut connection = self.connection.lock().map_err(|_| {
|
||||||
FeatureStorageError::Operation("Feature database lock was poisoned".to_string())
|
FeatureStorageError::Operation("Feature database lock was poisoned".to_string())
|
||||||
})?;
|
})?;
|
||||||
@@ -1264,8 +1290,14 @@ mod tests {
|
|||||||
let manager = manager.clone();
|
let manager = manager.clone();
|
||||||
thread::spawn(move || manager.shutdown())
|
thread::spawn(move || manager.shutdown())
|
||||||
};
|
};
|
||||||
thread::sleep(Duration::from_millis(20));
|
while database.inner.accepting_operations.load(Ordering::Acquire) {
|
||||||
|
thread::yield_now();
|
||||||
|
}
|
||||||
assert!(!shutdown.is_finished());
|
assert!(!shutdown.is_finished());
|
||||||
|
assert!(matches!(
|
||||||
|
values(&database).unwrap_err(),
|
||||||
|
FeatureStorageError::DatabaseClosed
|
||||||
|
));
|
||||||
release.wait();
|
release.wait();
|
||||||
operation.join().unwrap().unwrap();
|
operation.join().unwrap().unwrap();
|
||||||
shutdown.join().unwrap().unwrap();
|
shutdown.join().unwrap().unwrap();
|
||||||
|
|||||||
Reference in New Issue
Block a user