recursive space map

Makefile

@@ -15,6 +15,7 @@ SOURCE=\
 	restore_emitter.cc \
 	space_map.cc \
 	space_map_disk.cc \
+	space_map_recursive.cc \
 	thin_pool.cc \
 	transaction_manager.cc \
 	xml_format.cc

block_counter.h

@@ -0,0 +1,41 @@
+#ifndef BLOCK_COUNTER_H
+#define BLOCK_COUNTER_H
+
+#include "block.h"
+
+//----------------------------------------------------------------
+
+namespace persistent_data {
+	//----------------------------------------------------------------
+	// Little helper class that keeps track of how many times blocks
+	// are referenced.
+	//----------------------------------------------------------------
+	class block_counter {
+	public:
+		typedef std::map<block_address, unsigned> count_map;
+
+		void inc(block_address b) {
+			count_map::iterator it = counts_.find(b);
+			if (it == counts_.end())
+				counts_.insert(make_pair(b, 1));
+			else
+				it->second++;
+		}
+
+		unsigned get_count(block_address b) const {
+			count_map::const_iterator it = counts_.find(b);
+			return (it == counts_.end()) ? 0 : it->second;
+		}
+
+		count_map const &get_counts() const {
+			return counts_;
+		}
+
+	private:
+		count_map counts_;
+	};
+}
+
+//----------------------------------------------------------------
+
+#endif

btree_checker.h

@@ -1,8 +1,8 @@
 #ifndef BTREE_CHECKER_H
 #define BTREE_CHECKER_H
 
+#include "block_counter.h"
 #include "btree.h"
-
 #include "checksum.h"
 #include "error_set.h"
 
