Implement services
Sayan Nandan
1 year ago
parent
e309e35b63
commit
5cafc61231
@ -0,0 +1,47 @@
|
||||
/*
|
||||
* Created on Sun Sep 10 2023
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2023, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
use crate::engine::storage::v1::header_meta::HostRunMode;
|
||||
|
||||
pub struct ServerConfig {
|
||||
host_settings_version: u32,
|
||||
host_run_mode: HostRunMode,
|
||||
host_startup_counter: u64,
|
||||
}
|
||||
|
||||
impl ServerConfig {
|
||||
pub fn new(
|
||||
host_settings_version: u32,
|
||||
host_run_mode: HostRunMode,
|
||||
host_startup_counter: u64,
|
||||
) -> Self {
|
||||
Self {
|
||||
host_settings_version,
|
||||
host_run_mode,
|
||||
host_startup_counter,
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,88 @@
|
||||
/*
|
||||
* Created on Sun Sep 10 2023
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2023, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
use {
|
||||
super::util,
|
||||
crate::engine::{
|
||||
storage::v1::{data_batch::DataBatchPersistDriver, LocalFS},
|
||||
txn::gns::GNSTransactionDriverAnyFS,
|
||||
},
|
||||
parking_lot::Mutex,
|
||||
std::sync::Arc,
|
||||
};
|
||||
|
||||
/// GNS driver
|
||||
pub(super) struct FractalGNSDriver {
|
||||
status: util::Status,
|
||||
txn_driver: Mutex<GNSTransactionDriverAnyFS<LocalFS>>,
|
||||
}
|
||||
|
||||
impl FractalGNSDriver {
|
||||
pub(super) fn new(txn_driver: GNSTransactionDriverAnyFS<LocalFS>) -> Self {
|
||||
Self {
|
||||
status: util::Status::new_okay(),
|
||||
txn_driver: Mutex::new(txn_driver),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Model driver
|
||||
pub struct FractalModelDriver {
|
||||
hooks: Arc<FractalModelHooks>,
|
||||
batch_driver: Mutex<DataBatchPersistDriver<LocalFS>>,
|
||||
}
|
||||
|
||||
impl FractalModelDriver {
|
||||
/// Initialize a model driver with default settings
|
||||
pub fn init(batch_driver: DataBatchPersistDriver<LocalFS>) -> Self {
|
||||
Self {
|
||||
hooks: Arc::new(FractalModelHooks::new()),
|
||||
batch_driver: Mutex::new(batch_driver),
|
||||
}
|
||||
}
|
||||
/// Returns a reference to the batch persist driver
|
||||
pub fn batch_driver(&self) -> &Mutex<DataBatchPersistDriver<LocalFS>> {
|
||||
&self.batch_driver
|
||||
}
|
||||
}
|
||||
|
||||
/// Model hooks
|
||||
#[derive(Debug)]
|
||||
pub struct FractalModelHooks {
|
||||
status: util::Status,
|
||||
}
|
||||
|
||||
impl FractalModelHooks {
|
||||
#[cfg(test)]
|
||||
pub fn test() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
status: util::Status::new_okay(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,303 @@
|
||||
/*
|
||||
* Created on Sat Sep 09 2023
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2023, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
use {
|
||||
super::ModelUniqueID,
|
||||
crate::{
|
||||
engine::core::model::{delta::DataDelta, Model},
|
||||
util::os,
|
||||
},
|
||||
std::path::PathBuf,
|
||||
tokio::{
|
||||
fs,
|
||||
sync::mpsc::{UnboundedReceiver, UnboundedSender},
|
||||
task::JoinHandle,
|
||||
},
|
||||
};
|
||||
|
||||
/// A task for the [`FractalMgr`] to perform
|
||||
pub struct Task<T> {
|
||||
threshold: usize,
|
||||
task: T,
|
||||
}
|
||||
|
||||
impl<T> Task<T> {
|
||||
const THRESHOLD: usize = 10;
|
||||
/// Create a new task with the default threshold
|
||||
pub fn new(task: T) -> Self {
|
||||
Self::with_threshold(task, Self::THRESHOLD)
|
||||
}
|
||||
/// Create a task with the given threshold
|
||||
fn with_threshold(task: T, threshold: usize) -> Self {
|
||||
Self { threshold, task }
|
||||
}
|
||||
}
|
||||
|
||||
/// A general task
|
||||
pub enum GenericTask {
|
||||
/// Delete a single file
|
||||
DeleteFile(PathBuf),
|
||||
/// Delete a directory (and all its children)
|
||||
DeleteDirAll(PathBuf),
|
||||
}
|
||||
|
||||
/// A critical task
|
||||
pub enum CriticalTask {
|
||||
/// Write a new data batch
|
||||
WriteBatch(ModelUniqueID, usize),
|
||||
}
|
||||
|
||||
/// The task manager
|
||||
pub(super) struct FractalMgr {
|
||||
hp_dispatcher: UnboundedSender<Task<CriticalTask>>,
|
||||
general_dispatcher: UnboundedSender<Task<GenericTask>>,
|
||||
runtime_stats: FractalRTStat,
|
||||
}
|
||||
|
||||
pub(super) struct FractalRTStat {
|
||||
mem_free_bytes: u64,
|
||||
per_mdl_delta_max_size: usize,
|
||||
}
|
||||
|
||||
impl FractalRTStat {
|
||||
fn init(model_cnt: usize) -> Self {
|
||||
let mem_free_bytes = os::free_memory_in_bytes();
|
||||
let allowed_delta_limit = mem_free_bytes as f64 * 0.02;
|
||||
let per_model_limit = allowed_delta_limit / model_cnt.max(1) as f64;
|
||||
Self {
|
||||
mem_free_bytes,
|
||||
per_mdl_delta_max_size: per_model_limit as usize / sizeof!(DataDelta),
|
||||
}
|
||||
}
|
||||
pub(super) fn mem_free_bytes(&self) -> u64 {
|
||||
self.mem_free_bytes
|
||||
}
|
||||
pub(super) fn per_mdl_delta_max_size(&self) -> usize {
|
||||
self.per_mdl_delta_max_size
|
||||
}
|
||||
}
|
||||
|
||||
impl FractalMgr {
|
||||
pub(super) fn new(
|
||||
hp_dispatcher: UnboundedSender<Task<CriticalTask>>,
|
||||
general_dispatcher: UnboundedSender<Task<GenericTask>>,
|
||||
model_count: usize,
|
||||
) -> Self {
|
||||
Self {
|
||||
hp_dispatcher,
|
||||
general_dispatcher,
|
||||
runtime_stats: FractalRTStat::init(model_count),
|
||||
}
|
||||
}
|
||||
pub fn get_rt_stat(&self) -> &FractalRTStat {
|
||||
&self.runtime_stats
|
||||
}
|
||||
/// Add a high priority task to the queue
|
||||
///
|
||||
/// ## Panics
|
||||
///
|
||||
/// This will panic if the high priority executor has crashed or exited
|
||||
pub fn post_high_priority(&self, task: Task<CriticalTask>) {
|
||||
self.hp_dispatcher.send(task).unwrap()
|
||||
}
|
||||
/// Add a low priority task to the queue
|
||||
///
|
||||
/// ## Panics
|
||||
///
|
||||
/// This will panic if the low priority executor has crashed or exited
|
||||
pub fn post_low_priority(&self, task: Task<GenericTask>) {
|
||||
self.general_dispatcher.send(task).unwrap()
|
||||
}
|
||||
}
|
||||
|
||||
/// Handles to all the services that fractal needs. These are spawned on the default runtime
|
||||
pub struct FractalServiceHandles {
|
||||
pub hp_handle: JoinHandle<()>,
|
||||
pub lp_handle: JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl FractalMgr {
|
||||
/// Start all background services, and return their handles
|
||||
pub(super) fn start_all(
|
||||
global: super::Global,
|
||||
lp_receiver: UnboundedReceiver<Task<GenericTask>>,
|
||||
hp_receiver: UnboundedReceiver<Task<CriticalTask>>,
|
||||
) -> FractalServiceHandles {
|
||||
let fractal_mgr = global.get_state().fractal_mgr();
|
||||
let hp_handle = tokio::spawn(async move {
|
||||
FractalMgr::hp_executor_svc(fractal_mgr, global, hp_receiver).await
|
||||
});
|
||||
let lp_handle = tokio::spawn(async move {
|
||||
FractalMgr::general_executor_svc(fractal_mgr, global, lp_receiver).await
|
||||
});
|
||||
FractalServiceHandles {
|
||||
hp_handle,
|
||||
lp_handle,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// services
|
||||
impl FractalMgr {
|
||||
const GENERAL_EXECUTOR_WINDOW: u64 = 5 * 60;
|
||||
/// The high priority executor service runs in the background to take care of high priority tasks and take any
|
||||
/// appropriate action. It will exclusively own the high priority queue since it is the only broker that is
|
||||
/// allowed to perform HP tasks
|
||||
pub async fn hp_executor_svc(
|
||||
&'static self,
|
||||
global: super::Global,
|
||||
mut receiver: UnboundedReceiver<Task<CriticalTask>>,
|
||||
) {
|
||||
loop {
|
||||
let Some(Task { threshold, task }) = receiver.recv().await else {
|
||||
return; // all handles closed; nothing left to do
|
||||
};
|
||||
// TODO(@ohsayan): check threshold and update hooks
|
||||
match task {
|
||||
CriticalTask::WriteBatch(model_id, observed_size) => {
|
||||
let mdl_drivers = global.get_state().get_mdl_drivers().read();
|
||||
let Some(mdl_driver) = mdl_drivers.get(&model_id) else {
|
||||
// because we maximize throughput, the model driver may have been already removed but this task
|
||||
// was way behind in the queue
|
||||
continue;
|
||||
};
|
||||
let res = global.namespace().with_model(
|
||||
(model_id.space(), model_id.model()),
|
||||
|model| {
|
||||
if model.get_uuid() != model_id.uuid() {
|
||||
// once again, throughput maximization will lead to, in extremely rare cases, this
|
||||
// branch returning. but it is okay
|
||||
return Ok(());
|
||||
}
|
||||
// mark that we're taking these deltas
|
||||
model.delta_state().__fractal_take_from_data_delta(
|
||||
observed_size,
|
||||
super::FractalToken::new(),
|
||||
);
|
||||
Self::try_write_model_data_batch(model, observed_size, mdl_driver)
|
||||
},
|
||||
);
|
||||
match res {
|
||||
Ok(()) => {}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"Error writing data batch for model {}. Retrying...",
|
||||
model_id.uuid()
|
||||
);
|
||||
// enqueue again for retrying
|
||||
self.hp_dispatcher
|
||||
.send(Task::with_threshold(
|
||||
CriticalTask::WriteBatch(model_id, observed_size),
|
||||
threshold - 1,
|
||||
))
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
/// The general priority task or simply the general queue takes of care of low priority and other standard priority
|
||||
/// tasks (such as those running on a schedule). A low priority task can be promoted to a high priority task, and the
|
||||
/// discretion of the GP executor. Similarly, the executor owns the general purpose task queue since it is the sole broker
|
||||
/// for such tasks
|
||||
pub async fn general_executor_svc(
|
||||
&'static self,
|
||||
global: super::Global,
|
||||
mut lpq: UnboundedReceiver<Task<GenericTask>>,
|
||||
) {
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = tokio::time::sleep(std::time::Duration::from_secs(Self::GENERAL_EXECUTOR_WINDOW)) => {
|
||||
let mdl_drivers = global.get_state().get_mdl_drivers().read();
|
||||
for (model_id, driver) in mdl_drivers.iter() {
|
||||
let mut observed_len = 0;
|
||||
let res = global.namespace().with_model((model_id.space(), model_id.model()), |model| {
|
||||
if model.get_uuid() != model_id.uuid() {
|
||||
// once again, throughput maximization will lead to, in extremely rare cases, this
|
||||
// branch returning. but it is okay
|
||||
return Ok(());
|
||||
}
|
||||
// mark that we're taking these deltas
|
||||
observed_len = model.delta_state().__fractal_take_full_from_data_delta(super::FractalToken::new());
|
||||
Self::try_write_model_data_batch(model, observed_len, driver)
|
||||
});
|
||||
match res {
|
||||
Ok(()) => {}
|
||||
Err(_) => {
|
||||
// this failure is *not* good, so we want to promote this to a critical task
|
||||
self.hp_dispatcher.send(Task::new(CriticalTask::WriteBatch(model_id.clone(), observed_len))).unwrap()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
task = lpq.recv() => {
|
||||
let Some(Task { threshold, task }) = task else {
|
||||
return;
|
||||
};
|
||||
// TODO(@ohsayan): threshold
|
||||
match task {
|
||||
GenericTask::DeleteFile(f) => {
|
||||
if let Err(_) = fs::remove_file(&f).await {
|
||||
self.general_dispatcher.send(
|
||||
Task::with_threshold(GenericTask::DeleteFile(f), threshold - 1)
|
||||
).unwrap();
|
||||
}
|
||||
}
|
||||
GenericTask::DeleteDirAll(dir) => {
|
||||
if let Err(_) = fs::remove_dir_all(&dir).await {
|
||||
self.general_dispatcher.send(
|
||||
Task::with_threshold(GenericTask::DeleteDirAll(dir), threshold - 1)
|
||||
).unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// util
|
||||
impl FractalMgr {
|
||||
/// Attempt to write a model data batch with the observed size.
|
||||
///
|
||||
/// The zero check is essential
|
||||
fn try_write_model_data_batch(
|
||||
model: &Model,
|
||||
observed_size: usize,
|
||||
mdl_driver: &super::FractalModelDriver,
|
||||
) -> Result<(), crate::engine::error::DatabaseError> {
|
||||
if observed_size == 0 {
|
||||
// no changes, all good
|
||||
return Ok(());
|
||||
}
|
||||
// try flushing the batch
|
||||
let mut batch_driver = mdl_driver.batch_driver().lock();
|
||||
batch_driver.write_new_batch(model, observed_size)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,204 @@
|
||||
/*
|
||||
* Created on Sat Sep 09 2023
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2023, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
use {
|
||||
super::{
|
||||
core::GlobalNS, data::uuid::Uuid, storage::v1::LocalFS, txn::gns::GNSTransactionDriverAnyFS,
|
||||
},
|
||||
parking_lot::RwLock,
|
||||
std::{collections::HashMap, mem::MaybeUninit},
|
||||
tokio::sync::mpsc::unbounded_channel,
|
||||
};
|
||||
|
||||
mod config;
|
||||
mod drivers;
|
||||
mod mgr;
|
||||
mod util;
|
||||
pub use {
|
||||
config::ServerConfig,
|
||||
drivers::FractalModelDriver,
|
||||
mgr::{CriticalTask, GenericTask, Task},
|
||||
util::FractalToken,
|
||||
};
|
||||
|
||||
pub type ModelDrivers = HashMap<ModelUniqueID, drivers::FractalModelDriver>;
|
||||
|
||||
static mut GLOBAL: MaybeUninit<GlobalState> = MaybeUninit::uninit();
|
||||
|
||||
/*
|
||||
global state init
|
||||
*/
|
||||
|
||||
/// Returned by [`enable_and_start_all`]. This contains a [`Global`] handle that can be used to easily access global
|
||||
/// data
|
||||
pub struct GlobalStateStart {
|
||||
pub global: Global,
|
||||
pub mgr_handles: mgr::FractalServiceHandles,
|
||||
}
|
||||
|
||||
/// Enable all drivers and start all engines
|
||||
///
|
||||
/// ## Safety
|
||||
///
|
||||
/// Must be called iff this is the only thread calling it
|
||||
pub unsafe fn enable_and_start_all(
|
||||
gns: GlobalNS,
|
||||
config: config::ServerConfig,
|
||||
gns_driver: GNSTransactionDriverAnyFS<LocalFS>,
|
||||
model_drivers: ModelDrivers,
|
||||
) -> GlobalStateStart {
|
||||
let model_cnt_on_boot = model_drivers.len();
|
||||
let gns_driver = drivers::FractalGNSDriver::new(gns_driver);
|
||||
let mdl_driver = RwLock::new(model_drivers);
|
||||
let (hp_sender, hp_recv) = unbounded_channel();
|
||||
let (lp_sender, lp_recv) = unbounded_channel();
|
||||
let global_state = GlobalState::new(
|
||||
gns,
|
||||
gns_driver,
|
||||
mdl_driver,
|
||||
mgr::FractalMgr::new(hp_sender, lp_sender, model_cnt_on_boot),
|
||||
config,
|
||||
);
|
||||
GLOBAL = MaybeUninit::new(global_state);
|
||||
let token = Global::new();
|
||||
GlobalStateStart {
|
||||
global: token,
|
||||
mgr_handles: mgr::FractalMgr::start_all(token, lp_recv, hp_recv),
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
global access
|
||||
*/
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
/// A handle to the global state
|
||||
pub struct Global(());
|
||||
|
||||
impl Global {
|
||||
fn new() -> Self {
|
||||
Self(())
|
||||
}
|
||||
pub(self) fn get_state(&self) -> &'static GlobalState {
|
||||
unsafe { GLOBAL.assume_init_ref() }
|
||||
}
|
||||
/// Returns a handle to the [`GlobalNS`]
|
||||
pub fn namespace(&self) -> &'static GlobalNS {
|
||||
&unsafe { GLOBAL.assume_init_ref() }.gns
|
||||
}
|
||||
/// Post an urgent task
|
||||
pub fn post_high_priority_task(&self, task: Task<CriticalTask>) {
|
||||
self.get_state().fractal_mgr().post_high_priority(task)
|
||||
}
|
||||
/// Post a task with normal priority
|
||||
///
|
||||
/// NB: It is not guaranteed that the task will remain as a low priority task because the scheduler can choose
|
||||
/// to promote the task to a high priority task, if it deems necessary.
|
||||
pub fn post_standard_priority_task(&self, task: Task<GenericTask>) {
|
||||
self.get_state().fractal_mgr().post_low_priority(task)
|
||||
}
|
||||
/// Returns the maximum size a model's delta size can hit before it should immediately issue a batch write request
|
||||
/// to avoid memory pressure
|
||||
pub fn get_max_delta_size(&self) -> usize {
|
||||
self.get_state()
|
||||
.fractal_mgr()
|
||||
.get_rt_stat()
|
||||
.per_mdl_delta_max_size()
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
global state
|
||||
*/
|
||||
|
||||
/// The global state
|
||||
struct GlobalState {
|
||||
gns: GlobalNS,
|
||||
gns_driver: drivers::FractalGNSDriver,
|
||||
mdl_driver: RwLock<ModelDrivers>,
|
||||
task_mgr: mgr::FractalMgr,
|
||||
config: config::ServerConfig,
|
||||
}
|
||||
|
||||
impl GlobalState {
|
||||
fn new(
|
||||
gns: GlobalNS,
|
||||
gns_driver: drivers::FractalGNSDriver,
|
||||
mdl_driver: RwLock<ModelDrivers>,
|
||||
task_mgr: mgr::FractalMgr,
|
||||
config: config::ServerConfig,
|
||||
) -> Self {
|
||||
Self {
|
||||
gns,
|
||||
gns_driver,
|
||||
mdl_driver,
|
||||
task_mgr,
|
||||
config,
|
||||
}
|
||||
}
|
||||
pub(self) fn get_mdl_drivers(&self) -> &RwLock<ModelDrivers> {
|
||||
&self.mdl_driver
|
||||
}
|
||||
pub(self) fn fractal_mgr(&self) -> &mgr::FractalMgr {
|
||||
&self.task_mgr
|
||||
}
|
||||
}
|
||||
|
||||
// these impls are completely fine
|
||||
unsafe impl Send for GlobalState {}
|
||||
unsafe impl Sync for GlobalState {}
|
||||
|
||||
/// An unique signature that identifies a model, and only that model (guaranteed by the OS's random source)
|
||||
// NB(@ohsayan): if there are collisions, which I absolutely do not expect any instances of, pool in the space's UUID
|
||||
#[derive(Debug, PartialEq, Eq, Hash, Clone)]
|
||||
pub struct ModelUniqueID {
|
||||
space: Box<str>,
|
||||
model: Box<str>,
|
||||
uuid: Uuid,
|
||||
}
|
||||
|
||||
impl ModelUniqueID {
|
||||
/// Create a new unique model ID
|
||||
pub fn new(space: &str, model: &str, uuid: Uuid) -> Self {
|
||||
Self {
|
||||
space: space.into(),
|
||||
model: model.into(),
|
||||
uuid,
|
||||
}
|
||||
}
|
||||
/// Returns the space name
|
||||
pub fn space(&self) -> &str {
|
||||
self.space.as_ref()
|
||||
}
|
||||
/// Returns the model name
|
||||
pub fn model(&self) -> &str {
|
||||
self.model.as_ref()
|
||||
}
|
||||
/// Returns the uuid
|
||||
pub fn uuid(&self) -> Uuid {
|
||||
self.uuid
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,78 @@
|
||||
/*
|
||||
* Created on Sat Sep 09 2023
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2023, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct Status {
|
||||
okay: AtomicBool,
|
||||
}
|
||||
|
||||
impl Status {
|
||||
pub const fn new_okay() -> Self {
|
||||
Self::new(true)
|
||||
}
|
||||
pub const fn new_iffy() -> Self {
|
||||
Self::new(false)
|
||||
}
|
||||
const fn new(v: bool) -> Self {
|
||||
Self {
|
||||
okay: AtomicBool::new(v),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Status {
|
||||
pub fn is_iffy(&self) -> bool {
|
||||
!self._get()
|
||||
}
|
||||
pub fn is_healthy(&self) -> bool {
|
||||
self._get()
|
||||
}
|
||||
fn _get(&self) -> bool {
|
||||
self.okay.load(Ordering::Acquire)
|
||||
}
|
||||
}
|
||||
|
||||
impl Status {
|
||||
pub(super) fn set_okay(&self) {
|
||||
self._set(true)
|
||||
}
|
||||
pub(super) fn set_iffy(&self) {
|
||||
self._set(false)
|
||||
}
|
||||
fn _set(&self, v: bool) {
|
||||
self.okay.store(v, Ordering::Release)
|
||||
}
|
||||
}
|
||||
|
||||
/// A special token for fractal calls
|
||||
pub struct FractalToken(());
|
||||
impl FractalToken {
|
||||
pub(super) fn new() -> Self {
|
||||
Self(())
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,100 @@
|
||||
/*
|
||||
* Created on Sun Sep 10 2023
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2023, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
use crate::engine::{
|
||||
core::GlobalNS,
|
||||
data::uuid::Uuid,
|
||||
fractal::{FractalModelDriver, ModelDrivers, ModelUniqueID},
|
||||
storage::v1::{batch_jrnl, header_meta::HostRunMode, LocalFS, SDSSErrorContext, SDSSResult},
|
||||
txn::gns::GNSTransactionDriverAnyFS,
|
||||
};
|
||||
|
||||
const GNS_FILE_PATH: &str = "gns.db-tlog";
|
||||
|
||||
pub struct SEInitState {
|
||||
pub txn_driver: GNSTransactionDriverAnyFS<super::LocalFS>,
|
||||
pub model_drivers: ModelDrivers,
|
||||
}
|
||||
|
||||
impl SEInitState {
|
||||
pub fn new(
|
||||
txn_driver: GNSTransactionDriverAnyFS<super::LocalFS>,
|
||||
model_drivers: ModelDrivers,
|
||||
) -> Self {
|
||||
Self {
|
||||
txn_driver,
|
||||
model_drivers,
|
||||
}
|
||||
}
|
||||
pub fn try_init(
|
||||
host_setting_version: u32,
|
||||
host_run_mode: HostRunMode,
|
||||
host_startup_counter: u64,
|
||||
) -> SDSSResult<Self> {
|
||||
let gns = GlobalNS::empty();
|
||||
let gns_txn_driver = GNSTransactionDriverAnyFS::<LocalFS>::open_or_reinit_with_name(
|
||||
&gns,
|
||||
GNS_FILE_PATH,
|
||||
host_setting_version,
|
||||
host_run_mode,
|
||||
host_startup_counter,
|
||||
)?;
|
||||
let mut model_drivers = ModelDrivers::new();
|
||||
for (space_name, space) in gns.spaces().read().iter() {
|
||||
let space_uuid = space.get_uuid();
|
||||
for (model_name, model) in space.models().read().iter() {
|
||||
let path = Self::model_path(space_name, space_uuid, model_name, model.get_uuid());
|
||||
let persist_driver = match batch_jrnl::open_or_reinit(
|
||||
&path,
|
||||
model,
|
||||
host_setting_version,
|
||||
host_run_mode,
|
||||
host_startup_counter,
|
||||
) {
|
||||
Ok(j) => j,
|
||||
Err(e) => {
|
||||
return Err(e.with_extra(format!(
|
||||
"failed to restore model data from journal in `{path}`"
|
||||
)))
|
||||
}
|
||||
};
|
||||
let _ = model_drivers.insert(
|
||||
ModelUniqueID::new(space_name, model_name, model.get_uuid()),
|
||||
FractalModelDriver::init(persist_driver),
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(SEInitState::new(gns_txn_driver, model_drivers))
|
||||
}
|
||||
fn model_path(
|
||||
space_name: &str,
|
||||
space_uuid: Uuid,
|
||||
model_name: &str,
|
||||
model_uuid: Uuid,
|
||||
) -> String {
|
||||
format!("data/{space_name}-{space_uuid}/{model_name}-{model_uuid}/data.db-btlog")
|
||||
}
|
||||
}
|
||||
@ -1,70 +0,0 @@
|
||||
/*
|
||||
* Created on Mon Aug 28 2023
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2023, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
use crate::{
|
||||
engine::core::{model::delta::DataDelta, GlobalNS},
|
||||
util::os,
|
||||
};
|
||||
|
||||
type Buf = Vec<u8>;
|
||||
|
||||
/*
|
||||
memory adjustments
|
||||
*/
|
||||
|
||||
/// free memory in bytes
|
||||
static mut FREEMEM_BYTES: u64 = 0;
|
||||
/// capacity in bytes, per linked list
|
||||
static mut CAP_PER_LL_BYTES: u64 = 0;
|
||||
/// maximum number of nodes in linked list
|
||||
static mut MAX_NODES_IN_LL_CNT: usize = 0;
|
||||
|
||||
/// Set the free memory and cap for deltas so that we don't bust through memory
|
||||
///
|
||||
/// ## Safety
|
||||
/// - All models must have been loaded
|
||||
/// - This must be called **before** the arbiter spawns threads for connections
|
||||
pub unsafe fn set_limits(gns: &GlobalNS) {
|
||||
let model_cnt: usize = gns
|
||||
.spaces()
|
||||
.read()
|
||||
.values()
|
||||
.map(|space| space.models().read().len())
|
||||
.sum();
|
||||
let available_mem = os::free_memory_in_bytes();
|
||||
FREEMEM_BYTES = available_mem;
|
||||
CAP_PER_LL_BYTES =
|
||||
((available_mem / core::cmp::max(1, model_cnt) as u64) as f64 * 0.002) as u64;
|
||||
MAX_NODES_IN_LL_CNT = CAP_PER_LL_BYTES as usize / (sizeof!(DataDelta) + sizeof!(u64));
|
||||
}
|
||||
|
||||
/// Returns the maximum number of nodes that can be stored inside a delta queue for a model
|
||||
///
|
||||
/// Currently hardcoded to 0.2% of free memory after all datasets have been loaded
|
||||
pub unsafe fn get_max_delta_queue_size() -> usize {
|
||||
// TODO(@ohsayan): dynamically approximate this limit
|
||||
MAX_NODES_IN_LL_CNT
|
||||
}
|
||||
Loading…
Reference in New Issue