garage/src/server.rs

233 lines
5.6 KiB
Rust
Raw Normal View History

use std::path::PathBuf;
use std::sync::Arc;
use futures_util::future::*;
use tokio::sync::watch;
use crate::background::*;
2020-04-23 17:05:46 +00:00
use crate::config::*;
use crate::error::Error;
2020-04-23 17:05:46 +00:00
use crate::rpc::membership::System;
use crate::rpc::rpc_client::RpcHttpClient;
use crate::rpc::rpc_server::RpcServer;
2020-04-23 17:05:46 +00:00
use crate::table::table_fullcopy::*;
use crate::table::table_sharded::*;
use crate::table::*;
2020-04-23 17:05:46 +00:00
use crate::store::block::*;
use crate::store::block_ref_table::*;
use crate::store::bucket_table::*;
use crate::store::object_table::*;
use crate::store::version_table::*;
2020-04-12 11:03:55 +00:00
2020-04-23 17:05:46 +00:00
use crate::api::api_server;
2020-04-12 11:03:55 +00:00
2020-04-23 17:05:46 +00:00
use crate::admin_rpc::*;
2020-04-08 20:00:41 +00:00
pub struct Garage {
2020-04-23 17:05:46 +00:00
pub config: Config,
2020-04-08 20:00:41 +00:00
pub db: sled::Db,
pub background: Arc<BackgroundRunner>,
2020-04-08 20:00:41 +00:00
pub system: Arc<System>,
2020-04-12 11:03:55 +00:00
pub block_manager: Arc<BlockManager>,
2020-04-08 20:00:41 +00:00
pub bucket_table: Arc<Table<BucketTable, TableFullReplication>>,
pub object_table: Arc<Table<ObjectTable, TableShardedReplication>>,
pub version_table: Arc<Table<VersionTable, TableShardedReplication>>,
pub block_ref_table: Arc<Table<BlockRefTable, TableShardedReplication>>,
2020-04-08 20:00:41 +00:00
}
impl Garage {
pub async fn new(
config: Config,
db: sled::Db,
background: Arc<BackgroundRunner>,
2020-04-18 17:21:34 +00:00
rpc_server: &mut RpcServer,
) -> Arc<Self> {
2020-04-21 12:54:55 +00:00
info!("Initialize membership management system...");
2020-04-23 17:05:46 +00:00
let rpc_http_client = Arc::new(
RpcHttpClient::new(config.max_concurrent_rpc_requests, &config.rpc_tls)
.expect("Could not create RPC client"),
);
let system = System::new(
config.metadata_dir.clone(),
rpc_http_client,
background.clone(),
rpc_server,
);
2020-04-08 20:00:41 +00:00
let data_rep_param = TableShardedReplication {
2020-04-23 17:05:46 +00:00
replication_factor: config.data_replication_factor,
write_quorum: (config.data_replication_factor + 1) / 2,
2020-04-12 11:03:55 +00:00
read_quorum: 1,
};
let meta_rep_param = TableShardedReplication {
2020-04-23 17:05:46 +00:00
replication_factor: config.meta_replication_factor,
write_quorum: (config.meta_replication_factor + 1) / 2,
read_quorum: (config.meta_replication_factor + 1) / 2,
2020-04-08 20:00:41 +00:00
};
let control_rep_param = TableFullReplication::new(
2020-04-23 17:05:46 +00:00
config.meta_epidemic_factor,
(config.meta_epidemic_factor + 1) / 2,
);
info!("Initialize block manager...");
let block_manager = BlockManager::new(
&db,
config.data_dir.clone(),
data_rep_param.clone(),
system.clone(),
rpc_server,
);
2020-04-21 12:54:55 +00:00
info!("Initialize block_ref_table...");
2020-04-12 20:24:53 +00:00
let block_ref_table = Table::new(
2020-04-12 11:03:55 +00:00
BlockRefTable {
background: background.clone(),
block_manager: block_manager.clone(),
},
data_rep_param.clone(),
2020-04-08 20:00:41 +00:00
system.clone(),
&db,
2020-04-12 11:03:55 +00:00
"block_ref".to_string(),
2020-04-18 17:21:34 +00:00
rpc_server,
2020-04-16 15:04:28 +00:00
)
.await;
2020-04-21 12:54:55 +00:00
info!("Initialize version_table...");
2020-04-12 20:24:53 +00:00
let version_table = Table::new(
VersionTable {
2020-04-12 11:03:55 +00:00
background: background.clone(),
block_ref_table: block_ref_table.clone(),
},
meta_rep_param.clone(),
2020-04-09 21:45:07 +00:00
system.clone(),
&db,
"version".to_string(),
2020-04-18 17:21:34 +00:00
rpc_server,
2020-04-16 15:04:28 +00:00
)
.await;
2020-04-21 12:54:55 +00:00
info!("Initialize object_table...");
2020-04-12 20:24:53 +00:00
let object_table = Table::new(
2020-04-12 11:03:55 +00:00
ObjectTable {
background: background.clone(),
version_table: version_table.clone(),
2020-04-10 21:11:52 +00:00
},
meta_rep_param.clone(),
2020-04-10 21:11:52 +00:00
system.clone(),
&db,
2020-04-12 11:03:55 +00:00
"object".to_string(),
2020-04-18 17:21:34 +00:00
rpc_server,
2020-04-16 15:04:28 +00:00
)
.await;
2020-04-08 20:00:41 +00:00
2020-04-21 12:54:55 +00:00
info!("Initialize bucket_table...");
let bucket_table = Table::new(
BucketTable,
control_rep_param.clone(),
system.clone(),
&db,
"bucket".to_string(),
rpc_server,
)
.await;
2020-04-21 12:54:55 +00:00
info!("Initialize Garage...");
2020-04-18 17:21:34 +00:00
let garage = Arc::new(Self {
2020-04-23 17:05:46 +00:00
config,
2020-04-08 20:00:41 +00:00
db,
system: system.clone(),
2020-04-11 21:00:26 +00:00
block_manager,
background,
bucket_table,
2020-04-09 15:32:28 +00:00
object_table,
2020-04-09 21:45:07 +00:00
version_table,
2020-04-10 21:11:52 +00:00
block_ref_table,
2020-04-18 17:21:34 +00:00
});
2020-04-21 12:54:55 +00:00
info!("Crate admin RPC handler...");
AdminRpcHandler::new(garage.clone()).register_handler(rpc_server);
2020-04-21 12:54:55 +00:00
info!("Start block manager background thread...");
garage.block_manager.garage.swap(Some(garage.clone()));
garage.block_manager.clone().spawn_background_worker().await;
garage
2020-04-08 20:00:41 +00:00
}
}
async fn shutdown_signal(send_cancel: watch::Sender<bool>) -> Result<(), Error> {
// Wait for the CTRL+C signal
tokio::signal::ctrl_c()
.await
.expect("failed to install CTRL+C signal handler");
2020-04-21 12:54:55 +00:00
info!("Received CTRL+C, shutting down.");
send_cancel.broadcast(true)?;
Ok(())
}
async fn wait_from(mut chan: watch::Receiver<bool>) -> () {
while let Some(exit_now) = chan.recv().await {
if exit_now {
return;
}
}
}
pub async fn run_server(config_file: PathBuf) -> Result<(), Error> {
2020-04-21 12:54:55 +00:00
info!("Loading configuration...");
let config = read_config(config_file).expect("Unable to read config file");
2020-04-21 12:54:55 +00:00
info!("Opening database...");
2020-04-08 20:00:41 +00:00
let mut db_path = config.metadata_dir.clone();
2020-04-09 16:43:53 +00:00
db_path.push("db");
let db = sled::open(db_path).expect("Unable to open DB");
2020-04-08 20:00:41 +00:00
2020-04-21 12:54:55 +00:00
info!("Initialize RPC server...");
let mut rpc_server = RpcServer::new(config.rpc_bind_addr.clone(), config.rpc_tls.clone());
2020-04-21 12:54:55 +00:00
info!("Initializing background runner...");
2020-04-18 17:21:34 +00:00
let (send_cancel, watch_cancel) = watch::channel(false);
let background = BackgroundRunner::new(16, watch_cancel.clone());
2020-04-23 17:05:46 +00:00
let garage = Garage::new(config, db, background.clone(), &mut rpc_server).await;
2020-04-21 12:54:55 +00:00
info!("Initializing RPC and API servers...");
2020-04-18 17:21:34 +00:00
let run_rpc_server = Arc::new(rpc_server).run(wait_from(watch_cancel.clone()));
let api_server = api_server::run_api_server(garage.clone(), wait_from(watch_cancel.clone()));
futures::try_join!(
2020-04-23 17:05:46 +00:00
garage
.system
.clone()
.bootstrap(&garage.config.bootstrap_peers[..])
.map(|rv| {
info!("Bootstrap done");
Ok(rv)
}),
run_rpc_server.map(|rv| {
2020-04-21 12:54:55 +00:00
info!("RPC server exited");
rv
}),
api_server.map(|rv| {
2020-04-21 12:54:55 +00:00
info!("API server exited");
rv
}),
background.run().map(|rv| {
2020-04-21 12:54:55 +00:00
info!("Background runner exited");
Ok(rv)
}),
shutdown_signal(send_cancel),
)?;
2020-04-21 12:54:55 +00:00
info!("Cleaning up...");
Ok(())
}