@@ -16,35 +16,6 @@ using namespace std;
 //----------------------------------------------------------------
 
 namespace persistent_data {
-	//----------------------------------------------------------------
-	// Little helper class that keeps track of how many times blocks
-	// are referenced.
-	//----------------------------------------------------------------
-	class block_counter {
-	public:
-		typedef std::map<block_address, unsigned> count_map;
-
-		void inc(block_address b) {
-			count_map::iterator it = counts_.find(b);
-			if (it == counts_.end())
-				counts_.insert(make_pair(b, 1));
-			else
-				it->second++;
-		}
-
-		unsigned get_count(block_address b) const {
-			count_map::const_iterator it = counts_.find(b);
-			return (it == counts_.end()) ? 0 : it->second;
-		}
-
-		count_map const &get_counts() const {
-			return counts_;
-		}
-
-	private:
-		count_map counts_;
-	};
-
 	//----------------------------------------------------------------
 	// This class implements consistency checking for the btrees in
 	// general.  Derive from this if you want some additional checks.

metadata.cc

@@ -1,8 +1,8 @@
 #include "metadata.h"
 
 #include "math_utils.h"
-#include "space_map_disk.h"
 #include "space_map_core.h"
+#include "space_map_disk.h"
 
 #include <linux/fs.h>
 #include <sys/ioctl.h>

space_map.h

@@ -2,6 +2,7 @@
 #define SPACE_MAP_H
 
 #include "block.h"
+#include "block_counter.h"
 
 #include <boost/shared_ptr.hpp>
 
@@ -49,6 +50,15 @@ namespace persistent_data {
 		virtual void copy_root(void *dest, size_t len) = 0;
 	};
 
+	class checked_space_map : public persistent_space_map {
+	public:
+		typedef boost::shared_ptr<checked_space_map> ptr;
+
+		virtual void check(block_counter &counter) const {
+			throw std::runtime_error("not implemented");
+		}
+	};
+
 	class sm_adjust {
 	public:
 		sm_adjust(space_map::ptr sm, block_address b, int delta)

space_map_disk.cc

@@ -4,6 +4,7 @@
 #include "endian_utils.h"
 #include "math_utils.h"
 #include "space_map_disk_structures.h"
+#include "space_map_recursive.h"
 #include "transaction_manager.h"
 
 using namespace boost;
@@ -648,7 +649,7 @@ persistent_data::create_metadata_sm(transaction_manager::ptr tm, block_address n
 {
 	checked_space_map::ptr sm(new sm_metadata(tm));
 	sm->extend(nr_blocks);
-	return sm;
+	return create_recursive_sm(sm);
 }
 
 checked_space_map::ptr
@@ -660,7 +661,7 @@ persistent_data::open_metadata_sm(transaction_manager::ptr tm, void *root)
 	::memcpy(&d, root, sizeof(d));
 	sm_root_traits::unpack(d, v);
 
-	return checked_space_map::ptr(new sm_metadata(tm, v));
+	return create_recursive_sm(checked_space_map::ptr(new sm_metadata(tm, v)));
 }
 
 //----------------------------------------------------------------

space_map_disk.h

@@ -7,13 +7,6 @@
 //----------------------------------------------------------------
 
 namespace persistent_data {
-	class checked_space_map : public persistent_space_map {
-	public:
-		typedef boost::shared_ptr<checked_space_map> ptr;
-
-		virtual void check(block_counter &counter) const = 0;
-	};
-
 	checked_space_map::ptr
 	create_disk_sm(transaction_manager::ptr tm, block_address nr_blocks);
 

space_map_recursive.cc

@@ -0,0 +1,220 @@
+#include "space_map_recursive.h"
+
+using namespace persistent_data;
+
+//----------------------------------------------------------------
+
+namespace {
+	struct block_op {
+		enum op {
+			INC,
+			DEC,
+			SET
+		};
+
+		block_op(op o, block_address b)
+			: op_(o),
+			  b_(b) {
+			if (o == SET)
+				throw runtime_error("SET must take an operand");
+		}
+
+		block_op(op o, block_address b, uint32_t rc)
+			: op_(o),
+			  b_(b),
+			  rc_(rc) {
+			if (o != SET)
+				throw runtime_error("only SET takes an operand");
+		}
+
+		op op_;
+		block_address b_;
+		uint32_t rc_;
+	};
+
+	class sm_recursive : public checked_space_map {
+	public:
+		sm_recursive(checked_space_map::ptr sm)
+			: sm_(sm),
+			  recursing_(false) {
+		}
+
+		virtual block_address get_nr_blocks() const {
+			return sm_->get_nr_blocks();
+		}
+
+		virtual block_address get_nr_free() const {
+			return sm_->get_nr_free();
+		}
+
+		virtual ref_t get_count(block_address b) const {
+			cant_recurse("get_count");
+			recursing_const_lock lock(*this);
+			return sm_->get_count(b);
+		}
+
+		virtual void set_count(block_address b, ref_t c) {
+			if (recursing_)
+				add_op(block_op(block_op::SET, b, c));
+			else {
+				recursing_lock lock(*this);
+				return sm_->set_count(b, c);
+			}
+		}
+
+		virtual void commit() {
+			cant_recurse("commit");
+			sm_->commit();
+		}
+
+		virtual void inc(block_address b) {
+			if (recursing_)
+				add_op(block_op(block_op::INC, b));
+			else {
+				recursing_lock lock(*this);
+				return sm_->inc(b);
+			}
+		}
+
+		virtual void dec(block_address b) {
+			if (recursing_)
+				add_op(block_op(block_op::DEC, b));
+			else {
+				recursing_lock lock(*this);
+				return sm_->dec(b);
+			}
+		}
+
+		virtual block_address new_block() {
+			cant_recurse("new_block");
+			recursing_lock lock(*this);
+			return sm_->new_block();
+		}
+
+		virtual bool count_possibly_greater_than_one(block_address b) const {
+			if (recursing_)
+				return true;
+
+			else {
+				recursing_const_lock lock(*this);
+				return sm_->count_possibly_greater_than_one(b);
+			}
+		}
+
+		virtual void extend(block_address extra_blocks) {
+			cant_recurse("extend");
+			recursing_lock lock(*this);
+			return sm_->extend(extra_blocks);
+		}
+
+		virtual size_t root_size() {
+			cant_recurse("root_size");
+			recursing_const_lock lock(*this);
+			return sm_->root_size();
+		}
+
+		virtual void copy_root(void *dest, size_t len) {
+			cant_recurse("copy_root");
+			recursing_const_lock lock(*this);
+			return sm_->copy_root(dest, len);
+		}
+
+		virtual void check(block_counter &counter) const {
+			cant_recurse("check");
+			recursing_const_lock lock(*this);
+			return sm_->check(counter);
+		}
+
+
+		void flush_ops() {
+			op_map::const_iterator it, end = ops_.end();
+			for (it = ops_.begin(); it != end; ++it) {
+				list<block_op> const &ops = it->second;
+				list<block_op>::const_iterator op_it, op_end = ops.end();
+				for (op_it = ops.begin(); op_it != op_end; ++op_it) {
+					recursing_lock lock(*this);
+					switch (op_it->op_) {
+					case block_op::INC:
+						sm_->inc(op_it->b_);
+						break;
+
+					case block_op::DEC:
+						sm_->dec(op_it->b_);
+						break;
+
+					case block_op::SET:
+						sm_->set_count(op_it->b_, op_it->rc_);
+						break;
+					}
+				}
+			}
+
+			ops_.clear();
+		}
+
+	private:
+		void add_op(block_op const &op) {
+			ops_[op.b_].push_back(op);
+		}
+
+		void cant_recurse(string const &method) const {
+			if (recursing_)
+				throw runtime_error("recursive '" + method + "' not supported");
+		}
+
+		void set_recursing() const {
+			recursing_ = true;
+		}
+
+		struct recursing_lock {
+			recursing_lock(sm_recursive &smr)
+				: smr_(smr) {
+				smr_.recursing_ = true;
+			}
+
+			~recursing_lock() {
+				smr_.flush_ops();
+				smr_.recursing_ = false;
+			}
+
+		private:
+			sm_recursive &smr_;
+		};
+
+		struct recursing_const_lock {
+			recursing_const_lock(sm_recursive const &smr)
+				: smr_(smr) {
+				smr_.recursing_ = true;
+			}
+
+			~recursing_const_lock() {
+				smr_.recursing_ = false;
+			}
+
+		private:
+			sm_recursive const &smr_;
+		};
+
+		checked_space_map::ptr sm_;
+		mutable bool recursing_;
+
+		enum op {
+			BOP_INC,
+			BOP_DEC,
+			BOP_SET
+		};
+
+		typedef map<block_address, list<block_op> > op_map;
+		op_map ops_;
+	};
+}
+
+//----------------------------------------------------------------
+
+checked_space_map::ptr
+persistent_data::create_recursive_sm(checked_space_map::ptr sm)
+{
+	return checked_space_map::ptr(new sm_recursive(sm));
+}
+
+//----------------------------------------------------------------

space_map_recursive.h

@@ -0,0 +1,14 @@
+#ifndef SPACE_MAP_RECURSIVE_H
+#define SPACE_MAP_RECURSIVE_H
+
+#include "space_map.h"
+
+//----------------------------------------------------------------
+
+namespace persistent_data {
+	checked_space_map::ptr create_recursive_sm(checked_space_map::ptr sm);
+}
+
+//----------------------------------------------------------------
+
+#endif

unit-tests/space_map_disk_t.cc

@@ -113,6 +113,36 @@ void _test_set_affects_nr_allocated(space_map::ptr sm)
 	}
 }
 
+// Ref counts below 3 gets stored as bitmaps, above 3 they go into a btree
+// with uint32_t values.  Worth checking this thoroughly, especially for
+// the metadata format which may have complications due to recursion.
+void _test_high_ref_counts(space_map::ptr sm)
+{
+	srand(1234);
+	for (unsigned i = 0; i < NR_BLOCKS; i++)
+		sm->set_count(i, rand() % 6789);
+	sm->commit();
+
+	for (unsigned i = 0; i < NR_BLOCKS; i++) {
+		sm->inc(i);
+		sm->inc(i);
+		if (i % 1000)
+			sm->commit();
+	}
+	sm->commit();
+
+	srand(1234);
+	for (unsigned i = 0; i < NR_BLOCKS; i++)
+		BOOST_CHECK_EQUAL(sm->get_count(i), (rand() % 6789) + 2);
+
+	for (unsigned i = 0; i < NR_BLOCKS; i++)
+		sm->dec(i);
+
+	srand(1234);
+	for (unsigned i = 0; i < NR_BLOCKS; i++)
+		BOOST_CHECK_EQUAL(sm->get_count(i), (rand() % 6789) + 1);
+}
+
 //----------------------------------------------------------------
 
 #if 0
@@ -164,6 +194,12 @@ BOOST_AUTO_TEST_CASE(test_disk_set_affects_nr_allocated)
 	_test_set_affects_nr_allocated(sm);
 }
 
+BOOST_AUTO_TEST_CASE(test_disk_high_ref_counts)
+{
+	space_map::ptr sm = create_sm_disk();
+	_test_high_ref_counts(sm);
+}
+
 BOOST_AUTO_TEST_CASE(test_disk_reopen)
 {
 	unsigned char buffer[128];
@@ -232,6 +268,12 @@ BOOST_AUTO_TEST_CASE(test_metadata_set_affects_nr_allocated)
 	_test_set_affects_nr_allocated(sm);
 }
 
+BOOST_AUTO_TEST_CASE(test_metadata_high_ref_counts)
+{
+	space_map::ptr sm = create_sm_metadata();
+	_test_high_ref_counts(sm);
+}
+
 BOOST_AUTO_TEST_CASE(test_metadata_reopen)
 {
 	unsigned char buffer[128];