Group rocksdb.sst.read.micros stat by different user read IOActivity …

…+ misc (#11444)

Summary:
**Context/Summary:**
- Similar to #11288 but for user read such as `Get(), MultiGet(), DBIterator::XXX(), Verify(File)Checksum()`.
   - For this, I refactored some user-facing `MultiGet` calls in `TransactionBase` and various types of `DB` so that it does not call a user-facing `Get()` but `GetImpl()` for passing the `ReadOptions::io_activity` check (see PR conversation)
   - New user read stats breakdown are guarded by `kExceptDetailedTimers` since measurement shows they have 4-5% regression to the upstream/main.

- Misc
   - More refactoring: with #11288, we complete passing `ReadOptions/IOOptions` to FS level. So we can now replace the previously [added](#9424) `rate_limiter_priority` parameter in `RandomAccessFileReader`'s `Read/MultiRead/Prefetch()` with `IOOptions::rate_limiter_priority`
   - Also, `ReadAsync()` call time is measured in `SST_READ_MICRO` now

Pull Request resolved: #11444

Test Plan:
- CI fake db crash/stress test
- Microbenchmarking

**Build** `make clean && ROCKSDB_NO_FBCODE=1 DEBUG_LEVEL=0 make -jN db_basic_bench`
- google benchmark version: google/benchmark@604f6fd
- db_basic_bench_base: upstream
- db_basic_bench_pr: db_basic_bench_base + this PR
- asyncread_db_basic_bench_base: upstream + [db basic bench patch for IteratorNext](main...hx235:rocksdb:micro_bench_async_read)
- asyncread_db_basic_bench_pr: asyncread_db_basic_bench_base + this PR

**Test**

Get
```
TEST_TMPDIR=/dev/shm ./db_basic_bench_{null_stat|base|pr} --benchmark_filter=DBGet/comp_style:0/max_data:134217728/per_key_size:256/enable_statistics:1/negative_query:0/enable_filter:0/mmap:1/threads:1 --benchmark_repetitions=1000
```

Result
```
Coming soon
```

AsyncRead
```
TEST_TMPDIR=/dev/shm ./asyncread_db_basic_bench_{base|pr} --benchmark_filter=IteratorNext/comp_style:0/max_data:134217728/per_key_size:256/enable_statistics:1/async_io:1/include_detailed_timers:0 --benchmark_repetitions=1000 > syncread_db_basic_bench_{base|pr}.out
```

Result
```
Base:
1956,1956,1968,1977,1979,1986,1988,1988,1988,1990,1991,1991,1993,1993,1993,1993,1994,1996,1997,1997,1997,1998,1999,2001,2001,2002,2004,2007,2007,2008,

PR (2.3% regression, due to measuring `SST_READ_MICRO` that wasn't measured before):
1993,2014,2016,2022,2024,2027,2027,2028,2028,2030,2031,2031,2032,2032,2038,2039,2042,2044,2044,2047,2047,2047,2048,2049,2050,2052,2052,2052,2053,2053,
```

Reviewed By: ajkr

Differential Revision: D45918925

Pulled By: hx235

fbshipit-source-id: 58a54560d9ebeb3a59b6d807639692614dad058a

db/blob/blob_file_reader.cc

@@ -154,11 +154,9 @@ Status BlobFileReader::ReadHeader(const RandomAccessFileReader* file_reader,
     constexpr uint64_t read_offset = 0;
     constexpr size_t read_size = BlobLogHeader::kSize;
 
-    // TODO: rate limit reading headers from blob files.
     const Status s =
         ReadFromFile(file_reader, read_options, read_offset, read_size,
-                     statistics, &header_slice, &buf, &aligned_buf,
-                     Env::IO_TOTAL /* rate_limiter_priority */);
+                     statistics, &header_slice, &buf, &aligned_buf);
     if (!s.ok()) {
       return s;
     }
@@ -207,11 +205,9 @@ Status BlobFileReader::ReadFooter(const RandomAccessFileReader* file_reader,
     const uint64_t read_offset = file_size - BlobLogFooter::kSize;
     constexpr size_t read_size = BlobLogFooter::kSize;
 
-    // TODO: rate limit reading footers from blob files.
     const Status s =
         ReadFromFile(file_reader, read_options, read_offset, read_size,
-                     statistics, &footer_slice, &buf, &aligned_buf,
-                     Env::IO_TOTAL /* rate_limiter_priority */);
+                     statistics, &footer_slice, &buf, &aligned_buf);
     if (!s.ok()) {
       return s;
     }
@@ -242,8 +238,7 @@ Status BlobFileReader::ReadFromFile(const RandomAccessFileReader* file_reader,
                                     const ReadOptions& read_options,
                                     uint64_t read_offset, size_t read_size,
                                     Statistics* statistics, Slice* slice,
-                                    Buffer* buf, AlignedBuf* aligned_buf,
-                                    Env::IOPriority rate_limiter_priority) {
+                                    Buffer* buf, AlignedBuf* aligned_buf) {
   assert(slice);
   assert(buf);
   assert(aligned_buf);
@@ -264,13 +259,13 @@ Status BlobFileReader::ReadFromFile(const RandomAccessFileReader* file_reader,
     constexpr char* scratch = nullptr;
 
     s = file_reader->Read(io_options, read_offset, read_size, slice, scratch,
-                          aligned_buf, rate_limiter_priority);
+                          aligned_buf);
   } else {
     buf->reset(new char[read_size]);
     constexpr AlignedBuf* aligned_scratch = nullptr;
 
     s = file_reader->Read(io_options, read_offset, read_size, slice, buf->get(),
-                          aligned_scratch, rate_limiter_priority);
+                          aligned_scratch);
   }
 
   if (!s.ok()) {
@@ -345,8 +340,7 @@ Status BlobFileReader::GetBlob(
     }
     prefetched = prefetch_buffer->TryReadFromCache(
         io_options, file_reader_.get(), record_offset,
-        static_cast<size_t>(record_size), &record_slice, &s,
-        read_options.rate_limiter_priority, for_compaction);
+        static_cast<size_t>(record_size), &record_slice, &s, for_compaction);
     if (!s.ok()) {
       return s;
     }
