Arrange block manager

src/model/block.rs

@@ -20,7 +20,7 @@ use garage_rpc::rpc_client::*;
 use garage_rpc::rpc_server::*;
 
 use garage_table::table_sharded::TableShardedReplication;
-use garage_table::{DeletedFilter, TableReplication};
+use garage_table::TableReplication;
 
 use crate::block_ref_table::*;
 
@@ -303,56 +303,49 @@ impl BlockManager {
 		}
 
 		if exists && !needed {
-			let garage = self.garage.load_full().unwrap();
+			trace!("Offloading block {:?}", hash);
-			let active_refs = garage
-				.block_ref_table
-				.get_range(&hash, None, Some(DeletedFilter::NotDeleted), 1)
-				.await?;
-			let needed_by_others = !active_refs.is_empty();
-			if needed_by_others {
-				let ring = self.system.ring.borrow().clone();
-				let who = self.replication.replication_nodes(&hash, &ring);
-				let msg = Arc::new(Message::NeedBlockQuery(*hash));
-				let who_needs_fut = who.iter().map(|to| {
-					self.rpc_client
-						.call_arc(*to, msg.clone(), NEED_BLOCK_QUERY_TIMEOUT)
-				});
-				let who_needs = join_all(who_needs_fut).await;
 
-				let mut need_nodes = vec![];
+			let ring = self.system.ring.borrow().clone();
-				for (node, needed) in who.into_iter().zip(who_needs.iter()) {
+
-					match needed {
+			let mut who = self.replication.replication_nodes(&hash, &ring);
-						Ok(Message::NeedBlockReply(needed)) => {
+			who.retain(|id| *id != self.system.id);
-							if *needed {
+
-								need_nodes.push(node);
+			let msg = Arc::new(Message::NeedBlockQuery(*hash));
-							}
+			let who_needs_fut = who.iter().map(|to| {
-						}
+				self.rpc_client
-						Err(e) => {
+					.call_arc(*to, msg.clone(), NEED_BLOCK_QUERY_TIMEOUT)
-							return Err(Error::Message(format!(
+			});
-								"Should delete block, but unable to confirm that all other nodes that need it have it: {}",
+			let who_needs_resps = join_all(who_needs_fut).await;
-								e
+
-							)));
+			let mut need_nodes = vec![];
-						}
+			for (node, needed) in who.iter().zip(who_needs_resps.into_iter()) {
-						Ok(_) => {
+				match needed? {
-							return Err(Error::Message(format!(
+					Message::NeedBlockReply(needed) => {
-								"Unexpected response to NeedBlockQuery RPC"
+						if needed {
-							)));
+							need_nodes.push(*node);
 						}
 					}
-				}
+					_ => {
-
+						return Err(Error::Message(format!(
-				if need_nodes.len() > 0 {
+							"Unexpected response to NeedBlockQuery RPC"
-					let put_block_message = self.read_block(hash).await?;
+						)));
-					self.rpc_client
+					}
-						.try_call_many(
-							&need_nodes[..],
-							put_block_message,
-							RequestStrategy::with_quorum(need_nodes.len())
-								.with_timeout(BLOCK_RW_TIMEOUT),
-						)
-						.await?;
 				}
 			}
+
+			if need_nodes.len() > 0 {
+				trace!("Block {:?} neede by {} nodes, sending", hash, need_nodes.len());
+
+				let put_block_message = Arc::new(self.read_block(hash).await?);
+				let put_resps = join_all(need_nodes.iter().map(|to| {
+					self.rpc_client.call_arc(*to, put_block_message.clone(), BLOCK_RW_TIMEOUT)
+				})).await;
+				for resp in put_resps {
+					resp?;
+				}
+			}
+			trace!("Deleting block {:?}, offload finished ({} / {})", hash, need_nodes.len(), who.len());
+
 			fs::remove_file(path).await?;
 			self.resync_queue.remove(&hash)?;
 		}
@@ -427,53 +420,42 @@ impl BlockManager {
 		}
 
 		// 2. Repair blocks actually on disk
-		let mut ls_data_dir = fs::read_dir(&self.data_dir).await?;
+		self.repair_aux_read_dir_rec(&self.data_dir, must_exit).await?;
-		while let Some(data_dir_ent) = ls_data_dir.next().await {
-			let data_dir_ent = data_dir_ent?;
-			let dir_name = data_dir_ent.file_name();
-			let dir_name = match dir_name.into_string() {
-				Ok(x) => x,
-				Err(_) => continue,
-			};
-			if dir_name.len() != 2 || hex::decode(&dir_name).is_err() {
-				continue;
-			}
 
-			let mut ls_data_dir_2 = match fs::read_dir(data_dir_ent.path()).await {
+		Ok(())
-				Err(e) => {
+	}
-					warn!(
+
-						"Warning: could not list dir {:?}: {}",
+	fn repair_aux_read_dir_rec<'a>(&'a self, path: &'a PathBuf, must_exit: &'a watch::Receiver<bool>) -> BoxFuture<'a, Result<(), Error>> {
-						data_dir_ent.path().to_str(),
+		async move {
-						e
+			let mut ls_data_dir = fs::read_dir(path).await?;
-					);
+			while let Some(data_dir_ent) = ls_data_dir.next().await {
-					continue;
+				let data_dir_ent = data_dir_ent?;
-				}
+				let name = data_dir_ent.file_name();
-				Ok(x) => x,
+				let name = match name.into_string() {
-			};
-			while let Some(file) = ls_data_dir_2.next().await {
-				let file = file?;
-				let file_name = file.file_name();
-				let file_name = match file_name.into_string() {
 					Ok(x) => x,
 					Err(_) => continue,
 				};
-				if file_name.len() != 64 {
+				let ent_type = data_dir_ent.file_type().await?;
-					continue;
+				println!("name: {}, path: {:?}", name, data_dir_ent.path().to_str());
+
+				if name.len() == 2 && hex::decode(&name).is_ok() && ent_type.is_dir() {
+					self.repair_aux_read_dir_rec(&data_dir_ent.path(), must_exit).await?;
+				} else if name.len() == 64 {
+					let hash_bytes = match hex::decode(&name) {
+						Ok(h) => h,
+						Err(_) => continue,
+					};
+					let mut hash = [0u8; 32];
+					hash.copy_from_slice(&hash_bytes[..]);
+					self.put_to_resync(&hash.into(), 0)?;
 				}
-				let hash_bytes = match hex::decode(&file_name) {
-					Ok(h) => h,
-					Err(_) => continue,
-				};
-				let mut hash = [0u8; 32];
-				hash.copy_from_slice(&hash_bytes[..]);
-				self.put_to_resync(&hash.into(), 0)?;
 
 				if *must_exit.borrow() {
-					return Ok(());
+					break;
 				}
 			}
-		}
+			Ok(())
-		Ok(())
+		}.boxed()
 	}
 }