streamline and test transaction_manager

Makefile

@@ -3,13 +3,21 @@ SOURCE=\
 
 PROGRAM_SOURCE=\
 	block_t.cc \
-	space_map_t.cc
+	space_map_t.cc \
+	transaction_manager_t.cc
 
 OBJECTS=$(subst .cc,.o,$(SOURCE))
 CPPFLAGS=-Wall -std=c++0x
 INCLUDES=
 LIBS=-lstdc++
 
+.PHONEY: unit-tests
+
+unit-tests: block_t space_map_t transaction_manager_t
+	./block_t
+	./space_map_t
+	./transaction_manager_t
+
 .SUFFIXES: .cc .o .d
 
 %.d: %.cc
@@ -21,13 +29,16 @@ LIBS=-lstdc++
 	g++ -c $(CPPFLAGS) $(INCLUDES) -o $@ $<
 
 multisnap_display: $(OBJECTS) main.o
-	g++ -o $@ $+ $(LIBS)
+	g++ $(CPPFLAGS) -o $@ $+ $(LIBS)
 
 block_t: block_t.o
-	g++ -o $@ $+ $(LIBS)
+	g++ $(CPPFLAGS) -o $@ $+ $(LIBS)
 
 space_map_t: space_map_t.o
-	g++ -o $@ $+ $(LIBS)
+	g++ $(CPPFLAGS) -o $@ $+ $(LIBS)
+
+transaction_manager_t: transaction_manager_t.o
+	g++ $(CPPFLAGS) -o $@ $+ $(LIBS)
 
 include $(subst .cc,.d,$(SOURCE))
 include $(subst .cc,.d,$(PROGRAM_SOURCE))

transaction_manager.h

