Fix a bug that accesses invalid address in iterator cleanup function

Summary: Reported in T11889874. When registering the cleanup function we should copy the option so that we can still access it if ReadOptions is deleted.

Test Plan: Add a unit test to reproduce this bug.

Reviewers: sdong

Reviewed By: sdong

Subscribers: andrewkr, dhruba

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

db/db_impl.cc

@@ -3542,16 +3542,16 @@ bool DBImpl::MCOverlap(ManualCompaction* m, ManualCompaction* m1) {
 namespace {
 struct IterState {
   IterState(DBImpl* _db, InstrumentedMutex* _mu, SuperVersion* _super_version,
-            const ReadOptions* _read_options)
+            bool _background_purge)
       : db(_db),
         mu(_mu),
         super_version(_super_version),
-        read_options(_read_options) {}
+        background_purge(_background_purge) {}
 
   DBImpl* db;
   InstrumentedMutex* mu;
   SuperVersion* super_version;
-  const ReadOptions* read_options;
+  bool background_purge;
 };
 
 static void CleanupIteratorState(void* arg1, void* arg2) {
@@ -3561,8 +3561,6 @@ static void CleanupIteratorState(void* arg1, void* arg2) {
     // Job id == 0 means that this is not our background process, but rather
     // user thread
     JobContext job_context(0);
-    bool background_purge =
-        state->read_options->background_purge_on_iterator_cleanup;
 
     state->mu->Lock();
     state->super_version->Cleanup();
@@ -3571,7 +3569,7 @@ static void CleanupIteratorState(void* arg1, void* arg2) {
 
     delete state->super_version;
     if (job_context.HaveSomethingToDelete()) {
-      if (background_purge) {
+      if (state->background_purge) {
         // PurgeObsoleteFiles here does not delete files. Instead, it adds the
         // files to be deleted to a job queue, and deletes it in a separate
         // background thread.
@@ -3608,7 +3606,8 @@ InternalIterator* DBImpl::NewInternalIterator(const ReadOptions& read_options,
                                        &merge_iter_builder);
   internal_iter = merge_iter_builder.Finish();
   IterState* cleanup =
-      new IterState(this, &mutex_, super_version, &read_options);
+      new IterState(this, &mutex_, super_version,
+                    read_options.background_purge_on_iterator_cleanup);
   internal_iter->RegisterCleanup(CleanupIteratorState, cleanup, nullptr);
 
   return internal_iter;

db/deletefile_test.cc

@@ -279,6 +279,43 @@ TEST_F(DeleteFileTest, BackgroundPurgeTest) {
   CloseDB();
 }
 
+// This test is to reproduce a bug that read invalid ReadOption in iterator
+// cleanup function
+TEST_F(DeleteFileTest, BackgroundPurgeCopyOptions) {
+  std::string first("0"), last("999999");
+  CompactRangeOptions compact_options;
+  compact_options.change_level = true;
+  compact_options.target_level = 2;
+  Slice first_slice(first), last_slice(last);
+
+  // We keep an iterator alive
+  Iterator* itr = 0;
+  CreateTwoLevels();
+  ReadOptions* options = new ReadOptions();
+  options->background_purge_on_iterator_cleanup = true;
+  itr = db_->NewIterator(*options);
+  // ReadOptions is deleted, but iterator cleanup function should not be
+  // affected
+  delete options;
+
+  db_->CompactRange(compact_options, &first_slice, &last_slice);
+  // 3 sst after compaction with live iterator
+  CheckFileTypeCounts(dbname_, 0, 3, 1);
+  delete itr;
+
+  test::SleepingBackgroundTask sleeping_task_after;
+  env_->Schedule(&test::SleepingBackgroundTask::DoSleepTask,
+                 &sleeping_task_after, Env::Priority::HIGH);
+
+  // Make sure all background purges are executed
+  sleeping_task_after.WakeUp();
+  sleeping_task_after.WaitUntilDone();
+  // 1 sst after iterator deletion
+  CheckFileTypeCounts(dbname_, 0, 1, 1);
+
+  CloseDB();
+}
+
 TEST_F(DeleteFileTest, BackgroundPurgeTestMultipleJobs) {
   std::string first("0"), last("999999");
   CompactRangeOptions compact_options;