Bump up version to 7.5.4

Summary:
During compaction, blobs are currently read using the default
`ReadOptions`, which has the `fill_cache` flag set to true. Earlier,
this didn't make any difference since we didn't have a blob cache;
however, now we have to explicitly set this flag to false to avoid
polluting the cache during compaction.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/10457

Test Plan: `make check`

Reviewed By: riversand963

Differential Revision: D38333528

Pulled By: ltamasi

fbshipit-source-id: 5b4d49a1e39543bee73c7df2aa9194fb101875e2

HISTORY.md

@@ -1,5 +1,17 @@
 # Rocksdb Change Log
-## Unreleased
+## 7.5.4 (11/03/2022)
+* Fixed a bug where blobs read during compaction would pollute the cache.
+
+## 7.5.3 (08/24/2022)
+* Fixed a regression where L0's compaction score could become very large.
+
+## 7.5.2 (08/02/2022)
+### Bug Fixes
+* Fix a bug starting in 7.4.0 in which some fsync operations might be skipped in a DB after any DropColumnFamily on that DB, until it is re-opened. This can lead to data loss on power loss. (For custom FileSystem implementations, this could lead to `FSDirectory::Fsync` or `FSDirectory::Close` after the first `FSDirectory::Close`; Also, valgrind could report call to `close()` with `fd=-1`.)
+
+## 7.5.1 (08/01/2022)
+### Bug Fixes
+* Fix a bug where rate_limiter_parameter is not passed into `PartitionedFilterBlockReader::GetFilterPartitionBlock`.
 
 ## 7.5.0 (07/15/2022)
 ### New Features
@@ -22,12 +34,14 @@
 * When using block cache strict capacity limit (`LRUCache` with `strict_capacity_limit=true`), DB operations now fail with Status code `kAborted` subcode `kMemoryLimit` (`IsMemoryLimit()`) instead of `kIncomplete` (`IsIncomplete()`) when the capacity limit is reached, because Incomplete can mean other specific things for some operations. In more detail, `Cache::Insert()` now returns the updated Status code and this usually propagates through RocksDB to the user on failure.
 * NewClockCache calls temporarily return an LRUCache (with similar characteristics as the desired ClockCache). This is because ClockCache is being replaced by a new version (the old one had unknown bugs) but this is still under development.
 * Add two functions `int ReserveThreads(int threads_to_be_reserved)` and `int ReleaseThreads(threads_to_be_released)` into `Env` class. In the default implementation, both return 0. Newly added `xxxEnv` class that inherits `Env` should implement these two functions for thread reservation/releasing features.
+* Removed Customizable support for RateLimiter and removed its CreateFromString() and Type() functions.
 
 ### Bug Fixes
 * Fix a bug in which backup/checkpoint can include a WAL deleted by RocksDB.
 * Fix a bug where concurrent compactions might cause unnecessary further write stalling. In some cases, this might cause write rate to drop to minimum.
 * Fix a bug in Logger where if dbname and db_log_dir are on different filesystems, dbname creation would fail wrt to db_log_dir path returning an error and fails to open the DB.
-* Fix a CPU and memory efficiency issue introduce by https://github.com/facebook/rocksdb/pull/8336 which made InternalKeyComparator configurable as an unintended side effect.
+* Fix a CPU and memory efficiency issue introduce by https://github.com/facebook/rocksdb/pull/8336 which made InternalKeyComparator configurable as an unintended side effect
+* Fix a bug where `GenericRateLimiter` could revert the bandwidth set dynamically using `SetBytesPerSecond()` when a user configures a structure enclosing it, e.g., using `GetOptionsFromString()` to configure an `Options` that references an existing `RateLimiter` object.
 
 ## Behavior Change
 * In leveled compaction with dynamic levelling, level multiplier is not anymore adjusted due to oversized L0. Instead, compaction score is adjusted by increasing size level target by adding incoming bytes from upper levels. This would deprioritize compactions from upper levels if more data from L0 is coming. This is to fix some unnecessary full stalling due to drastic change of level targets, while not wasting write bandwidth for compaction while writes are overloaded.

db/blob/db_blob_compaction_test.cc

@@ -798,6 +798,42 @@ TEST_F(DBBlobCompactionTest, CompactionReadaheadMerge) {
   Close();
 }
 
+TEST_F(DBBlobCompactionTest, CompactionDoNotFillCache) {
+  Options options = GetDefaultOptions();
+
+  options.enable_blob_files = true;
+  options.min_blob_size = 0;
+  options.enable_blob_garbage_collection = true;
+  options.blob_garbage_collection_age_cutoff = 1.0;
+  options.disable_auto_compactions = true;
+  options.statistics = CreateDBStatistics();
+
+  LRUCacheOptions cache_options;
+  cache_options.capacity = 1 << 20;
+  cache_options.metadata_charge_policy = kDontChargeCacheMetadata;
+
+  options.blob_cache = NewLRUCache(cache_options);
+
+  Reopen(options);
+
+  ASSERT_OK(Put("key", "lime"));
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_OK(Flush());
+
+  ASSERT_OK(Put("key", "pie"));
+  ASSERT_OK(Put("foo", "baz"));
+  ASSERT_OK(Flush());
+
+  constexpr Slice* begin = nullptr;
+  constexpr Slice* end = nullptr;
+
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), begin, end));
+
+  ASSERT_EQ(options.statistics->getTickerCount(BLOB_DB_CACHE_ADD), 0);
+
+  Close();
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

db/column_family.cc

@@ -699,19 +699,6 @@ ColumnFamilyData::~ColumnFamilyData() {
           id_, name_.c_str());
     }
   }
-
-  if (data_dirs_.size()) {  // Explicitly close data directories
-    Status s = Status::OK();
-    for (auto& data_dir_ptr : data_dirs_) {
-      if (data_dir_ptr) {
-        s = data_dir_ptr->Close(IOOptions(), nullptr);
-        if (!s.ok()) {
-          // TODO(zichen): add `Status Close()` and `CloseDirectories()
-          s.PermitUncheckedError();
-        }
-      }
-    }
-  }
 }
 
 bool ColumnFamilyData::UnrefAndTryDelete() {

db/compaction/compaction_iterator.cc

@@ -1284,7 +1284,10 @@ std::unique_ptr<BlobFetcher> CompactionIterator::CreateBlobFetcherIfNeeded(
     return nullptr;
   }
 
-  return std::unique_ptr<BlobFetcher>(new BlobFetcher(version, ReadOptions()));
+  ReadOptions read_options;
+  read_options.fill_cache = false;
+
+  return std::unique_ptr<BlobFetcher>(new BlobFetcher(version, read_options));
 }
 
 std::unique_ptr<PrefetchBufferCollection>

db/db_basic_test.cc

@@ -1209,9 +1209,9 @@ TEST_F(DBBasicTest, DBCloseAllDirectoryFDs) {
   s = db->Close();
   auto* counted_fs =
       options.env->GetFileSystem()->CheckedCast<CountedFileSystem>();
-  assert(counted_fs);
-  ASSERT_TRUE(counted_fs->counters()->dir_opens ==
-              counted_fs->counters()->dir_closes);
+  ASSERT_TRUE(counted_fs != nullptr);
+  ASSERT_EQ(counted_fs->counters()->dir_opens,
+            counted_fs->counters()->dir_closes);
   ASSERT_OK(s);
   delete db;
 }

db/db_impl/db_impl.cc

@@ -823,16 +823,29 @@ Status DBImpl::RegisterRecordSeqnoTimeWorker() {
         min_time_duration / SeqnoToTimeMapping::kMaxSeqnoTimePairsPerCF;
   }
 