@@ -14,37 +14,41 @@ namespace persistent_data {
 	class transaction_manager : public boost::noncopyable {
 	public:
 		typedef boost::shared_ptr<transaction_manager<MetadataBlockSize> > ptr;
+		typedef typename block_manager<MetadataBlockSize>::read_ref read_ref;
+		typedef typename block_manager<MetadataBlockSize>::write_ref write_ref;
+		typedef typename block_manager<MetadataBlockSize>::validator::ptr validator;
 
+		// If the space map is persistent, then the caller should
+		// hold onto a reference and remember to call sm_->commit()
+		// and update the superblock before dropping the superblock
+		// reference.
 		transaction_manager(typename block_manager<MetadataBlockSize>::ptr bm,
 				    space_map::ptr sm);
 		~transaction_manager();
 
-		typedef typename block_manager<MetadataBlockSize>::read_ref read_ref;
+		// Drop the superblock reference to commit
-		typedef typename block_manager<MetadataBlockSize>::write_ref write_ref;
+		write_ref begin(block_address superblock);
-		typedef typename block_manager<MetadataBlockSize>::block_validator block_validator;
+		write_ref begin(block_address superblock, validator v);
 
-		void reserve_block(block_address location);
-		void begin();
-		void pre_commit();
-		void commit(write_ref superblock);
-
-		block_address alloc_block();
 		write_ref new_block();
-		write_ref new_block(block_validator const &v);
+		write_ref new_block(validator v);
 
-		write_ref shadow(block_address orig, bool &inc_children);
+		// shadowing returns a new write_ref, and a boolean which
-		write_ref shadow(block_address orig, block_validator const &v, bool &inc_children);
+		// indicates whether the children should be incremented.
+		std::pair<write_ref, bool> shadow(block_address orig);
+		std::pair<write_ref, bool> shadow(block_address orig, validator v);
 
 		read_ref read_lock(block_address b);
-		read_ref read_lock(block_address b, block_validator const &v);
+		read_ref read_lock(block_address b, validator v);
 
-		void inc(block_address b);
+		space_map::ptr get_sm() {
-		void dec(block_address b);
+			return sm_;
-		uint32_t ref_count(block_address b) const;
+		}
 
 	private:
 		void add_shadow(block_address b);
-	        bool is_shadow(block_address b);
+		void remove_shadow(block_address b);
+	        bool is_shadow(block_address b) const;
 		void wipe_shadow_table();
 
 		typename block_manager<MetadataBlockSize>::ptr bm_;

transaction_manager.tcc

@@ -4,6 +4,7 @@
 
 using namespace boost;
 using namespace persistent_data;
+using namespace std;
 
 //----------------------------------------------------------------
 
@@ -21,98 +22,104 @@ transaction_manager<BlockSize>::~transaction_manager()
 }
 
 template <uint32_t BlockSize>
-void
+typename transaction_manager<BlockSize>::write_ref
-transaction_manager<BlockSize>::reserve_block(block_address location)
+transaction_manager<BlockSize>::begin(block_address superblock)
 {
-	sm_->inc_block(location);
+	auto wr = bm_->superblock(superblock);
-}
-
-template <uint32_t BlockSize>
-void
-transaction_manager<BlockSize>::begin()
-{
-}
-
-template <uint32_t BlockSize>
-void
-transaction_manager<BlockSize>::pre_commit()
-{
-	sm_->commit();
-}
-
-template <uint32_t BlockSize>
-void
-transaction_manager<BlockSize>::commit(write_ref superblock)
-{
-	bm_->flush(superblock);
 	wipe_shadow_table();
+	return wr;
 }
 
 template <uint32_t BlockSize>
-block_address
+typename transaction_manager<BlockSize>::write_ref
-transaction_manager<BlockSize>::alloc_block()
+transaction_manager<BlockSize>::begin(block_address superblock,
+				      validator v)
 {
-	return sm_->new_block();
+	auto wr = bm_->superblock(superblock, v);
+	wipe_shadow_table();
+	return wr;
 }
 
+// FIXME: these explicit try/catches are gross
 template <uint32_t BlockSize>
 typename transaction_manager<BlockSize>::write_ref
 transaction_manager<BlockSize>::new_block()
 {
 	block_address b = sm_->new_block();
-	add_shadow(b);
+	try {
-	return bm_->write_lock_zero(b);
+		add_shadow(b);
+		try {
+			return bm_->write_lock_zero(b);
+		} catch (...) {
+			remove_shadow(b);
+			throw;
+		}
+
+	} catch (...) {
+		sm_->dec(b);
+		throw;
+	}
 }
 
 template <uint32_t BlockSize>
 typename transaction_manager<BlockSize>::write_ref
-transaction_manager<BlockSize>::new_block(block_validator const &v)
+transaction_manager<BlockSize>::new_block(validator v)
 {
 	block_address b = sm_->new_block();
-	add_shadow(b);
+	try {
-	return bm_->write_lock_zero(b, v);
+		add_shadow(b);
+		try {
+			return bm_->write_lock_zero(b, v);
+		} catch (...) {
+			remove_shadow(b);
+			throw;
+		}
+	} catch (...) {
+		sm_->dec(b);
+		throw;
+	}
 }
 
+// FIXME: make exception safe
 template <uint32_t BlockSize>
-typename transaction_manager<BlockSize>::write_ref
+pair<typename transaction_manager<BlockSize>::write_ref, bool>
-transaction_manager<BlockSize>::shadow(block_address orig, bool &inc_children)
+transaction_manager<BlockSize>::shadow(block_address orig)
 {
 	if (is_shadow(orig) &&
-	    sm_->count_possibly_greater_than_one(orig)) {
+	    !sm_->count_possibly_greater_than_one(orig))
-		inc_children = false;
+		return make_pair(bm_->write_lock(orig), false);
-		return bm_->write_lock(orig);
-	}
 
 	auto src = bm_->read_lock(orig);
 	auto dest = bm_->write_lock_zero(sm_->new_block());
-	::memcpy(dest->data_, src->data_, BlockSize);
+	::memcpy(dest.data(), src.data(), BlockSize);
 
 	ref_t count = sm_->get_count(orig);
-	sm_->dec_block(orig);
+	if (count == 0)
-	inc_children = count > 1;
+		throw runtime_error("shadowing free block");
-	add_shadow(dest->location_);
+	sm_->dec(orig);
-	return dest;
+	add_shadow(dest.get_location());
+	return make_pair(dest, count > 1);
 }
 
+// FIXME: duplicate code
 template <uint32_t BlockSize>
-typename transaction_manager<BlockSize>::write_ref
+pair<typename transaction_manager<BlockSize>::write_ref, bool>
-transaction_manager<BlockSize>::shadow(block_address orig, block_validator const &v, bool &inc_children)
+transaction_manager<BlockSize>::shadow(block_address orig, validator v)
 {
 	if (is_shadow(orig) &&
-	    sm_->count_possibly_greater_than_one(orig)) {
+	    sm_->count_possibly_greater_than_one(orig))
-		inc_children = false;
+		return make_pair(bm_->write_lock(orig), false);
-		return bm_->write_lock(orig);
-	}
 
 	auto src = bm_->read_lock(orig, v);
 	auto dest = bm_->write_lock_zero(sm_->new_block(), v);
 	::memcpy(dest->data_, src->data_, BlockSize);
 
 	ref_t count = sm_->get_count(orig);
-	sm_->dec_block(orig);
+	if (count == 0)
-	inc_children = count > 1;
+		throw runtime_error("shadowing free block");
+	sm_->dec(orig);
 	add_shadow(dest->location_);
-	return dest;
+	return make_pair(dest, count > 1);
 }
 
 template <uint32_t BlockSize>
