diff --git a/chain/src/chain.rs b/chain/src/chain.rs index 13325fad..11057398 100644 --- a/chain/src/chain.rs +++ b/chain/src/chain.rs @@ -1293,19 +1293,24 @@ impl Chain { return Ok(()); } - let mut count = 0; let tail_hash = header_pmmr.get_header_hash_by_height(head.height - horizon)?; let tail = batch.get_block_header(&tail_hash)?; - // Remove old blocks (including short lived fork blocks) which height < tail.height - for block in batch.blocks_iter()? { + // Remove old blocks (including short-lived fork blocks) which height < tail.height + let mut blocks_to_delete = vec![]; + let iter = batch.blocks_iter()?; + for block in iter { if let Ok(block) = block { if block.header.height < tail.height { - let _ = batch.delete_block(&block.hash()); - count += 1; + blocks_to_delete.push(block.hash()); } } } + let mut count = 0; + for bh in blocks_to_delete { + let _ = batch.delete_block(&bh); + count += 1; + } batch.save_body_tail(&Tip::from_header(&tail))?; diff --git a/chain/src/linked_list.rs b/chain/src/linked_list.rs index 11b0a114..40b9b36b 100644 --- a/chain/src/linked_list.rs +++ b/chain/src/linked_list.rs @@ -405,22 +405,36 @@ impl RewindableListIndex for MultiIndex { impl PruneableListIndex for MultiIndex { fn clear(&self, batch: &mut Batch<'_>) -> Result<(), Error> { - let mut list_count = 0; - let mut entry_count = 0; + let mut lists_to_delete = vec![]; let list_db_key = Some(self.list_prefix); for key in batch.db.iter(list_db_key, |k, _| Ok(k.to_vec()))? { if let Ok(key) = key { - let _ = batch.delete(list_db_key, &key); - list_count += 1; + lists_to_delete.push(key); } } + let mut list_count = 0; + for l in lists_to_delete { + match batch.delete(list_db_key, &l) { + Ok(_) => list_count += 1, + Err(_) => {} + } + } + + let mut entries_to_delete = vec![]; let entry_db_key = Some(self.entry_prefix); for key in batch.db.iter(entry_db_key, |k, _| Ok(k.to_vec()))? { if let Ok(key) = key { - let _ = batch.delete(entry_db_key, &key); - entry_count += 1; + entries_to_delete.push(key); } } + let mut entry_count = 0; + for e in entries_to_delete { + match batch.delete(entry_db_key, &e) { + Ok(_) => entry_count += 1, + Err(_) => {} + } + } + debug!( "clear: lists deleted: {}, entries deleted: {}", list_count, entry_count diff --git a/chain/src/store.rs b/chain/src/store.rs index e5ed2c60..d8df65f2 100644 --- a/chain/src/store.rs +++ b/chain/src/store.rs @@ -340,8 +340,8 @@ impl<'a> Batch<'a> { /// Iterator over the output_pos index. pub fn output_pos_iter( - &self, - ) -> Result, CommitPos), Error>>, Error> { + &'a self, + ) -> Result, CommitPos), Error>> + 'a, Error> { let protocol_version = self.db.protocol_version(); self.db.iter(Some(OUTPUT_POS_PREFIX), move |k, mut v| { ser::deserialize(&mut v, protocol_version, DeserializationMode::default()) @@ -461,7 +461,7 @@ impl<'a> Batch<'a> { /// Iterator over all full blocks in the db. /// Uses default db serialization strategy via db protocol version. - pub fn blocks_iter(&self) -> Result> + 'a, Error> { + pub fn blocks_iter(&'a self) -> Result> + 'a, Error> { let protocol_version = self.db.protocol_version(); self.db.iter(Some(BLOCK_PREFIX), move |_, mut v| { ser::deserialize(&mut v, protocol_version, DeserializationMode::default()) @@ -472,8 +472,8 @@ impl<'a> Batch<'a> { /// Iterator over raw data for full blocks in the db. /// Used during block migration (we need flexibility around deserialization). pub fn blocks_raw_iter( - &self, - ) -> Result, Vec), Error>>, Error> { + &'a self, + ) -> Result, Vec), Error>> + 'a, Error> { self.db .iter(Some(BLOCK_PREFIX), |k, v| Ok((k.to_vec(), v.to_vec()))) } diff --git a/chain/src/txhashset/txhashset.rs b/chain/src/txhashset/txhashset.rs index 8155a547..5d88cd29 100644 --- a/chain/src/txhashset/txhashset.rs +++ b/chain/src/txhashset/txhashset.rs @@ -643,7 +643,7 @@ impl TxHashSet { // Iterate over the current output_pos index, removing any entries that // do not point to to the expected output. - let mut removed_count = 0; + let mut pos_to_delete = vec![]; for kp in batch.output_pos_iter()? { if let Ok((key, pos1)) = kp { let pos0 = pos1.pos - 1; @@ -658,10 +658,14 @@ impl TxHashSet { } } } - batch.delete(Some(store::OUTPUT_POS_PREFIX), &key)?; - removed_count += 1; + pos_to_delete.push(key); } } + let mut removed_count = 0; + for p in pos_to_delete { + batch.delete(Some(store::OUTPUT_POS_PREFIX), &p)?; + removed_count += 1; + } debug!( "init_output_pos_index: removed {} stale index entries", removed_count diff --git a/p2p/src/store.rs b/p2p/src/store.rs index fe404e05..39018ad5 100644 --- a/p2p/src/store.rs +++ b/p2p/src/store.rs @@ -195,7 +195,9 @@ pub struct PeersIterBatch<'a> { impl<'a> PeersIterBatch<'a> { /// Iterator over all known peers. - pub fn peers_iter(&self) -> Result>, Error> { + pub fn peers_iter( + &'a self, + ) -> Result> + 'a, Error> { let protocol_version = self.db.protocol_version(); self.db.iter(Some(PEER_PREFIX), move |_, mut v| { ser::deserialize(&mut v, protocol_version, DeserializationMode::default()) diff --git a/store/src/lmdb.rs b/store/src/lmdb.rs index ee4c977f..9e8fd4cc 100644 --- a/store/src/lmdb.rs +++ b/store/src/lmdb.rs @@ -252,9 +252,15 @@ impl Store { let migrate_from = Path::new(root_path).join(env_name); if migrate_from.exists() { match s.migrate_to_default_env(db_name, &migrate_from) { - Ok(_) => { - let _ = fs::remove_dir_all(&migrate_from); - } + Ok(_) => match fs::remove_dir_all(&migrate_from) { + Ok(_) => {} + Err(e) => { + return Err(Error::FileErr(format!( + "Can not remove old DB file: {:?}", + e + ))); + } + }, Err(e) => { error!("DB {} migration error: {:?}", env_name, e); match s.clear() { @@ -310,7 +316,7 @@ impl Store { db.put(&mut write_to, key, &v)?; count += 1; } else { - error!("Migration: unknown DB key: {}", db_name[0]); + warn!("Migration: unknown DB key: {}", db_name[0]); } } else { self.def_db.put(&mut write_to, k, &v)?; @@ -560,11 +566,11 @@ impl Store { } /// Produces an iterator from the provided database key. - pub fn iter( + pub fn iter<'a, F, T>( &self, db_key: Option, deserialize: F, - ) -> Result, Error> + ) -> Result, Error> where F: Fn(&[u8], &[u8]) -> Result, { @@ -662,7 +668,8 @@ impl<'a> Batch<'a> { /// Writes a single key/value pair to the provided database key. pub fn put(&mut self, db_key: Option, key: &[u8], value: &[u8]) -> Result<(), Error> { let db = self.store.get_db(db_key)?; - db.put(&mut self.write, key, value)?; + let w = &mut self.write; + db.put(w, key, value)?; Ok(()) } @@ -724,14 +731,38 @@ impl<'a> Batch<'a> { /// Produces an iterator from the provided database key. pub fn iter( - &self, + &'a self, db_key: Option, deserialize: F, - ) -> Result, Error> + ) -> Result, Error> where F: Fn(&[u8], &[u8]) -> Result, { - self.store.iter(db_key, deserialize) + self.store.wait_for_resize(); + + TxCounter::on_change_tx_count(&self.store.env_path, true); + let read = self.write.nested_read_txn(); + match read { + Ok(read) => { + let db_res = self.store.get_db(db_key); + match db_res { + Ok(db) => Ok(DatabaseIterator::new( + self.store, + Arc::new(db.clone()), + read, + deserialize, + )), + Err(e) => { + TxCounter::on_change_tx_count(&self.store.env_path, false); + Err(Error::from(e)) + } + } + } + Err(e) => { + TxCounter::on_change_tx_count(&self.store.env_path, false); + Err(Error::from(e)) + } + } } /// Gets a `Readable` value from the database by provided key and deserialization strategy. @@ -788,12 +819,12 @@ impl<'a> Batch<'a> { /// An iterator based on database key. /// Caller is responsible for deserialization of the data. -pub struct DatabaseIterator +pub struct DatabaseIterator<'a, F, T> where F: Fn(&[u8], &[u8]) -> Result, { db: Arc>, - read: Arc>, + read: Arc>, keys: Vec>, total_keys: usize, skip_cur: usize, @@ -803,7 +834,7 @@ where tx_counter: TxCounter, } -impl Iterator for DatabaseIterator +impl Iterator for DatabaseIterator<'_, F, T> where F: Fn(&[u8], &[u8]) -> Result, { @@ -811,15 +842,13 @@ where fn next(&mut self) -> Option { if let Some(k) = self.keys.iter().skip(self.skip_cur).next() { + self.skip_total += 1; + self.skip_cur += 1; match self.db.get(&self.read, k) { Ok(v) => { if let Some(v) = v { return match (self.deserialize)(k, v) { - Ok(v) => { - self.skip_total += 1; - self.skip_cur += 1; - Some(Ok(v)) - } + Ok(v) => Some(Ok(v)), Err(e) => { error!("db iter: error deserializing: {}", e); Some(Err(Error::from(e))) @@ -853,7 +882,7 @@ where } } -impl DatabaseIterator +impl<'a, F, T> DatabaseIterator<'a, F, T> where F: Fn(&[u8], &[u8]) -> Result, { @@ -861,9 +890,9 @@ where pub fn new( store: &Store, db: Arc>, - read: RoTxn<'static, WithoutTls>, + read: RoTxn<'a, WithoutTls>, deserialize: F, - ) -> DatabaseIterator { + ) -> DatabaseIterator<'a, F, T> { let (keys, total_keys) = if let Ok(iter) = db.iter(&read) { let total = iter.move_between_keys().count(); let keys = if let Ok(iter) = db.iter(&read) { diff --git a/store/tests/lmdb.rs b/store/tests/lmdb.rs index 797614db..7aebb528 100644 --- a/store/tests/lmdb.rs +++ b/store/tests/lmdb.rs @@ -118,13 +118,12 @@ fn test_iter() -> Result<(), store::Error> { let mut batch = store.batch()?; batch.put(Some(prefix), &key, &value)?; - // TODO - This is not currently possible (and we need to be aware of this). - // Currently our SerIterator is limited to using a ReadTransaction only. - // // Check we can see the new entry via an iterator using the uncommitted batch. - // let mut iter: SerIterator> = batch.iter(&[0])?; - // assert_eq!(iter.next(), Some((key.to_vec(), value.to_vec()))); - // assert_eq!(iter.next(), None); + { + let mut iter = batch.iter(Some(prefix), |_, v| Ok(v.to_vec()))?; + assert_eq!(iter.next(), Some(Ok(value.to_vec()))); + assert_eq!(iter.next(), None); + } // Check we can not yet see the new entry via an iterator outside the uncommitted batch. let mut iter = store.iter(Some(prefix), |_, v| Ok(v.to_vec()))?;