Deuxfleurs/garage
44
56
Fork 74
Code Issues 100 Pull requests 14 Releases 30 Wiki Activity

WIP add content defined chunking #42

Closed
trinity-1686a wants to merge 42 commits from content-defined-chunking into master
pull from: content-defined-chunking
merge into: Deuxfleurs:master
Conversation 3 Commits 42 Files changed 60 +7 -9
3 changed files with 7 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit 0aad2f2e06 - Show all commits

6
src/table/gc.rs
View file

@ -25,8 +25,8 @@ const TABLE_GC_BATCH_SIZE: usize = 1024;
const TABLE_GC_RPC_TIMEOUT: Duration = Duration::from_secs(30); const TABLE_GC_RPC_TIMEOUT: Duration = Duration::from_secs(30);
pub struct TableGC<F: TableSchema, R: TableReplication> { pub struct TableGC<F: TableSchema, R: TableReplication> {
data: Arc<TableData<F, R>>,
system: Arc<System>, system: Arc<System>,
data: Arc<TableData<F, R>>,
rpc_client: Arc<RpcClient<GcRPC>>, rpc_client: Arc<RpcClient<GcRPC>>,
} }
@ -46,16 +46,16 @@ where
R: TableReplication + 'static, R: TableReplication + 'static,
{ {
pub(crate) fn launch( pub(crate) fn launch(
data: Arc<TableData<F, R>>,
system: Arc<System>, system: Arc<System>,
data: Arc<TableData<F, R>>,
rpc_server: &mut RpcServer, rpc_server: &mut RpcServer,
) -> Arc<Self> { ) -> Arc<Self> {
let rpc_path = format!("table_{}/gc", data.name); let rpc_path = format!("table_{}/gc", data.name);
let rpc_client = system.rpc_client::<GcRPC>(&rpc_path); let rpc_client = system.rpc_client::<GcRPC>(&rpc_path);
let gc = Arc::new(Self { let gc = Arc::new(Self {
data: data.clone(),
system: system.clone(), system: system.clone(),
data: data.clone(),
rpc_client, rpc_client,
}); });

6
src/table/merkle.rs
View file

@ -38,7 +38,6 @@ pub fn hash_of_merkle_partition_opt(p: Option<MerklePartition>) -> Hash {
pub struct MerkleUpdater<F: TableSchema, R: TableReplication> { pub struct MerkleUpdater<F: TableSchema, R: TableReplication> {
data: Arc<TableData<F, R>>, data: Arc<TableData<F, R>>,
background: Arc<BackgroundRunner>,
// Content of the todo tree: items where // Content of the todo tree: items where
// - key = the key of an item in the main table, ie hash(partition_key)+sort_key // - key = the key of an item in the main table, ie hash(partition_key)+sort_key
@ -86,19 +85,18 @@ where
R: TableReplication + 'static, R: TableReplication + 'static,
{ {
pub(crate) fn launch( pub(crate) fn launch(
background: &BackgroundRunner,
data: Arc<TableData<F, R>>, data: Arc<TableData<F, R>>,
background: Arc<BackgroundRunner>,
) -> Arc<Self> { ) -> Arc<Self> {
let empty_node_hash = blake2sum(&rmp_to_vec_all_named(&MerkleNode::Empty).unwrap()[..]); let empty_node_hash = blake2sum(&rmp_to_vec_all_named(&MerkleNode::Empty).unwrap()[..]);
let ret = Arc::new(Self { let ret = Arc::new(Self {
data, data,
background,
empty_node_hash, empty_node_hash,
}); });
let ret2 = ret.clone(); let ret2 = ret.clone();
ret.background.spawn_worker( background.spawn_worker(
format!("Merkle tree updater for {}", ret.data.name), format!("Merkle tree updater for {}", ret.data.name),
|must_exit: watch::Receiver<bool>| ret2.updater_loop(must_exit), |must_exit: watch::Receiver<bool>| ret2.updater_loop(must_exit),
); );

4
src/table/table.rs
View file

@ -66,7 +66,7 @@ where
let data = TableData::new(name, instance, replication, db); let data = TableData::new(name, instance, replication, db);
let merkle_updater = MerkleUpdater::launch(data.clone(), system.background.clone()); let merkle_updater = MerkleUpdater::launch(&system.background, data.clone());
let syncer = TableSyncer::launch( let syncer = TableSyncer::launch(
system.clone(), system.clone(),
@ -74,7 +74,7 @@ where
merkle_updater.clone(), merkle_updater.clone(),
rpc_server, rpc_server,
); );
TableGC::launch(data.clone(), system.clone(), rpc_server); TableGC::launch(system.clone(), data.clone(), rpc_server);
let table = Arc::new(Self { let table = Arc::new(Self {
system, system,