From 40c48e6a590c887586664957da2dca0f368cdcb2 Mon Sep 17 00:00:00 2001
From: Alex Auvolat <alex@adnab.me>
Date: Fri, 17 Apr 2020 20:58:10 +0200
Subject: [PATCH] Several resync workers; add delay on retry resync

---
 src/block.rs | 31 +++++++++++++++----------------
 src/table.rs | 10 +++++-----
 2 files changed, 20 insertions(+), 21 deletions(-)

diff --git a/src/block.rs b/src/block.rs
index ff38d65c..200b4201 100644
--- a/src/block.rs
+++ b/src/block.rs
@@ -18,6 +18,7 @@ use crate::rpc_client::*;
 use crate::server::Garage;
 
 const NEED_BLOCK_QUERY_TIMEOUT: Duration = Duration::from_secs(5);
+const RESYNC_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
 
 pub struct BlockManager {
 	pub data_dir: PathBuf,
@@ -50,11 +51,14 @@ impl BlockManager {
 	}
 
 	pub async fn spawn_background_worker(self: Arc<Self>) {
-		let bm2 = self.clone();
-		self.system
-			.background
-			.spawn_worker(move |must_exit| bm2.resync_loop(must_exit))
-			.await;
+		// Launch 2 simultaneous workers for background resync loop preprocessing
+		for _i in 0..2usize {
+			let bm2 = self.clone();
+			self.system
+				.background
+				.spawn_worker(move |must_exit| bm2.resync_loop(must_exit))
+				.await;
+		}
 	}
 
 	pub async fn write_block(&self, hash: &Hash, data: &[u8]) -> Result<Message, Error> {
@@ -158,7 +162,7 @@ impl BlockManager {
 
 	async fn resync_loop(self: Arc<Self>, must_exit: watch::Receiver<bool>) -> Result<(), Error> {
 		while !*must_exit.borrow() {
-			if let Some((time_bytes, hash_bytes)) = self.resync_queue.get_gt(&[])? {
+			if let Some((time_bytes, hash_bytes)) = self.resync_queue.pop_min()? {
 				let time_msec = u64_from_bytes(&time_bytes[0..8]);
 				eprintln!(
 					"First in resync queue: {} (now = {})",
@@ -170,18 +174,13 @@ impl BlockManager {
 					hash.copy_from_slice(hash_bytes.as_ref());
 					let hash = Hash::from(hash);
 
-					match self.resync_iter(&hash).await {
-						Ok(_) => {
-							self.resync_queue.remove(&time_bytes)?;
-						}
-						Err(e) => {
-							eprintln!(
-								"Failed to resync hash {:?}, leaving it in queue: {}",
-								hash, e
-							);
-						}
+					if let Err(e) = self.resync_iter(&hash).await {
+						eprintln!("Failed to resync block {:?}, retrying later: {}", hash, e);
+						self.put_to_resync(&hash, RESYNC_RETRY_TIMEOUT.as_millis() as u64)?;
 					}
 					continue;
+				} else {
+					self.resync_queue.insert(time_bytes, hash_bytes)?;
 				}
 			}
 			tokio::time::delay_for(Duration::from_secs(1)).await;
diff --git a/src/table.rs b/src/table.rs
index 9b67ad94..d2c7de94 100644
--- a/src/table.rs
+++ b/src/table.rs
@@ -326,7 +326,7 @@ impl<F: TableSchema + 'static> Table<F> {
 		rpc: &TableRPC<F>,
 		quorum: usize,
 	) -> Result<Vec<TableRPC<F>>, Error> {
-		eprintln!("Table RPC to {:?}: {}", who, serde_json::to_string(&rpc)?);
+		//eprintln!("Table RPC to {:?}: {}", who, serde_json::to_string(&rpc)?);
 
 		let rpc_bytes = rmp_to_vec_all_named(rpc)?;
 		let rpc_msg = Message::TableRPC(self.name.to_string(), rpc_bytes);
@@ -353,10 +353,10 @@ impl<F: TableSchema + 'static> Table<F> {
 				resp
 			)));
 		}
-		eprintln!(
-			"Table RPC responses: {}",
-			serde_json::to_string(&resps_vals)?
-		);
+		//eprintln!(
+		//	"Table RPC responses: {}",
+		//	serde_json::to_string(&resps_vals)?
+		//);
 		Ok(resps_vals)
 	}