aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/block.rs31
-rw-r--r--src/table.rs10
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)
}