@@ -357,10 +351,10 @@ Status BlobFileReader::GetBlob(
     PERF_COUNTER_ADD(blob_read_count, 1);
     PERF_COUNTER_ADD(blob_read_byte, record_size);
     PERF_TIMER_GUARD(blob_read_time);
-    const Status s = ReadFromFile(
-        file_reader_.get(), read_options, record_offset,
-        static_cast<size_t>(record_size), statistics_, &record_slice, &buf,
-        &aligned_buf, read_options.rate_limiter_priority);
+    const Status s =
+        ReadFromFile(file_reader_.get(), read_options, record_offset,
+                     static_cast<size_t>(record_size), statistics_,
+                     &record_slice, &buf, &aligned_buf);
     if (!s.ok()) {
       return s;
     }
@@ -468,9 +462,12 @@ void BlobFileReader::MultiGetBlob(
   TEST_SYNC_POINT("BlobFileReader::MultiGetBlob:ReadFromFile");
   PERF_COUNTER_ADD(blob_read_count, num_blobs);
   PERF_COUNTER_ADD(blob_read_byte, total_len);
-  s = file_reader_->MultiRead(IOOptions(), read_reqs.data(), read_reqs.size(),
-                              direct_io ? &aligned_buf : nullptr,
-                              read_options.rate_limiter_priority);
+  IOOptions opts;
+  s = file_reader_->PrepareIOOptions(read_options, opts);
+  if (s.ok()) {
+    s = file_reader_->MultiRead(opts, read_reqs.data(), read_reqs.size(),
+                                direct_io ? &aligned_buf : nullptr);
+  }
   if (!s.ok()) {
     for (auto& req : read_reqs) {
       req.status.PermitUncheckedError();

db/blob/blob_file_reader.h

@@ -89,8 +89,7 @@ class BlobFileReader {
                              const ReadOptions& read_options,
                              uint64_t read_offset, size_t read_size,
                              Statistics* statistics, Slice* slice, Buffer* buf,
-                             AlignedBuf* aligned_buf,
-                             Env::IOPriority rate_limiter_priority);
+                             AlignedBuf* aligned_buf);
 
   static Status VerifyBlob(const Slice& record_slice, const Slice& user_key,
                            uint64_t value_size);

db/blob/blob_log_sequential_reader.cc

@@ -29,9 +29,8 @@ Status BlobLogSequentialReader::ReadSlice(uint64_t size, Slice* slice,
 
   StopWatch read_sw(clock_, statistics_, BLOB_DB_BLOB_FILE_READ_MICROS);
   // TODO: rate limit `BlobLogSequentialReader` reads (it appears unused?)
-  Status s =
-      file_->Read(IOOptions(), next_byte_, static_cast<size_t>(size), slice,
-                  buf, nullptr, Env::IO_TOTAL /* rate_limiter_priority */);
+  Status s = file_->Read(IOOptions(), next_byte_, static_cast<size_t>(size),
+                         slice, buf, nullptr);
   next_byte_ += size;
   if (!s.ok()) {
     return s;

db/convenience.cc

@@ -7,6 +7,7 @@
 
 #include "rocksdb/convenience.h"
 
+#include "db/convenience_impl.h"
 #include "db/db_impl/db_impl.h"
 #include "util/cast_util.h"
 
@@ -39,9 +40,25 @@ Status VerifySstFileChecksum(const Options& options,
 }
 Status VerifySstFileChecksum(const Options& options,
                              const EnvOptions& env_options,
-                             const ReadOptions& read_options,
+                             const ReadOptions& _read_options,
                              const std::string& file_path,
                              const SequenceNumber& largest_seqno) {
+  if (_read_options.io_activity != Env::IOActivity::kUnknown) {
+    return Status::InvalidArgument(
+        "Can only call VerifySstFileChecksum with `ReadOptions::io_activity` "
+        "is "
+        "`Env::IOActivity::kUnknown`");
+  }
+  ReadOptions read_options(_read_options);
+  return VerifySstFileChecksumInternal(options, env_options, read_options,
+                                       file_path, largest_seqno);
+}
+
+Status VerifySstFileChecksumInternal(const Options& options,
+                                     const EnvOptions& env_options,
+                                     const ReadOptions& read_options,
+                                     const std::string& file_path,
+                                     const SequenceNumber& largest_seqno) {
   std::unique_ptr<FSRandomAccessFile> file;
   uint64_t file_size;
   InternalKeyComparator internal_comparator(options.comparator);
@@ -68,8 +85,8 @@ Status VerifySstFileChecksum(const Options& options,
       !kImmortal, false /* force_direct_prefetch */, -1 /* level */);
   reader_options.largest_seqno = largest_seqno;
   s = ioptions.table_factory->NewTableReader(
-      reader_options, std::move(file_reader), file_size, &table_reader,
-      false /* prefetch_index_and_filter_in_cache */);
+      read_options, reader_options, std::move(file_reader), file_size,
+      &table_reader, false /* prefetch_index_and_filter_in_cache */);
   if (!s.ok()) {
     return s;
   }

db/convenience_impl.h

@@ -0,0 +1,15 @@
+// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+#include "rocksdb/db.h"
+
+namespace ROCKSDB_NAMESPACE {
+Status VerifySstFileChecksumInternal(const Options& options,
+                                     const EnvOptions& env_options,
+                                     const ReadOptions& read_options,
+                                     const std::string& file_path,
+                                     const SequenceNumber& largest_seqno = 0);
+}  // namespace ROCKSDB_NAMESPACE

db/db_impl/compacted_db_impl.cc

@@ -43,18 +43,25 @@ Status CompactedDBImpl::Get(const ReadOptions& options, ColumnFamilyHandle*,
              /*timestamp*/ nullptr);
 }
 
-Status CompactedDBImpl::Get(const ReadOptions& options, ColumnFamilyHandle*,
-                            const Slice& key, PinnableSlice* value,
-                            std::string* timestamp) {
-  if (options.io_activity != Env::IOActivity::kUnknown) {
+Status CompactedDBImpl::Get(const ReadOptions& _read_options,
+                            ColumnFamilyHandle*, const Slice& key,
+                            PinnableSlice* value, std::string* timestamp) {
+  if (_read_options.io_activity != Env::IOActivity::kUnknown &&
+      _read_options.io_activity != Env::IOActivity::kGet) {
     return Status::InvalidArgument(
-        "Cannot call Get with `ReadOptions::io_activity` != "
-        "`Env::IOActivity::kUnknown`");
+        "Can only call Get with `ReadOptions::io_activity` is "
+        "`Env::IOActivity::kUnknown` or `Env::IOActivity::kGet`");
+  }
+  ReadOptions read_options(_read_options);
+  if (read_options.io_activity == Env::IOActivity::kUnknown) {
+    read_options.io_activity = Env::IOActivity::kGet;
   }
+
   assert(user_comparator_);
-  if (options.timestamp) {
-    const Status s = FailIfTsMismatchCf(
-        DefaultColumnFamily(), *(options.timestamp), /*ts_for_read=*/true);
+  if (read_options.timestamp) {
+    const Status s =
+        FailIfTsMismatchCf(DefaultColumnFamily(), *(read_options.timestamp),
+                           /*ts_for_read=*/true);
     if (!s.ok()) {
       return s;
     }
@@ -74,7 +81,7 @@ Status CompactedDBImpl::Get(const ReadOptions& options, ColumnFamilyHandle*,
   GetWithTimestampReadCallback read_cb(kMaxSequenceNumber);
   std::string* ts =
       user_comparator_->timestamp_size() > 0 ? timestamp : nullptr;
-  LookupKey lkey(key, kMaxSequenceNumber, options.timestamp);
+  LookupKey lkey(key, kMaxSequenceNumber, read_options.timestamp);
   GetContext get_context(user_comparator_, nullptr, nullptr, nullptr,
                          GetContext::kNotFound, lkey.user_key(), value,
                          /*columns=*/nullptr, ts, nullptr, nullptr, true,
@@ -88,8 +95,8 @@ Status CompactedDBImpl::Get(const ReadOptions& options, ColumnFamilyHandle*,
           /*b_has_ts=*/false) < 0) {
     return Status::NotFound();
   }
-  Status s = f.fd.table_reader->Get(options, lkey.internal_key(), &get_context,
-                                    nullptr);
+  Status s = f.fd.table_reader->Get(read_options, lkey.internal_key(),
+                                    &get_context, nullptr);
   if (!s.ok() && !s.IsNotFound()) {
     return s;
   }
@@ -106,15 +113,28 @@ std::vector<Status> CompactedDBImpl::MultiGet(
 }
 
 std::vector<Status> CompactedDBImpl::MultiGet(
-    const ReadOptions& options, const std::vector<ColumnFamilyHandle*>&,
+    const ReadOptions& _read_options, const std::vector<ColumnFamilyHandle*>&,
     const std::vector<Slice>& keys, std::vector<std::string>* values,
     std::vector<std::string>* timestamps) {
   assert(user_comparator_);
   size_t num_keys = keys.size();
+  if (_read_options.io_activity != Env::IOActivity::kUnknown &&
+      _read_options.io_activity != Env::IOActivity::kMultiGet) {
+    Status s = Status::InvalidArgument(
+        "Can only call MultiGet with `ReadOptions::io_activity` is "
+        "`Env::IOActivity::kUnknown` or `Env::IOActivity::kMultiGet`");
+    return std::vector<Status>(num_keys, s);
+  }
+
+  ReadOptions read_options(_read_options);
+  if (read_options.io_activity == Env::IOActivity::kUnknown) {
+    read_options.io_activity = Env::IOActivity::kMultiGet;
+  }
 
-  if (options.timestamp) {
-    Status s = FailIfTsMismatchCf(DefaultColumnFamily(), *(options.timestamp),
-                                  /*ts_for_read=*/true);
+  if (read_options.timestamp) {
+    Status s =
+        FailIfTsMismatchCf(DefaultColumnFamily(), *(read_options.timestamp),
+                           /*ts_for_read=*/true);
     if (!s.ok()) {
       return std::vector<Status>(num_keys, s);
     }
@@ -136,7 +156,7 @@ std::vector<Status> CompactedDBImpl::MultiGet(
   GetWithTimestampReadCallback read_cb(kMaxSequenceNumber);
   autovector<TableReader*, 16> reader_list;
   for (const auto& key : keys) {
-    LookupKey lkey(key, kMaxSequenceNumber, options.timestamp);
+    LookupKey lkey(key, kMaxSequenceNumber, read_options.timestamp);
     const FdWithKeyRange& f = files_.files[FindFile(lkey.user_key())];
     if (user_comparator_->CompareWithoutTimestamp(
             key, /*a_has_ts=*/false,
@@ -159,14 +179,15 @@ std::vector<Status> CompactedDBImpl::MultiGet(
     if (r != nullptr) {
       PinnableSlice pinnable_val;
       std::string& value = (*values)[idx];
-      LookupKey lkey(keys[idx], kMaxSequenceNumber, options.timestamp);
+      LookupKey lkey(keys[idx], kMaxSequenceNumber, read_options.timestamp);
       std::string* timestamp = timestamps ? &(*timestamps)[idx] : nullptr;
       GetContext get_context(
           user_comparator_, nullptr, nullptr, nullptr, GetContext::kNotFound,
           lkey.user_key(), &pinnable_val, /*columns=*/nullptr,
           user_comparator_->timestamp_size() > 0 ? timestamp : nullptr, nullptr,
           nullptr, true, nullptr, nullptr, nullptr, nullptr, &read_cb);
-      Status s = r->Get(options, lkey.internal_key(), &get_context, nullptr);
+      Status s =
+          r->Get(read_options, lkey.internal_key(), &get_context, nullptr);
       assert(static_cast<size_t>(idx) < statuses.size());
       if (!s.ok() && !s.IsNotFound()) {
         statuses[idx] = s;

db/db_impl/compacted_db_impl.h

@@ -30,9 +30,9 @@ class CompactedDBImpl : public DBImpl {
                      ColumnFamilyHandle* column_family, const Slice& key,
                      PinnableSlice* value) override;
 
-  Status Get(const ReadOptions& options, ColumnFamilyHandle* column_family,
-             const Slice& key, PinnableSlice* value,
-             std::string* timestamp) override;
+  Status Get(const ReadOptions& _read_options,
+             ColumnFamilyHandle* column_family, const Slice& key,
+             PinnableSlice* value, std::string* timestamp) override;
 
   using DB::MultiGet;
   // Note that CompactedDBImpl::MultiGet is not the optimized version of
@@ -43,7 +43,7 @@ class CompactedDBImpl : public DBImpl {
       const std::vector<Slice>& keys,
       std::vector<std::string>* values) override;
 
-  std::vector<Status> MultiGet(const ReadOptions& options,
+  std::vector<Status> MultiGet(const ReadOptions& _read_options,
                                const std::vector<ColumnFamilyHandle*>&,
                                const std::vector<Slice>& keys,
                                std::vector<std::string>* values,