Don't Finalize in CompactionPicker

Summary:
Finalize re-sorts (read: mutates) the files_ in Version* and it is called by CompactionPicker during normal runtime. At the same time, this same Version* lives in the SuperVersion* and is accessed without the mutex in GetImpl() code path.

Mutating the files_ in one thread and reading the same files_ in another thread is a bad idea. It caused this issue: http://ci-builds.fb.com/job/rocksdb_crashtest/285/console

Long-term, we need to be more careful with method contracts and clearly document what state can be mutated when. Now that we are much faster because we don't lock in GetImpl(), we keep running into data races that were not a problem before when we were slower. db_stress has been very helpful in detecting those.

Short-term, I removed Finalize() from CompactionPicker.

Note: I believe this is an issue in current 2.7 version running in production.

Test Plan:
make check
Will also run db_stress to see if issue is gone

Reviewers: sdong, ljin, dhruba, haobo

Reviewed By: sdong

CC: leveldb

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

db/compaction_picker.cc

@@ -373,12 +373,6 @@ Compaction* LevelCompactionPicker::PickCompaction(Version* version,
   Compaction* c = nullptr;
   int level = -1;
 
-  // Compute the compactions needed. It is better to do it here
-  // and also in LogAndApply(), otherwise the values could be stale.
-  std::vector<uint64_t> size_being_compacted(NumberLevels() - 1);
-  SizeBeingCompacted(size_being_compacted);
-  version->Finalize(size_being_compacted);
-
   // We prefer compactions triggered by too much data in a level over
   // the compactions triggered by seeks.
   //

db/version_set.cc

@@ -461,6 +461,7 @@ Version::Version(VersionSet* vset, uint64_t version_number)
       prev_(this),
       refs_(0),
       num_levels_(vset->num_levels_),
+      finalized_(false),
       files_(new std::vector<FileMetaData*>[num_levels_]),
       files_by_size_(num_levels_),
       next_file_to_compact_by_size_(num_levels_),
@@ -478,6 +479,7 @@ void Version::Get(const ReadOptions& options,
                   GetStats* stats,
                   const Options& db_options,
                   bool* value_found) {
+  assert(finalized_);
   Slice ikey = k.internal_key();
   Slice user_key = k.user_key();
   const Comparator* ucmp = vset_->icmp_.user_comparator();
@@ -642,6 +644,8 @@ bool Version::UpdateStats(const GetStats& stats) {
 }
 
 void Version::Finalize(std::vector<uint64_t>& size_being_compacted) {
+  assert(!finalized_);
+  finalized_ = true;
   // Pre-sort level0 for Get()
   if (vset_->options_->compaction_style == kCompactionStyleUniversal) {
     std::sort(files_[0].begin(), files_[0].end(), NewestFirstBySeqNo);

db/version_set.h

@@ -227,6 +227,7 @@ class Version {
   Version* prev_;               // Previous version in linked list
   int refs_;                    // Number of live refs to this version
   int num_levels_;              // Number of levels
+  bool finalized_;              // True if Finalized is called
 
   // List of files per level, files in each level are arranged
   // in increasing order of keys