@@ -124,32 +131,11 @@ transaction_manager<BlockSize>::read_lock(block_address b)
 
 template <uint32_t BlockSize>
 typename transaction_manager<BlockSize>::read_ref
-transaction_manager<BlockSize>::read_lock(block_address b, block_validator const &v)
+transaction_manager<BlockSize>::read_lock(block_address b, validator v)
 {
 	return bm_->read_lock(b, v);
 }
 
-template <uint32_t BlockSize>
-void
-transaction_manager<BlockSize>::inc(block_address b)
-{
-	sm_->inc_block(b);
-}
-
-template <uint32_t BlockSize>
-void
-transaction_manager<BlockSize>::dec(block_address b)
-{
-	sm_->dec_block(b);
-}
-
-template <uint32_t BlockSize>
-uint32_t
-transaction_manager<BlockSize>::ref_count(block_address b) const
-{
-	return sm_->get_count(b);
-}
-
 template <uint32_t BlockSize>
 void
 transaction_manager<BlockSize>::add_shadow(block_address b)
@@ -157,9 +143,16 @@ transaction_manager<BlockSize>::add_shadow(block_address b)
 	shadows_.insert(b);
 }
 
+template <uint32_t BlockSize>
+void
+transaction_manager<BlockSize>::remove_shadow(block_address b)
+{
+	shadows_.erase(b);
+}
+
 template <uint32_t BlockSize>
 bool
-transaction_manager<BlockSize>::is_shadow(block_address b)
+transaction_manager<BlockSize>::is_shadow(block_address b) const
 {
 	return shadows_.count(b) > 0;
 }

transaction_manager_t.cc

@@ -0,0 +1,91 @@
+#include "transaction_manager.h"
+#include "core_map.h"
+
+#define BOOST_TEST_MODULE TransactionManagerTests
+#include <boost/test/included/unit_test.hpp>
+
+using namespace std;
+using namespace boost;
+using namespace persistent_data;
+
+//----------------------------------------------------------------
+
+namespace {
+	block_address const NR_BLOCKS = 1024;
+
+	transaction_manager<4096>::ptr
+	create_tm() {
+		block_manager<4096>::ptr bm(new block_manager<4096>("./test.data", NR_BLOCKS));
+		space_map::ptr sm(new core_map(NR_BLOCKS));
+		transaction_manager<4096>::ptr tm(new transaction_manager<4096>(bm, sm));
+		return tm;
+	}
+}
+
+//----------------------------------------------------------------
+
+BOOST_AUTO_TEST_CASE(commit_succeeds)
+{
+	auto tm = create_tm();
+	tm->begin(0);
+}
+
+BOOST_AUTO_TEST_CASE(shadowing)
+{
+	auto tm = create_tm();
+	auto superblock = tm->begin(0);
+
+	auto sm = tm->get_sm();
+	sm->inc(1);
+	auto p = tm->shadow(1);
+	auto b = p.first.get_location();
+	BOOST_CHECK(b != 1);
+	BOOST_CHECK(!p.second);
+	BOOST_CHECK(sm->get_count(1) == 0);
+
+	p = tm->shadow(b);
+	BOOST_CHECK(p.first.get_location() == b);
+	BOOST_CHECK(!p.second);
+
+	sm->inc(b);
+	p = tm->shadow(b);
+	BOOST_CHECK(p.first.get_location() != b);
+	BOOST_CHECK(p.second);
+}
+
+BOOST_AUTO_TEST_CASE(multiple_shadowing)
+{
+	auto tm = create_tm();
+	auto superblock = tm->begin(0);
+
+	auto sm = tm->get_sm();
+	sm->set_count(1, 3);
+
+	auto p = tm->shadow(1);
+	auto b = p.first.get_location();
+	BOOST_CHECK(b != 1);
+	BOOST_CHECK(p.second);
+
+	p = tm->shadow(1);
+	auto b2 = p.first.get_location();
+	BOOST_CHECK(b2 != 1);
+	BOOST_CHECK(b2 != b);
+	BOOST_CHECK(p.second);
+
+	p = tm->shadow(1);
+	auto b3 = p.first.get_location();
+	BOOST_CHECK(b3 != b2);
+	BOOST_CHECK(b3 != b);
+	BOOST_CHECK(b3 != 1);
+	BOOST_CHECK(!p.second);
+}
+
+BOOST_AUTO_TEST_CASE(shadow_free_block_fails)
+{
+	auto tm = create_tm();
+	auto superblock = tm->begin(0);
+	BOOST_CHECK_THROW(tm->shadow(1), runtime_error);
+}
+
+// First shadow in a transaction returns a different block
+// Second shadow in a transaction rtu