aboutsummaryrefslogtreecommitdiff
path: root/src/model/garage.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/model/garage.rs')
-rw-r--r--src/model/garage.rs89
1 files changed, 80 insertions, 9 deletions
diff --git a/src/model/garage.rs b/src/model/garage.rs
index 3daa1b33..721d5e3a 100644
--- a/src/model/garage.rs
+++ b/src/model/garage.rs
@@ -7,6 +7,7 @@ use garage_db as db;
use garage_util::background::*;
use garage_util::config::*;
use garage_util::error::*;
+use garage_util::persister::PersisterShared;
use garage_rpc::replication_mode::ReplicationMode;
use garage_rpc::system::System;
@@ -17,6 +18,8 @@ use garage_table::replication::TableShardedReplication;
use garage_table::*;
use crate::s3::block_ref_table::*;
+use crate::s3::lifecycle_worker;
+use crate::s3::mpu_table::*;
use crate::s3::object_table::*;
use crate::s3::version_table::*;
@@ -57,11 +60,18 @@ pub struct Garage {
pub object_table: Arc<Table<ObjectTable, TableShardedReplication>>,
/// Counting table containing object counters
pub object_counter_table: Arc<IndexCounter<Object>>,
+ /// Table containing S3 multipart uploads
+ pub mpu_table: Arc<Table<MultipartUploadTable, TableShardedReplication>>,
+ /// Counting table containing multipart object counters
+ pub mpu_counter_table: Arc<IndexCounter<MultipartUpload>>,
/// Table containing S3 object versions
pub version_table: Arc<Table<VersionTable, TableShardedReplication>>,
/// Table containing S3 block references (not blocks themselves)
pub block_ref_table: Arc<Table<BlockRefTable, TableShardedReplication>>,
+ /// Persister for lifecycle worker info
+ pub lifecycle_persister: PersisterShared<lifecycle_worker::LifecycleWorkerPersisted>,
+
#[cfg(feature = "k2v")]
pub k2v: GarageK2V,
}
@@ -82,8 +92,22 @@ impl Garage {
// Create meta dir and data dir if they don't exist already
std::fs::create_dir_all(&config.metadata_dir)
.ok_or_message("Unable to create Garage metadata directory")?;
- std::fs::create_dir_all(&config.data_dir)
- .ok_or_message("Unable to create Garage data directory")?;
+ match &config.data_dir {
+ DataDirEnum::Single(data_dir) => {
+ std::fs::create_dir_all(data_dir).ok_or_message(format!(
+ "Unable to create Garage data directory: {}",
+ data_dir.to_string_lossy()
+ ))?;
+ }
+ DataDirEnum::Multiple(data_dirs) => {
+ for dir in data_dirs {
+ std::fs::create_dir_all(&dir.path).ok_or_message(format!(
+ "Unable to create Garage data directory: {}",
+ dir.path.to_string_lossy()
+ ))?;
+ }
+ }
+ }
info!("Opening database...");
let mut db_path = config.metadata_dir.clone();
@@ -91,6 +115,11 @@ impl Garage {
// ---- Sled DB ----
#[cfg(feature = "sled")]
"sled" => {
+ if config.metadata_fsync {
+ return Err(Error::Message(format!(
+ "`metadata_fsync = true` is not supported with the Sled database engine"
+ )));
+ }
db_path.push("db");
info!("Opening Sled database at: {}", db_path.display());
let db = db::sled_adapter::sled::Config::default()
@@ -109,6 +138,15 @@ impl Garage {
db_path.push("db.sqlite");
info!("Opening Sqlite database at: {}", db_path.display());
let db = db::sqlite_adapter::rusqlite::Connection::open(db_path)
+ .and_then(|db| {
+ db.pragma_update(None, "journal_mode", &"WAL")?;
+ if config.metadata_fsync {
+ db.pragma_update(None, "synchronous", &"NORMAL")?;
+ } else {
+ db.pragma_update(None, "synchronous", &"OFF")?;
+ }
+ Ok(db)
+ })
.ok_or_message("Unable to open sqlite DB")?;
db::sqlite_adapter::SqliteDb::init(db)
}
@@ -133,8 +171,10 @@ impl Garage {
env_builder.max_readers(500);
env_builder.map_size(map_size);
unsafe {
- env_builder.flag(heed::flags::Flags::MdbNoSync);
env_builder.flag(heed::flags::Flags::MdbNoMetaSync);
+ if !config.metadata_fsync {
+ env_builder.flag(heed::flags::Flags::MdbNoSync);
+ }
}
let db = match env_builder.open(&db_path) {
Err(heed::Error::Io(e)) if e.kind() == std::io::ErrorKind::OutOfMemory => {
@@ -178,6 +218,9 @@ impl Garage {
let replication_mode = ReplicationMode::parse(&config.replication_mode)
.ok_or_message("Invalid replication_mode in config file.")?;
+ info!("Initialize background variable system...");
+ let mut bg_vars = vars::BgVars::new();
+
info!("Initialize membership management system...");
let system = System::new(network_key, replication_mode, &config)?;
@@ -204,10 +247,12 @@ impl Garage {
let block_manager = BlockManager::new(
&db,
config.data_dir.clone(),
+ config.data_fsync,
config.compression_level,
data_rep_param,
system.clone(),
- );
+ )?;
+ block_manager.register_bg_vars(&mut bg_vars);
// ---- admin tables ----
info!("Initialize bucket_table...");
@@ -244,6 +289,20 @@ impl Garage {
&db,
);
+ info!("Initialize multipart upload counter table...");
+ let mpu_counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), &db);
+
+ info!("Initialize multipart upload table...");
+ let mpu_table = Table::new(
+ MultipartUploadTable {
+ version_table: version_table.clone(),
+ mpu_counter_table: mpu_counter_table.clone(),
+ },
+ meta_rep_param.clone(),
+ system.clone(),
+ &db,
+ );
+
info!("Initialize object counter table...");
let object_counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), &db);
@@ -252,6 +311,7 @@ impl Garage {
let object_table = Table::new(
ObjectTable {
version_table: version_table.clone(),
+ mpu_table: mpu_table.clone(),
object_counter_table: object_counter_table.clone(),
},
meta_rep_param.clone(),
@@ -259,14 +319,15 @@ impl Garage {
&db,
);
+ info!("Load lifecycle worker state...");
+ let lifecycle_persister =
+ PersisterShared::new(&system.metadata_dir, "lifecycle_worker_state");
+ lifecycle_worker::register_bg_vars(&lifecycle_persister, &mut bg_vars);
+
// ---- K2V ----
#[cfg(feature = "k2v")]
let k2v = GarageK2V::new(system.clone(), &db, meta_rep_param);
- // Initialize bg vars
- let mut bg_vars = vars::BgVars::new();
- block_manager.register_bg_vars(&mut bg_vars);
-
// -- done --
Ok(Arc::new(Self {
config,
@@ -280,14 +341,17 @@ impl Garage {
key_table,
object_table,
object_counter_table,
+ mpu_table,
+ mpu_counter_table,
version_table,
block_ref_table,
+ lifecycle_persister,
#[cfg(feature = "k2v")]
k2v,
}))
}
- pub fn spawn_workers(&self, bg: &BackgroundRunner) {
+ pub fn spawn_workers(self: &Arc<Self>, bg: &BackgroundRunner) {
self.block_manager.spawn_workers(bg);
self.bucket_table.spawn_workers(bg);
@@ -296,9 +360,16 @@ impl Garage {
self.object_table.spawn_workers(bg);
self.object_counter_table.spawn_workers(bg);
+ self.mpu_table.spawn_workers(bg);
+ self.mpu_counter_table.spawn_workers(bg);
self.version_table.spawn_workers(bg);
self.block_ref_table.spawn_workers(bg);
+ bg.spawn_worker(lifecycle_worker::LifecycleWorker::new(
+ self.clone(),
+ self.lifecycle_persister.clone(),
+ ));
+
#[cfg(feature = "k2v")]
self.k2v.spawn_workers(bg);
}