Update HISTORY and version.h for 7.4.5

Summary:
TL;DR: due to a recent change, if you drop a column family,
often that DB will no longer fsync after writing new SST files
to remaining or new column families, which could lead to data
loss on power loss.

More bug detail:
The intent of https://github.com/facebook/rocksdb/issues/10049 was to Close FSDirectory objects at
DB::Close time rather than waiting for DB object destruction.
Unfortunately, it also closes shared FSDirectory objects on
DropColumnFamily (& destroy remaining handles), which can lead
to use-after-Close on FSDirectory shared with remaining column
families. Those "uses" are only Fsyncs (or redundant Closes). In
the default Posix filesystem, an Fsync on a closed FSDirectory is a
quiet no-op. Consequently (under most configurations), if you drop
a column family, that DB will no longer fsync after writing new SST
files to column families sharing the same directory (true under most
configurations).

More fix detail:
Basically, this removes unnecessary Close ops on destroying
ColumnFamilyData. We let `shared_ptr` take care of calling the
destructor at the right time. If the intent was to require Close be
called before destroying FSDirectory, that was not made clear by the
author of FileSystem and was not at all enforced by https://github.com/facebook/rocksdb/issues/10049, which
could have added `assert(fd_ == -1)` to `~PosixDirectory()` but did
not. To keep this fix simple, we relax the unit test for https://github.com/facebook/rocksdb/issues/10049 to allow
timely destruction of FSDirectory to suffice as Close (in
CountedFileSystem). Added a TODO to revisit that.

Also in this PR:
* Added a TODO to share FSDirectory instances between DB and its column
families. (Already shared among column families.)
* Made DB::Close attempt to close all its open FSDirectory objects even
if there is a failure in closing one. Also code clean-up around this
logic.

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

Test Plan:
add an assert to check for use-after-Close. With that
existing tests can detect the misuse. With fix, tests pass (except noted
relaxing of unit test for https://github.com/facebook/rocksdb/issues/10049)

Reviewed By: ajkr

Differential Revision: D38357922

Pulled By: pdillinger

fbshipit-source-id: d42079cadbedf0a969f03389bf586b3b4e1f9137

HISTORY.md

@@ -1,4 +1,27 @@
 # Rocksdb Change Log
+## 7.4.5 (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.4.4 (07/19/2022)
+### Public API changes
+* Removed Customizable support for RateLimiter and removed its CreateFromString() and Type() functions.
+
+### Bug Fixes
+* 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.
+
+## 7.4.3 (07/13/2022)
+### Behavior Changes
+* For track_and_verify_wals_in_manifest, revert to the original behavior before #10087: syncing of live WAL file is not tracked, and we track only the synced sizes of **closed** WALs. (PR #10330).
+
+## 7.4.2 (06/30/2022)
+### Bug Fixes
+* 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.
+
+## 7.4.1 (06/28/2022)
+### Bug Fixes
+* Pass `rate_limiter_priority` through filter block reader functions to `FileSystem`.
+
 ## 7.4.0 (06/19/2022)
 ### Bug Fixes
 * Fixed a bug in calculating key-value integrity protection for users of in-place memtable updates. In particular, the affected users would be those who configure `protection_bytes_per_key > 0` on `WriteBatch` or `WriteOptions`, and configure `inplace_callback != nullptr`.
@@ -13,6 +36,8 @@
 * Avoid a crash if the IDENTITY file is accidentally truncated to empty. A new DB ID will be written and generated on Open.
 * Fixed a possible corruption for users of `manual_wal_flush` and/or `FlushWAL(true /* sync */)`, together with `track_and_verify_wals_in_manifest == true`. For those users, losing unsynced data (e.g., due to power loss) could make future DB opens fail with a `Status::Corruption` complaining about missing WAL data.
 * Fixed a bug in `WriteBatchInternal::Append()` where WAL termination point in write batch was not considered and the function appends an incorrect number of checksums.
+* Fixed a crash bug introduced in 7.3.0 affecting users of MultiGet with `kDataBlockBinaryAndHash`.
+* Add some fixes in async_io which was doing extra prefetching in shorter scans.
 
 ### Public API changes
 * Add new API GetUnixTime in Snapshot class which returns the unix time at which Snapshot is taken.

build_tools/check-sources.sh

@@ -17,12 +17,18 @@ if [ "$?" != "1" ]; then
   BAD=1
 fi
 
-git grep -n '<rocksdb/' -- ':!build_tools/check-sources.sh'
+git grep -n 'include <rocksdb/' -- ':!build_tools/check-sources.sh'
 if [ "$?" != "1" ]; then
   echo '^^^^^ Use double-quotes as in #include "rocksdb/something.h"'
   BAD=1
 fi
 
+git grep -n 'include "include/rocksdb/' -- ':!build_tools/check-sources.sh'
+if [ "$?" != "1" ]; then
+  echo '^^^^^ Use #include "rocksdb/something.h" instead of #include "include/rocksdb/something.h"'
+  BAD=1
+fi
+
 git grep -n 'using namespace' -- ':!build_tools' ':!docs' \
     ':!third-party/folly/folly/lang/Align.h' \
     ':!third-party/gtest-1.8.1/fused-src/gtest/gtest.h'

db/blob/blob_source.h

@@ -10,7 +10,7 @@
 #include "cache/cache_helpers.h"
 #include "cache/cache_key.h"
 #include "db/blob/blob_file_cache.h"
-#include "include/rocksdb/cache.h"
+#include "rocksdb/cache.h"
 #include "rocksdb/rocksdb_namespace.h"
 #include "table/block_based/cachable_entry.h"
 

db/column_family.cc

@@ -695,19 +695,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/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;
 }
