[block_counter] Add the stop-on-error option to prevent silent errors

The stop-on-error option aims to solve the lack of damage_visitor support
in space_map::count_metadata and btree counting_visitor. To prevent
possibly silent errors during space map or btree counting, the option
stops the counting process immediately when a btree or bitmap error
was found.

The bitmap blocks are also validated to avoid potential false-alarm,
if broken bitmap is not counted, and the corresponding ref-count is
also zeroed.

persistent-data/block_counter.h

@@ -33,6 +33,12 @@ namespace persistent_data {
 	public:
 		typedef std::map<block_address, unsigned> count_map;
 
+		block_counter(): stop_on_error_(false) {}
+
+		block_counter(bool stop_on_error)
+			: stop_on_error_(stop_on_error) {
+		}
+
 		virtual ~block_counter() {}
 
 		virtual void inc(block_address b) {
@@ -52,8 +58,13 @@ namespace persistent_data {
 			return counts_;
 		}
 
+		virtual bool stop_on_error() {
+			return stop_on_error_;
+		}
+
 	private:
 		count_map counts_;
+		bool stop_on_error_;
 	};
 
 	//----------------------------------------------------------------

persistent-data/data-structures/btree_counter.h

@@ -16,7 +16,10 @@ namespace persistent_data {
 
 			counting_visitor(block_counter &bc, ValueCounter &vc)
 				: bc_(bc),
-				  vc_(vc) {
+				  vc_(vc),
+				  error_outcome_(bc.stop_on_error() ?
+						 tree::visitor::RETHROW_EXCEPTION :
+						 tree::visitor::EXCEPTION_HANDLED) {
 			}
 
 			virtual bool visit_internal(node_location const &l,
@@ -51,7 +54,7 @@ namespace persistent_data {
 
 			error_outcome error_accessing_node(node_location const &l, block_address b,
 							   std::string const &what) {
-				return btree<Levels, ValueTraits>::visitor::EXCEPTION_HANDLED;
+				return error_outcome_;
 			}
 
 		private:
@@ -110,6 +113,7 @@ namespace persistent_data {
 			ValueCounter &vc_;
 			btree_node_checker checker_;
 			boost::optional<uint64_t> last_leaf_key_[Levels];
+			error_outcome error_outcome_;
 		};
 	}
 

persistent-data/space-maps/disk.cc

@@ -639,21 +639,30 @@ namespace {
 		//--------------------------------
 
 		struct index_entry_counter {
-			index_entry_counter(block_counter &bc)
-			: bc_(bc) {
+			index_entry_counter(transaction_manager &tm,
+					    block_counter &bc)
+			: tm_(tm),
+			  bc_(bc),
+			  bitmap_validator_(new bitmap_block_validator()) {
 			}
 
 			void visit(btree_detail::node_location const &loc, index_entry const &ie) {
-				if (ie.blocknr_ != 0)
+				if (!ie.blocknr_)
+					return;
+
+				block_manager::read_ref rr = tm_.read_lock(ie.blocknr_, bitmap_validator_);
+				if (rr.data())
 					bc_.inc(ie.blocknr_);
 			}
 
 		private:
+			transaction_manager &tm_;
 			block_counter &bc_;
+			bcache::validator::ptr bitmap_validator_;
 		};
 
 		virtual void count_metadata(block_counter &bc) const {
-			index_entry_counter vc(bc);
+			index_entry_counter vc(tm_, bc);
 			count_btree_blocks(bitmaps_, bc, vc);
 		}
 
@@ -705,14 +714,16 @@ namespace {
 		typedef std::shared_ptr<metadata_index_store> ptr;
 
 		metadata_index_store(transaction_manager &tm)
-			: tm_(tm) {
+			: tm_(tm),
+			  bitmap_validator_(new bitmap_block_validator()) {
 			block_manager::write_ref wr = tm_.new_block(index_validator());
 			bitmap_root_ = wr.get_location();
 		}
 
 		metadata_index_store(transaction_manager &tm, block_address root, block_address nr_indexes)
 			: tm_(tm),
-			  bitmap_root_(root) {
+			  bitmap_root_(root),
+			  bitmap_validator_(new bitmap_block_validator()) {
 			resize(nr_indexes);
 			load_ies();
 		}
@@ -723,8 +734,17 @@ namespace {
 			for (unsigned i = 0; i < entries_.size(); i++) {
 				block_address b = entries_[i].blocknr_;
 
-				if (b != 0)
-					bc.inc(b);
+				if (b == 0)
+					continue;
+
+				try {
+					block_manager::read_ref rr = tm_.read_lock(b, bitmap_validator_);
+					if (rr.data())
+						bc.inc(b);
+				} catch (std::exception &e) {
+					if (bc.stop_on_error())
+						throw;
+				}
 			}
 		}
 
@@ -786,6 +806,7 @@ namespace {
 		transaction_manager &tm_;
 		block_address bitmap_root_;
 		std::vector<index_entry> entries_;
+		bcache::validator::ptr bitmap_validator_;
 	};
 }