preload table handle on Recover() when max_open_files == -1

Summary: This covers existing table files before DB open happens and avoids contention on table cache

Test Plan: db_test

Reviewers: haobo, sdong, igor, dhruba

Reviewed By: haobo

CC: leveldb

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

db/db_impl.cc

@@ -3592,6 +3592,21 @@ void DBImpl::GetLiveFilesMetaData(std::vector<LiveFileMetaData> *metadata) {
   return versions_->GetLiveFilesMetaData(metadata);
 }
 
+void DBImpl::TEST_GetFilesMetaData(
+    std::vector<std::vector<FileMetaData>>* metadata) {
+  MutexLock l(&mutex_);
+  metadata->resize(NumberLevels());
+  for (int level = 0; level < NumberLevels(); level++) {
+    const std::vector<FileMetaData*>& files =
+      versions_->current()->files_[level];
+
+    (*metadata)[level].clear();
+    for (const auto& f : files) {
+      (*metadata)[level].push_back(*f);
+    }
+  }
+}
+
 Status DBImpl::GetDbIdentity(std::string& identity) {
   std::string idfilename = IdentityFileName(dbname_);
   unique_ptr<SequentialFile> idfile;

db/db_impl.h

@@ -140,6 +140,8 @@ class DBImpl : public DB {
     default_interval_to_delete_obsolete_WAL_ = default_interval_to_delete_obsolete_WAL;
   }
 
+  void TEST_GetFilesMetaData(std::vector<std::vector<FileMetaData>>* metadata);
+
   // holds references to memtable, all immutable memtables and version
   struct SuperVersion {
     MemTable* mem;

db/db_test.cc

@@ -1661,6 +1661,42 @@ TEST(DBTest, Recover) {
   } while (ChangeOptions());
 }
 
+TEST(DBTest, RecoverWithTableHandle) {
+  do {
+    Options options = CurrentOptions();
+    options.create_if_missing = true;
+    options.write_buffer_size = 100;
+    options.disable_auto_compactions = true;
+    DestroyAndReopen(&options);
+
+    ASSERT_OK(Put("foo", "v1"));
+    ASSERT_OK(Put("bar", "v2"));
+    dbfull()->TEST_FlushMemTable();
+    ASSERT_OK(Put("foo", "v3"));
+    ASSERT_OK(Put("bar", "v4"));
+    dbfull()->TEST_FlushMemTable();
+    ASSERT_OK(Put("big", std::string(100, 'a')));
+    Reopen();
+
+    std::vector<std::vector<FileMetaData>> files;
+    dbfull()->TEST_GetFilesMetaData(&files);
+    int total_files = 0;
+    for (const auto& level : files) {
+      total_files += level.size();
+    }
+    ASSERT_EQ(total_files, 3);
+    for (const auto& level : files) {
+      for (const auto& file : level) {
+        if (kInfiniteMaxOpenFiles == option_config_) {
+          ASSERT_TRUE(file.table_reader_handle != nullptr);
+        } else {
+          ASSERT_TRUE(file.table_reader_handle == nullptr);
+        }
+      }
+    }
+  } while (ChangeOptions());
+}
+
 TEST(DBTest, IgnoreRecoveredLog) {
   std::string backup_logs = dbname_ + "/backup_logs";
 

db/version_set.cc

@@ -1749,6 +1749,12 @@ Status VersionSet::Recover() {
   }
 
   if (s.ok()) {
+    if (options_->max_open_files == -1) {
+      // unlimited table cache. Pre-load table handle now.
+      // Need to do it out of the mutex.
+      builder.LoadTableHandlers();
+    }
+
     Version* v = new Version(this, current_version_number_++);
     builder.SaveTo(v);