aboutsummaryrefslogtreecommitdiff
path: root/src/block.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/block.rs')
-rw-r--r--src/block.rs100
1 files changed, 63 insertions, 37 deletions
diff --git a/src/block.rs b/src/block.rs
index e898ad19..25a10910 100644
--- a/src/block.rs
+++ b/src/block.rs
@@ -6,6 +6,7 @@ use futures::stream::*;
use tokio::fs;
use tokio::prelude::*;
use tokio::sync::{watch, Mutex};
+use arc_swap::ArcSwapOption;
use crate::data;
use crate::data::*;
@@ -13,6 +14,7 @@ use crate::error::Error;
use crate::membership::System;
use crate::proto::*;
use crate::rpc_client::*;
+use crate::server::Garage;
pub struct BlockManager {
pub data_dir: PathBuf,
@@ -20,10 +22,11 @@ pub struct BlockManager {
pub resync_queue: sled::Tree,
pub lock: Mutex<()>,
pub system: Arc<System>,
+ pub garage: ArcSwapOption<Garage>,
}
impl BlockManager {
- pub async fn new(db: &sled::Db, data_dir: PathBuf, system: Arc<System>) -> Arc<Self> {
+ pub fn new(db: &sled::Db, data_dir: PathBuf, system: Arc<System>) -> Arc<Self> {
let rc = db
.open_tree("block_local_rc")
.expect("Unable to open block_local_rc tree");
@@ -33,20 +36,23 @@ impl BlockManager {
.open_tree("block_local_resync_queue")
.expect("Unable to open block_local_resync_queue tree");
- let block_manager = Arc::new(Self {
+ Arc::new(Self {
rc,
resync_queue,
data_dir,
lock: Mutex::new(()),
system,
- });
- let bm2 = block_manager.clone();
- block_manager
+ garage: ArcSwapOption::from(None),
+ })
+ }
+
+ 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;
- block_manager
}
pub async fn write_block(&self, hash: &Hash, data: &[u8]) -> Result<Message, Error> {
@@ -80,7 +86,7 @@ impl BlockManager {
let _lock = self.lock.lock().await;
eprintln!("Block {:?} is corrupted. Deleting and resyncing.", hash);
fs::remove_file(path).await?;
- self.resync_queue.insert(hash.to_vec(), vec![1u8])?;
+ self.put_to_resync(&hash, 0)?;
return Err(Error::CorruptData(hash.clone()));
}
@@ -98,38 +104,55 @@ impl BlockManager {
}
pub fn block_incref(&self, hash: &Hash) -> Result<(), Error> {
- self.rc.merge(&hash, vec![1])?;
+ let new_rc = self.rc.merge(&hash, vec![1])?;
+ if new_rc.map(|x| u64_from_bytes(&x[..]) == 0).unwrap_or(true) {
+ self.put_to_resync(&hash, BLOCK_RW_TIMEOUT.as_millis() as u64)?;
+ }
Ok(())
}
pub fn block_decref(&self, hash: &Hash) -> Result<(), Error> {
- if self.rc.merge(&hash, vec![0])?.is_none() {
- self.resync_queue.insert(hash.to_vec(), vec![1u8])?;
+ let new_rc = self.rc.merge(&hash, vec![0])?;
+ if new_rc.is_none() {
+ self.put_to_resync(&hash, 2 * BLOCK_RW_TIMEOUT.as_millis() as u64)?;
}
Ok(())
}
+ fn put_to_resync(&self, hash: &Hash, delay_millis: u64) -> Result<(), Error> {
+ let when = now_msec() + delay_millis;
+ eprintln!("Put resync_queue: {} {:?}", when, hash);
+ let mut key = u64::to_be_bytes(when).to_vec();
+ key.extend(hash.as_ref());
+ self.resync_queue.insert(key, hash.as_ref())?;
+ Ok(())
+ }
+
async fn resync_loop(self: Arc<Self>, must_exit: watch::Receiver<bool>) -> Result<(), Error> {
while !*must_exit.borrow() {
- if let Some((hash_bytes, _v)) = self.resync_queue.get_gt(&[])? {
- let mut hash = [0u8; 32];
- hash.copy_from_slice(hash_bytes.as_ref());
- let hash = Hash::from(hash);
-
- match self.resync_iter(&hash).await {
- Ok(_) => {
- self.resync_queue.remove(&hash_bytes)?;
- }
- Err(e) => {
- eprintln!(
- "Failed to resync hash {:?}, leaving it in queue: {}",
- hash, e
- );
+ if let Some((time_bytes, hash_bytes)) = self.resync_queue.get_gt(&[])? {
+ let time_msec = u64_from_bytes(&time_bytes[0..8]);
+ eprintln!("First in resync queue: {} (now = {})", time_msec, now_msec());
+ if now_msec() >= time_msec {
+ let mut hash = [0u8; 32];
+ hash.copy_from_slice(hash_bytes.as_ref());
+ let hash = Hash::from(hash);
+
+ match self.resync_iter(&hash).await {
+ Ok(_) => {
+ self.resync_queue.remove(&hash_bytes)?;
+ }
+ Err(e) => {
+ eprintln!(
+ "Failed to resync hash {:?}, leaving it in queue: {}",
+ hash, e
+ );
+ }
}
+ continue;
}
- } else {
- tokio::time::delay_for(Duration::from_secs(1)).await;
}
+ tokio::time::delay_for(Duration::from_secs(1)).await;
}
Ok(())
}
@@ -145,14 +168,23 @@ impl BlockManager {
.map(|x| u64_from_bytes(x.as_ref()) > 0)
.unwrap_or(false);
+ eprintln!("Resync block {:?}: exists {}, needed {}", hash, exists, needed);
+
if exists && !needed {
- // TODO: verify that other nodes that might need it have it ?
+ let garage = self.garage.load_full().unwrap();
+ let active_refs = garage.block_ref_table.get_range(&hash, &[0u8; 32].into(), Some(()), 1).await?;
+ let needed_by_others = !active_refs.is_empty();
+ if needed_by_others {
+ // TODO check they have it and send it if not
+ }
fs::remove_file(path).await?;
self.resync_queue.remove(&hash)?;
}
if needed && !exists {
// TODO find a way to not do this if they are sending it to us
+ // Let's suppose this isn't an issue for now with the BLOCK_RW_TIMEOUT delay
+ // between the RC being incremented and this part being called.
let block_data = rpc_get_block(&self.system, &hash).await?;
self.write_block(hash, &block_data[..]).await?;
}
@@ -190,11 +222,8 @@ fn rc_merge(_key: &[u8], old: Option<&[u8]>, new: &[u8]) -> Option<Vec<u8>> {
}
pub async fn rpc_get_block(system: &Arc<System>, hash: &Hash) -> Result<Vec<u8>, Error> {
- let who = system
- .ring
- .borrow()
- .clone()
- .walk_ring(&hash, system.config.data_replication_factor);
+ let ring = system.ring.borrow().clone();
+ let who = ring.walk_ring(&hash, system.config.data_replication_factor);
let msg = Message::GetBlock(hash.clone());
let mut resp_stream = who
.iter()
@@ -215,11 +244,8 @@ pub async fn rpc_get_block(system: &Arc<System>, hash: &Hash) -> Result<Vec<u8>,
}
pub async fn rpc_put_block(system: &Arc<System>, hash: Hash, data: Vec<u8>) -> Result<(), Error> {
- let who = system
- .ring
- .borrow()
- .clone()
- .walk_ring(&hash, system.config.data_replication_factor);
+ let ring = system.ring.borrow().clone();
+ let who = ring.walk_ring(&hash, system.config.data_replication_factor);
rpc_try_call_many(
system.clone(),
&who[..],