RocksDB: Ensure Snapshots are released when no longer needed

Change-Id: I4e9722be036390adc52fa2f365e8215f9fb6126b
Reviewed-on: http://review.couchbase.org/82916
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: jameseh96

engines/ep/src/rocksdb-kvstore/rocksdb-kvstore.cc

@@ -670,6 +670,7 @@ ScanContext* RocksDBKVStore::initScanContext(
         DocumentFilter options,
         ValueFilter valOptions) {
     size_t scanId = scanCounter++;
+    scanSnapshots.emplace(scanId, SnapshotPtr(db->GetSnapshot(), *db));
     // As we cannot efficiently determine how many documents this scan will
     // find, we approximate this value with the seqno difference + 1
     // as scan is supposed to be inclusive at both ends,
@@ -707,7 +708,7 @@ scan_error_t RocksDBKVStore::scan(ScanContext* ctx) {
                                      : GetMetaOnly::No;
 
     rocksdb::ReadOptions snapshotOpts{rocksdb::ReadOptions()};
-    snapshotOpts.snapshot = db->GetSnapshot();
+    snapshotOpts.snapshot = scanSnapshots.at(ctx->scanId).get();
 
     std::unique_ptr<rocksdb::Iterator> it(
             db->NewIterator(snapshotOpts, seqnoFamilyHandle.get()));
@@ -799,3 +800,13 @@ scan_error_t RocksDBKVStore::scan(ScanContext* ctx) {
 
     return scan_success;
 }
+
+void RocksDBKVStore::destroyScanContext(ScanContext* ctx) {
+    // TODO RDB: Might be nice to have the snapshot in the ctx and
+    // release it on destruction
+    auto it = scanSnapshots.find(ctx->scanId);
+    if (it != scanSnapshots.end()) {
+        scanSnapshots.erase(it);
+    }
+    delete ctx;
+}

engines/ep/src/rocksdb-kvstore/rocksdb-kvstore.h

@@ -262,10 +262,7 @@ class RocksDBKVStore : public KVStore {
 
     scan_error_t scan(ScanContext* sctx) override;
 
-    void destroyScanContext(ScanContext* ctx) override {
-        // TODO vmx 2016-10-29: implement
-        delete ctx;
-    }
+    void destroyScanContext(ScanContext* ctx) override;
 
     bool persistCollectionsManifestItem(uint16_t vbid,
                                         const Item& manifestItem) override {
@@ -356,5 +353,17 @@ class RocksDBKVStore : public KVStore {
 
     std::atomic<size_t> scanCounter; // atomic counter for generating scan id
 
+    struct SnapshotDeleter {
+        SnapshotDeleter(rocksdb::DB& db) : db(db) {
+        }
+        void operator()(const rocksdb::Snapshot* s) {
+            db.ReleaseSnapshot(s);
+        }
+        rocksdb::DB& db;
+    };
+    using SnapshotPtr =
+            std::unique_ptr<const rocksdb::Snapshot, SnapshotDeleter>;
+    std::map<size_t, SnapshotPtr> scanSnapshots;
+
     Logger& logger;
 };