@@ -4143,7 +4143,7 @@ TEST_F(DBBasicTest, FailOpenIfLoggerCreationFail) {
 
   Status s = TryReopen(options);
   ASSERT_EQ(nullptr, options.info_log);
-  ASSERT_TRUE(s.IsAborted());
+  ASSERT_TRUE(s.IsIOError());
 
   SyncPoint::GetInstance()->DisableProcessing();
   SyncPoint::GetInstance()->ClearAllCallBacks();
@@ -4193,7 +4193,9 @@ TEST_F(DBBasicTest, VerifyFileChecksums) {
   ASSERT_TRUE(db_->VerifyFileChecksums(ReadOptions()).IsInvalidArgument());
 }
 
-TEST_F(DBBasicTest, ManualWalSync) {
+// TODO: re-enable after we provide finer-grained control for WAL tracking to
+// meet the needs of different use cases, durability levels and recovery modes.
+TEST_F(DBBasicTest, DISABLED_ManualWalSync) {
   Options options = CurrentOptions();
   options.track_and_verify_wals_in_manifest = true;
   options.wal_recovery_mode = WALRecoveryMode::kAbsoluteConsistency;

db/db_impl/db_impl.cc

@@ -156,7 +156,9 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname,
                bool read_only)
     : dbname_(dbname),
       own_info_log_(options.info_log == nullptr),
-      initial_db_options_(SanitizeOptions(dbname, options, read_only)),
+      init_logger_creation_s_(),
+      initial_db_options_(SanitizeOptions(dbname, options, read_only,
+                                          &init_logger_creation_s_)),
       env_(initial_db_options_.env),
       io_tracer_(std::make_shared<IOTracer>()),
       immutable_db_options_(initial_db_options_),
@@ -747,6 +749,9 @@ Status DBImpl::CloseHelper() {
 Status DBImpl::CloseImpl() { return CloseHelper(); }
 
 DBImpl::~DBImpl() {
+  // TODO: remove this.
+  init_logger_creation_s_.PermitUncheckedError();
+
   InstrumentedMutexLock closing_lock_guard(&closing_mutex_);
   if (closed_) {
     return;
@@ -1471,12 +1476,12 @@ Status DBImpl::MarkLogsSynced(uint64_t up_to, bool synced_dir) {
   for (auto it = logs_.begin(); it != logs_.end() && it->number <= up_to;) {
     auto& wal = *it;
     assert(wal.IsSyncing());
-    if (immutable_db_options_.track_and_verify_wals_in_manifest &&
-        wal.GetPreSyncSize() > 0) {
-      synced_wals.AddWal(wal.number, WalMetadata(wal.GetPreSyncSize()));
-    }
 
     if (logs_.size() > 1) {
+      if (immutable_db_options_.track_and_verify_wals_in_manifest &&
+          wal.GetPreSyncSize() > 0) {
+        synced_wals.AddWal(wal.number, WalMetadata(wal.GetPreSyncSize()));
+      }
       logs_to_free_.push_back(wal.ReleaseWriter());
       // To modify logs_ both mutex_ and log_write_mutex_ must be held
       InstrumentedMutexLock l(&log_write_mutex_);

db/db_impl/db_impl.h

@@ -117,7 +117,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
@@ -126,53 +125,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;
   }
 
@@ -1248,6 +1229,7 @@ class DBImpl : public DB {
   std::unique_ptr<VersionSet> versions_;
   // Flag to check whether we allocated and own the info log file
   bool own_info_log_;
+  Status init_logger_creation_s_;
   const DBOptions initial_db_options_;
   Env* const env_;
   std::shared_ptr<IOTracer> io_tracer_;
@@ -2590,10 +2572,12 @@ class GetWithTimestampReadCallback : public ReadCallback {
 };
 
 extern Options SanitizeOptions(const std::string& db, const Options& src,
-                               bool read_only = false);
+                               bool read_only = false,
+                               Status* logger_creation_s = nullptr);
 
 extern DBOptions SanitizeOptions(const std::string& db, const DBOptions& src,
-                                 bool read_only = false);
+                                 bool read_only = false,
+                                 Status* logger_creation_s = nullptr);
 
 extern CompressionType GetCompressionFlush(
     const ImmutableCFOptions& ioptions,

db/db_impl/db_impl_open.cc

@@ -27,8 +27,9 @@
 
 namespace ROCKSDB_NAMESPACE {
 Options SanitizeOptions(const std::string& dbname, const Options& src,
-                        bool read_only) {
-  auto db_options = SanitizeOptions(dbname, DBOptions(src), read_only);
+                        bool read_only, Status* logger_creation_s) {
+  auto db_options =
+      SanitizeOptions(dbname, DBOptions(src), read_only, logger_creation_s);
   ImmutableDBOptions immutable_db_options(db_options);
   auto cf_options =
       SanitizeOptions(immutable_db_options, ColumnFamilyOptions(src));
@@ -36,7 +37,7 @@ Options SanitizeOptions(const std::string& dbname, const Options& src,
 }
 
 DBOptions SanitizeOptions(const std::string& dbname, const DBOptions& src,
-                          bool read_only) {
+                          bool read_only, Status* logger_creation_s) {
   DBOptions result(src);
 
   if (result.env == nullptr) {
@@ -59,6 +60,9 @@ DBOptions SanitizeOptions(const std::string& dbname, const DBOptions& src,
     if (!s.ok()) {
       // No place suitable for logging
       result.info_log = nullptr;
+      if (logger_creation_s) {
+        *logger_creation_s = s;
+      }
     }
   }
 
@@ -534,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);
@@ -1747,9 +1752,11 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
 
   DBImpl* impl = new DBImpl(db_options, dbname, seq_per_batch, batch_per_txn);
   if (!impl->immutable_db_options_.info_log) {
-    s = Status::Aborted("Failed to create logger");
+    s = impl->init_logger_creation_s_;
     delete impl;
     return s;
+  } else {
+    assert(impl->init_logger_creation_s_.ok());
   }
   s = impl->env_->CreateDirIfMissing(impl->immutable_db_options_.GetWalDir());
   if (s.ok()) {

db/db_test.cc

@@ -4092,9 +4092,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_stress_tool/db_stress_common.h

@@ -157,6 +157,7 @@ DECLARE_bool(partition_filters);
 DECLARE_bool(optimize_filters_for_memory);
 DECLARE_bool(detect_filter_construct_corruption);
 DECLARE_int32(index_type);
+DECLARE_int32(data_block_index_type);
 DECLARE_string(db);
 DECLARE_string(secondaries_base);
 DECLARE_bool(test_secondary);

db_stress_tool/db_stress_gflags.cc

@@ -493,9 +493,15 @@ DEFINE_bool(
 DEFINE_int32(
     index_type,
     static_cast<int32_t>(
-        ROCKSDB_NAMESPACE::BlockBasedTableOptions::kBinarySearch),
+        ROCKSDB_NAMESPACE::BlockBasedTableOptions().index_type),
     "Type of block-based table index (see `enum IndexType` in table.h)");
 
+DEFINE_int32(
+    data_block_index_type,
+    static_cast<int32_t>(
+        ROCKSDB_NAMESPACE::BlockBasedTableOptions().data_block_index_type),
+    "Index type for data blocks (see `enum DataBlockIndexType` in table.h)");
+
 DEFINE_string(db, "", "Use the db with the following name.");
 
 DEFINE_string(secondaries_base, "",

db_stress_tool/db_stress_test_base.cc

@@ -2775,6 +2775,9 @@ void InitializeOptionsFromFlags(
       FLAGS_detect_filter_construct_corruption;
   block_based_options.index_type =
       static_cast<BlockBasedTableOptions::IndexType>(FLAGS_index_type);
+  block_based_options.data_block_index_type =
+      static_cast<BlockBasedTableOptions::DataBlockIndexType>(
+          FLAGS_data_block_index_type);
   block_based_options.prepopulate_block_cache =
       static_cast<BlockBasedTableOptions::PrepopulateBlockCache>(
           FLAGS_prepopulate_block_cache);

env/io_posix.cc

@@ -1713,6 +1713,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/options.h

@@ -483,11 +483,17 @@ struct DBOptions {
   bool flush_verify_memtable_count = true;
 
   // If true, the log numbers and sizes of the synced WALs are tracked
-  // in MANIFEST, then during DB recovery, if a synced WAL is missing
+  // in MANIFEST. During DB recovery, if a synced WAL is missing
   // from disk, or the WAL's size does not match the recorded size in
   // MANIFEST, an error will be reported and the recovery will be aborted.
   //
+  // This is one additional protection against WAL corruption besides the
+  // per-WAL-entry checksum.
+  //
   // Note that this option does not work with secondary instance.
+  // Currently, only syncing closed WALs are tracked. Calling `DB::SyncWAL()`,
+  // etc. or writing with `WriteOptions::sync=true` to sync the live WAL is not
+  // tracked for performance/efficiency reasons.
   //
   // Default: false
   bool track_and_verify_wals_in_manifest = false;

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 4
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 5
 
 // 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

logging/auto_roll_logger.cc

@@ -278,11 +278,21 @@ Status CreateLoggerFromOptions(const std::string& dbname,
       InfoLogFileName(dbname, db_absolute_path, options.db_log_dir);
 
   const auto& clock = env->GetSystemClock();
-  // In case it does not exist
+  // In case it does not exist.
   s = env->CreateDirIfMissing(dbname);
   if (!s.ok()) {
-    return s;
+    if (options.db_log_dir.empty()) {
+      return s;
+    } else {
+      // Ignore the error returned during creation of dbname because dbname and
+      // db_log_dir can be on different filesystems in which case dbname will
+      // not exist and error should be ignored. db_log_dir creation will handle
+      // the error in case there is any error in the creation of dbname on same
+      // filesystem.
+      s = Status::OK();
+    }
   }
+  assert(s.ok());
 
   if (!options.db_log_dir.empty()) {
     s = env->CreateDirIfMissing(options.db_log_dir);

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

@@ -422,12 +422,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/block_based_table_reader.cc

@@ -1902,7 +1902,8 @@ bool BlockBasedTable::PrefixRangeMayMatch(
     may_match = filter->RangeMayExist(
         read_options.iterate_upper_bound, user_key_without_ts, prefix_extractor,
         rep_->internal_comparator.user_comparator(), const_ikey_ptr,
-        &filter_checked, need_upper_bound_check, no_io, lookup_context);
+        &filter_checked, need_upper_bound_check, no_io, lookup_context,
+        read_options.rate_limiter_priority);
   }
 
   if (filter_checked) {
@@ -1974,7 +1975,8 @@ FragmentedRangeTombstoneIterator* BlockBasedTable::NewRangeTombstoneIterator(
 bool BlockBasedTable::FullFilterKeyMayMatch(
     FilterBlockReader* filter, const Slice& internal_key, const bool no_io,
     const SliceTransform* prefix_extractor, GetContext* get_context,
-    BlockCacheLookupContext* lookup_context) const {
+    BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) const {
   if (filter == nullptr) {
     return true;
   }
@@ -1984,13 +1986,15 @@ bool BlockBasedTable::FullFilterKeyMayMatch(
   size_t ts_sz = rep_->internal_comparator.user_comparator()->timestamp_size();
   Slice user_key_without_ts = StripTimestampFromUserKey(user_key, ts_sz);
   if (rep_->whole_key_filtering) {
-    may_match = filter->KeyMayMatch(user_key_without_ts, no_io, const_ikey_ptr,
-                                    get_context, lookup_context);
+    may_match =
+        filter->KeyMayMatch(user_key_without_ts, no_io, const_ikey_ptr,
+                            get_context, lookup_context, rate_limiter_priority);
   } else if (!PrefixExtractorChanged(prefix_extractor) &&
              prefix_extractor->InDomain(user_key_without_ts) &&
              !filter->PrefixMayMatch(
                  prefix_extractor->Transform(user_key_without_ts), no_io,
-                 const_ikey_ptr, get_context, lookup_context)) {
+                 const_ikey_ptr, get_context, lookup_context,
+                 rate_limiter_priority)) {
     // FIXME ^^^: there should be no reason for Get() to depend on current
     // prefix_extractor at all. It should always use table_prefix_extractor.
     may_match = false;
@@ -2005,14 +2009,15 @@ bool BlockBasedTable::FullFilterKeyMayMatch(
 void BlockBasedTable::FullFilterKeysMayMatch(
     FilterBlockReader* filter, MultiGetRange* range, const bool no_io,
     const SliceTransform* prefix_extractor,
-    BlockCacheLookupContext* lookup_context) const {
+    BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) const {
   if (filter == nullptr) {
     return;
   }
   uint64_t before_keys = range->KeysLeft();
   assert(before_keys > 0);  // Caller should ensure
   if (rep_->whole_key_filtering) {
-    filter->KeysMayMatch(range, no_io, lookup_context);
+    filter->KeysMayMatch(range, no_io, lookup_context, rate_limiter_priority);
     uint64_t after_keys = range->KeysLeft();
     if (after_keys) {
       RecordTick(rep_->ioptions.stats, BLOOM_FILTER_FULL_POSITIVE, after_keys);
@@ -2028,7 +2033,8 @@ void BlockBasedTable::FullFilterKeysMayMatch(
   } else if (!PrefixExtractorChanged(prefix_extractor)) {
     // FIXME ^^^: there should be no reason for MultiGet() to depend on current
     // prefix_extractor at all. It should always use table_prefix_extractor.
-    filter->PrefixesMayMatch(range, prefix_extractor, false, lookup_context);
+    filter->PrefixesMayMatch(range, prefix_extractor, false, lookup_context,
+                             rate_limiter_priority);
     RecordTick(rep_->ioptions.stats, BLOOM_FILTER_PREFIX_CHECKED, before_keys);
     uint64_t after_keys = range->KeysLeft();
     uint64_t filtered_keys = before_keys - after_keys;
@@ -2065,7 +2071,8 @@ Status BlockBasedTable::Get(const ReadOptions& read_options, const Slice& key,
   }
   TEST_SYNC_POINT("BlockBasedTable::Get:BeforeFilterMatch");
   const bool may_match = FullFilterKeyMayMatch(
-      filter, key, no_io, prefix_extractor, get_context, &lookup_context);
+      filter, key, no_io, prefix_extractor, get_context, &lookup_context,
+      read_options.rate_limiter_priority);
   TEST_SYNC_POINT("BlockBasedTable::Get:AfterFilterMatch");
   if (!may_match) {
     RecordTick(rep_->ioptions.stats, BLOOM_FILTER_USEFUL);

table/block_based/block_based_table_reader.h

@@ -454,12 +454,14 @@ class BlockBasedTable : public TableReader {
                              const bool no_io,
                              const SliceTransform* prefix_extractor,
                              GetContext* get_context,
-                             BlockCacheLookupContext* lookup_context) const;
+                             BlockCacheLookupContext* lookup_context,
+                             Env::IOPriority rate_limiter_priority) const;
 
   void FullFilterKeysMayMatch(FilterBlockReader* filter, MultiGetRange* range,
                               const bool no_io,
                               const SliceTransform* prefix_extractor,
-                              BlockCacheLookupContext* lookup_context) const;
+                              BlockCacheLookupContext* lookup_context,
+                              Env::IOPriority rate_limiter_priority) const;
 
   // If force_direct_prefetch is true, always prefetching to RocksDB
   //    buffer, rather than calling RandomAccessFile::Prefetch().

table/block_based/block_based_table_reader_sync_and_async.h

@@ -335,7 +335,7 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::MultiGet)
       TableReaderCaller::kUserMultiGet, tracing_mget_id,
       /*_get_from_user_specified_snapshot=*/read_options.snapshot != nullptr};
   FullFilterKeysMayMatch(filter, &sst_file_range, no_io, prefix_extractor,
-                         &lookup_context);
+                         &lookup_context, read_options.rate_limiter_priority);
 
   if (!sst_file_range.empty()) {
     IndexBlockIter iiter_on_stack;
@@ -610,15 +610,6 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::MultiGet)
           break;
         }
 
-        bool may_exist = biter->SeekForGet(key);
-        if (!may_exist) {
-          // HashSeek cannot find the key this block and the the iter is not
-          // the end of the block, i.e. cannot be in the following blocks
-          // either. In this case, the seek_key cannot be found, so we break
-          // from the top level for-loop.
-          break;
-        }
-
         // Reusing blocks complicates pinning/Cleanable, because the cache
         // entry referenced by biter can only be released once all returned
         // pinned values are released. This code previously did an extra
@@ -661,6 +652,15 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::MultiGet)
           value_pinner = nullptr;
         }
 
+        bool may_exist = biter->SeekForGet(key);
+        if (!may_exist) {
+          // HashSeek cannot find the key this block and the the iter is not
+          // the end of the block, i.e. cannot be in the following blocks
+          // either. In this case, the seek_key cannot be found, so we break
+          // from the top level for-loop.
+          break;
+        }
+
         // Call the *saver function on each entry/block until it returns false
         for (; biter->Valid(); biter->Next()) {
           ParsedInternalKey parsed_key;

table/block_based/filter_block.h

@@ -112,16 +112,18 @@ class FilterBlockReader {
   virtual bool KeyMayMatch(const Slice& key, const bool no_io,
                            const Slice* const const_ikey_ptr,
                            GetContext* get_context,
-                           BlockCacheLookupContext* lookup_context) = 0;
+                           BlockCacheLookupContext* lookup_context,
+                           Env::IOPriority rate_limiter_priority) = 0;
 
   virtual void KeysMayMatch(MultiGetRange* range, const bool no_io,
-                            BlockCacheLookupContext* lookup_context) {
+                            BlockCacheLookupContext* lookup_context,
+                            Env::IOPriority rate_limiter_priority) {
     for (auto iter = range->begin(); iter != range->end(); ++iter) {
       const Slice ukey_without_ts = iter->ukey_without_ts;
       const Slice ikey = iter->ikey;
       GetContext* const get_context = iter->get_context;
       if (!KeyMayMatch(ukey_without_ts, no_io, &ikey, get_context,
-                       lookup_context)) {
+                       lookup_context, rate_limiter_priority)) {
         range->SkipKey(iter);
       }
     }
@@ -133,19 +135,22 @@ class FilterBlockReader {
   virtual bool PrefixMayMatch(const Slice& prefix, const bool no_io,
                               const Slice* const const_ikey_ptr,
                               GetContext* get_context,
-                              BlockCacheLookupContext* lookup_context) = 0;
+                              BlockCacheLookupContext* lookup_context,
+                              Env::IOPriority rate_limiter_priority) = 0;
 
   virtual void PrefixesMayMatch(MultiGetRange* range,
                                 const SliceTransform* prefix_extractor,
                                 const bool no_io,
-                                BlockCacheLookupContext* lookup_context) {
+                                BlockCacheLookupContext* lookup_context,
+                                Env::IOPriority rate_limiter_priority) {
     for (auto iter = range->begin(); iter != range->end(); ++iter) {
       const Slice ukey_without_ts = iter->ukey_without_ts;
       const Slice ikey = iter->ikey;
       GetContext* const get_context = iter->get_context;
       if (prefix_extractor->InDomain(ukey_without_ts) &&
           !PrefixMayMatch(prefix_extractor->Transform(ukey_without_ts), no_io,
-                          &ikey, get_context, lookup_context)) {
+                          &ikey, get_context, lookup_context,
+                          rate_limiter_priority)) {
         range->SkipKey(iter);
       }
     }
@@ -170,7 +175,8 @@ class FilterBlockReader {
                              const Slice* const const_ikey_ptr,
                              bool* filter_checked, bool need_upper_bound_check,
                              bool no_io,
-                             BlockCacheLookupContext* lookup_context) = 0;
+                             BlockCacheLookupContext* lookup_context,
+                             Env::IOPriority rate_limiter_priority) = 0;
 };
 
 }  // namespace ROCKSDB_NAMESPACE

table/block_based/filter_block_reader_common.cc

@@ -67,7 +67,8 @@ template <typename TBlocklike>
 Status FilterBlockReaderCommon<TBlocklike>::GetOrReadFilterBlock(
     bool no_io, GetContext* get_context,
     BlockCacheLookupContext* lookup_context,
-    CachableEntry<TBlocklike>* filter_block, BlockType block_type) const {
+    CachableEntry<TBlocklike>* filter_block, BlockType block_type,
+    Env::IOPriority rate_limiter_priority) const {
   assert(filter_block);
 
   if (!filter_block_.IsEmpty()) {
@@ -76,6 +77,7 @@ Status FilterBlockReaderCommon<TBlocklike>::GetOrReadFilterBlock(
   }
 
   ReadOptions read_options;
+  read_options.rate_limiter_priority = rate_limiter_priority;
   if (no_io) {
     read_options.read_tier = kBlockCacheTier;
   }
@@ -100,7 +102,8 @@ bool FilterBlockReaderCommon<TBlocklike>::RangeMayExist(
     const SliceTransform* prefix_extractor, const Comparator* comparator,
     const Slice* const const_ikey_ptr, bool* filter_checked,
     bool need_upper_bound_check, bool no_io,
-    BlockCacheLookupContext* lookup_context) {
+    BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) {
   if (!prefix_extractor || !prefix_extractor->InDomain(user_key_without_ts)) {
     *filter_checked = false;
     return true;
@@ -113,7 +116,8 @@ bool FilterBlockReaderCommon<TBlocklike>::RangeMayExist(
   } else {
     *filter_checked = true;
     return PrefixMayMatch(prefix, no_io, const_ikey_ptr,
-                          /* get_context */ nullptr, lookup_context);
+                          /* get_context */ nullptr, lookup_context,
+                          rate_limiter_priority);
   }
 }
 

table/block_based/filter_block_reader_common.h

@@ -40,7 +40,8 @@ class FilterBlockReaderCommon : public FilterBlockReader {
                      const Comparator* comparator,
                      const Slice* const const_ikey_ptr, bool* filter_checked,
                      bool need_upper_bound_check, bool no_io,
-                     BlockCacheLookupContext* lookup_context) override;
+                     BlockCacheLookupContext* lookup_context,
+                     Env::IOPriority rate_limiter_priority) override;
 
  protected:
   static Status ReadFilterBlock(const BlockBasedTable* table,
@@ -59,7 +60,8 @@ class FilterBlockReaderCommon : public FilterBlockReader {
   Status GetOrReadFilterBlock(bool no_io, GetContext* get_context,
                               BlockCacheLookupContext* lookup_context,
                               CachableEntry<TBlocklike>* filter_block,
-                              BlockType block_type) const;
+                              BlockType block_type,
+                              Env::IOPriority rate_limiter_priority) const;
 
   size_t ApproximateFilterBlockMemoryUsage() const;
 

table/block_based/full_filter_block.cc

@@ -124,13 +124,16 @@ FullFilterBlockReader::FullFilterBlockReader(
     : FilterBlockReaderCommon(t, std::move(filter_block)) {
 }
 
-bool FullFilterBlockReader::KeyMayMatch(
-    const Slice& key, const bool no_io, const Slice* const /*const_ikey_ptr*/,
-    GetContext* get_context, BlockCacheLookupContext* lookup_context) {
+bool FullFilterBlockReader::KeyMayMatch(const Slice& key, const bool no_io,
+                                        const Slice* const /*const_ikey_ptr*/,
+                                        GetContext* get_context,
+                                        BlockCacheLookupContext* lookup_context,
+                                        Env::IOPriority rate_limiter_priority) {
   if (!whole_key_filtering()) {
     return true;
   }
-  return MayMatch(key, no_io, get_context, lookup_context);
+  return MayMatch(key, no_io, get_context, lookup_context,
+                  rate_limiter_priority);
 }
 
 std::unique_ptr<FilterBlockReader> FullFilterBlockReader::Create(
@@ -163,17 +166,21 @@ std::unique_ptr<FilterBlockReader> FullFilterBlockReader::Create(
 bool FullFilterBlockReader::PrefixMayMatch(
     const Slice& prefix, const bool no_io,
     const Slice* const /*const_ikey_ptr*/, GetContext* get_context,
-    BlockCacheLookupContext* lookup_context) {
-  return MayMatch(prefix, no_io, get_context, lookup_context);
+    BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) {
+  return MayMatch(prefix, no_io, get_context, lookup_context,
+                  rate_limiter_priority);
 }
 
 bool FullFilterBlockReader::MayMatch(
     const Slice& entry, bool no_io, GetContext* get_context,
-    BlockCacheLookupContext* lookup_context) const {
+    BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) const {
   CachableEntry<ParsedFullFilterBlock> filter_block;
 
-  const Status s = GetOrReadFilterBlock(no_io, get_context, lookup_context,
-                                        &filter_block, BlockType::kFilter);
+  const Status s =
+      GetOrReadFilterBlock(no_io, get_context, lookup_context, &filter_block,
+                           BlockType::kFilter, rate_limiter_priority);
   if (!s.ok()) {
     IGNORE_STATUS_IF_ERROR(s);
     return true;
@@ -198,29 +205,33 @@ bool FullFilterBlockReader::MayMatch(
 
 void FullFilterBlockReader::KeysMayMatch(
     MultiGetRange* range, const bool no_io,
-    BlockCacheLookupContext* lookup_context) {
+    BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) {
   if (!whole_key_filtering()) {
     // Simply return. Don't skip any key - consider all keys as likely to be
     // present
     return;
   }
-  MayMatch(range, no_io, nullptr, lookup_context);
+  MayMatch(range, no_io, nullptr, lookup_context, rate_limiter_priority);
 }
 
 void FullFilterBlockReader::PrefixesMayMatch(
     MultiGetRange* range, const SliceTransform* prefix_extractor,
-    const bool no_io, BlockCacheLookupContext* lookup_context) {
-  MayMatch(range, no_io, prefix_extractor, lookup_context);
+    const bool no_io, BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) {
+  MayMatch(range, no_io, prefix_extractor, lookup_context,
+           rate_limiter_priority);
 }
 
 void FullFilterBlockReader::MayMatch(
     MultiGetRange* range, bool no_io, const SliceTransform* prefix_extractor,
-    BlockCacheLookupContext* lookup_context) const {
+    BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) const {
   CachableEntry<ParsedFullFilterBlock> filter_block;
 
-  const Status s =
-      GetOrReadFilterBlock(no_io, range->begin()->get_context, lookup_context,
-                           &filter_block, BlockType::kFilter);
+  const Status s = GetOrReadFilterBlock(
+      no_io, range->begin()->get_context, lookup_context, &filter_block,
+      BlockType::kFilter, rate_limiter_priority);
   if (!s.ok()) {
     IGNORE_STATUS_IF_ERROR(s);
     return;

table/block_based/full_filter_block.h

@@ -107,34 +107,40 @@ class FullFilterBlockReader
 
   bool KeyMayMatch(const Slice& key, const bool no_io,
                    const Slice* const const_ikey_ptr, GetContext* get_context,
-                   BlockCacheLookupContext* lookup_context) override;
+                   BlockCacheLookupContext* lookup_context,
+                   Env::IOPriority rate_limiter_priority) override;
 
   bool PrefixMayMatch(const Slice& prefix, const bool no_io,
                       const Slice* const const_ikey_ptr,
                       GetContext* get_context,
-                      BlockCacheLookupContext* lookup_context) override;
+                      BlockCacheLookupContext* lookup_context,
+                      Env::IOPriority rate_limiter_priority) override;
 
   void KeysMayMatch(MultiGetRange* range, const bool no_io,
-                    BlockCacheLookupContext* lookup_context) override;
+                    BlockCacheLookupContext* lookup_context,
+                    Env::IOPriority rate_limiter_priority) override;
   // Used in partitioned filter code
   void KeysMayMatch2(MultiGetRange* range,
                      const SliceTransform* /*prefix_extractor*/,
-                     const bool no_io,
-                     BlockCacheLookupContext* lookup_context) {
-    KeysMayMatch(range, no_io, lookup_context);
+                     const bool no_io, BlockCacheLookupContext* lookup_context,
+                     Env::IOPriority rate_limiter_priority) {
+    KeysMayMatch(range, no_io, lookup_context, rate_limiter_priority);
   }
 
   void PrefixesMayMatch(MultiGetRange* range,
                         const SliceTransform* prefix_extractor,
                         const bool no_io,
-                        BlockCacheLookupContext* lookup_context) override;
+                        BlockCacheLookupContext* lookup_context,
+                        Env::IOPriority rate_limiter_priority) override;
   size_t ApproximateMemoryUsage() const override;
  private:
   bool MayMatch(const Slice& entry, bool no_io, GetContext* get_context,
-                BlockCacheLookupContext* lookup_context) const;
+                BlockCacheLookupContext* lookup_context,
+                Env::IOPriority rate_limiter_priority) const;
   void MayMatch(MultiGetRange* range, bool no_io,
                 const SliceTransform* prefix_extractor,
-                BlockCacheLookupContext* lookup_context) const;
+                BlockCacheLookupContext* lookup_context,
+                Env::IOPriority rate_limiter_priority) const;
 };
 
 }  // namespace ROCKSDB_NAMESPACE

table/block_based/full_filter_block_test.cc

@@ -118,7 +118,7 @@ TEST_F(PluginFullFilterBlockTest, PluginEmptyBuilder) {
   ASSERT_TRUE(reader.KeyMayMatch("foo",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr, Env::IO_TOTAL));
 }
 
 TEST_F(PluginFullFilterBlockTest, PluginSingleChunk) {
@@ -136,34 +136,42 @@ TEST_F(PluginFullFilterBlockTest, PluginSingleChunk) {
       nullptr /* cache */, nullptr /* cache_handle */, true /* own_value */);
 
   FullFilterBlockReader reader(table_.get(), std::move(block));
+  Env::IOPriority rate_limiter_priority = Env::IO_TOTAL;
   ASSERT_TRUE(reader.KeyMayMatch("foo",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr,
+                                 rate_limiter_priority));
   ASSERT_TRUE(reader.KeyMayMatch("bar",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr,
+                                 rate_limiter_priority));
   ASSERT_TRUE(reader.KeyMayMatch("box",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr,
+                                 rate_limiter_priority));
   ASSERT_TRUE(reader.KeyMayMatch("hello",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr,
+                                 rate_limiter_priority));
   ASSERT_TRUE(reader.KeyMayMatch("foo",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr,
+                                 rate_limiter_priority));
   ASSERT_TRUE(!reader.KeyMayMatch("missing",
                                   /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                   /*get_context=*/nullptr,
-                                  /*lookup_context=*/nullptr));
+                                  /*lookup_context=*/nullptr,
+                                  rate_limiter_priority));
   ASSERT_TRUE(!reader.KeyMayMatch("other",
                                   /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                   /*get_context=*/nullptr,
-                                  /*lookup_context=*/nullptr));
+                                  /*lookup_context=*/nullptr,
+                                  rate_limiter_priority));
 }
 
 class FullFilterBlockTest : public mock::MockBlockBasedTableTester,
@@ -188,7 +196,7 @@ TEST_F(FullFilterBlockTest, EmptyBuilder) {
   ASSERT_TRUE(reader.KeyMayMatch("foo",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr, Env::IO_TOTAL));
 }
 
 class CountUniqueFilterBitsBuilderWrapper : public FilterBitsBuilder {
@@ -285,34 +293,42 @@ TEST_F(FullFilterBlockTest, SingleChunk) {
       nullptr /* cache */, nullptr /* cache_handle */, true /* own_value */);
 
   FullFilterBlockReader reader(table_.get(), std::move(block));
+  Env::IOPriority rate_limiter_priority = Env::IO_TOTAL;
   ASSERT_TRUE(reader.KeyMayMatch("foo",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr,
+                                 rate_limiter_priority));
   ASSERT_TRUE(reader.KeyMayMatch("bar",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr,
+                                 rate_limiter_priority));
   ASSERT_TRUE(reader.KeyMayMatch("box",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr,
+                                 rate_limiter_priority));
   ASSERT_TRUE(reader.KeyMayMatch("hello",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr,
+                                 rate_limiter_priority));
   ASSERT_TRUE(reader.KeyMayMatch("foo",
                                  /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                  /*get_context=*/nullptr,
-                                 /*lookup_context=*/nullptr));
+                                 /*lookup_context=*/nullptr,
+                                 rate_limiter_priority));
   ASSERT_TRUE(!reader.KeyMayMatch("missing",
                                   /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                   /*get_context=*/nullptr,
-                                  /*lookup_context=*/nullptr));
+                                  /*lookup_context=*/nullptr,
+                                  rate_limiter_priority));
   ASSERT_TRUE(!reader.KeyMayMatch("other",
                                   /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                                   /*get_context=*/nullptr,
-                                  /*lookup_context=*/nullptr));
+                                  /*lookup_context=*/nullptr,
+                                  rate_limiter_priority));
 }
 
 }  // namespace ROCKSDB_NAMESPACE

table/block_based/partitioned_filter_block.cc

@@ -217,41 +217,46 @@ std::unique_ptr<FilterBlockReader> PartitionedFilterBlockReader::Create(
 
 bool PartitionedFilterBlockReader::KeyMayMatch(
     const Slice& key, const bool no_io, const Slice* const const_ikey_ptr,
-    GetContext* get_context, BlockCacheLookupContext* lookup_context) {
+    GetContext* get_context, BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) {
   assert(const_ikey_ptr != nullptr);
   if (!whole_key_filtering()) {
     return true;
   }
 
   return MayMatch(key, no_io, const_ikey_ptr, get_context, lookup_context,
-                  &FullFilterBlockReader::KeyMayMatch);
+                  rate_limiter_priority, &FullFilterBlockReader::KeyMayMatch);
 }
 
 void PartitionedFilterBlockReader::KeysMayMatch(
     MultiGetRange* range, const bool no_io,
-    BlockCacheLookupContext* lookup_context) {
+    BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) {
   if (!whole_key_filtering()) {
     return;  // Any/all may match
   }
 
-  MayMatch(range, nullptr, no_io, lookup_context,
+  MayMatch(range, nullptr, no_io, lookup_context, rate_limiter_priority,
            &FullFilterBlockReader::KeysMayMatch2);
 }
 
 bool PartitionedFilterBlockReader::PrefixMayMatch(
     const Slice& prefix, const bool no_io, const Slice* const const_ikey_ptr,
-    GetContext* get_context, BlockCacheLookupContext* lookup_context) {
+    GetContext* get_context, BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) {
   assert(const_ikey_ptr != nullptr);
   return MayMatch(prefix, no_io, const_ikey_ptr, get_context, lookup_context,
+                  rate_limiter_priority,
                   &FullFilterBlockReader::PrefixMayMatch);
 }
 
 void PartitionedFilterBlockReader::PrefixesMayMatch(
     MultiGetRange* range, const SliceTransform* prefix_extractor,
-    const bool no_io, BlockCacheLookupContext* lookup_context) {
+    const bool no_io, BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority) {
   assert(prefix_extractor);
   MayMatch(range, prefix_extractor, no_io, lookup_context,
-           &FullFilterBlockReader::PrefixesMayMatch);
+           rate_limiter_priority, &FullFilterBlockReader::PrefixesMayMatch);
 }
 
 BlockHandle PartitionedFilterBlockReader::GetFilterPartitionHandle(
@@ -316,11 +321,12 @@ Status PartitionedFilterBlockReader::GetFilterPartitionBlock(
 bool PartitionedFilterBlockReader::MayMatch(
     const Slice& slice, bool no_io, const Slice* const_ikey_ptr,
     GetContext* get_context, BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority,
     FilterFunction filter_function) const {
   CachableEntry<Block> filter_block;
-  Status s =
-      GetOrReadFilterBlock(no_io, get_context, lookup_context, &filter_block,
-                           BlockType::kFilterPartitionIndex);
+  Status s = GetOrReadFilterBlock(
+      no_io, get_context, lookup_context, &filter_block,
+      BlockType::kFilterPartitionIndex, rate_limiter_priority);
   if (UNLIKELY(!s.ok())) {
     IGNORE_STATUS_IF_ERROR(s);
     return true;
@@ -347,17 +353,19 @@ bool PartitionedFilterBlockReader::MayMatch(
   FullFilterBlockReader filter_partition(table(),
                                          std::move(filter_partition_block));
   return (filter_partition.*filter_function)(slice, no_io, const_ikey_ptr,
-                                             get_context, lookup_context);
+                                             get_context, lookup_context,
+                                             rate_limiter_priority);
 }
 
 void PartitionedFilterBlockReader::MayMatch(
     MultiGetRange* range, const SliceTransform* prefix_extractor, bool no_io,
     BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority,
     FilterManyFunction filter_function) const {
   CachableEntry<Block> filter_block;
-  Status s =
-      GetOrReadFilterBlock(no_io, range->begin()->get_context, lookup_context,
-                           &filter_block, BlockType::kFilterPartitionIndex);
+  Status s = GetOrReadFilterBlock(
+      no_io, range->begin()->get_context, lookup_context, &filter_block,
+      BlockType::kFilterPartitionIndex, rate_limiter_priority);
   if (UNLIKELY(!s.ok())) {
     IGNORE_STATUS_IF_ERROR(s);
     return;  // Any/all may match
@@ -381,7 +389,7 @@ void PartitionedFilterBlockReader::MayMatch(
         this_filter_handle != prev_filter_handle) {
       MultiGetRange subrange(*range, start_iter_same_handle, iter);
       MayMatchPartition(&subrange, prefix_extractor, prev_filter_handle, no_io,
-                        lookup_context, filter_function);
+                        lookup_context, rate_limiter_priority, filter_function);
       range->AddSkipsFrom(subrange);
       start_iter_same_handle = iter;
     }
@@ -397,7 +405,7 @@ void PartitionedFilterBlockReader::MayMatch(
   if (!prev_filter_handle.IsNull()) {
     MultiGetRange subrange(*range, start_iter_same_handle, range->end());
     MayMatchPartition(&subrange, prefix_extractor, prev_filter_handle, no_io,
-                      lookup_context, filter_function);
+                      lookup_context, rate_limiter_priority, filter_function);
     range->AddSkipsFrom(subrange);
   }
 }
@@ -406,6 +414,7 @@ void PartitionedFilterBlockReader::MayMatchPartition(
     MultiGetRange* range, const SliceTransform* prefix_extractor,
     BlockHandle filter_handle, bool no_io,
     BlockCacheLookupContext* lookup_context,
+    Env::IOPriority rate_limiter_priority,
     FilterManyFunction filter_function) const {
   CachableEntry<ParsedFullFilterBlock> filter_partition_block;
   Status s = GetFilterPartitionBlock(
@@ -419,7 +428,7 @@ void PartitionedFilterBlockReader::MayMatchPartition(
   FullFilterBlockReader filter_partition(table(),
                                          std::move(filter_partition_block));
   (filter_partition.*filter_function)(range, prefix_extractor, no_io,
-                                      lookup_context);
+                                      lookup_context, rate_limiter_priority);
 }
 
 size_t PartitionedFilterBlockReader::ApproximateMemoryUsage() const {
@@ -447,7 +456,8 @@ Status PartitionedFilterBlockReader::CacheDependencies(const ReadOptions& ro,
 
   Status s = GetOrReadFilterBlock(false /* no_io */, nullptr /* get_context */,
                                   &lookup_context, &filter_block,
-                                  BlockType::kFilterPartitionIndex);
+                                  BlockType::kFilterPartitionIndex,
+                                  ro.rate_limiter_priority);
   if (!s.ok()) {
     ROCKS_LOG_ERROR(rep->ioptions.logger,
                     "Error retrieving top-level filter block while trying to "

table/block_based/partitioned_filter_block.h

@@ -111,18 +111,22 @@ class PartitionedFilterBlockReader : public FilterBlockReaderCommon<Block> {
 
   bool KeyMayMatch(const Slice& key, const bool no_io,
                    const Slice* const const_ikey_ptr, GetContext* get_context,
-                   BlockCacheLookupContext* lookup_context) override;
+                   BlockCacheLookupContext* lookup_context,
+                   Env::IOPriority rate_limiter_priority) override;
   void KeysMayMatch(MultiGetRange* range, const bool no_io,
-                    BlockCacheLookupContext* lookup_context) override;
+                    BlockCacheLookupContext* lookup_context,
+                    Env::IOPriority rate_limiter_priority) override;
 
   bool PrefixMayMatch(const Slice& prefix, const bool no_io,
                       const Slice* const const_ikey_ptr,
                       GetContext* get_context,
-                      BlockCacheLookupContext* lookup_context) override;
+                      BlockCacheLookupContext* lookup_context,
+                      Env::IOPriority rate_limiter_priority) override;
   void PrefixesMayMatch(MultiGetRange* range,
                         const SliceTransform* prefix_extractor,
                         const bool no_io,
-                        BlockCacheLookupContext* lookup_context) override;
+                        BlockCacheLookupContext* lookup_context,
+                        Env::IOPriority rate_limiter_priority) override;
 
   size_t ApproximateMemoryUsage() const override;
 
@@ -137,21 +141,26 @@ class PartitionedFilterBlockReader : public FilterBlockReaderCommon<Block> {
 
   using FilterFunction = bool (FullFilterBlockReader::*)(
       const Slice& slice, const bool no_io, const Slice* const const_ikey_ptr,
-      GetContext* get_context, BlockCacheLookupContext* lookup_context);
+      GetContext* get_context, BlockCacheLookupContext* lookup_context,
+      Env::IOPriority rate_limiter_priority);
   bool MayMatch(const Slice& slice, bool no_io, const Slice* const_ikey_ptr,
                 GetContext* get_context,
                 BlockCacheLookupContext* lookup_context,
+                Env::IOPriority rate_limiter_priority,
                 FilterFunction filter_function) const;
   using FilterManyFunction = void (FullFilterBlockReader::*)(
       MultiGetRange* range, const SliceTransform* prefix_extractor,
-      const bool no_io, BlockCacheLookupContext* lookup_context);
+      const bool no_io, BlockCacheLookupContext* lookup_context,
+      Env::IOPriority rate_limiter_priority);
   void MayMatch(MultiGetRange* range, const SliceTransform* prefix_extractor,
                 bool no_io, BlockCacheLookupContext* lookup_context,
+                Env::IOPriority rate_limiter_priority,
                 FilterManyFunction filter_function) const;
   void MayMatchPartition(MultiGetRange* range,
                          const SliceTransform* prefix_extractor,
                          BlockHandle filter_handle, bool no_io,
                          BlockCacheLookupContext* lookup_context,
+                         Env::IOPriority rate_limiter_priority,
                          FilterManyFunction filter_function) const;
   Status CacheDependencies(const ReadOptions& ro, bool pin) override;
 

table/block_based/partitioned_filter_block_test.cc

@@ -164,6 +164,7 @@ class PartitionedFilterBlockTest
                     PartitionedIndexBuilder* pib, bool empty = false) {
     std::unique_ptr<PartitionedFilterBlockReader> reader(
         NewReader(builder, pib));
+    Env::IOPriority rate_limiter_priority = Env::IO_TOTAL;
     // Querying added keys
     const bool no_io = true;
     for (auto key : keys) {
@@ -171,7 +172,8 @@ class PartitionedFilterBlockTest
       const Slice ikey_slice = Slice(*ikey.rep());
       ASSERT_TRUE(reader->KeyMayMatch(key, !no_io, &ikey_slice,
                                       /*get_context=*/nullptr,
-                                      /*lookup_context=*/nullptr));
+                                      /*lookup_context=*/nullptr,
+                                      rate_limiter_priority));
     }
     {
       // querying a key twice
@@ -179,7 +181,8 @@ class PartitionedFilterBlockTest
       const Slice ikey_slice = Slice(*ikey.rep());
       ASSERT_TRUE(reader->KeyMayMatch(keys[0], !no_io, &ikey_slice,
                                       /*get_context=*/nullptr,
-                                      /*lookup_context=*/nullptr));
+                                      /*lookup_context=*/nullptr,
+                                      rate_limiter_priority));
     }
     // querying missing keys
     for (auto key : missing_keys) {
@@ -188,12 +191,14 @@ class PartitionedFilterBlockTest
       if (empty) {
         ASSERT_TRUE(reader->KeyMayMatch(key, !no_io, &ikey_slice,
                                         /*get_context=*/nullptr,
-                                        /*lookup_context=*/nullptr));
+                                        /*lookup_context=*/nullptr,
+                                        rate_limiter_priority));
       } else {
         // assuming a good hash function
         ASSERT_FALSE(reader->KeyMayMatch(key, !no_io, &ikey_slice,
                                          /*get_context=*/nullptr,
-                                         /*lookup_context=*/nullptr));
+                                         /*lookup_context=*/nullptr,
+                                         rate_limiter_priority));
       }
     }
   }
