fixed leaking log::Writers

Summary: Fixes valgrind errors in column_family_test.

Test Plan: `make check`, `make valgrind_check`

Reviewers: igor, yhchiang

Reviewed By: yhchiang

Subscribers: dhruba

Differential Revision: https://reviews.facebook.net/D41181
(cherry picked from commit 4ce5be4255)

db/column_family.cc

@@ -59,6 +59,7 @@ ColumnFamilyHandleImpl::~ColumnFamilyHandleImpl() {
     if (job_context.HaveSomethingToDelete()) {
       db_->PurgeObsoleteFiles(job_context);
     }
+    job_context.Clean();
   }
 }
 

db/db_impl.cc

@@ -542,6 +542,7 @@ void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force,
   }
 
   // We're just cleaning up for DB::Write().
+  assert(job_context->logs_to_free.empty());
   job_context->logs_to_free = logs_to_free_;
   logs_to_free_.clear();
 

db/db_test.cc

@@ -12670,6 +12670,7 @@ TEST_F(DBTest, DontDeletePendingOutputs) {
     dbfull()->FindObsoleteFiles(&job_context, true /*force*/);
     dbfull()->TEST_UnlockMutex();
     dbfull()->PurgeObsoleteFiles(job_context);
+    job_context.Clean();
   };
 
   env_->table_write_callback_ = &purge_obsolete_files_function;

db/forward_iterator.cc

@@ -169,6 +169,7 @@ void ForwardIterator::Cleanup(bool release_sv) {
       if (job_context.HaveSomethingToDelete()) {
         db_->PurgeObsoleteFiles(job_context);
       }
+      job_context.Clean();
     }
   }
 }

db/job_context.h

@@ -83,6 +83,10 @@ struct JobContext {
     new_superversion = create_superversion ? new SuperVersion() : nullptr;
   }
 
+  // For non-empty JobContext Clean() has to be called at least once before
+  // before destruction (see asserts in ~JobContext()). Should be called with
+  // unlocked DB mutex. Destructor doesn't call Clean() to avoid accidentally
+  // doing potentially slow Clean() with locked DB mutex.
   void Clean() {
     // free pending memtables
     for (auto m : memtables_to_free) {
@@ -109,6 +113,7 @@ struct JobContext {
     assert(memtables_to_free.size() == 0);
     assert(superversions_to_free.size() == 0);
     assert(new_superversion == nullptr);
+    assert(logs_to_free.size() == 0);
   }
 };