diff --git a/src/api/admin/repair.rs b/src/api/admin/repair.rs index 1d5665d1..de62c75a 100644 --- a/src/api/admin/repair.rs +++ b/src/api/admin/repair.rs @@ -360,14 +360,7 @@ impl Worker for BlockRcRepair { _must_exit: &mut watch::Receiver, ) -> Result { for _i in 0..RC_REPAIR_ITER_COUNT { - let next1 = self - .block_manager - .rc - .rc_table - .range(self.cursor.as_slice()..)? - .next() - .transpose()? - .map(|(k, _)| Hash::try_from(k.as_slice()).unwrap()); + let next1 = self.block_manager.rc.get_first_hash_from(self.cursor)?; let next2 = self .block_ref_table .data diff --git a/src/block/manager.rs b/src/block/manager.rs index ce14f625..c7a4c317 100644 --- a/src/block/manager.rs +++ b/src/block/manager.rs @@ -144,7 +144,7 @@ impl BlockManager { // Open metadata tables let rc = db - .open_tree("block_local_rc") + .open_typed_tree("block_local_rc") .expect("Unable to open block_local_rc tree"); let rc = BlockRc::new(rc); @@ -158,9 +158,9 @@ impl BlockManager { let metrics = BlockManagerMetrics::new( config.compression_level, - rc.rc_table.clone(), - resync.queue.clone(), - resync.errors.clone(), + rc.rc_table.untyped().clone(), + resync.queue.untyped().clone(), + resync.errors.untyped().clone(), buffer_kb_semaphore.clone(), ); @@ -449,9 +449,8 @@ impl BlockManager { let mut blocks = Vec::with_capacity(self.resync.errors.approximate_len()?); for ent in self.resync.errors.iter()? { let (hash, cnt) = ent?; - let cnt = ErrorCounter::decode(&cnt); blocks.push(BlockResyncErrorInfo { - hash: Hash::try_from(&hash).unwrap(), + hash, refcount: 0, error_count: cnt.errors, last_try: cnt.last_try, diff --git a/src/block/rc.rs b/src/block/rc.rs index 881df68b..3675ad8d 100644 --- a/src/block/rc.rs +++ b/src/block/rc.rs @@ -14,12 +14,12 @@ pub type CalculateRefcount = Box db::TxResult + Send + Sync>; pub struct BlockRc { - pub rc_table: db::Tree, + pub(crate) rc_table: db::TypedTree, pub(crate) recalc_rc: ArcSwapOption>, } impl BlockRc { - pub(crate) fn new(rc: db::Tree) -> Self { + pub(crate) fn new(rc: db::TypedTree) -> Self { Self { rc_table: rc, recalc_rc: ArcSwapOption::new(None), @@ -33,11 +33,8 @@ impl BlockRc { tx: &mut db::Transaction, hash: &Hash, ) -> db::TxOpResult { - let old_rc = RcEntry::parse_opt(tx.get(&self.rc_table, hash)?); - match old_rc.increment().serialize() { - Some(x) => tx.insert(&self.rc_table, hash, x)?, - None => unreachable!(), - } + let old_rc = self.rc_table.tx_get(tx, hash)?.unwrap_or(RcEntry::Absent); + self.rc_table.tx_insert(tx, hash, &old_rc.increment())?; Ok(old_rc.is_zero()) } @@ -48,17 +45,27 @@ impl BlockRc { tx: &mut db::Transaction, hash: &Hash, ) -> db::TxOpResult { - let new_rc = RcEntry::parse_opt(tx.get(&self.rc_table, hash)?).decrement(); - match new_rc.serialize() { - Some(x) => tx.insert(&self.rc_table, hash, x)?, - None => tx.remove(&self.rc_table, hash)?, + let new_rc = self.rc_table.tx_get(tx, hash)?.unwrap_or(RcEntry::Absent).decrement(); + match new_rc { + RcEntry::Absent => self.rc_table.tx_remove(tx, hash)?, + _ => self.rc_table.tx_insert(tx, hash, &new_rc)?, } Ok(matches!(new_rc, RcEntry::Deletable { .. })) } /// Read a block's reference count pub(crate) fn get_block_rc(&self, hash: &Hash) -> Result { - Ok(RcEntry::parse_opt(self.rc_table.get(hash.as_ref())?)) + Ok(self.rc_table.get(hash)?.unwrap_or(RcEntry::Absent)) + } + + /// Return the first hash stored in the RC table at or after `cursor` + pub fn get_first_hash_from(&self, cursor: Hash) -> Result, Error> { + Ok(self + .rc_table + .range(cursor..)? + .next() + .transpose()? + .map(|(k, _)| k)) } /// Delete an entry in the RC table if it is deletable and the @@ -66,12 +73,11 @@ impl BlockRc { pub(crate) fn clear_deleted_block_rc(&self, hash: &Hash) -> Result<(), Error> { let now = now_msec(); self.rc_table.db().transaction(|tx| { - let rcval = RcEntry::parse_opt(tx.get(&self.rc_table, hash)?); - match rcval { - RcEntry::Deletable { at_time } if now > at_time => { - tx.remove(&self.rc_table, hash)?; + let rcval = self.rc_table.tx_get(tx, hash)?.unwrap_or(RcEntry::Absent); + if let RcEntry::Deletable { at_time } = rcval { + if now > at_time { + self.rc_table.tx_remove(tx, hash)?; } - _ => (), } Ok(()) })?; @@ -91,7 +97,7 @@ impl BlockRc { for f in recalc_fns.iter() { cnt += f(tx, hash)?; } - let old_rc = RcEntry::parse_opt(tx.get(&self.rc_table, hash)?); + let old_rc = self.rc_table.tx_get(tx, hash)?.unwrap_or(RcEntry::Absent); trace!( "Block RC for {:?}: stored={}, calculated={}", hash, @@ -112,7 +118,7 @@ impl BlockRc { at_time: now_msec() + BLOCK_GC_DELAY.as_millis() as u64, } }; - tx.insert(&self.rc_table, hash, new_rc.serialize().unwrap())?; + self.rc_table.tx_insert(tx, hash, &new_rc)?; Ok((cnt, true)) } else { Ok((cnt, false)) @@ -131,6 +137,38 @@ impl BlockRc { } } +impl db::DbBytes for RcEntry { + fn encode(&self) -> Vec { + match self { + RcEntry::Present { count } => u64::to_be_bytes(*count).to_vec(), + RcEntry::Deletable { at_time } => { + [u64::to_be_bytes(0), u64::to_be_bytes(*at_time)].concat() + } + RcEntry::Absent => panic!("cannot encode RcEntry::Absent"), + } + } + + fn decode(bytes: &[u8]) -> db::Result { + if bytes.len() == 8 { + Ok(RcEntry::Present { + count: u64::from_be_bytes(bytes.try_into().unwrap()), + }) + } else if bytes.len() == 16 { + Ok(RcEntry::Deletable { + at_time: u64::from_be_bytes(bytes[8..16].try_into().unwrap()), + }) + } else { + Err(db::Error::Decode( + format!( + "invalid RC entry: expected 8 or 16 bytes, got {}", + bytes.len() + ) + .into(), + )) + } + } +} + /// Describes the state of the reference counter for a block #[derive(Clone, Copy, Debug)] pub(crate) enum RcEntry { @@ -154,38 +192,6 @@ pub(crate) enum RcEntry { } impl RcEntry { - fn parse(bytes: &[u8]) -> Self { - if bytes.len() == 8 { - RcEntry::Present { - count: u64::from_be_bytes(bytes.try_into().unwrap()), - } - } else if bytes.len() == 16 { - RcEntry::Deletable { - at_time: u64::from_be_bytes(bytes[8..16].try_into().unwrap()), - } - } else { - panic!("Invalid RC entry: {:?}, database is corrupted. This is an error Garage is currently unable to recover from. Sorry, and also please report a bug.", - bytes - ) - } - } - - fn parse_opt>(bytes: Option) -> Self { - bytes - .map(|b| Self::parse(b.as_ref())) - .unwrap_or(Self::Absent) - } - - fn serialize(self) -> Option> { - match self { - RcEntry::Present { count } => Some(u64::to_be_bytes(count).to_vec()), - RcEntry::Deletable { at_time } => { - Some([u64::to_be_bytes(0), u64::to_be_bytes(at_time)].concat()) - } - RcEntry::Absent => None, - } - } - fn increment(self) -> Self { let old_count = match self { RcEntry::Present { count } => count, diff --git a/src/block/repair.rs b/src/block/repair.rs index 9499d4ab..6c443a1c 100644 --- a/src/block/repair.rs +++ b/src/block/repair.rs @@ -102,16 +102,15 @@ impl Worker for RepairWorker { let mut batch_of_hashes = vec![]; let start_bound = match self.next_start.as_ref() { None => Bound::Unbounded, - Some(x) => Bound::Excluded(x.as_slice()), + Some(x) => Bound::Excluded(*x), }; for entry in self .manager .rc .rc_table - .range::<&[u8], _>((start_bound, Bound::Unbounded))? + .range((start_bound, Bound::Unbounded))? { let (hash, _) = entry?; - let hash = Hash::try_from(&hash[..]).unwrap(); batch_of_hashes.push(hash); if batch_of_hashes.len() >= 1000 { break; diff --git a/src/block/resync.rs b/src/block/resync.rs index 003c1172..08a9ef59 100644 --- a/src/block/resync.rs +++ b/src/block/resync.rs @@ -44,15 +44,58 @@ pub(crate) const MAX_RESYNC_WORKERS: usize = 8; const INITIAL_RESYNC_TRANQUILITY: u32 = 2; pub struct BlockResyncManager { - pub(crate) queue: db::Tree, + pub(crate) queue: db::TypedTree, pub(crate) notify: Arc, - pub(crate) errors: db::Tree, + pub(crate) errors: db::TypedTree, busy_set: BusySet, persister: PersisterShared, } +/// Key of the resync queue tree: blocks are resynced in order of increasing +/// `when` (msec timestamp of the next try), with the block hash as tie-breaker. +/// +// CAREFUL: this type implements `DbOrdKey`, so its byte encoding must be +// order-preserving. +// The derived `Ord` compares fields in declaration order, which must match +// the order in which `encode()` writes them; and `when` must remain an +// *unsigned* integer, as the big-endian encoding is only order-preserving +// for unsigned types. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub(crate) struct ResyncQueueKey { + pub(crate) when: u64, + pub(crate) hash: Hash, +} + +impl db::DbBytes for ResyncQueueKey { + fn encode(&self) -> Vec { + let mut v = Vec::with_capacity(40); + v.extend_from_slice(&u64::to_be_bytes(self.when)); + v.extend_from_slice(self.hash.as_slice()); + v + } + + fn decode(bytes: &[u8]) -> std::result::Result { + if bytes.len() != 40 { + return Err(db::DecodeError( + format!( + "invalid resync queue key: expected 40 bytes, got {}", + bytes.len() + ) + .into(), + )); + } + Ok(ResyncQueueKey { + when: u64::from_be_bytes(bytes[..8].try_into().unwrap()), + hash: Hash::try_from(&bytes[8..]) + .ok_or_else(|| db::DecodeError("invalid resync queue key: bad hash".into()))?, + }) + } +} + +impl db::DbOrdKey for ResyncQueueKey {} + #[derive(Serialize, Deserialize, Clone, Copy)] struct ResyncPersistedConfig { n_workers: usize, @@ -74,22 +117,21 @@ enum ResyncIterResult { IdleFor(Duration), } -type BusySet = Arc>>>; +type BusySet = Arc>>; struct BusyBlock { - time_bytes: Vec, - hash_bytes: Vec, + key: ResyncQueueKey, busy_set: BusySet, } impl BlockResyncManager { pub(crate) fn new(db: &db::Db, system: &System) -> Self { let queue = db - .open_tree("block_local_resync_queue") + .open_typed_tree("block_local_resync_queue") .expect("Unable to open block_local_resync_queue tree"); let errors = db - .open_tree("block_local_resync_errors") + .open_typed_tree("block_local_resync_errors") .expect("Unable to open block_local_resync_errors tree"); let persister = PersisterShared::new(&system.metadata_dir, "resync_cfg"); @@ -116,11 +158,10 @@ impl BlockResyncManager { /// Clear the error counter for a block and put it in queue immediately pub fn clear_backoff(&self, hash: &Hash) -> Result<(), Error> { let now = now_msec(); - if let Some(ec) = self.errors.get(hash)? { - let mut ec = ErrorCounter::decode(&ec); + if let Some(mut ec) = self.errors.get(hash)? { if ec.errors > 0 { ec.last_try = now - ec.delay_msec(); - self.errors.insert(hash, ec.encode())?; + self.errors.insert(hash, &ec)?; self.put_to_resync_at(hash, now)?; return Ok(()); } @@ -251,23 +292,21 @@ impl BlockResyncManager { pub(crate) fn put_to_resync_at(&self, hash: &Hash, when: u64) -> db::Result<()> { trace!("Put resync_queue: {} {:?}", when, hash); - let mut key = u64::to_be_bytes(when).to_vec(); - key.extend(hash.as_ref()); - self.queue.insert(key, hash.as_ref())?; + let qkey = ResyncQueueKey { when, hash: *hash }; + self.queue.insert(&qkey, hash)?; self.notify.notify_waiters(); Ok(()) } async fn resync_iter(&self, manager: &BlockManager) -> Result { if let Some(block) = self.get_block_to_resync()? { - let time_msec = u64::from_be_bytes(block.time_bytes[0..8].try_into().unwrap()); + let time_msec = block.key.when; let now = now_msec(); if now >= time_msec { - let hash = Hash::try_from(&block.hash_bytes[..]).unwrap(); + let hash = block.key.hash; - if let Some(ec) = self.errors.get(hash.as_slice())? { - let ec = ErrorCounter::decode(&ec); + if let Some(ec) = self.errors.get(&hash)? { if now < ec.next_try() { // if next retry after an error is not yet, // don't do resync and return early, but still @@ -277,7 +316,7 @@ impl BlockResyncManager { // is not removing the one we added just above // (we want to do the remove after the insert to ensure // that the item is not lost if we crash in-between) - self.queue.remove(&block.time_bytes)?; + self.queue.remove(&block.key)?; return Ok(ResyncIterResult::BusyDidNothing); } } @@ -307,21 +346,21 @@ impl BlockResyncManager { manager.metrics.resync_error_counter.add(1); error!("Error when resyncing {:?}: {}", hash, e); - let err_counter = match self.errors.get(hash.as_slice())? { - Some(ec) => ErrorCounter::decode(&ec).add1(now + 1), + let err_counter = match self.errors.get(&hash)? { + Some(ec) => ec.add1(now + 1), None => ErrorCounter::new(now + 1), }; - self.errors.insert(hash.as_slice(), err_counter.encode())?; + self.errors.insert(&hash, &err_counter)?; self.put_to_resync_at(&hash, err_counter.next_try())?; // err_counter.next_try() >= now + 1 > now, // the entry we remove from the queue is not // the entry we inserted with put_to_resync_at - self.queue.remove(&block.time_bytes)?; + self.queue.remove(&block.key)?; } else { - self.errors.remove(hash.as_slice())?; - self.queue.remove(&block.time_bytes)?; + self.errors.remove(&hash)?; + self.queue.remove(&block.key)?; } Ok(ResyncIterResult::BusyDidSomething) @@ -344,12 +383,11 @@ impl BlockResyncManager { fn get_block_to_resync(&self) -> Result, db::Error> { let mut busy = self.busy_set.lock().unwrap(); for it in self.queue.iter()? { - let (time_bytes, hash_bytes) = it?; - if !busy.contains(&time_bytes) { - busy.insert(time_bytes.clone()); + let (key, _) = it?; + if !busy.contains(&key) { + busy.insert(key); return Ok(Some(BusyBlock { - time_bytes, - hash_bytes, + key, busy_set: self.busy_set.clone(), })); } @@ -506,7 +544,7 @@ impl BlockResyncManager { impl Drop for BusyBlock { fn drop(&mut self) { let mut busy = self.busy_set.lock().unwrap(); - busy.remove(&self.time_bytes); + busy.remove(&self.key); } } @@ -602,6 +640,7 @@ impl Worker for ResyncWorker { /// Counts the number of errors when resyncing a block, /// and the time of the last try. +/// /// Used to implement exponential backoff. #[derive(Clone, Copy, Debug)] pub(crate) struct ErrorCounter { @@ -609,6 +648,31 @@ pub(crate) struct ErrorCounter { pub(crate) last_try: u64, } +impl db::DbBytes for ErrorCounter { + fn encode(&self) -> Vec { + let mut v = Vec::with_capacity(16); + v.extend_from_slice(&u64::to_be_bytes(self.errors)); + v.extend_from_slice(&u64::to_be_bytes(self.last_try)); + v + } + + fn decode(bytes: &[u8]) -> std::result::Result { + if bytes.len() != 16 { + return Err(db::DecodeError( + format!( + "invalid error counter: expected 16 bytes, got {}", + bytes.len() + ) + .into(), + )); + } + Ok(Self { + errors: u64::from_be_bytes(bytes[..8].try_into().unwrap()), + last_try: u64::from_be_bytes(bytes[8..].try_into().unwrap()), + }) + } +} + impl ErrorCounter { fn new(now: u64) -> Self { Self { @@ -617,21 +681,6 @@ impl ErrorCounter { } } - pub(crate) fn decode(data: &[u8]) -> Self { - Self { - errors: u64::from_be_bytes(data[0..8].try_into().unwrap()), - last_try: u64::from_be_bytes(data[8..16].try_into().unwrap()), - } - } - - fn encode(&self) -> Vec { - [ - u64::to_be_bytes(self.errors), - u64::to_be_bytes(self.last_try), - ] - .concat() - } - fn add1(self, now: u64) -> Self { Self { errors: self.errors + 1, diff --git a/src/util/data.rs b/src/util/data.rs index 63c4379f..4b500b14 100644 --- a/src/util/data.rs +++ b/src/util/data.rs @@ -155,6 +155,21 @@ pub fn gen_uuid() -> Uuid { rand::rng().random::<[u8; 32]>().into() } +impl garage_db::DbBytes for FixedBytes32 { + fn encode(&self) -> Vec { + self.0.into() + } + fn decode(bytes: &[u8]) -> std::result::Result { + Self::try_from(bytes).ok_or_else(|| { + garage_db::DecodeError( + format!("invalid hash: expected 32 bytes, got {}", bytes.len()).into(), + ) + }) + } +} + +impl garage_db::DbOrdKey for FixedBytes32 {} + #[cfg(test)] mod test { use super::*; diff --git a/src/util/error.rs b/src/util/error.rs index c3d4ab58..7cbe0a4a 100644 --- a/src/util/error.rs +++ b/src/util/error.rs @@ -79,6 +79,18 @@ impl Error { } } +impl From for Error { + fn from(e: garage_db::DbError) -> Error { + Error::Db(e.into()) + } +} + +impl From for Error { + fn from(e: garage_db::DecodeError) -> Error { + Error::Db(e.into()) + } +} + impl From> for Error { fn from(e: garage_db::TxError) -> Error { match e {