[array] damage reported if the array is smaller than claimed in the ctr

persistent-data/data-structures/array.h

@@ -146,27 +146,37 @@ namespace persistent_data {
 		struct block_value_visitor {
 			block_value_visitor(array<ValueTraits> const &a, ValueVisitor &vv)
 				: a_(a),
-				  vv_(vv) {
+				  vv_(vv),
+				  highest_index_() {
 			}
 
 			void visit(btree_path const &p,
-				   typename block_traits::value_type const &v) {
+				   typename block_traits::value_type const &block) {
-				a_.visit_value(vv_, p, v);
+				highest_index_ = max<unsigned>(highest_index_,
+							       a_.visit_array_block(vv_, p, block));
+			}
+
+			unsigned get_highest_seen() const {
+				return highest_index_;
 			}
 
 		private:
 			array<ValueTraits> const &a_;
 			ValueVisitor &vv_;
+			unsigned highest_index_;
 		};
 
+		// Returns the highest index visited
 		template <typename ValueVisitor>
-		void visit_value(ValueVisitor &vv,
+		unsigned visit_array_block(ValueVisitor &vv,
-				 btree_path const &p,
+					   btree_path const &p,
-				 typename block_traits::value_type const &v) const {
+					   typename block_traits::value_type const &v) const {
 			rblock rb(tm_->read_lock(v, validator_), rc_);
 
 			for (uint32_t i = 0; i < rb.nr_entries(); i++)
 				vv.visit(p[0] * rb.max_entries() + i, rb.get(i));
+
+			return p[0] * rb.max_entries() + (rb.nr_entries() - 1);
 		}
 
 		template <typename DamageVisitor>
@@ -269,6 +279,13 @@ namespace persistent_data {
 			block_value_visitor<ValueVisitor> bvisitor(*this, value_visitor);
 			block_damage_visitor<DamageVisitor> dvisitor(damage_visitor, entries_per_block_);
 			btree_visit_values(block_tree_, counter, bvisitor, dvisitor);
+
+			// check that all blocks were seen
+			unsigned h = bvisitor.get_highest_seen();
+			if (h != nr_entries_ - 1) {
+				array_detail::damage d(run<unsigned>(h + 1, nr_entries_), "missing blocks");
+				damage_visitor.visit(d);
+			}
 		}
 
 	private:

unit-tests/array_t.cc

@@ -33,66 +33,109 @@ namespace {
 	block_address const NR_BLOCKS = 102400;
 	typedef persistent_data::array<uint64_traits> array64;
 
-	transaction_manager::ptr
+	class ArrayTests : public Test {
-	create_tm() {
+	public:
-		block_manager<>::ptr bm(new block_manager<>("./test.data", NR_BLOCKS, 4, block_io<>::READ_WRITE));
+		ArrayTests()
-		space_map::ptr sm(new core_map(NR_BLOCKS));
+			: tm_(create_tm()) {
-		transaction_manager::ptr tm(new transaction_manager(bm, sm));
+		}
-		return tm;
-	}
 
-	typename array64::ptr
+		void
-	create_array(unsigned nr_entries, uint64_t default_value) {
+		create_array(unsigned nr_entries, uint64_t default_value) {
-		uint64_traits::ref_counter rc;
+			uint64_traits::ref_counter rc;
 
-		typename array64::ptr a(new array64(create_tm(), rc));
+			a_.reset(new array64(tm_, rc));
 
-		if (nr_entries)
+			if (nr_entries)
-			a->grow(nr_entries, default_value);
+				a_->grow(nr_entries, default_value);
+		}
 
-		return a;
+		void
-	}
+		reopen_array() {
+			uint64_traits::ref_counter rc;
+			unsigned nr_entries = a_->get_nr_entries();
+			block_address root = a_->get_root();
+			a_.reset(new array64(tm_, rc, root, nr_entries));
+		}
 
-	typename array64::ptr
+		void
-	open_array(block_address root, unsigned nr_entries) {
+		reopen_array(unsigned nr_entries) {
-		uint64_traits::ref_counter rc;
+			uint64_traits::ref_counter rc;
+			block_address root = a_->get_root();
+			a_.reset(new array64(tm_, rc, root, nr_entries));
+		}
 
-		typename array64::ptr a(new array64(create_tm(), rc, root, nr_entries));
+		uint64_t get(unsigned index) const {
-		return a;
+			return a_->get(index);
-	}
+		}
+
+		void set(unsigned index, uint64_t const &value) {
+			a_->set(index, value);
+		}
+
+		array64::ptr a_;
+
+	private:
+		static transaction_manager::ptr
+		create_tm() {
+			block_manager<>::ptr bm(new block_manager<>("./test.data", NR_BLOCKS, 4, block_io<>::READ_WRITE));
+			space_map::ptr sm(new core_map(NR_BLOCKS));
+			transaction_manager::ptr tm(new transaction_manager(bm, sm));
+			return tm;
+		}
+
+		transaction_manager::ptr tm_;
+	};
+
+	class value_visitor {
+	public:
+		void visit(unsigned index, uint64_t value) {
+			m.insert(make_pair(index, value));
+		}
+
+		std::map<unsigned, uint64_t> m;
+	};
+
+	class damage_visitor {
+	public:
+		void visit(array_detail::damage const &d) {
+			ds.push_back(d);
+		}
+
+		std::list<array_detail::damage> ds;
+	};
 }
 
 //----------------------------------------------------------------
 
-TEST(ArrayTests, can_create_an_empty_array)
+TEST_F(ArrayTests, can_create_an_empty_array)
 {
-	persistent_data::array<uint64_traits>::ptr tree = create_array(0, 0);
+	create_array(0, 0);
-	ASSERT_THROW(tree->get(0), runtime_error);
+	ASSERT_THROW(get(0), runtime_error);
 }
 
-TEST(ArrayTests, get_elements)
+TEST_F(ArrayTests, get_elements)
 {
 	unsigned const COUNT = 10000;
-	persistent_data::array<uint64_traits>::ptr tree = create_array(COUNT, 123);
+	create_array(COUNT, 123);
 
 	for (unsigned i = 0; i < COUNT; i++)
-		ASSERT_THAT(tree->get(i), Eq(123u));
+		ASSERT_THAT(get(i), Eq(123u));
 
-	ASSERT_THROW(tree->get(COUNT), runtime_error);
+	ASSERT_THROW(get(COUNT), runtime_error);
 }
 
-TEST(ArrayTests, set_elements)
+TEST_F(ArrayTests, set_elements)
 {
 	unsigned const COUNT = 10000;
-	persistent_data::array<uint64_traits>::ptr tree = create_array(COUNT, 123);
+	create_array(COUNT, 123);
 
 	for (unsigned i = 0; i < COUNT; i++)
-		tree->set(i, 124);
+		set(i, 124);
 
 	for (unsigned i = 0; i < COUNT; i++)
-		ASSERT_THAT(tree->get(i), Eq(124u));
+		ASSERT_THAT(get(i), Eq(124u));
 
-	ASSERT_THROW(tree->get(COUNT), runtime_error);
+	ASSERT_THROW(get(COUNT), runtime_error);
 }
 
 template <typename T, unsigned size>
@@ -100,7 +143,7 @@ unsigned array_size(T (&)[size]) {
 	return size;
 }
 
-TEST(ArrayTests, grow)
+TEST_F(ArrayTests, grow)
 {
 	unsigned const COUNT = 10000;
 	unsigned const STEPS[] = {
@@ -116,50 +159,77 @@ TEST(ArrayTests, grow)
 			chunks.push_back(c);
 		chunks.push_back(COUNT);
 
-		persistent_data::array<uint64_traits>::ptr a = create_array(0, 123);
+		create_array(0, 123);
 
 		for (unsigned i = 1; i < chunks.size(); i++) {
 			if (i > 1)
-				ASSERT_THAT(a->get(chunks[i - 1] - 1), Eq(i - 1));
+				ASSERT_THAT(get(chunks[i - 1] - 1), Eq(i - 1));
 
-			a->grow(chunks[i], i);
+			a_->grow(chunks[i], i);
 
 			if (i > 1)
-				ASSERT_THAT(a->get(chunks[i - 1] - 1), Eq(i - 1));
+				ASSERT_THAT(get(chunks[i - 1] - 1), Eq(i - 1));
 
 			for (unsigned j = chunks[i - 1]; j < chunks[i]; j++)
-				ASSERT_THAT(a->get(j), Eq(i));
+				ASSERT_THAT(get(j), Eq(i));
 
-			ASSERT_THROW(a->get(chunks[i] + 1), runtime_error);
+			ASSERT_THROW(get(chunks[i] + 1), runtime_error);
 		}
 	}
 }
 
-TEST(ArrayTests, reopen_array)
+TEST_F(ArrayTests, reopen_array)
 {
 	unsigned const COUNT = 10000;
-	block_address root;
 
-	{
+	create_array(COUNT, 123);
-		typename array64::ptr a = create_array(COUNT, 123);
+	for (unsigned i = 0; i < COUNT; i += 7)
+		set(i, 234);
 
-		for (unsigned i = 0; i < COUNT; i += 7)
+	reopen_array();
-			a->set(i, 234);
+	for (unsigned i = 0; i < COUNT; i++)
+		ASSERT_THAT(get(i), Eq(i % 7 ? 123u : 234u));
+}
 
-		root = a->get_root();
+TEST_F(ArrayTests, visit_values)
-	}
+{
+	unsigned const COUNT = 10000;
 
-	{
+	create_array(COUNT, 123);
-		typename array64::ptr a = open_array(root, COUNT);
 
-		for (unsigned i = 0; i < COUNT; i++)
+	value_visitor vv;
-			ASSERT_THAT(a->get(i), Eq(i % 7 ? 123u : 234u));
+	damage_visitor dv;
+	a_->visit_values(vv, dv);
+
+	for (unsigned i = 0; i < COUNT; i++) {
+		unsigned c = vv.m.count(i);
+		uint64_t v = vv.m[i];
+		ASSERT_THAT(c, Eq(1u));
+		ASSERT_THAT(v, Eq(123ull));
 	}
 }
 
-TEST(Array_Tests, destroy)
+TEST_F(ArrayTests, visit_values_with_too_few_entries)
 {
-	// FIXME: pending
+	unsigned const COUNT = 10000;
+	unsigned const EXTRA = 2 * COUNT;
+
+	create_array(COUNT, 123);
+	reopen_array(EXTRA);
+
+	value_visitor vv;
+	damage_visitor dv;
+	a_->visit_values(vv, dv);
+
+	for (unsigned i = 0; i < COUNT; i++) {
+		unsigned c = vv.m.count(i);
+		uint64_t v = vv.m[i];
+		ASSERT_THAT(c, Eq(1u));
+		ASSERT_THAT(v, Eq(123ull));
+	}
+
+	ASSERT_THAT(dv.ds.size(), Eq(1ul));
+	ASSERT_THAT(dv.ds.front().lost_keys_, Eq(run<uint32_t>(COUNT, EXTRA)));
 }
 
 //----------------------------------------------------------------