@@ -345,7 +350,8 @@ TEST_P(PartitionedFilterBlockTest, SamePrefixInMultipleBlocks) {
     ASSERT_TRUE(reader->PrefixMayMatch(prefix_extractor->Transform(key),
                                        /*no_io=*/false, &ikey_slice,
                                        /*get_context=*/nullptr,
-                                       /*lookup_context=*/nullptr));
+                                       /*lookup_context=*/nullptr,
+                                       Env::IO_TOTAL));
   }
   // Non-existent keys but with the same prefix
   const std::string pnonkeys[4] = {"p-key9", "p-key11", "p-key21", "p-key31"};
@@ -355,7 +361,8 @@ TEST_P(PartitionedFilterBlockTest, SamePrefixInMultipleBlocks) {
     ASSERT_TRUE(reader->PrefixMayMatch(prefix_extractor->Transform(key),
                                        /*no_io=*/false, &ikey_slice,
                                        /*get_context=*/nullptr,
-                                       /*lookup_context=*/nullptr));
+                                       /*lookup_context=*/nullptr,
+                                       Env::IO_TOTAL));
   }
 }
 
@@ -386,6 +393,7 @@ TEST_P(PartitionedFilterBlockTest, PrefixInWrongPartitionBug) {
   CutABlock(pib.get(), pkeys[4]);
   std::unique_ptr<PartitionedFilterBlockReader> reader(
       NewReader(builder.get(), pib.get()));
+  Env::IOPriority rate_limiter_priority = Env::IO_TOTAL;
   for (auto key : pkeys) {
     auto prefix = prefix_extractor->Transform(key);
     auto ikey = InternalKey(prefix, 0, ValueType::kTypeValue);
@@ -393,7 +401,8 @@ TEST_P(PartitionedFilterBlockTest, PrefixInWrongPartitionBug) {
     ASSERT_TRUE(reader->PrefixMayMatch(prefix,
                                        /*no_io=*/false, &ikey_slice,
                                        /*get_context=*/nullptr,
-                                       /*lookup_context=*/nullptr));
+                                       /*lookup_context=*/nullptr,
+                                       rate_limiter_priority));
   }
 }
 