-  Status s = periodic_work_scheduler_->RegisterRecordSeqnoTimeWorker(
-      this, seqno_time_cadence);
-  if (s.IsNotSupported()) {
-    // TODO: Fix the timer cannot cancel and re-add the same task
-    ROCKS_LOG_WARN(
-        immutable_db_options_.info_log,
-        "Updating seqno to time worker cadence is not supported yet, to make "
-        "the change effective, please reopen the DB instance.");
-    s = Status::OK();
+  Status s;
+  if (seqno_time_cadence != record_seqno_time_cadence_) {
+    if (seqno_time_cadence == 0) {
+      periodic_work_scheduler_->UnregisterRecordSeqnoTimeWorker(this);
+    } else {
+      s = periodic_work_scheduler_->RegisterRecordSeqnoTimeWorker(
+          this, seqno_time_cadence);
+    }
+
+    if (s.ok()) {
+      record_seqno_time_cadence_ = seqno_time_cadence;
+    }
+
+    if (s.IsNotSupported()) {
+      // TODO: Fix the timer cannot cancel and re-add the same task
+      ROCKS_LOG_WARN(
+          immutable_db_options_.info_log,
+          "Updating seqno to time worker cadence is not supported yet, to make "
+          "the change effective, please reopen the DB instance.");
+      s = Status::OK();
+    }
   }
+
   return s;
 #else
   return Status::OK();

db/db_impl/db_impl.h

@@ -118,7 +118,6 @@ class Directories {
   IOStatus Close(const IOOptions& options, IODebugContext* dbg) {
     // close all directories for all database paths
     IOStatus s = IOStatus::OK();
-    IOStatus temp_s = IOStatus::OK();
 
     // The default implementation for Close() in Directory/FSDirectory class
     // "NotSupported" status, the upper level interface should be able to
@@ -127,53 +126,35 @@ class Directories {
     // `FSDirectory::Close()` yet
 
     if (db_dir_) {
-      temp_s = db_dir_->Close(options, dbg);
-      if (!temp_s.ok()) {
-        if (temp_s.IsNotSupported()) {
-          temp_s.PermitUncheckedError();
-        } else {
-          s = temp_s;
-        }
+      IOStatus temp_s = db_dir_->Close(options, dbg);
+      if (!temp_s.ok() && !temp_s.IsNotSupported() && s.ok()) {
+        s = std::move(temp_s);
       }
     }
 
-    if (!s.ok()) {
-      return s;
-    }
+    // Attempt to close everything even if one fails
+    s.PermitUncheckedError();
 
     if (wal_dir_) {
-      s = wal_dir_->Close(options, dbg);
-      if (!temp_s.ok()) {
-        if (temp_s.IsNotSupported()) {
-          temp_s.PermitUncheckedError();
-        } else {
-          s = temp_s;
+      IOStatus temp_s = wal_dir_->Close(options, dbg);
+      if (!temp_s.ok() && !temp_s.IsNotSupported() && s.ok()) {
+        s = std::move(temp_s);
+      }
+    }
+
+    s.PermitUncheckedError();
+
+    for (auto& data_dir_ptr : data_dirs_) {
+      if (data_dir_ptr) {
+        IOStatus temp_s = data_dir_ptr->Close(options, dbg);
+        if (!temp_s.ok() && !temp_s.IsNotSupported() && s.ok()) {
+          s = std::move(temp_s);
         }
       }
     }
 
-    if (!s.ok()) {
-      return s;
-    }
-
-    if (data_dirs_.size() > 0 && s.ok()) {
-      for (auto& data_dir_ptr : data_dirs_) {
-        if (data_dir_ptr) {
-          temp_s = data_dir_ptr->Close(options, dbg);
-          if (!temp_s.ok()) {
-            if (temp_s.IsNotSupported()) {
-              temp_s.PermitUncheckedError();
-            } else {
-              return temp_s;
-            }
-          }
-        }
-      }
-    }
-
-    // Mark temp_s as checked when temp_s is still the initial status
-    // (IOStatus::OK(), not checked yet)
-    temp_s.PermitUncheckedError();
+    // Ready for caller
+    s.MustCheck();
     return s;
   }
 
@@ -1160,7 +1141,7 @@ class DBImpl : public DB {
   int TEST_BGFlushesAllowed() const;
   size_t TEST_GetWalPreallocateBlockSize(uint64_t write_buffer_size) const;
   void TEST_WaitForPeridicWorkerRun(std::function<void()> callback) const;
-  const SeqnoToTimeMapping& TEST_GetSeqnoToTimeMapping() const;
+  SeqnoToTimeMapping TEST_GetSeqnoToTimeMapping() const;
   size_t TEST_EstimateInMemoryStatsHistorySize() const;
 
   uint64_t TEST_GetCurrentLogNumber() const {
@@ -2544,6 +2525,10 @@ class DBImpl : public DB {
   // PeriodicWorkScheduler::Default(). Only in unittest, it can be overrided by
   // PeriodicWorkTestScheduler.
   PeriodicWorkScheduler* periodic_work_scheduler_;
+
+  // Current cadence of the periodic worker for recording sequence number to
+  // time.
+  uint64_t record_seqno_time_cadence_ = 0;
 #endif
 
   // When set, we use a separate queue for writes that don't write to memtable.

db/db_impl/db_impl_debug.cc

@@ -314,7 +314,8 @@ PeriodicWorkTestScheduler* DBImpl::TEST_GetPeriodicWorkScheduler() const {
   return static_cast<PeriodicWorkTestScheduler*>(periodic_work_scheduler_);
 }
 
-const SeqnoToTimeMapping& DBImpl::TEST_GetSeqnoToTimeMapping() const {
+SeqnoToTimeMapping DBImpl::TEST_GetSeqnoToTimeMapping() const {
+  InstrumentedMutexLock l(&mutex_);
   return seqno_time_mapping_;
 }
 

db/db_impl/db_impl_open.cc

@@ -538,6 +538,7 @@ Status DBImpl::Recover(
     s = CheckConsistency();
   }
   if (s.ok() && !read_only) {
+    // TODO: share file descriptors (FSDirectory) with SetDirectories above
     std::map<std::string, std::shared_ptr<FSDirectory>> created_dirs;
     for (auto cfd : *versions_->GetColumnFamilySet()) {
       s = cfd->AddDirectories(&created_dirs);

db/db_test.cc

@@ -4101,9 +4101,6 @@ class MockedRateLimiterWithNoOptionalAPIImpl : public RateLimiter {
 
   ~MockedRateLimiterWithNoOptionalAPIImpl() override {}
 
-  const char* Name() const override {
-    return "MockedRateLimiterWithNoOptionalAPI";
-  }
   void SetBytesPerSecond(int64_t bytes_per_second) override {
     (void)bytes_per_second;
   }

db/periodic_work_scheduler.cc

@@ -64,14 +64,6 @@ Status PeriodicWorkScheduler::Register(DBImpl* dbi,
 Status PeriodicWorkScheduler::RegisterRecordSeqnoTimeWorker(
     DBImpl* dbi, uint64_t record_cadence_sec) {
   MutexLock l(&timer_mu_);
-  if (record_seqno_time_cadence_ == record_cadence_sec) {
-    return Status::OK();
-  }
-  if (record_cadence_sec == 0) {
-    timer->Cancel(GetTaskName(dbi, PeriodicWorkTaskNames::kRecordSeqnoTime));
-    record_seqno_time_cadence_ = record_cadence_sec;
-    return Status::OK();
-  }
   timer->Start();
   static std::atomic_uint64_t initial_delay(0);
   bool succeeded = timer->Add(
@@ -83,7 +75,6 @@ Status PeriodicWorkScheduler::RegisterRecordSeqnoTimeWorker(
     return Status::NotSupported(
         "Updating seqno to time worker cadence is not supported yet");
   }
-  record_seqno_time_cadence_ = record_cadence_sec;
   return Status::OK();
 }
 
@@ -93,7 +84,6 @@ void PeriodicWorkScheduler::UnregisterRecordSeqnoTimeWorker(DBImpl* dbi) {
   if (!timer->HasPendingTask()) {
     timer->Shutdown();
   }
-  record_seqno_time_cadence_ = 0;
 }
 
 void PeriodicWorkScheduler::Unregister(DBImpl* dbi) {

db/periodic_work_scheduler.h

@@ -57,9 +57,6 @@ class PeriodicWorkScheduler {
   // Get the unique task name (prefix with db session id)
   std::string GetTaskName(const DBImpl* dbi,
                           const std::string& func_name) const;
-
- private:
-  uint64_t record_seqno_time_cadence_ = 0;
 };
 
 #ifndef NDEBUG

db/seqno_time_test.cc

@@ -639,6 +639,38 @@ TEST_F(SeqnoTimeTest, DISABLED_MultiCFs) {
   Close();
 }
 
+TEST_F(SeqnoTimeTest, MultiInstancesBasic) {
+  const int kInstanceNum = 2;
+
+  Options options = CurrentOptions();
+  options.preclude_last_level_data_seconds = 10000;
+  options.env = mock_env_.get();
+  options.stats_dump_period_sec = 0;
+  options.stats_persist_period_sec = 0;
+
+  auto dbs = std::vector<DB*>(kInstanceNum);
+  for (int i = 0; i < kInstanceNum; i++) {
+    ASSERT_OK(
+        DB::Open(options, test::PerThreadDBPath(std::to_string(i)), &(dbs[i])));
+  }
+
+  // Make sure the second instance has the worker enabled
+  auto dbi = static_cast_with_check<DBImpl>(dbs[1]);
+  WriteOptions wo;
+  for (int i = 0; i < 200; i++) {
+    ASSERT_OK(dbi->Put(wo, Key(i), "value"));
+    dbfull()->TEST_WaitForPeridicWorkerRun(
+        [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(100)); });
+  }
+  SeqnoToTimeMapping seqno_to_time_mapping = dbi->TEST_GetSeqnoToTimeMapping();
+  ASSERT_GT(seqno_to_time_mapping.Size(), 10);
+
+  for (int i = 0; i < kInstanceNum; i++) {
+    ASSERT_OK(dbs[i]->Close());
+    delete dbs[i];
+  }
+}
+
 TEST_F(SeqnoTimeTest, SeqnoToTimeMappingUniversal) {
   Options options = CurrentOptions();
   options.compaction_style = kCompactionStyleUniversal;

db/version_set.cc

@@ -2733,24 +2733,43 @@ void VersionStorageInfo::ComputeCompactionScore(
           // Level-based involves L0->L0 compactions that can lead to oversized
           // L0 files. Take into account size as well to avoid later giant
           // compactions to the base level.
-          uint64_t l0_target_size = mutable_cf_options.max_bytes_for_level_base;
-          if (immutable_options.level_compaction_dynamic_level_bytes &&
-              level_multiplier_ != 0.0) {
-            // Prevent L0 to Lbase fanout from growing larger than
-            // `level_multiplier_`. This prevents us from getting stuck picking
-            // L0 forever even when it is hurting write-amp. That could happen
-            // in dynamic level compaction's write-burst mode where the base
-            // level's target size can grow to be enormous.
-            l0_target_size =
-                std::max(l0_target_size,
-                         static_cast<uint64_t>(level_max_bytes_[base_level_] /
-                                               level_multiplier_));
-          }
-          score =
-              std::max(score, static_cast<double>(total_size) / l0_target_size);
-          if (immutable_options.level_compaction_dynamic_level_bytes &&
-              score > 1.0) {
-            score *= kScoreScale;
+          // If score in L0 is always too high, L0->L1 will always be
+          // prioritized over L1->L2 compaction and L1 will accumulate to
+          // too large. But if L0 score isn't high enough, L0 will accumulate
+          // and data is not moved to L1 fast enough. With potential L0->L0
+          // compaction, number of L0 files aren't always an indication of
+          // L0 oversizing, and we also need to consider total size of L0.
+          if (immutable_options.level_compaction_dynamic_level_bytes) {
+            if (total_size >= mutable_cf_options.max_bytes_for_level_base) {
+              // When calculating estimated_compaction_needed_bytes, we assume
+              // L0 is qualified as pending compactions. We will need to make
+              // sure that it qualifies for compaction.
+              // It might be guafanteed by logic below anyway, but we are
+              // explicit here to make sure we don't stop writes with no
+              // compaction scheduled.
+              score = std::max(score, 1.01);
+            }
+            if (total_size > level_max_bytes_[base_level_]) {
+              // In this case, we compare L0 size with actual L1 size and make
+              // sure score is more than 1.0 (10.0 after scaled) if L0 is larger
+              // than L1. Since in this case L1 score is lower than 10.0, L0->L1
+              // is prioritized over L1->L2.
+              uint64_t base_level_size = 0;
+              for (auto f : files_[base_level_]) {
+                base_level_size += f->compensated_file_size;
+              }
+              score = std::max(score, static_cast<double>(total_size) /
+                                          static_cast<double>(std::max(
+                                              base_level_size,
+                                              level_max_bytes_[base_level_])));
+            }
+            if (score > 1.0) {
+              score *= kScoreScale;
+            }
+          } else {
+            score = std::max(score,
+                             static_cast<double>(total_size) /
+                                 mutable_cf_options.max_bytes_for_level_base);
           }
         }
       }

env/env_test.cc

@@ -1401,7 +1401,7 @@ TEST_P(EnvPosixTestWithParam, MultiRead) {
           }
         });
 
-    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
     std::unique_ptr<RandomAccessFile> file;
     std::vector<ReadRequest> reqs(3);
     std::vector<std::unique_ptr<char, Deleter>> data;
@@ -1522,6 +1522,79 @@ TEST_F(EnvPosixTest, MultiReadNonAlignedLargeNum) {
   }
 }
 
+#ifndef ROCKSDB_LITE
+TEST_F(EnvPosixTest, NonAlignedDirectIOMultiReadBeyondFileSize) {
+  EnvOptions soptions;
+  soptions.use_direct_reads = true;
+  soptions.use_direct_writes = false;
+  std::string fname = test::PerThreadDBPath(env_, "testfile");
+
+  Random rnd(301);
+  std::unique_ptr<WritableFile> wfile;
+  size_t alignment = 0;
+  // Create file.
+  {
+    ASSERT_OK(env_->NewWritableFile(fname, &wfile, soptions));
+    auto data_ptr = NewAligned(4095, 'b');
+    Slice data_b(data_ptr.get(), 4095);
+    ASSERT_OK(wfile->PositionedAppend(data_b, 0U));
+    ASSERT_OK(wfile->Close());
+  }
+
+#if !defined(OS_MACOSX) && !defined(OS_WIN) && !defined(OS_SOLARIS) && \
+    !defined(OS_AIX) && !defined(OS_OPENBSD) && !defined(OS_FREEBSD)
+  if (soptions.use_direct_reads) {
+    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+        "NewRandomAccessFile:O_DIRECT", [&](void* arg) {
+          int* val = static_cast<int*>(arg);
+          *val &= ~O_DIRECT;
+        });
+  }
+#endif
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  const int num_reads = 2;
+  // Create requests
+  std::vector<std::string> scratches;
+  scratches.reserve(num_reads);
+  std::vector<ReadRequest> reqs(num_reads);
+
+  std::unique_ptr<RandomAccessFile> file;
+  ASSERT_OK(env_->NewRandomAccessFile(fname, &file, soptions));
+  alignment = file->GetRequiredBufferAlignment();
+  ASSERT_EQ(num_reads, reqs.size());
+
+  std::vector<std::unique_ptr<char, Deleter>> data;
+
+  std::vector<size_t> offsets = {0, 2047};
+  std::vector<size_t> lens = {2047, 4096 - 2047};
+
+  for (size_t i = 0; i < num_reads; i++) {
+    // Do alignment
+    reqs[i].offset = static_cast<uint64_t>(TruncateToPageBoundary(
+        alignment, static_cast<size_t>(/*offset=*/offsets[i])));
+    reqs[i].len =
+        Roundup(static_cast<size_t>(/*offset=*/offsets[i]) + /*length=*/lens[i],
+                alignment) -
+        reqs[i].offset;
+
+    size_t new_capacity = Roundup(reqs[i].len, alignment);
+    data.emplace_back(NewAligned(new_capacity, 0));
+    reqs[i].scratch = data.back().get();
+  }
+
+  // Query the data
+  ASSERT_OK(file->MultiRead(reqs.data(), reqs.size()));
+
+  // Validate results
+  for (size_t i = 0; i < num_reads; ++i) {
+    ASSERT_OK(reqs[i].status);
+  }
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+}
+#endif  // ROCKSDB_LITE
+
 #if defined(ROCKSDB_IOURING_PRESENT)
 void GenerateFilesAndRequest(Env* env, const std::string& fname,
                              std::vector<ReadRequest>* ret_reqs,

env/io_posix.cc

@@ -750,14 +750,17 @@ IOStatus PosixRandomAccessFile::MultiRead(FSReadRequest* reqs,
                    bytes_read, read_again);
       int32_t res = cqe->res;
       if (res >= 0) {
-        if (bytes_read == 0 && read_again) {
-          Slice tmp_slice;
-          req->status =
-              Read(req->offset + req_wrap->finished_len,
-                   req->len - req_wrap->finished_len, options, &tmp_slice,
-                   req->scratch + req_wrap->finished_len, dbg);
-          req->result =
-              Slice(req->scratch, req_wrap->finished_len + tmp_slice.size());
+        if (bytes_read == 0) {
+          if (read_again) {
+            Slice tmp_slice;
+            req->status =
+                Read(req->offset + req_wrap->finished_len,
+                     req->len - req_wrap->finished_len, options, &tmp_slice,
+                     req->scratch + req_wrap->finished_len, dbg);
+            req->result =
+                Slice(req->scratch, req_wrap->finished_len + tmp_slice.size());
+          }
+          // else It means EOF so no need to do anything.
         } else if (bytes_read < req_wrap->iov.iov_len) {
           incomplete_rq_list.push_back(req_wrap);
         }
@@ -1681,6 +1684,7 @@ IOStatus PosixDirectory::Close(const IOOptions& /*opts*/,
 IOStatus PosixDirectory::FsyncWithDirOptions(
     const IOOptions& /*opts*/, IODebugContext* /*dbg*/,
     const DirFsyncOptions& dir_fsync_options) {
+  assert(fd_ >= 0);  // Check use after close
   IOStatus s = IOStatus::OK();
 #ifndef OS_AIX
   if (is_btrfs_) {

include/rocksdb/rate_limiter.h

@@ -9,7 +9,6 @@
 
 #pragma once
 
-#include "rocksdb/customizable.h"
 #include "rocksdb/env.h"
 #include "rocksdb/statistics.h"
 #include "rocksdb/status.h"
@@ -19,7 +18,7 @@ namespace ROCKSDB_NAMESPACE {
 // Exceptions MUST NOT propagate out of overridden functions into RocksDB,
 // because RocksDB is not exception-safe. This could cause undefined behavior
 // including data loss, unreported corruption, deadlocks, and more.
-class RateLimiter : public Customizable {
+class RateLimiter {
  public:
   enum class OpType {
     kRead,
@@ -32,20 +31,11 @@ class RateLimiter : public Customizable {
     kAllIo,
   };
 
-  static const char* Type() { return "RateLimiter"; }
-  static Status CreateFromString(const ConfigOptions& options,
-                                 const std::string& value,
-                                 std::shared_ptr<RateLimiter>* result);
-
   // For API compatibility, default to rate-limiting writes only.
-  explicit RateLimiter(Mode mode = Mode::kWritesOnly);
+  explicit RateLimiter(Mode mode = Mode::kWritesOnly) : mode_(mode) {}
 
   virtual ~RateLimiter() {}
 
-  // Deprecated. Will be removed in a major release. Derived classes
-  // should implement this method.
-  virtual const char* Name() const override { return ""; }
-
   // This API allows user to dynamically change rate limiter's bytes per second.
   // REQUIRED: bytes_per_second > 0
   virtual void SetBytesPerSecond(int64_t bytes_per_second) = 0;
@@ -135,7 +125,7 @@ class RateLimiter : public Customizable {
   Mode GetMode() { return mode_; }
 
  private:
-  Mode mode_;
+  const Mode mode_;
 };
 
 // Create a RateLimiter object, which can be shared among RocksDB instances to

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 7
 #define ROCKSDB_MINOR 5
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 4
 
 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these

options/customizable_test.cc

@@ -27,7 +27,6 @@
 #include "rocksdb/filter_policy.h"
 #include "rocksdb/flush_block_policy.h"
 #include "rocksdb/memory_allocator.h"
-#include "rocksdb/rate_limiter.h"
 #include "rocksdb/secondary_cache.h"
 #include "rocksdb/slice_transform.h"
 #include "rocksdb/sst_partitioner.h"
@@ -42,7 +41,6 @@
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
 #include "util/file_checksum_helper.h"
-#include "util/rate_limiter.h"
 #include "util/string_util.h"
 #include "utilities/compaction_filters/remove_emptyvalue_compactionfilter.h"
 #include "utilities/memory_allocators.h"
@@ -1472,21 +1470,6 @@ class MockFileChecksumGenFactory : public FileChecksumGenFactory {
   }
 };
 
-class MockRateLimiter : public RateLimiter {
- public:
-  static const char* kClassName() { return "MockRateLimiter"; }
-  const char* Name() const override { return kClassName(); }
-  void SetBytesPerSecond(int64_t /*bytes_per_second*/) override {}
-  int64_t GetBytesPerSecond() const override { return 0; }
-  int64_t GetSingleBurstBytes() const override { return 0; }
-  int64_t GetTotalBytesThrough(const Env::IOPriority /*pri*/) const override {
-    return 0;
-  }
-  int64_t GetTotalRequests(const Env::IOPriority /*pri*/) const override {
-    return 0;
-  }
-};
-
 class MockFilterPolicy : public FilterPolicy {
  public:
   static const char* kClassName() { return "MockFilterPolicy"; }
@@ -1618,14 +1601,6 @@ static int RegisterLocalObjects(ObjectLibrary& library,
         return guard->get();
       });
 
-  library.AddFactory<RateLimiter>(
-      MockRateLimiter::kClassName(),
-      [](const std::string& /*uri*/, std::unique_ptr<RateLimiter>* guard,
-         std::string* /* errmsg */) {
-        guard->reset(new MockRateLimiter());
-        return guard->get();
-      });
-
   library.AddFactory<const FilterPolicy>(
       MockFilterPolicy::kClassName(),
       [](const std::string& /*uri*/, std::unique_ptr<const FilterPolicy>* guard,
@@ -2149,37 +2124,6 @@ TEST_F(LoadCustomizableTest, LoadMemoryAllocatorTest) {
   }
 }
 
-TEST_F(LoadCustomizableTest, LoadRateLimiterTest) {
-#ifndef ROCKSDB_LITE
-  ASSERT_OK(TestSharedBuiltins<RateLimiter>(MockRateLimiter::kClassName(),
-                                            GenericRateLimiter::kClassName()));
-#else
-  ASSERT_OK(TestSharedBuiltins<RateLimiter>(MockRateLimiter::kClassName(), ""));
-#endif  // ROCKSDB_LITE
-
-  std::shared_ptr<RateLimiter> result;
-  ASSERT_OK(RateLimiter::CreateFromString(
-      config_options_, std::string(GenericRateLimiter::kClassName()) + ":1234",
-      &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_TRUE(result->IsInstanceOf(GenericRateLimiter::kClassName()));
-#ifndef ROCKSDB_LITE
-  ASSERT_OK(GetDBOptionsFromString(
-      config_options_, db_opts_,
-      std::string("rate_limiter=") + GenericRateLimiter::kClassName(),
-      &db_opts_));
-  ASSERT_NE(db_opts_.rate_limiter, nullptr);
-  if (RegisterTests("Test")) {
-    ExpectCreateShared<RateLimiter>(MockRateLimiter::kClassName());
-    ASSERT_OK(GetDBOptionsFromString(
-        config_options_, db_opts_,
-        std::string("rate_limiter=") + MockRateLimiter::kClassName(),
-        &db_opts_));
-    ASSERT_NE(db_opts_.rate_limiter, nullptr);
-  }
-#endif  // ROCKSDB_LITE
-}
-
 TEST_F(LoadCustomizableTest, LoadFilterPolicyTest) {
   const std::string kAutoBloom = BloomFilterPolicy::kClassName();
   const std::string kAutoRibbon = RibbonFilterPolicy::kClassName();

options/db_options.cc

@@ -421,12 +421,11 @@ static std::unordered_map<std::string, OptionTypeInfo>
         {"db_host_id",
          {offsetof(struct ImmutableDBOptions, db_host_id), OptionType::kString,
           OptionVerificationType::kNormal, OptionTypeFlags::kCompareNever}},
+        // Temporarily deprecated due to race conditions (examples in PR 10375).
         {"rate_limiter",
-         OptionTypeInfo::AsCustomSharedPtr<RateLimiter>(
-             offsetof(struct ImmutableDBOptions, rate_limiter),
-             OptionVerificationType::kNormal,
-             OptionTypeFlags::kCompareNever | OptionTypeFlags::kAllowNull)},
-
+         {offsetof(struct ImmutableDBOptions, rate_limiter),
+          OptionType::kUnknown, OptionVerificationType::kDeprecated,
+          OptionTypeFlags::kDontSerialize | OptionTypeFlags::kCompareNever}},
         // The following properties were handled as special cases in ParseOption
         // This means that the properties could be read from the options file
         // but never written to the file or compared to each other.

table/block_based/partitioned_filter_block.cc

@@ -288,6 +288,7 @@ Status PartitionedFilterBlockReader::GetFilterPartitionBlock(
     FilePrefetchBuffer* prefetch_buffer, const BlockHandle& fltr_blk_handle,
     bool no_io, GetContext* get_context,
     BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority,
     CachableEntry<ParsedFullFilterBlock>* filter_block) const {
   assert(table());
   assert(filter_block);
@@ -304,6 +305,7 @@ Status PartitionedFilterBlockReader::GetFilterPartitionBlock(
   }
 
   ReadOptions read_options;
+  read_options.rate_limiter_priority = rate_limiter_priority;
   if (no_io) {
     read_options.read_tier = kBlockCacheTier;
   }
@@ -344,7 +346,7 @@ bool PartitionedFilterBlockReader::MayMatch(
   CachableEntry<ParsedFullFilterBlock> filter_partition_block;
   s = GetFilterPartitionBlock(nullptr /* prefetch_buffer */, filter_handle,
                               no_io, get_context, lookup_context,
-                              &filter_partition_block);
+                              rate_limiter_priority, &filter_partition_block);
   if (UNLIKELY(!s.ok())) {
     IGNORE_STATUS_IF_ERROR(s);
     return true;
@@ -419,7 +421,8 @@ void PartitionedFilterBlockReader::MayMatchPartition(
   CachableEntry<ParsedFullFilterBlock> filter_partition_block;
   Status s = GetFilterPartitionBlock(
       nullptr /* prefetch_buffer */, filter_handle, no_io,
-      range->begin()->get_context, lookup_context, &filter_partition_block);
+      range->begin()->get_context, lookup_context, rate_limiter_priority,
+      &filter_partition_block);
   if (UNLIKELY(!s.ok())) {
     IGNORE_STATUS_IF_ERROR(s);
     return;  // Any/all may match

table/block_based/partitioned_filter_block.h

@@ -137,6 +137,7 @@ class PartitionedFilterBlockReader : public FilterBlockReaderCommon<Block> {
       FilePrefetchBuffer* prefetch_buffer, const BlockHandle& handle,
       bool no_io, GetContext* get_context,
       BlockCacheLookupContext* lookup_context,
+      Env::IOPriority rate_limiter_priority,
       CachableEntry<ParsedFullFilterBlock>* filter_block) const;
 
   using FilterFunction = bool (FullFilterBlockReader::*)(

util/rate_limiter.cc

@@ -13,14 +13,9 @@
 
 #include "monitoring/statistics.h"
 #include "port/port.h"
-#include "rocksdb/convenience.h"
 #include "rocksdb/system_clock.h"
-#include "rocksdb/utilities/customizable_util.h"
-#include "rocksdb/utilities/object_registry.h"
-#include "rocksdb/utilities/options_type.h"
 #include "test_util/sync_point.h"
 #include "util/aligned_buffer.h"
-#include "util/string_util.h"
 
 namespace ROCKSDB_NAMESPACE {
 size_t RateLimiter::RequestToken(size_t bytes, size_t alignment,
@@ -50,68 +45,33 @@ struct GenericRateLimiter::Req {
   bool granted;
 };
 
-static std::unordered_map<std::string, OptionTypeInfo>
-    generic_rate_limiter_type_info = {
-#ifndef ROCKSDB_LITE
-        {"rate_bytes_per_sec",
-         {offsetof(struct GenericRateLimiter::GenericRateLimiterOptions,
-                   max_bytes_per_sec),
-          OptionType::kInt64T}},
-        {"refill_period_us",
-         {offsetof(struct GenericRateLimiter::GenericRateLimiterOptions,
-                   refill_period_us),
-          OptionType::kInt64T}},
-        {"fairness",
-         {offsetof(struct GenericRateLimiter::GenericRateLimiterOptions,
-                   fairness),
-          OptionType::kInt32T}},
-        {"auto_tuned",
-         {offsetof(struct GenericRateLimiter::GenericRateLimiterOptions,
-                   auto_tuned),
-          OptionType::kBoolean}},
-        {"clock",
-         OptionTypeInfo::AsCustomSharedPtr<SystemClock>(
-             offsetof(struct GenericRateLimiter::GenericRateLimiterOptions,
-                      clock),
-             OptionVerificationType::kByNameAllowFromNull,
-             OptionTypeFlags::kAllowNull)},
-#endif  // ROCKSDB_LITE
-};
-
 GenericRateLimiter::GenericRateLimiter(
     int64_t rate_bytes_per_sec, int64_t refill_period_us, int32_t fairness,
     RateLimiter::Mode mode, const std::shared_ptr<SystemClock>& clock,
     bool auto_tuned)
     : RateLimiter(mode),
-      options_(rate_bytes_per_sec, refill_period_us, fairness, clock,
-               auto_tuned),
+      refill_period_us_(refill_period_us),
+      rate_bytes_per_sec_(auto_tuned ? rate_bytes_per_sec / 2
+                                     : rate_bytes_per_sec),
+      refill_bytes_per_period_(
+          CalculateRefillBytesPerPeriod(rate_bytes_per_sec_)),
+      clock_(clock),
       stop_(false),
       exit_cv_(&request_mutex_),
       requests_to_wait_(0),
       available_bytes_(0),
+      next_refill_us_(NowMicrosMonotonic()),
+      fairness_(fairness > 100 ? 100 : fairness),
       rnd_((uint32_t)time(nullptr)),
       wait_until_refill_pending_(false),
-      num_drains_(0) {
-  RegisterOptions(&options_, &generic_rate_limiter_type_info);
+      auto_tuned_(auto_tuned),
+      num_drains_(0),
+      max_bytes_per_sec_(rate_bytes_per_sec),
+      tuned_time_(NowMicrosMonotonic()) {
   for (int i = Env::IO_LOW; i < Env::IO_TOTAL; ++i) {
     total_requests_[i] = 0;
     total_bytes_through_[i] = 0;
   }
-  Initialize();
-}
-void GenericRateLimiter::Initialize() {
-  if (options_.clock == nullptr) {
-    options_.clock = SystemClock::Default();
-  }
-  options_.fairness = std::min(options_.fairness, 100);
-  next_refill_us_ = NowMicrosMonotonic();
-  tuned_time_ = std::chrono::microseconds(NowMicrosMonotonic());
-  if (options_.auto_tuned) {
-    rate_bytes_per_sec_ = options_.max_bytes_per_sec / 2;
-  } else {
-    rate_bytes_per_sec_ = options_.max_bytes_per_sec;
-  }
-  refill_bytes_per_period_ = CalculateRefillBytesPerPeriod(rate_bytes_per_sec_);
 }
 
 GenericRateLimiter::~GenericRateLimiter() {
@@ -135,18 +95,6 @@ GenericRateLimiter::~GenericRateLimiter() {
   }
 }
 
-Status GenericRateLimiter::PrepareOptions(const ConfigOptions& options) {
-  if (options_.fairness <= 0) {
-    return Status::InvalidArgument("Fairness must be > 0");
-  } else if (options_.max_bytes_per_sec <= 0) {
-    return Status::InvalidArgument("max_bytes_per_sec must be > 0");
-  } else if (options_.refill_period_us <= 0) {
-    return Status::InvalidArgument("Refill_period_us must be > 0");
-  }
-  Initialize();
-  return RateLimiter::PrepareOptions(options);
-}
-
 // This API allows user to dynamically change rate limiter's bytes per second.
 void GenericRateLimiter::SetBytesPerSecond(int64_t bytes_per_second) {
   assert(bytes_per_second > 0);
@@ -165,11 +113,11 @@ void GenericRateLimiter::Request(int64_t bytes, const Env::IOPriority pri,
                            &rate_bytes_per_sec_);
   MutexLock g(&request_mutex_);
 
-  if (options_.auto_tuned) {
+  if (auto_tuned_) {
     static const int kRefillsPerTune = 100;
     std::chrono::microseconds now(NowMicrosMonotonic());
-    if (now - tuned_time_ >= kRefillsPerTune * std::chrono::microseconds(
-                                                   options_.refill_period_us)) {
+    if (now - tuned_time_ >=
+        kRefillsPerTune * std::chrono::microseconds(refill_period_us_)) {
       Status s = Tune();
       s.PermitUncheckedError();  //**TODO: What to do on error?
     }
@@ -213,7 +161,7 @@ void GenericRateLimiter::Request(int64_t bytes, const Env::IOPriority pri,
       } else {
         // Whichever thread reaches here first performs duty (1) as described
         // above.
-        int64_t wait_until = options_.clock->NowMicros() + time_until_refill_us;
+        int64_t wait_until = clock_->NowMicros() + time_until_refill_us;
         RecordTick(stats, NUMBER_RATE_LIMITER_DRAINS);
         ++num_drains_;
         wait_until_refill_pending_ = true;
@@ -273,12 +221,12 @@ GenericRateLimiter::GeneratePriorityIterationOrder() {
   // first
   pri_iteration_order[0] = Env::IO_USER;
 
-  bool high_pri_iterated_after_mid_low_pri = rnd_.OneIn(options_.fairness);
+  bool high_pri_iterated_after_mid_low_pri = rnd_.OneIn(fairness_);
   TEST_SYNC_POINT_CALLBACK(
       "GenericRateLimiter::GeneratePriorityIterationOrder::"
       "PostRandomOneInFairnessForHighPri",
       &high_pri_iterated_after_mid_low_pri);
-  bool mid_pri_itereated_after_low_pri = rnd_.OneIn(options_.fairness);
+  bool mid_pri_itereated_after_low_pri = rnd_.OneIn(fairness_);
   TEST_SYNC_POINT_CALLBACK(
       "GenericRateLimiter::GeneratePriorityIterationOrder::"
       "PostRandomOneInFairnessForMidPri",
@@ -307,7 +255,7 @@ GenericRateLimiter::GeneratePriorityIterationOrder() {
 
 void GenericRateLimiter::RefillBytesAndGrantRequests() {
   TEST_SYNC_POINT("GenericRateLimiter::RefillBytesAndGrantRequests");
-  next_refill_us_ = NowMicrosMonotonic() + options_.refill_period_us;
+  next_refill_us_ = NowMicrosMonotonic() + refill_period_us_;
   // Carry over the left over quota from the last period
   auto refill_bytes_per_period =
       refill_bytes_per_period_.load(std::memory_order_relaxed);
@@ -348,12 +296,12 @@ void GenericRateLimiter::RefillBytesAndGrantRequests() {
 int64_t GenericRateLimiter::CalculateRefillBytesPerPeriod(
     int64_t rate_bytes_per_sec) {
   if (std::numeric_limits<int64_t>::max() / rate_bytes_per_sec <
-      options_.refill_period_us) {
+      refill_period_us_) {
     // Avoid unexpected result in the overflow case. The result now is still
     // inaccurate but is a number that is large enough.
     return std::numeric_limits<int64_t>::max() / 1000000;
   } else {
-    return rate_bytes_per_sec * options_.refill_period_us / 1000000;
+    return rate_bytes_per_sec * refill_period_us_ / 1000000;
   }
 }
 
@@ -368,11 +316,10 @@ Status GenericRateLimiter::Tune() {
   std::chrono::microseconds prev_tuned_time = tuned_time_;
   tuned_time_ = std::chrono::microseconds(NowMicrosMonotonic());
 
-  int64_t elapsed_intervals =
-      (tuned_time_ - prev_tuned_time +
-       std::chrono::microseconds(options_.refill_period_us) -
-       std::chrono::microseconds(1)) /
-      std::chrono::microseconds(options_.refill_period_us);
+  int64_t elapsed_intervals = (tuned_time_ - prev_tuned_time +
+                               std::chrono::microseconds(refill_period_us_) -
+                               std::chrono::microseconds(1)) /
+                              std::chrono::microseconds(refill_period_us_);
   // We tune every kRefillsPerTune intervals, so the overflow and division-by-
   // zero conditions should never happen.
   assert(num_drains_ <= std::numeric_limits<int64_t>::max() / 100);
@@ -382,13 +329,13 @@ Status GenericRateLimiter::Tune() {
   int64_t prev_bytes_per_sec = GetBytesPerSecond();
   int64_t new_bytes_per_sec;
   if (drained_pct == 0) {
-    new_bytes_per_sec = options_.max_bytes_per_sec / kAllowedRangeFactor;
+    new_bytes_per_sec = max_bytes_per_sec_ / kAllowedRangeFactor;
   } else if (drained_pct < kLowWatermarkPct) {
     // sanitize to prevent overflow
     int64_t sanitized_prev_bytes_per_sec =
         std::min(prev_bytes_per_sec, std::numeric_limits<int64_t>::max() / 100);
     new_bytes_per_sec =
-        std::max(options_.max_bytes_per_sec / kAllowedRangeFactor,
+        std::max(max_bytes_per_sec_ / kAllowedRangeFactor,
                  sanitized_prev_bytes_per_sec * 100 / (100 + kAdjustFactorPct));
   } else if (drained_pct > kHighWatermarkPct) {
     // sanitize to prevent overflow
@@ -396,7 +343,7 @@ Status GenericRateLimiter::Tune() {
         std::min(prev_bytes_per_sec, std::numeric_limits<int64_t>::max() /
                                          (100 + kAdjustFactorPct));
     new_bytes_per_sec =
-        std::min(options_.max_bytes_per_sec,
+        std::min(max_bytes_per_sec_,
                  sanitized_prev_bytes_per_sec * (100 + kAdjustFactorPct) / 100);
   } else {
     new_bytes_per_sec = prev_bytes_per_sec;
@@ -419,79 +366,7 @@ RateLimiter* NewGenericRateLimiter(
   std::unique_ptr<RateLimiter> limiter(
       new GenericRateLimiter(rate_bytes_per_sec, refill_period_us, fairness,
                              mode, SystemClock::Default(), auto_tuned));
-  Status s = limiter->PrepareOptions(ConfigOptions());
-  if (s.ok()) {
-    return limiter.release();
-  } else {
-    assert(false);
-    return nullptr;
-  }
-}
-namespace {
-#ifndef ROCKSDB_LITE
-static int RegisterBuiltinRateLimiters(ObjectLibrary& library,
-                                       const std::string& /*arg*/) {
-  library.AddFactory<RateLimiter>(
-      GenericRateLimiter::kClassName(),
-      [](const std::string& /*uri*/, std::unique_ptr<RateLimiter>* guard,
-         std::string* /*errmsg*/) {
-        guard->reset(
-            new GenericRateLimiter(std::numeric_limits<int64_t>::max()));
-        return guard->get();
-      });
-  size_t num_types;
-  return static_cast<int>(library.GetFactoryCount(&num_types));
-}
-
-static std::unordered_map<std::string, RateLimiter::Mode>
-    rate_limiter_mode_map = {
-        {"kReadsOnly", RateLimiter::Mode::kReadsOnly},
-        {"kWritesOnly", RateLimiter::Mode::kWritesOnly},
-        {"kAllIo", RateLimiter::Mode::kAllIo},
-};
-#endif  // ROCKSDB_LITE
-static bool LoadRateLimiter(const std::string& name,
-                            std::shared_ptr<RateLimiter>* limiter) {
-  auto plen = strlen(GenericRateLimiter::kClassName());
-  if (name.size() > plen + 2 && name[plen] == ':' &&
-      StartsWith(name, GenericRateLimiter::kClassName())) {
-    auto rate = ParseInt64(name.substr(plen + 1));
-    limiter->reset(new GenericRateLimiter(rate));
-    return true;
-  } else {
-    return false;
-  }
-}
-
-static std::unordered_map<std::string, OptionTypeInfo> rate_limiter_type_info =
-    {
-#ifndef ROCKSDB_LITE
-        {"mode",
-         OptionTypeInfo::Enum<RateLimiter::Mode>(0, &rate_limiter_mode_map)},
-#endif  // ROCKSDB_LITE
-};
-}  // namespace
-
-RateLimiter::RateLimiter(Mode mode) : mode_(mode) {
-  RegisterOptions("", &mode_, &rate_limiter_type_info);
-}
-
-Status RateLimiter::CreateFromString(const ConfigOptions& config_options,
-                                     const std::string& value,
-                                     std::shared_ptr<RateLimiter>* result) {
-  if (value.empty()) {
-    result->reset();
-    return Status::OK();
-  } else {
-#ifndef ROCKSDB_LITE
-    static std::once_flag once;
-    std::call_once(once, [&]() {
-      RegisterBuiltinRateLimiters(*(ObjectLibrary::Default().get()), "");
-    });
-#endif  // ROCKSDB_LITE
-    return LoadSharedObject<RateLimiter>(config_options, value, LoadRateLimiter,
-                                         result);
-  }
+  return limiter.release();
 }
 
 }  // namespace ROCKSDB_NAMESPACE

util/rate_limiter.h

@@ -26,38 +26,13 @@ namespace ROCKSDB_NAMESPACE {
 
 class GenericRateLimiter : public RateLimiter {
  public:
-  struct GenericRateLimiterOptions {
-    static const char* kName() { return "GenericRateLimiterOptions"; }
-    GenericRateLimiterOptions(int64_t _rate_bytes_per_sec,
-                              int64_t _refill_period_us, int32_t _fairness,
-                              const std::shared_ptr<SystemClock>& _clock,
-                              bool _auto_tuned)
-        : max_bytes_per_sec(_rate_bytes_per_sec),
-          refill_period_us(_refill_period_us),
-          clock(_clock),
-          fairness(_fairness > 100 ? 100 : _fairness),
-          auto_tuned(_auto_tuned) {}
-    int64_t max_bytes_per_sec;
-    int64_t refill_period_us;
-    std::shared_ptr<SystemClock> clock;
-    int32_t fairness;
-    bool auto_tuned;
-  };
-
- public:
-  explicit GenericRateLimiter(
-      int64_t refill_bytes, int64_t refill_period_us = 100 * 1000,
-      int32_t fairness = 10,
-      RateLimiter::Mode mode = RateLimiter::Mode::kWritesOnly,
-      const std::shared_ptr<SystemClock>& clock = nullptr,
-      bool auto_tuned = false);
+  GenericRateLimiter(int64_t refill_bytes, int64_t refill_period_us,
+                     int32_t fairness, RateLimiter::Mode mode,
+                     const std::shared_ptr<SystemClock>& clock,
+                     bool auto_tuned);
 
   virtual ~GenericRateLimiter();
 
-  static const char* kClassName() { return "GenericRateLimiter"; }
-  const char* Name() const override { return kClassName(); }
-  Status PrepareOptions(const ConfigOptions& options) override;
-
   // This API allows user to dynamically change rate limiter's bytes per second.
   virtual void SetBytesPerSecond(int64_t bytes_per_second) override;
 
@@ -120,25 +95,29 @@ class GenericRateLimiter : public RateLimiter {
     return rate_bytes_per_sec_;
   }
 
+  virtual void TEST_SetClock(std::shared_ptr<SystemClock> clock) {
+    MutexLock g(&request_mutex_);
+    clock_ = std::move(clock);
+    next_refill_us_ = NowMicrosMonotonic();
+  }
+
  private:
-  void Initialize();
   void RefillBytesAndGrantRequests();
   std::vector<Env::IOPriority> GeneratePriorityIterationOrder();
   int64_t CalculateRefillBytesPerPeriod(int64_t rate_bytes_per_sec);
   Status Tune();
 
-  uint64_t NowMicrosMonotonic() {
-    return options_.clock->NowNanos() / std::milli::den;
-  }
+  uint64_t NowMicrosMonotonic() { return clock_->NowNanos() / std::milli::den; }
 
   // This mutex guard all internal states
   mutable port::Mutex request_mutex_;
 
-  GenericRateLimiterOptions options_;
+  const int64_t refill_period_us_;
 
   int64_t rate_bytes_per_sec_;
   // This variable can be changed dynamically.
   std::atomic<int64_t> refill_bytes_per_period_;
+  std::shared_ptr<SystemClock> clock_;
 
   bool stop_;
   port::CondVar exit_cv_;
@@ -149,13 +128,16 @@ class GenericRateLimiter : public RateLimiter {
   int64_t available_bytes_;
   int64_t next_refill_us_;
 
+  int32_t fairness_;
   Random rnd_;
 
   struct Req;
   std::deque<Req*> queue_[Env::IO_TOTAL];
   bool wait_until_refill_pending_;
 
+  bool auto_tuned_;
   int64_t num_drains_;
+  const int64_t max_bytes_per_sec_;
   std::chrono::microseconds tuned_time_;
 };
 

util/rate_limiter_test.cc

@@ -15,11 +15,8 @@
 #include <limits>
 
 #include "db/db_test_util.h"
-#include "options/options_parser.h"
 #include "port/port.h"
-#include "rocksdb/convenience.h"
 #include "rocksdb/system_clock.h"
-#include "rocksdb/utilities/options_type.h"
 #include "test_util/sync_point.h"
 #include "test_util/testharness.h"
 #include "util/random.h"
@@ -466,95 +463,6 @@ TEST_F(RateLimiterTest, AutoTuneIncreaseWhenFull) {
   ASSERT_LT(new_bytes_per_sec, orig_bytes_per_sec);
 }
 
-TEST_F(RateLimiterTest, CreateGenericRateLimiterFromString) {
-  std::shared_ptr<RateLimiter> limiter;
-  ConfigOptions config_options;
-  std::string limiter_id = GenericRateLimiter::kClassName();
-  ASSERT_OK(RateLimiter::CreateFromString(config_options, limiter_id + ":1024",
-                                          &limiter));
-  ASSERT_NE(limiter, nullptr);
-  ASSERT_EQ(limiter->GetBytesPerSecond(), 1024U);
-#ifndef ROCKSDB_LITE
-  ASSERT_OK(RateLimiter::CreateFromString(
-      config_options, "rate_bytes_per_sec=2048;id=" + limiter_id, &limiter));
-  ASSERT_NE(limiter, nullptr);
-  ASSERT_EQ(limiter->GetBytesPerSecond(), 2048U);
-  ASSERT_NOK(RateLimiter::CreateFromString(
-      config_options, "rate_bytes_per_sec=0;id=" + limiter_id, &limiter));
-  ASSERT_NOK(RateLimiter::CreateFromString(
-      config_options, "rate_bytes_per_sec=2048;fairness=0;id=" + limiter_id,
-      &limiter));
-
-  ASSERT_OK(
-      RateLimiter::CreateFromString(config_options,
-                                    "rate_bytes_per_sec=2048;refill_period_us="
-                                    "1024;fairness=42;auto_tuned=true;"
-                                    "mode=kReadsOnly;id=" +
-                                        limiter_id,
-                                    &limiter));
-  ASSERT_NE(limiter, nullptr);
-  auto opts =
-      limiter->GetOptions<GenericRateLimiter::GenericRateLimiterOptions>();
-  ASSERT_NE(opts, nullptr);
-  ASSERT_EQ(opts->max_bytes_per_sec, 2048);
-  ASSERT_EQ(opts->refill_period_us, 1024);
-  ASSERT_EQ(opts->fairness, 42);
-  ASSERT_EQ(opts->auto_tuned, true);
-  ASSERT_TRUE(limiter->IsRateLimited(RateLimiter::OpType::kRead));
-  ASSERT_FALSE(limiter->IsRateLimited(RateLimiter::OpType::kWrite));
-#endif  // ROCKSDB_LITE
-}
-
-#ifndef ROCKSDB_LITE
-// This test is for a rate limiter that has no name (Name() returns "").
-// When the default Name() method is deprecated, this test should be removed.
-TEST_F(RateLimiterTest, NoNameRateLimiter) {
-  static std::unordered_map<std::string, OptionTypeInfo> dummy_limiter_options =
-      {
-          {"dummy",
-           {0, OptionType::kInt, OptionVerificationType::kNormal,
-            OptionTypeFlags::kNone}},
-      };
-  class NoNameRateLimiter : public RateLimiter {
-   public:
-    explicit NoNameRateLimiter(bool do_register) {
-      if (do_register) {
-        RegisterOptions("", &dummy, &dummy_limiter_options);
-      }
-    }
-    void SetBytesPerSecond(int64_t /*bytes_per_second*/) override {}
-    int64_t GetSingleBurstBytes() const override { return 0; }
-    int64_t GetTotalBytesThrough(const Env::IOPriority /*pri*/) const override {
-      return 0;
-    }
-    int64_t GetTotalRequests(const Env::IOPriority /*pri*/) const override {
-      return 0;
-    }
-    int64_t GetBytesPerSecond() const override { return 0; }
-
-   private:
-    int dummy;
-  };
-
-  ConfigOptions config_options;
-  DBOptions db_opts, copy;
-  db_opts.rate_limiter.reset(new NoNameRateLimiter(false));
-  ASSERT_EQ(db_opts.rate_limiter->GetId(), "");
-  ASSERT_EQ(db_opts.rate_limiter->ToString(config_options), "");
-  db_opts.rate_limiter.reset(new NoNameRateLimiter(true));
-  ASSERT_EQ(db_opts.rate_limiter->GetId(), "");
-  ASSERT_EQ(db_opts.rate_limiter->ToString(config_options), "");
-  std::string opt_str;
-  ASSERT_OK(GetStringFromDBOptions(config_options, db_opts, &opt_str));
-  ASSERT_OK(
-      GetDBOptionsFromString(config_options, DBOptions(), opt_str, &copy));
-  ASSERT_OK(
-      RocksDBOptionsParser::VerifyDBOptions(config_options, db_opts, copy));
-  ASSERT_EQ(copy.rate_limiter, nullptr);
-  ASSERT_NE(copy.rate_limiter, db_opts.rate_limiter);
-}
-#endif  // ROCKSDB_LITE
-
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

utilities/backup/backup_engine.cc

@@ -186,54 +186,13 @@ class BackupEngineImpl {
       const std::shared_ptr<SystemClock>& backup_rate_limiter_clock,
       const std::shared_ptr<SystemClock>& restore_rate_limiter_clock) {
     if (backup_rate_limiter_clock) {
-      assert(options_.backup_rate_limiter->IsInstanceOf(
-          GenericRateLimiter::kClassName()));
-      auto* backup_rate_limiter_options =
-          options_.backup_rate_limiter
-              ->GetOptions<GenericRateLimiter::GenericRateLimiterOptions>();
-
-      assert(backup_rate_limiter_options);
-      RateLimiter::Mode backup_rate_limiter_mode;
-      if (!options_.backup_rate_limiter->IsRateLimited(
-              RateLimiter::OpType::kRead)) {
-        backup_rate_limiter_mode = RateLimiter::Mode::kWritesOnly;
-      } else if (!options_.backup_rate_limiter->IsRateLimited(
-                     RateLimiter::OpType::kWrite)) {
-        backup_rate_limiter_mode = RateLimiter::Mode::kReadsOnly;
-      } else {
-        backup_rate_limiter_mode = RateLimiter::Mode::kAllIo;
-      }
-      options_.backup_rate_limiter.reset(new GenericRateLimiter(
-          backup_rate_limiter_options->max_bytes_per_sec,
-          backup_rate_limiter_options->refill_period_us,
-          backup_rate_limiter_options->fairness, backup_rate_limiter_mode,
-          backup_rate_limiter_clock, backup_rate_limiter_options->auto_tuned));
+      static_cast<GenericRateLimiter*>(options_.backup_rate_limiter.get())
+          ->TEST_SetClock(backup_rate_limiter_clock);
     }
 
     if (restore_rate_limiter_clock) {
-      assert(options_.restore_rate_limiter->IsInstanceOf(
-          GenericRateLimiter::kClassName()));
-      auto* restore_rate_limiter_options =
-          options_.restore_rate_limiter
-              ->GetOptions<GenericRateLimiter::GenericRateLimiterOptions>();
-      assert(restore_rate_limiter_options);
-
-      RateLimiter::Mode restore_rate_limiter_mode;
-      if (!options_.restore_rate_limiter->IsRateLimited(
-              RateLimiter::OpType::kRead)) {
-        restore_rate_limiter_mode = RateLimiter::Mode::kWritesOnly;
-      } else if (!options_.restore_rate_limiter->IsRateLimited(
-                     RateLimiter::OpType::kWrite)) {
-        restore_rate_limiter_mode = RateLimiter::Mode::kReadsOnly;
-      } else {
-        restore_rate_limiter_mode = RateLimiter::Mode::kAllIo;
-      }
-      options_.restore_rate_limiter.reset(new GenericRateLimiter(
-          restore_rate_limiter_options->max_bytes_per_sec,
-          restore_rate_limiter_options->refill_period_us,
-          restore_rate_limiter_options->fairness, restore_rate_limiter_mode,
-          restore_rate_limiter_clock,
-          restore_rate_limiter_options->auto_tuned));
+      static_cast<GenericRateLimiter*>(options_.restore_rate_limiter.get())
+          ->TEST_SetClock(restore_rate_limiter_clock);
     }
   }
 

utilities/counted_fs.cc

@@ -211,6 +211,7 @@ class CountedRandomRWFile : public FSRandomRWFileOwnerWrapper {
 class CountedDirectory : public FSDirectoryWrapper {
  private:
   mutable CountedFileSystem* fs_;
+  bool closed_ = false;
 
  public:
   CountedDirectory(std::unique_ptr<FSDirectory>&& f, CountedFileSystem* fs)
@@ -229,6 +230,7 @@ class CountedDirectory : public FSDirectoryWrapper {
     if (rv.ok()) {
       fs_->counters()->closes++;
       fs_->counters()->dir_closes++;
+      closed_ = true;
     }
     return rv;
   }
@@ -242,6 +244,14 @@ class CountedDirectory : public FSDirectoryWrapper {
     }
     return rv;
   }
+
+  ~CountedDirectory() {
+    if (!closed_) {
+      // TODO: fix DB+CF code to use explicit Close, not rely on destructor
+      fs_->counters()->closes++;
+      fs_->counters()->dir_closes++;
+    }
+  }
 };
 }  // anonymous namespace