os: KeyValueDB: implement snapshot iterators

Create a set of functions, to be implemented by derivative classes of
KeyValueDB, responsible for returning an iterator with strong
read-consistency guarantees. How this iterator is implemented, or by what
is it backed up, is implementation specific, but it must guarantee that
all reads made using this iterator are as if there were no subsequent
writes to the store since we created the iterator.

For instance, LevelDBStore will back this iterator with a leveldb Snapshot,
while KeyValueDBMemory will perform a copy of its in-memory map.

Signed-off-by: Joao Eduardo Luis <joao.luis@inktank.com>

4 files changed, 68 insertions, 0 deletions

diff --git a/src/os/KeyValueDB.h b/src/os/KeyValueDB.h
index 4be188f320c..f62bca996a5 100644
--- a/src/os/KeyValueDB.h
+++ b/src/os/KeyValueDB.h
@@ -155,10 +155,21 @@ public:
     );
   }
 
+  WholeSpaceIterator get_snapshot_iterator() {
+    return _get_snapshot_iterator();
+  }
+
+  Iterator get_snapshot_iterator(const string &prefix) {
+    return std::tr1::shared_ptr<IteratorImpl>(
+      new IteratorImpl(prefix, get_snapshot_iterator())
+    );
+  }
+
   virtual ~KeyValueDB() {}
 
 protected:
   virtual WholeSpaceIterator _get_iterator() = 0;
+  virtual WholeSpaceIterator _get_snapshot_iterator() = 0;
 };
 
 #endif
diff --git a/src/os/LevelDBStore.h b/src/os/LevelDBStore.h
index a24a57fa849..0fd9a11bced 100644
--- a/src/os/LevelDBStore.h
+++ b/src/os/LevelDBStore.h
@@ -154,6 +154,19 @@ public:
     }
   };
 
+  class LevelDBSnapshotIteratorImpl : public LevelDBWholeSpaceIteratorImpl {
+    leveldb::DB *db;
+    const leveldb::Snapshot *snapshot;
+  public:
+    LevelDBSnapshotIteratorImpl(leveldb::DB *db, const leveldb::Snapshot *s,
+				leveldb::Iterator *iter) :
+      LevelDBWholeSpaceIteratorImpl(iter), db(db), snapshot(s) { }
+
+    ~LevelDBSnapshotIteratorImpl() {
+      assert(snapshot != NULL);
+      db->ReleaseSnapshot(snapshot);
+    }
+  };
 
   /// Utility
   static string combine_strings(const string &prefix, const string &value);
@@ -177,6 +190,20 @@ protected:
       )
     );
   }
+
+  WholeSpaceIterator _get_snapshot_iterator() {
+    const leveldb::Snapshot *snapshot;
+    leveldb::ReadOptions options;
+
+    snapshot = db->GetSnapshot();
+    options.snapshot = snapshot;
+
+    return std::tr1::shared_ptr<KeyValueDB::WholeSpaceIteratorImpl>(
+      new LevelDBSnapshotIteratorImpl(db.get(), snapshot,
+	db->NewIterator(options))
+    );
+  }
+
 };
 
 #endif
diff --git a/src/test/ObjectMap/KeyValueDBMemory.cc b/src/test/ObjectMap/KeyValueDBMemory.cc
index 56aba4647ef..87671ccb1a7 100644
--- a/src/test/ObjectMap/KeyValueDBMemory.cc
+++ b/src/test/ObjectMap/KeyValueDBMemory.cc
@@ -218,3 +218,32 @@ KeyValueDB::WholeSpaceIterator KeyValueDBMemory::_get_iterator() {
     new WholeSpaceMemIterator(this)
   );
 }
+
+class WholeSpaceSnapshotMemIterator : public WholeSpaceMemIterator {
+public:
+
+  /**
+   * @note
+   * We perform a copy of the db map, which is populated by bufferlists.
+   *
+   * These are designed as shallow containers, thus there is a chance that
+   * changing the underlying memory pages will lead to the iterator seeing
+   * erroneous states.
+   *
+   * Although we haven't verified this yet, there is this chance, so we should
+   * keep it in mind.
+   */
+
+  WholeSpaceSnapshotMemIterator(KeyValueDBMemory *db) :
+    WholeSpaceMemIterator(db) { }
+  ~WholeSpaceSnapshotMemIterator() {
+    delete db;
+  }
+};
+
+KeyValueDB::WholeSpaceIterator KeyValueDBMemory::_get_snapshot_iterator() {
+  KeyValueDBMemory *snap_db = new KeyValueDBMemory(this);
+  return std::tr1::shared_ptr<KeyValueDB::WholeSpaceIteratorImpl>(
+    new WholeSpaceSnapshotMemIterator(snap_db)
+  );
+}
diff --git a/src/test/ObjectMap/KeyValueDBMemory.h b/src/test/ObjectMap/KeyValueDBMemory.h
index c7bb2878238..baed9de28e0 100644
--- a/src/test/ObjectMap/KeyValueDBMemory.h
+++ b/src/test/ObjectMap/KeyValueDBMemory.h
@@ -138,4 +138,5 @@ private:
 
 protected:
   WholeSpaceIterator _get_iterator();
+  WholeSpaceIterator _get_snapshot_iterator();
 };