|
|
|
@ -26,22 +26,20 @@
|
|
|
|
|
|
|
|
|
|
|
|
use {
|
|
|
|
use {
|
|
|
|
super::{
|
|
|
|
super::{
|
|
|
|
drivers::FractalGNSDriver, CriticalTask, GenericTask, GlobalInstanceLike, ModelUniqueID,
|
|
|
|
drivers::FractalGNSDriver, CriticalTask, FractalModelDriver, GenericTask,
|
|
|
|
Task,
|
|
|
|
GlobalInstanceLike, Task,
|
|
|
|
},
|
|
|
|
},
|
|
|
|
crate::engine::{
|
|
|
|
crate::engine::{
|
|
|
|
core::{EntityIDRef, GlobalNS},
|
|
|
|
core::{EntityIDRef, GNSData, GlobalNS},
|
|
|
|
data::uuid::Uuid,
|
|
|
|
data::uuid::Uuid,
|
|
|
|
error::ErrorKind,
|
|
|
|
error::ErrorKind,
|
|
|
|
fractal::drivers::FractalModelDriver,
|
|
|
|
|
|
|
|
storage::{
|
|
|
|
storage::{
|
|
|
|
safe_interfaces::{paths_v1, FileSystem, StdModelBatch},
|
|
|
|
safe_interfaces::{paths_v1, FileSystem, StdModelBatch},
|
|
|
|
BatchStats, GNSDriver, ModelDriver,
|
|
|
|
BatchStats, GNSDriver, ModelDriver,
|
|
|
|
},
|
|
|
|
},
|
|
|
|
RuntimeResult,
|
|
|
|
RuntimeResult,
|
|
|
|
},
|
|
|
|
},
|
|
|
|
parking_lot::{Mutex, RwLock},
|
|
|
|
parking_lot::RwLock,
|
|
|
|
std::collections::HashMap,
|
|
|
|
|
|
|
|
};
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
|
|
/// A `test` mode global implementation
|
|
|
|
/// A `test` mode global implementation
|
|
|
|
@ -50,19 +48,15 @@ pub struct TestGlobal {
|
|
|
|
lp_queue: RwLock<Vec<Task<GenericTask>>>,
|
|
|
|
lp_queue: RwLock<Vec<Task<GenericTask>>>,
|
|
|
|
#[allow(unused)]
|
|
|
|
#[allow(unused)]
|
|
|
|
max_delta_size: usize,
|
|
|
|
max_delta_size: usize,
|
|
|
|
txn_driver: Mutex<FractalGNSDriver>,
|
|
|
|
|
|
|
|
model_drivers: RwLock<HashMap<ModelUniqueID, super::drivers::FractalModelDriver>>,
|
|
|
|
|
|
|
|
max_data_pressure: usize,
|
|
|
|
max_data_pressure: usize,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
impl TestGlobal {
|
|
|
|
impl TestGlobal {
|
|
|
|
fn new(gns: GlobalNS, max_delta_size: usize, txn_driver: GNSDriver) -> Self {
|
|
|
|
fn new(gns: GlobalNS, max_delta_size: usize) -> Self {
|
|
|
|
Self {
|
|
|
|
Self {
|
|
|
|
gns,
|
|
|
|
gns,
|
|
|
|
lp_queue: RwLock::default(),
|
|
|
|
lp_queue: RwLock::default(),
|
|
|
|
max_delta_size,
|
|
|
|
max_delta_size,
|
|
|
|
txn_driver: Mutex::new(FractalGNSDriver::new(txn_driver)),
|
|
|
|
|
|
|
|
model_drivers: RwLock::default(),
|
|
|
|
|
|
|
|
max_data_pressure: usize::MAX,
|
|
|
|
max_data_pressure: usize::MAX,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
@ -72,25 +66,20 @@ impl TestGlobal {
|
|
|
|
/// Normally, model drivers are not loaded on startup because of shared global state. Calling this will attempt to load
|
|
|
|
/// Normally, model drivers are not loaded on startup because of shared global state. Calling this will attempt to load
|
|
|
|
/// all model drivers
|
|
|
|
/// all model drivers
|
|
|
|
pub fn load_model_drivers(&self) -> RuntimeResult<()> {
|
|
|
|
pub fn load_model_drivers(&self) -> RuntimeResult<()> {
|
|
|
|
let mut mdl_drivers = self.model_drivers.write();
|
|
|
|
let space_idx = self.gns.namespace().idx().read();
|
|
|
|
let space_idx = self.gns.idx().read();
|
|
|
|
for (model_name, model) in self.gns.namespace().idx_models().read().iter() {
|
|
|
|
for (model_name, model) in self.gns.idx_models().read().iter() {
|
|
|
|
let model_data = model.data();
|
|
|
|
let space_uuid = space_idx.get(model_name.space()).unwrap().get_uuid();
|
|
|
|
let space_uuid = space_idx.get(model_name.space()).unwrap().get_uuid();
|
|
|
|
let driver = ModelDriver::open_model_driver(
|
|
|
|
let driver = ModelDriver::open_model_driver(
|
|
|
|
model,
|
|
|
|
model_data,
|
|
|
|
&paths_v1::model_path(
|
|
|
|
&paths_v1::model_path(
|
|
|
|
model_name.space(),
|
|
|
|
model_name.space(),
|
|
|
|
space_uuid,
|
|
|
|
space_uuid,
|
|
|
|
model_name.entity(),
|
|
|
|
model_name.entity(),
|
|
|
|
model.get_uuid(),
|
|
|
|
model_data.get_uuid(),
|
|
|
|
),
|
|
|
|
),
|
|
|
|
)?;
|
|
|
|
)?;
|
|
|
|
assert!(mdl_drivers
|
|
|
|
model.driver().initialize_model_driver(driver);
|
|
|
|
.insert(
|
|
|
|
|
|
|
|
ModelUniqueID::new(model_name.space(), model_name.entity(), model.get_uuid()),
|
|
|
|
|
|
|
|
FractalModelDriver::init(driver)
|
|
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
.is_none());
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
}
|
|
|
|
@ -103,13 +92,13 @@ impl TestGlobal {
|
|
|
|
me
|
|
|
|
me
|
|
|
|
}
|
|
|
|
}
|
|
|
|
pub fn new_with_driver_id(log_name: &str) -> Self {
|
|
|
|
pub fn new_with_driver_id(log_name: &str) -> Self {
|
|
|
|
let gns = GlobalNS::empty();
|
|
|
|
let data = GNSData::empty();
|
|
|
|
let driver = match GNSDriver::create_gns_with_name(log_name) {
|
|
|
|
let driver = match GNSDriver::create_gns_with_name(log_name) {
|
|
|
|
Ok(drv) => Ok(drv),
|
|
|
|
Ok(drv) => Ok(drv),
|
|
|
|
Err(e) => match e.kind() {
|
|
|
|
Err(e) => match e.kind() {
|
|
|
|
ErrorKind::IoError(e_) => match e_.kind() {
|
|
|
|
ErrorKind::IoError(e_) => match e_.kind() {
|
|
|
|
std::io::ErrorKind::AlreadyExists => {
|
|
|
|
std::io::ErrorKind::AlreadyExists => {
|
|
|
|
GNSDriver::open_gns_with_name(log_name, &gns)
|
|
|
|
GNSDriver::open_gns_with_name(log_name, &data)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
_ => Err(e),
|
|
|
|
_ => Err(e),
|
|
|
|
},
|
|
|
|
},
|
|
|
|
@ -117,7 +106,7 @@ impl TestGlobal {
|
|
|
|
},
|
|
|
|
},
|
|
|
|
}
|
|
|
|
}
|
|
|
|
.unwrap();
|
|
|
|
.unwrap();
|
|
|
|
Self::new(gns, 0, driver)
|
|
|
|
Self::new(GlobalNS::new(data, FractalGNSDriver::new(driver)), 0)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
@ -125,24 +114,19 @@ impl GlobalInstanceLike for TestGlobal {
|
|
|
|
fn state(&self) -> &GlobalNS {
|
|
|
|
fn state(&self) -> &GlobalNS {
|
|
|
|
&self.gns
|
|
|
|
&self.gns
|
|
|
|
}
|
|
|
|
}
|
|
|
|
fn gns_driver(&self) -> &Mutex<FractalGNSDriver> {
|
|
|
|
|
|
|
|
&self.txn_driver
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
fn taskmgr_post_high_priority(&self, task: Task<CriticalTask>) {
|
|
|
|
fn taskmgr_post_high_priority(&self, task: Task<CriticalTask>) {
|
|
|
|
match task.into_task() {
|
|
|
|
match task.into_task() {
|
|
|
|
CriticalTask::WriteBatch(mdl_id, count) => {
|
|
|
|
CriticalTask::WriteBatch(mdl_id, count) => {
|
|
|
|
let models = self.gns.idx_models().read();
|
|
|
|
let models = self.gns.namespace().idx_models().read();
|
|
|
|
let mdl = models
|
|
|
|
let mdl = models
|
|
|
|
.get(&EntityIDRef::new(mdl_id.space(), mdl_id.model()))
|
|
|
|
.get(&EntityIDRef::new(mdl_id.space(), mdl_id.model()))
|
|
|
|
.unwrap();
|
|
|
|
.unwrap();
|
|
|
|
self.model_drivers
|
|
|
|
let mut mdl_driver = mdl.driver().batch_driver().lock();
|
|
|
|
.read()
|
|
|
|
mdl_driver
|
|
|
|
.get(&mdl_id)
|
|
|
|
.as_mut()
|
|
|
|
|
|
|
|
.unwrap()
|
|
|
|
|
|
|
|
.commit_with_ctx(StdModelBatch::new(mdl.data(), count), BatchStats::new())
|
|
|
|
.unwrap()
|
|
|
|
.unwrap()
|
|
|
|
.batch_driver()
|
|
|
|
|
|
|
|
.lock()
|
|
|
|
|
|
|
|
.commit_with_ctx(StdModelBatch::new(mdl, count), BatchStats::new())
|
|
|
|
|
|
|
|
.unwrap();
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
@ -158,18 +142,10 @@ impl GlobalInstanceLike for TestGlobal {
|
|
|
|
space_uuid: Uuid,
|
|
|
|
space_uuid: Uuid,
|
|
|
|
model_name: &str,
|
|
|
|
model_name: &str,
|
|
|
|
model_uuid: Uuid,
|
|
|
|
model_uuid: Uuid,
|
|
|
|
skip_delete: bool,
|
|
|
|
|
|
|
|
) {
|
|
|
|
) {
|
|
|
|
let id = ModelUniqueID::new(space_name, model_name, model_uuid);
|
|
|
|
self.taskmgr_post_standard_priority(Task::new(GenericTask::delete_model_dir(
|
|
|
|
self.model_drivers
|
|
|
|
space_name, space_uuid, model_name, model_uuid,
|
|
|
|
.write()
|
|
|
|
)));
|
|
|
|
.remove(&id)
|
|
|
|
|
|
|
|
.expect("tried to remove non-existent model");
|
|
|
|
|
|
|
|
if !skip_delete {
|
|
|
|
|
|
|
|
self.taskmgr_post_standard_priority(Task::new(GenericTask::delete_model_dir(
|
|
|
|
|
|
|
|
space_name, space_uuid, model_name, model_uuid,
|
|
|
|
|
|
|
|
)));
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
fn initialize_model_driver(
|
|
|
|
fn initialize_model_driver(
|
|
|
|
&self,
|
|
|
|
&self,
|
|
|
|
@ -177,7 +153,7 @@ impl GlobalInstanceLike for TestGlobal {
|
|
|
|
space_uuid: Uuid,
|
|
|
|
space_uuid: Uuid,
|
|
|
|
model_name: &str,
|
|
|
|
model_name: &str,
|
|
|
|
model_uuid: Uuid,
|
|
|
|
model_uuid: Uuid,
|
|
|
|
) -> crate::engine::error::RuntimeResult<()> {
|
|
|
|
) -> crate::engine::error::RuntimeResult<FractalModelDriver> {
|
|
|
|
// create model dir
|
|
|
|
// create model dir
|
|
|
|
FileSystem::create_dir_all(&paths_v1::model_dir(
|
|
|
|
FileSystem::create_dir_all(&paths_v1::model_dir(
|
|
|
|
space_name, space_uuid, model_name, model_uuid,
|
|
|
|
space_name, space_uuid, model_name, model_uuid,
|
|
|
|
@ -185,20 +161,16 @@ impl GlobalInstanceLike for TestGlobal {
|
|
|
|
let driver = ModelDriver::create_model_driver(&paths_v1::model_path(
|
|
|
|
let driver = ModelDriver::create_model_driver(&paths_v1::model_path(
|
|
|
|
space_name, space_uuid, model_name, model_uuid,
|
|
|
|
space_name, space_uuid, model_name, model_uuid,
|
|
|
|
))?;
|
|
|
|
))?;
|
|
|
|
self.model_drivers.write().insert(
|
|
|
|
Ok(super::drivers::FractalModelDriver::init(driver))
|
|
|
|
ModelUniqueID::new(space_name, model_name, model_uuid),
|
|
|
|
|
|
|
|
super::drivers::FractalModelDriver::init(driver),
|
|
|
|
|
|
|
|
);
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
impl Drop for TestGlobal {
|
|
|
|
impl Drop for TestGlobal {
|
|
|
|
fn drop(&mut self) {
|
|
|
|
fn drop(&mut self) {
|
|
|
|
let mut txn_driver = self.txn_driver.lock();
|
|
|
|
let mut txn_driver = self.gns.gns_driver().txn_driver.lock();
|
|
|
|
GNSDriver::close_driver(&mut txn_driver.txn_driver).unwrap();
|
|
|
|
GNSDriver::close_driver(&mut txn_driver).unwrap();
|
|
|
|
for (_, model_driver) in self.model_drivers.write().drain() {
|
|
|
|
for (_, model_driver) in self.gns.namespace().idx_models().write().drain() {
|
|
|
|
model_driver.close().unwrap();
|
|
|
|
model_driver.into_driver().close().unwrap();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
Sayan Nandan
7 months ago