tools/db_bench_tool.cc

@@ -1657,6 +1657,9 @@ DEFINE_uint32(write_batch_protection_bytes_per_key, 0,
               "Size of per-key-value checksum in each write batch. Currently "
               "only value 0 and 8 are supported.");
 
+DEFINE_bool(track_and_verify_wals_in_manifest, false,
+            "If true, enable WAL tracking in the MANIFEST");
+
 namespace ROCKSDB_NAMESPACE {
 namespace {
 static Status CreateMemTableRepFactory(
@@ -4459,6 +4462,8 @@ class Benchmark {
     }
 
     options.allow_data_in_errors = FLAGS_allow_data_in_errors;
+    options.track_and_verify_wals_in_manifest =
+        FLAGS_track_and_verify_wals_in_manifest;
 
     // Integrated BlobDB
     options.enable_blob_files = FLAGS_enable_blob_files;

tools/db_crashtest.py

@@ -63,6 +63,7 @@ default_params = {
     "clear_column_family_one_in": 0,
     "compact_files_one_in": 1000000,
     "compact_range_one_in": 1000000,
+    "data_block_index_type": lambda: random.choice([0, 1]),
     "delpercent": 4,
     "delrangepercent": 1,
     "destroy_db_initially": 0,

util/filter_bench.cc

@@ -20,6 +20,7 @@ int main() {
 #include "port/port.h"
 #include "port/stack_trace.h"
 #include "rocksdb/cache.h"
+#include "rocksdb/env.h"
 #include "rocksdb/system_clock.h"
 #include "rocksdb/table.h"
 #include "table/block_based/filter_policy_internal.h"
@@ -150,6 +151,7 @@ using ROCKSDB_NAMESPACE::Cache;
 using ROCKSDB_NAMESPACE::CacheEntryRole;
 using ROCKSDB_NAMESPACE::CacheEntryRoleOptions;
 using ROCKSDB_NAMESPACE::EncodeFixed32;
+using ROCKSDB_NAMESPACE::Env;
 using ROCKSDB_NAMESPACE::FastRange32;
 using ROCKSDB_NAMESPACE::FilterBitsReader;
 using ROCKSDB_NAMESPACE::FilterBuildingContext;
@@ -726,7 +728,7 @@ double FilterBench::RandomQueryTest(uint32_t inside_threshold, bool dry_run,
                 batch_slices[i],
                 /*no_io=*/false, /*const_ikey_ptr=*/nullptr,
                 /*get_context=*/nullptr,
-                /*lookup_context=*/nullptr);
+                /*lookup_context=*/nullptr, Env::IO_TOTAL);
           }
         } else {
           if (dry_run) {

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/backup/backup_engine_test.cc

@@ -3066,7 +3066,7 @@ TEST_F(BackupEngineTest, OpenBackupAsReadOnlyDB) {
   db = nullptr;
 
   // Now try opening read-write and make sure it fails, for safety.
-  ASSERT_TRUE(DB::Open(opts, name, &db).IsAborted());
+  ASSERT_TRUE(DB::Open(opts, name, &db).IsIOError());
 }
 
 TEST_F(BackupEngineTest, ProgressCallbackDuringBackup) {

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