Small fixes

This commit is contained in:
Alex 2021-03-15 19:51:16 +01:00
parent 642bed601f
commit 667e4e72a8
5 changed files with 37 additions and 16 deletions

View file

@ -48,7 +48,7 @@ async fn handle_delete_internal(
key.into(), key.into(),
vec![ObjectVersion { vec![ObjectVersion {
uuid: version_uuid, uuid: version_uuid,
timestamp: now_msec(), timestamp,
state: ObjectVersionState::Complete(ObjectVersionData::DeleteMarker), state: ObjectVersionState::Complete(ObjectVersionData::DeleteMarker),
}], }],
); );

View file

@ -432,13 +432,13 @@ impl AdminRpcHandler {
writeln!( writeln!(
&mut ret, &mut ret,
" number of blocks: {}", " number of blocks: {}",
self.garage.block_manager.rc.len() self.garage.block_manager.rc_len()
) )
.unwrap(); .unwrap();
writeln!( writeln!(
&mut ret, &mut ret,
" resync queue length: {}", " resync queue length: {}",
self.garage.block_manager.resync_queue.len() self.garage.block_manager.resync_queue_len()
) )
.unwrap(); .unwrap();
@ -460,19 +460,19 @@ impl AdminRpcHandler {
writeln!( writeln!(
to, to,
" Merkle updater todo queue length: {}", " Merkle updater todo queue length: {}",
t.data.merkle_updater.todo.len() t.data.merkle_updater.todo_len()
) )
.unwrap(); .unwrap();
writeln!( writeln!(
to, to,
" Merkle tree size: {}", " Merkle tree size: {}",
t.data.merkle_updater.merkle_tree.len() t.data.merkle_updater.merkle_tree_len()
) )
.unwrap(); .unwrap();
writeln!( writeln!(
to, to,
" GC todo queue length: {}", " GC todo queue length: {}",
t.data.gc_todo.len() t.data.gc_todo_len()
) )
.unwrap(); .unwrap();
Ok(()) Ok(())

View file

@ -27,6 +27,8 @@ use crate::garage::Garage;
pub const INLINE_THRESHOLD: usize = 3072; pub const INLINE_THRESHOLD: usize = 3072;
pub const BACKGROUND_WORKERS: u64 = 1;
const BLOCK_RW_TIMEOUT: Duration = Duration::from_secs(42); const BLOCK_RW_TIMEOUT: Duration = Duration::from_secs(42);
const BLOCK_GC_TIMEOUT: Duration = Duration::from_secs(60); const BLOCK_GC_TIMEOUT: Duration = Duration::from_secs(60);
const NEED_BLOCK_QUERY_TIMEOUT: Duration = Duration::from_secs(5); const NEED_BLOCK_QUERY_TIMEOUT: Duration = Duration::from_secs(5);
@ -56,14 +58,14 @@ pub struct BlockManager {
pub data_dir: PathBuf, pub data_dir: PathBuf,
pub data_dir_lock: Mutex<()>, pub data_dir_lock: Mutex<()>,
pub rc: sled::Tree, rc: sled::Tree,
pub resync_queue: sled::Tree, resync_queue: sled::Tree,
pub resync_notify: Notify, resync_notify: Notify,
pub system: Arc<System>, system: Arc<System>,
rpc_client: Arc<RpcClient<Message>>, rpc_client: Arc<RpcClient<Message>>,
pub garage: ArcSwapOption<Garage>, pub(crate) garage: ArcSwapOption<Garage>,
} }
impl BlockManager { impl BlockManager {
@ -128,7 +130,7 @@ impl BlockManager {
pub fn spawn_background_worker(self: Arc<Self>) { pub fn spawn_background_worker(self: Arc<Self>) {
// Launch 2 simultaneous workers for background resync loop preprocessing // Launch 2 simultaneous workers for background resync loop preprocessing
for i in 0..2u64 { for i in 0..BACKGROUND_WORKERS {
let bm2 = self.clone(); let bm2 = self.clone();
let background = self.system.background.clone(); let background = self.system.background.clone();
tokio::spawn(async move { tokio::spawn(async move {
@ -373,7 +375,6 @@ impl BlockManager {
); );
fs::remove_file(path).await?; fs::remove_file(path).await?;
self.resync_queue.remove(&hash)?;
} }
if needed && !exists { if needed && !exists {
@ -494,6 +495,14 @@ impl BlockManager {
} }
.boxed() .boxed()
} }
pub fn resync_queue_len(&self) -> usize {
self.resync_queue.len()
}
pub fn rc_len(&self) -> usize {
self.rc.len()
}
} }
fn u64_from_be_bytes<T: AsRef<[u8]>>(bytes: T) -> u64 { fn u64_from_be_bytes<T: AsRef<[u8]>>(bytes: T) -> u64 {

View file

@ -18,7 +18,7 @@ pub struct TableData<F: TableSchema> {
pub instance: F, pub instance: F,
pub store: sled::Tree, pub store: sled::Tree,
pub gc_todo: sled::Tree, pub(crate) gc_todo: sled::Tree,
pub merkle_updater: Arc<MerkleUpdater>, pub merkle_updater: Arc<MerkleUpdater>,
} }
@ -239,4 +239,8 @@ where
}, },
} }
} }
pub fn gc_todo_len(&self) -> usize {
self.gc_todo.len()
}
} }

View file

@ -40,13 +40,13 @@ pub struct MerkleUpdater {
// - 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
// - value = the hash of the full serialized item, if present, // - value = the hash of the full serialized item, if present,
// or an empty vec if item is absent (deleted) // or an empty vec if item is absent (deleted)
pub todo: sled::Tree, pub(crate) todo: sled::Tree,
pub(crate) todo_notify: Notify, pub(crate) todo_notify: Notify,
// Content of the merkle tree: items where // Content of the merkle tree: items where
// - key = .bytes() for MerkleNodeKey // - key = .bytes() for MerkleNodeKey
// - value = serialization of a MerkleNode, assumed to be MerkleNode::empty if not found // - value = serialization of a MerkleNode, assumed to be MerkleNode::empty if not found
pub merkle_tree: sled::Tree, pub(crate) merkle_tree: sled::Tree,
empty_node_hash: Hash, empty_node_hash: Hash,
} }
@ -311,6 +311,14 @@ impl MerkleUpdater {
Some(v) => Ok(rmp_serde::decode::from_read_ref::<_, MerkleNode>(&v[..])?), Some(v) => Ok(rmp_serde::decode::from_read_ref::<_, MerkleNode>(&v[..])?),
} }
} }
pub fn merkle_tree_len(&self) -> usize {
self.merkle_tree.len()
}
pub fn todo_len(&self) -> usize {
self.todo.len()
}
} }
impl MerkleNodeKey { impl MerkleNodeKey {