Explicitly clean JobContext

Summary: This way we can gurantee that old MemTables get destructed before DBImpl gets destructed, which might be useful if we want to make them depend on state from DBImpl.

Test Plan: make check with asserts in JobContext's destructor

Reviewers: ljin, sdong, yhchiang, rven, jonahcohen

Reviewed By: jonahcohen

Subscribers: dhruba, leveldb

Differential Revision: https://reviews.facebook.net/D28959

db/db_filesnapshot.cc

@@ -19,6 +19,7 @@
 #include <stdint.h>
 #include "db/db_impl.h"
 #include "db/filename.h"
+#include "db/job_context.h"
 #include "db/version_set.h"
 #include "rocksdb/db.h"
 #include "rocksdb/env.h"

db/db_impl.cc

@@ -32,6 +32,7 @@
 #include "db/db_iter.h"
 #include "db/dbformat.h"
 #include "db/filename.h"
+#include "db/job_context.h"
 #include "db/log_reader.h"
 #include "db/log_writer.h"
 #include "db/memtable.h"
@@ -287,6 +288,7 @@ DBImpl::~DBImpl() {
     if (job_context.HaveSomethingToDelete()) {
       PurgeObsoleteFiles(job_context);
     }
+    job_context.Clean();
   }
 
   // versions need to be destroyed before table_cache since it can hold
@@ -666,6 +668,7 @@ void DBImpl::DeleteObsoleteFiles() {
   if (job_context.HaveSomethingToDelete()) {
     PurgeObsoleteFiles(job_context);
   }
+  job_context.Clean();
 }
 
 Status DBImpl::Recover(
@@ -1343,6 +1346,7 @@ Status DBImpl::CompactFilesImpl(
     if (job_context.HaveSomethingToDelete()) {
       PurgeObsoleteFiles(job_context);
     }
+    job_context.Clean();
     mutex_.Lock();
   }
 
@@ -1808,6 +1812,7 @@ void DBImpl::BackgroundCallFlush() {
       if (job_context.HaveSomethingToDelete()) {
         PurgeObsoleteFiles(job_context);
       }
+      job_context.Clean();
       mutex_.Lock();
     }
 
@@ -1884,6 +1889,7 @@ void DBImpl::BackgroundCallCompaction() {
       if (job_context.HaveSomethingToDelete()) {
         PurgeObsoleteFiles(job_context);
       }
+      job_context.Clean();
       mutex_.Lock();
     }
 
@@ -2190,6 +2196,7 @@ static void CleanupIteratorState(void* arg1, void* arg2) {
     if (job_context.HaveSomethingToDelete()) {
       state->db->PurgeObsoleteFiles(job_context);
     }
+    job_context.Clean();
   }
 
   delete state;
@@ -3306,6 +3313,7 @@ Status DBImpl::DeleteFile(std::string name) {
   if (job_context.HaveSomethingToDelete()) {
     PurgeObsoleteFiles(job_context);
   }
+  job_context.Clean();
   {
     MutexLock l(&mutex_);
     // schedule flush if file deletion means we freed the space for flushes to

db/db_impl.h

@@ -38,7 +38,6 @@
 #include "db/write_controller.h"
 #include "db/flush_scheduler.h"
 #include "db/write_thread.h"
-#include "db/job_context.h"
 
 namespace rocksdb {
 
@@ -49,6 +48,7 @@ class VersionEdit;
 class VersionSet;
 class CompactionFilterV2;
 class Arena;
+struct JobContext;
 
 class DBImpl : public DB {
  public:

db/db_test.cc

@@ -19,6 +19,7 @@
 #include "db/dbformat.h"
 #include "db/db_impl.h"
 #include "db/filename.h"
+#include "db/job_context.h"
 #include "db/version_set.h"
 #include "db/write_batch_internal.h"
 #include "rocksdb/cache.h"

db/job_context.h

@@ -21,7 +21,8 @@ class MemTable;
 struct JobContext {
   inline bool HaveSomethingToDelete() const {
     return candidate_files.size() || sst_delete_files.size() ||
-           log_delete_files.size();
+           log_delete_files.size() || new_superversion != nullptr ||
+           superversions_to_free.size() > 0 || memtables_to_free.size() > 0;
   }
 
   // Structure to store information for candidate files to delete.
@@ -73,7 +74,7 @@ struct JobContext {
     new_superversion = create_superversion ? new SuperVersion() : nullptr;
   }
 
-  ~JobContext() {
+  void Clean() {
     // free pending memtables
     for (auto m : memtables_to_free) {
       delete m;
@@ -85,6 +86,16 @@ struct JobContext {
     // if new_superversion was not used, it will be non-nullptr and needs
     // to be freed here
     delete new_superversion;
+
+    memtables_to_free.clear();
+    superversions_to_free.clear();
+    new_superversion = nullptr;
+  }
+
+  ~JobContext() {
+    assert(memtables_to_free.size() == 0);
+    assert(superversions_to_free.size() == 0);
+    assert(new_superversion == nullptr);
   }
 };