Merge pull request #149 from mingnus/thin-check-fix-metadata-leaks

thin_check enhancements

base/io_generator.cc

@@ -1,5 +1,7 @@
 #include "base/io_generator.h"
 #include "base/sequence_generator.h"
+#include <chrono>
+#include <random>
 #include <stdexcept>
 #include <cstdlib>
 #include <cstring>
@@ -31,19 +33,23 @@ namespace {
 		typedef std::shared_ptr<op_generator> ptr;
 
 		op_generator(base::req_op op1)
-			: op1_(op1), op2_(op1), op1_pct_(100) {
+			: op1_(op1), op2_(op1), op1_pct_(100),
+			  rand_seed_(std::chrono::high_resolution_clock::now().time_since_epoch().count()),
+			  op_engine_(rand_seed_) {
 		}
 
 		op_generator(base::req_op op1,
 			     base::req_op op2,
 			     unsigned op1_pct)
-			: op1_(op1), op2_(op2), op1_pct_(op1_pct) {
+			: op1_(op1), op2_(op2), op1_pct_(op1_pct),
+			  rand_seed_(std::chrono::high_resolution_clock::now().time_since_epoch().count()),
+			  op_engine_(rand_seed_) {
 			if (op1_pct > 100)
 				throw std::runtime_error("invalid percentage");
 		}
 
 		base::req_op next_op() {
-			if (static_cast<unsigned>(std::rand()) % 100 > op1_pct_)
+			if (op_engine_() % 100 > op1_pct_)
 				return op2_;
 			return op1_;
 		}
@@ -52,6 +58,9 @@ namespace {
 		base::req_op op1_;
 		base::req_op op2_;
 		unsigned op1_pct_;
+		uint64_t rand_seed_;
+
+		std::mt19937 op_engine_;
 	};
 
 	//--------------------------------

base/sequence_generator.cc

@@ -1,5 +1,7 @@
 #include "base/run_set.h"
 #include "base/sequence_generator.h"
+#include <chrono>
+#include <random>
 #include <stdexcept>
 
 //----------------------------------------------------------------
@@ -80,6 +82,9 @@ namespace {
 			: begin_(begin),
 			  step_(step),
 			  max_forward_steps_(seq_nr),
+			  rand_seed_(std::chrono::high_resolution_clock::now().time_since_epoch().count()),
+			  results_engine_(rand_seed_),
+			  steps_engine_(rand_seed_),
 			  nr_generated_(0)
 		{
 			if (!size || !step || !seq_nr)
@@ -93,9 +98,15 @@ namespace {
 			if (!max_forward_steps_ || max_forward_steps_ > nr_steps_)
 				throw std::runtime_error("invalid number of forward steps");
 
-			if (max_forward_steps_ > 1)
+			results_distr_ = std::uniform_int_distribution<uint64_t>(0,
+					nr_steps_ - 1);
+
+			if (max_forward_steps_ > 1) {
+				steps_distr_ = std::uniform_int_distribution<uint64_t>(1,
+						max_forward_steps_);
 				reset_forward_generator();
 			}
+		}
 
 		uint64_t next() {
 			// FIXME: eliminate if-else
@@ -109,7 +120,7 @@ namespace {
 		void reset_forward_generator() {
 			uint64_t begin = peek_random_step();
 
-			unsigned seq_nr = (std::rand() % max_forward_steps_) + 1;
+			unsigned seq_nr = steps_distr_(steps_engine_);
 			base::run_set<uint64_t>::const_iterator it = rand_map_.upper_bound(begin);
 			if (it != rand_map_.end())
 				seq_nr = std::min<uint64_t>(seq_nr, *it->begin_ - begin);
@@ -148,12 +159,13 @@ namespace {
 			}
 		}
 
-		uint64_t peek_random_step() const {
+		// TODO: improve the complexity of generating unique sequence
+		uint64_t peek_random_step() {
 			uint64_t step_idx;
 
 			bool found = true;
 			while (found) {
-				step_idx = std::rand() % nr_steps_;
+				step_idx = results_distr_(results_engine_);
 				found = rand_map_.member(step_idx);
 			}
 
@@ -164,7 +176,12 @@ namespace {
 		uint64_t nr_steps_;
 		uint64_t step_;
 		unsigned max_forward_steps_;
+		uint64_t rand_seed_;
 
+		std::mt19937_64 results_engine_;
+		std::mt19937_64 steps_engine_;
+		std::uniform_int_distribution<uint64_t> results_distr_;
+		std::uniform_int_distribution<uint64_t> steps_distr_;
 		base::run_set<uint64_t> rand_map_;
 		uint64_t nr_generated_;
 		forward_sequence_generator forward_gen_;

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/core.cc

@@ -190,10 +190,12 @@ namespace {
 				set_bits_(b, c);
 		}
 		
-		if (old_c == 0 && c > 0)
+		if (old_c == 0 && c > 0) {
 			nr_free_--;
+			if (b == search_start_)
+				search_start_++;
 
-		else if (old_c > 0 && c == 0) {
+		} else if (old_c > 0 && c == 0) {
 			if (b < search_start_)
 				search_start_ = b;
 

persistent-data/space-maps/disk.cc

@@ -43,11 +43,19 @@ namespace {
 			bitmap_header const *data = reinterpret_cast<bitmap_header const *>(raw);
 			crc32c sum(BITMAP_CSUM_XOR);
 			sum.append(&data->not_used, MD_BLOCK_SIZE - sizeof(uint32_t));
-			if (sum.get_sum() != to_cpu<uint32_t>(data->csum))
+			if (sum.get_sum() != to_cpu<uint32_t>(data->csum)) {
-				throw checksum_error("bad checksum in space map bitmap");
+				std::ostringstream out;
+				out << "bad checksum in space map bitmap (block "
+				    << location << ")";
+				throw checksum_error(out.str());
+			}
 
-			if (to_cpu<uint64_t>(data->blocknr) != location)
+			if (to_cpu<uint64_t>(data->blocknr) != location) {
-				throw checksum_error("bad block nr in space map bitmap");
+				std::ostringstream out;
+				out << "bad block nr in space map bitmap (block "
+				    << location << ")";
+				throw checksum_error(out.str());
+			}
 		}
 
 		virtual bool check_raw(void const *raw) const {
@@ -77,11 +85,19 @@ namespace {
 			metadata_index const *mi = reinterpret_cast<metadata_index const *>(raw);
 			crc32c sum(INDEX_CSUM_XOR);
 			sum.append(&mi->padding_, MD_BLOCK_SIZE - sizeof(uint32_t));
-			if (sum.get_sum() != to_cpu<uint32_t>(mi->csum_))
+			if (sum.get_sum() != to_cpu<uint32_t>(mi->csum_)) {
-				throw checksum_error("bad checksum in metadata index block");
+				std::ostringstream out;
+				out << "bad checksum in metadata index block (block "
+				    << location << ")";
+				throw checksum_error(out.str());
+			}
 
-			if (to_cpu<uint64_t>(mi->blocknr_) != location)
+			if (to_cpu<uint64_t>(mi->blocknr_) != location) {
-				throw checksum_error("bad block nr in metadata index block");
+				std::ostringstream out;
+				out << "bad block nr in metadata index block (block "
+				    << location << ")";
+				throw checksum_error(out.str());
+			}
 		}
 
 		virtual bool check_raw(void const *raw) const {
@@ -639,21 +655,30 @@ namespace {
 		//--------------------------------
 
 		struct index_entry_counter {
-			index_entry_counter(block_counter &bc)
+			index_entry_counter(transaction_manager &tm,
-			: bc_(bc) {
+					    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 +730,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 +750,17 @@ namespace {
 			for (unsigned i = 0; i < entries_.size(); i++) {
 				block_address b = entries_[i].blocknr_;
 
-				if (b != 0)
+				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 +822,7 @@ namespace {
 		transaction_manager &tm_;
 		block_address bitmap_root_;
 		std::vector<index_entry> entries_;
+		bcache::validator::ptr bitmap_validator_;
 	};
 }
 

persistent-data/transaction_manager.h

@@ -69,6 +69,12 @@ namespace persistent_data {
 			bm_->prefetch(b);
 		}
 
+		// mark a block as shadowed,
+		// for the purpose of in-place modification
+		void mark_shadowed(block_address b) {
+			add_shadow(b);
+		}
+
 	private:
 		void add_shadow(block_address b);
 		void remove_shadow(block_address b);

thin-provisioning/damage_generator.cc

@@ -1,3 +1,5 @@
+#include <chrono>
+#include <random>
 #include "damage_generator.h"
 
 using namespace thin_provisioning;
@@ -13,12 +15,14 @@ namespace {
 		base::run_set<block_address> visited;
 		block_address nr_visited = 0;
 
-		srand(time(NULL));
+		uint64_t rand_seed(std::chrono::high_resolution_clock::now().time_since_epoch().count());
+		std::mt19937 rand_engine(rand_seed);
+
 		while (nr_blocks) {
 			if (nr_visited == sm_size)
 				break;
 
-			block_address b = rand() % sm_size;
+			block_address b = rand_engine() % sm_size;
 			if (visited.member(b))
 				continue;
 

thin-provisioning/metadata_checker.cc

@@ -274,6 +274,17 @@ namespace {
 		return err;
 	}
 
+	void mark_sm_shadowed(transaction_manager::ptr tm, persistent_space_map::ptr sm) {
+		block_counter bc(true);
+		sm->count_metadata(bc);
+
+		block_address nr_blocks = sm->get_nr_blocks();
+		for (block_address b = 0; b < nr_blocks; b++) {
+			if (bc.get_count(b))
+				tm->mark_shadowed(b);
+		}
+	}
+
 	error_state clear_leaked_blocks(space_map::ptr actual,
 					block_counter const &expected) {
 		error_state err = NO_ERROR;
@@ -296,7 +307,8 @@ namespace {
 		out << "checking space map counts" << end_message();
 		nested_output::nest _ = out.push();
 
-		count_metadata(tm, sb, bc);
+		if (!count_metadata(tm, sb, bc))
+			return FATAL;
 
 		// Finally we need to check the metadata space map agrees
 		// with the counts we've just calculated.
@@ -318,7 +330,7 @@ namespace {
 		} else {
 			for (block_address b = 0; b < nr_blocks; b++) {
 				auto a_count = actual->get_count(b);
-				auto e_count = actual->get_count(b);
+				auto e_count = expected->get_count(b);
 
 				if (a_count != e_count) {
 					out << "data reference counts differ for block " << b
@@ -358,6 +370,7 @@ namespace {
 			  options_(check_opts),
 			  out_(cerr, 2),
 			  info_out_(cout, 0),
+			  expected_rc_(true), // set stop on the first error
 			  err_(NO_ERROR) {
 
 			if (output_opts == OUTPUT_QUIET) {
@@ -386,8 +399,11 @@ namespace {
 			print_info(tm, sb, info_out_);
 
 			if (options_.sm_opts_ == check_options::SPACE_MAP_FULL) {
+				// data block reference counting is disabled
+				// until that there's a better solution in space
+				// and time complexity
 				space_map::ptr data_sm{open_disk_sm(*tm, &sb.data_space_map_root_)};
-				optional<space_map::ptr> core_sm{create_core_map(data_sm->get_nr_blocks())};
+				optional<space_map::ptr> core_sm;
 				err_ << examine_data_mappings(tm, sb, options_.data_mapping_opts_, out_, core_sm);
 
 				if (err_ == FATAL)
@@ -397,15 +413,15 @@ namespace {
 				// then we should check the space maps too.
 				err_ << examine_metadata_space_map(tm, sb, options_.sm_opts_, out_, expected_rc_);
 
-				// check the data space map
+				// verify ref-counts of data blocks
-				if (core_sm)
+				if (err_ != FATAL && core_sm)
 					err_ << compare_space_maps(data_sm, *core_sm, out_);
 			} else
 				err_ << examine_data_mappings(tm, sb, options_.data_mapping_opts_, out_,
 							      optional<space_map::ptr>());
 		}
 
-		bool fix_metadata_leaks() {
+		bool fix_metadata_leaks(bool open_transaction) {
 			if (!verify_preconditions_before_fixing()) {
 				out_ << "metadata has not been fully examined" << end_message();
 				return false;
@@ -424,6 +440,9 @@ namespace {
 				open_metadata_sm(*tm, static_cast<void const*>(&sb.metadata_space_map_root_));
 			tm->set_sm(metadata_sm);
 
+			if (!open_transaction)
+				mark_sm_shadowed(tm, metadata_sm);
+
 			err_ = clear_leaked_blocks(metadata_sm, expected_rc_);
 
 			if (err_ != NO_ERROR)
@@ -449,6 +468,10 @@ namespace {
 
 			block_manager::ptr bm = open_bm(path_, block_manager::READ_WRITE);
 			superblock_detail::superblock sb = read_superblock(bm);
+
+			if (!sb.get_needs_check_flag())
+				return true;
+
 			sb.set_needs_check_flag(false);
 			write_superblock(bm, sb);
 
@@ -553,7 +576,8 @@ check_options::check_options()
 	  sm_opts_(SPACE_MAP_FULL),
 	  ignore_non_fatal_(false),
 	  fix_metadata_leaks_(false),
-	  clear_needs_check_(false) {
+	  clear_needs_check_(false),
+	  open_transaction_(false) {
 }
 
 void check_options::set_superblock_only() {
@@ -625,7 +649,7 @@ thin_provisioning::check_metadata(std::string const &path,
 
 	checker.check();
 	if (check_opts.fix_metadata_leaks_)
-		checker.fix_metadata_leaks();
+		checker.fix_metadata_leaks(check_opts.open_transaction_);
 	if (check_opts.fix_metadata_leaks_ ||
 	    check_opts.clear_needs_check_)
 		checker.clear_needs_check_flag();

thin-provisioning/metadata_checker.h

@@ -55,6 +55,7 @@ namespace thin_provisioning {
 		bool ignore_non_fatal_;
 		bool fix_metadata_leaks_;
 		bool clear_needs_check_;
+		bool open_transaction_;
 	};
 
 	enum output_options {

thin-provisioning/metadata_counter.cc

@@ -8,7 +8,7 @@ using namespace thin_provisioning;
 //----------------------------------------------------------------
 
 namespace {
-	void count_trees(transaction_manager::ptr tm,
+	bool count_trees(transaction_manager::ptr tm,
 			 superblock_detail::superblock const &sb,
 			 block_counter &bc) {
 
@@ -27,11 +27,15 @@ namespace {
 					   mapping_tree_detail::block_traits::ref_counter(space_map::ptr()));
 			count_btree_blocks(mtree, bc, vc);
 		}
+
+		return true;
 	}
 
-	void count_space_maps(transaction_manager::ptr tm,
+	bool count_space_maps(transaction_manager::ptr tm,
 			      superblock_detail::superblock const &sb,
 			      block_counter &bc) {
+		bool ret = true;
+
 		// Count the metadata space map (no-throw)
 		try {
 			persistent_space_map::ptr metadata_sm =
@@ -39,36 +43,46 @@ namespace {
 			metadata_sm->count_metadata(bc);
 		} catch (std::exception &e) {
 			cerr << e.what() << endl;
+			ret = false;
 		}
 
 		// Count the data space map (no-throw)
-		{
+		try {
 			persistent_space_map::ptr data_sm =
 				open_disk_sm(*tm, static_cast<void const *>(&sb.data_space_map_root_));
 			data_sm->count_metadata(bc);
+		} catch (std::exception &e) {
+			cerr << e.what() << endl;
+			ret = false;
 		}
+
+		return ret;
 	}
 }
 
 //----------------------------------------------------------------
 
-void thin_provisioning::count_metadata(transaction_manager::ptr tm,
+bool thin_provisioning::count_metadata(transaction_manager::ptr tm,
 				       superblock_detail::superblock const &sb,
 				       block_counter &bc,
 				       bool skip_metadata_snap) {
+	bool ret = true;
+
 	// Count the superblock
 	bc.inc(superblock_detail::SUPERBLOCK_LOCATION);
-	count_trees(tm, sb, bc);
+	ret &= count_trees(tm, sb, bc);
 
 	// Count the metadata snap, if present
 	if (!skip_metadata_snap && sb.metadata_snap_ != superblock_detail::SUPERBLOCK_LOCATION) {
 		bc.inc(sb.metadata_snap_);
 
 		superblock_detail::superblock snap = read_superblock(tm->get_bm(), sb.metadata_snap_);
-		count_trees(tm, snap, bc);
+		ret &= count_trees(tm, snap, bc);
 	}
 
-	count_space_maps(tm, sb, bc);
+	ret &= count_space_maps(tm, sb, bc);
+
+	return ret;
 }
 
 //----------------------------------------------------------------

thin-provisioning/metadata_counter.h

@@ -7,7 +7,7 @@
 //----------------------------------------------------------------
 
 namespace thin_provisioning {
-	void count_metadata(transaction_manager::ptr tm,
+	bool count_metadata(transaction_manager::ptr tm,
 			    superblock_detail::superblock const &sb,
 			    block_counter &bc,
 			    bool skip_metadata_snap = false);

unit-tests/space_map_t.cc

@@ -20,7 +20,9 @@
 #include "persistent-data/space-maps/disk.h"
 #include "persistent-data/space-maps/core.h"
 #include "persistent-data/space-maps/careful_alloc.h"
+#include "persistent-data/space-maps/disk_structures.h"
 #include "persistent-data/space-maps/recursive.h"
+#include "test_utils.h"
 
 using namespace std;
 using namespace persistent_data;
@@ -29,14 +31,16 @@ using namespace testing;
 //----------------------------------------------------------------
 
 namespace {
-	block_address const NR_BLOCKS = 1000; // FIXME: bump up
+	unsigned const ENTRIES_PER_BLOCK = (MD_BLOCK_SIZE - sizeof(sm_disk_detail::bitmap_header)) * 4;
+	unsigned const NR_BITMAPS = 2;
+	unsigned const NR_BLOCKS = ENTRIES_PER_BLOCK * NR_BITMAPS;
 	block_address const SUPERBLOCK = 0;
-	block_address const MAX_LOCKS = 8;
+	unsigned const MAX_LOCKS = 8;
 
 	class SpaceMapTests : public Test {
 	public:
 		SpaceMapTests()
-			: bm_(new block_manager("./test.data", NR_BLOCKS, MAX_LOCKS, block_manager::READ_WRITE)),
+			: bm_(test::create_bm(NR_BLOCKS)),
 			  sm_(create_core_map(NR_BLOCKS)),
 			  tm_(bm_, sm_) {
 		}
@@ -229,8 +233,89 @@ namespace {
 			test_high_ref_counts(SMCreator::create(tm_));
 		}
 
-		void
+
-		copy_space_maps(space_map::ptr lhs, space_map::ptr rhs) {
+		block_manager::ptr bm_;
+		space_map::ptr sm_;
+		transaction_manager tm_;
+	};
+
+	//--------------------------------
+
+	class MetadataSpaceMapTests : public Test {
+	public:
+		MetadataSpaceMapTests() {
+			memset(metadata_sm_root_, 0, sizeof(metadata_sm_root_));
+			create_metadata_sm(NR_BLOCKS);
+		}
+
+		void commit() {
+			metadata_sm_->commit();
+
+			ASSERT_THAT(metadata_sm_->root_size(), Le(sizeof(metadata_sm_root_)));
+			metadata_sm_->copy_root(metadata_sm_root_, sizeof(metadata_sm_root_));
+		}
+
+		void open() {
+			bm_ = block_manager::ptr(new block_manager("./test.data", NR_BLOCKS, MAX_LOCKS, block_manager::READ_WRITE));
+			space_map::ptr core_sm{create_core_map(NR_BLOCKS)};
+			tm_ = transaction_manager::ptr(new transaction_manager(bm_, core_sm));
+			metadata_sm_ = persistent_data::open_metadata_sm(*tm_, metadata_sm_root_);
+			tm_->set_sm(metadata_sm_);
+		}
+
+		void create_data_sm(block_address nr_blocks) {
+			data_sm_ = create_disk_sm(*tm_, nr_blocks);
+		}
+
+		void load_ies(std::vector<sm_disk_detail::index_entry> &entries) {
+			sm_disk_detail::sm_root v;
+			get_root(v);
+
+			block_address nr_indexes = (v.nr_blocks_ + ENTRIES_PER_BLOCK - 1) / ENTRIES_PER_BLOCK;
+			entries.resize(nr_indexes);
+			ASSERT_EQ(entries.size(), NR_BITMAPS);
+
+			block_manager::read_ref rr =
+				bm_->read_lock(v.bitmap_root_, index_validator());
+			sm_disk_detail::metadata_index const *mdi = reinterpret_cast<sm_disk_detail::metadata_index const *>(rr.data());
+			for (block_address i = 0; i < nr_indexes; i++) {
+				sm_disk_detail::index_entry_traits::unpack(*(mdi->index + i), entries[i]);
+				ASSERT_EQ(metadata_sm_->get_count(entries[i].blocknr_), 1u);
+			}
+		}
+
+		// note that the index_block is not shadowed until space map commit
+		void get_root(sm_disk_detail::sm_root &v) const {
+			sm_disk_detail::sm_root_disk d;
+			metadata_sm_->copy_root(&d, sizeof(d));
+			sm_disk_detail::sm_root_traits::unpack(d, v);
+		}
+
+		void mark_shadowed() {
+			block_counter bc(true);
+			metadata_sm_->count_metadata(bc);
+
+			block_address nr_blocks = metadata_sm_->get_nr_blocks();
+			for (block_address b = 0; b < nr_blocks; b++) {
+				if (bc.get_count(b))
+					tm_->mark_shadowed(b);
+			}
+		}
+
+		checked_space_map::ptr metadata_sm_;
+		checked_space_map::ptr data_sm_;
+
+	private:
+		void create_metadata_sm(block_address nr_blocks) {
+			bm_ = test::create_bm(NR_BLOCKS);
+			space_map::ptr core_sm{create_core_map(nr_blocks)};
+			tm_ = transaction_manager::ptr(new transaction_manager(bm_, core_sm));
+			metadata_sm_ = persistent_data::create_metadata_sm(*tm_, nr_blocks);
+			copy_space_maps(metadata_sm_, core_sm);
+			tm_->set_sm(metadata_sm_);
+		}
+
+		void copy_space_maps(space_map::ptr lhs, space_map::ptr rhs) {
 			for (block_address b = 0; b < rhs->get_nr_blocks(); b++) {
 				uint32_t count = rhs->get_count(b);
 				if (count > 0)
@@ -239,8 +324,9 @@ namespace {
 		}
 
 		block_manager::ptr bm_;
-		space_map::ptr sm_;
+		transaction_manager::ptr tm_;
-		transaction_manager tm_;
+
+		unsigned char metadata_sm_root_[128];
 	};
 }
 
@@ -273,17 +359,157 @@ TEST_F(SpaceMapTests, test_sm_metadata)
 	test_sm_reopen<sm_metadata_creator>();
 }
 
-TEST_F(SpaceMapTests, test_metadata_and_disk)
+//----------------------------------------------------------------
-{
-	block_manager::ptr bm(
-		new block_manager("./test.data", NR_BLOCKS, MAX_LOCKS, block_manager::READ_WRITE));
-	space_map::ptr core_sm{create_core_map(NR_BLOCKS)};
-	transaction_manager::ptr tm(new transaction_manager(bm, core_sm));
-	persistent_space_map::ptr metadata_sm = persistent_data::create_metadata_sm(*tm, NR_BLOCKS);
-	copy_space_maps(metadata_sm, core_sm);
-	tm->set_sm(metadata_sm);
 
-	persistent_space_map::ptr data_sm_ = create_disk_sm(*tm, NR_BLOCKS * 2);
+TEST_F(MetadataSpaceMapTests, test_metadata_and_disk)
+{
+	create_data_sm(NR_BLOCKS * 2);
+}
+
+// Test whether sm_recursive protected allocated blocks in a recursion
+// (github issue #70)
+TEST_F(MetadataSpaceMapTests, test_allocate_blocks)
+{
+	commit();
+	open();
+
+	std::vector<sm_disk_detail::index_entry> entries;
+	load_ies(entries);
+
+	block_address nr_allocatable = metadata_sm_->get_nr_free() - NR_BITMAPS;
+	block_address nr_allocated = 0;
+	while (true) {
+		space_map::maybe_block b = metadata_sm_->new_block();
+		if (!b)
+			break;
+		++nr_allocated;
+
+		for (auto const &e : entries)
+			ASSERT_NE(*b, e.blocknr_);
+	}
+
+	ASSERT_EQ(nr_allocated, nr_allocatable);
+	for (auto const &e : entries)
+		ASSERT_EQ(metadata_sm_->get_count(e.blocknr_), 0u);
+}
+
+// Test whether sm_careful_alloc protects released blocks in a recursion.
+// (github issue #97)
+TEST_F(MetadataSpaceMapTests, test_multiple_shadows_by_dec)
+{
+	// Occupy all the blocks belonging to the first bitmap,
+	// to change the position of further allocated blocks.
+	for (unsigned i = 0; i < ENTRIES_PER_BLOCK; i++)
+		metadata_sm_->new_block();
+
+	// the extra two blocks are the index block and the ref-count tree root
+	ASSERT_EQ(metadata_sm_->get_nr_free(), ENTRIES_PER_BLOCK - NR_BITMAPS - 2);
+
+	commit();
+	open();
+
+	std::vector<sm_disk_detail::index_entry> entries;
+	load_ies(entries);
+
+	// Releasing the block belonging to the second bitmap results in
+	// shadowing the second, then the first bitmap blocks.
+	// The shadow operations must not reuse the released block.
+	block_address last = metadata_sm_->get_nr_blocks() - metadata_sm_->get_nr_free() - 1;
+	ASSERT_EQ(metadata_sm_->get_count(last), 1u);
+	metadata_sm_->dec(last);
+
+	// assert that the released blocks are not used
+	ASSERT_EQ(metadata_sm_->get_count(last), 0u);
+	for (auto const &e : entries)
+		ASSERT_EQ(metadata_sm_->get_count(e.blocknr_), 0u);
+
+	ASSERT_EQ(metadata_sm_->get_nr_free(), ENTRIES_PER_BLOCK - NR_BITMAPS - 1);
+
+	// check ref-counts of shadowed bitmaps
+	commit();
+	open();
+	load_ies(entries);
+}
+
+// Test whether sm_recursive protects allocated blocks in a recursion
+// (github issue #70)
+TEST_F(MetadataSpaceMapTests, test_multiple_shadows_by_inc)
+{
+	// Occupy all the blocks belonging to the first bitmap,
+	// to change the position of further allocated blocks.
+	for (unsigned i = 0; i < ENTRIES_PER_BLOCK; i++)
+		metadata_sm_->new_block();
+
+	// the extra two blocks are the index block and the ref-count tree root
+	ASSERT_EQ(metadata_sm_->get_nr_free(), ENTRIES_PER_BLOCK - NR_BITMAPS - 2);
+
+	commit();
+	open();
+
+	std::vector<sm_disk_detail::index_entry> entries;
+	load_ies(entries);
+
+	// allocating a block results in shadowing the second,
+	// then the first bitmap
+	space_map::maybe_block b = metadata_sm_->new_block();
+
+	// assert reference counts
+	ASSERT_EQ(metadata_sm_->get_count(*b), 1u);
+	for (auto const &e : entries)
+		ASSERT_EQ(metadata_sm_->get_count(e.blocknr_), 0u);
+
+	ASSERT_EQ(metadata_sm_->get_nr_free(), ENTRIES_PER_BLOCK - NR_BITMAPS - 3);
+
+	// check ref-counts of shadowed bitmaps
+	commit();
+	open();
+	load_ies(entries);
+}
+
+// Test whether intended in-place modification works
+TEST_F(MetadataSpaceMapTests, test_intended_in_place_modification)
+{
+	sm_disk_detail::sm_root root;
+	get_root(root);
+	std::vector<sm_disk_detail::index_entry> entries;
+	load_ies(entries);
+
+	commit();
+	open();
+	mark_shadowed();
+
+	// the bitmaps won't be shadowed,
+	// all the free blocks therefore become allocatable
+	block_address nr_allocatable = metadata_sm_->get_nr_free();
+	block_address nr_allocated = 0;
+	while (true) {
+		space_map::maybe_block b = metadata_sm_->new_block();
+		if (!b)
+			break;
+		++nr_allocated;
+
+		for (auto const &e : entries)
+			ASSERT_NE(*b, e.blocknr_);
+	}
+
+	ASSERT_EQ(nr_allocated, nr_allocatable);
+
+	commit();
+	open();
+
+	sm_disk_detail::sm_root root2;
+	get_root(root2);
+	std::vector<sm_disk_detail::index_entry> entries2;
+	load_ies(entries2);
+
+	// assert the space map block locations are not changed
+	ASSERT_EQ(root.bitmap_root_, root2.bitmap_root_);
+	ASSERT_EQ(root.ref_count_root_, root2.ref_count_root_);
+	ASSERT_EQ(entries.size(), entries2.size());
+	for (unsigned i = 0; i < entries.size(); i++)
+		ASSERT_EQ(entries[i].blocknr_, entries2[i].blocknr_);
+
+	ASSERT_EQ(root2.nr_allocated_, root.nr_allocated_ + nr_allocated);
 }
 
 //----------------------------------------------------------------