Fix some errors in async prefetching in FilePrefetchBuffer (#9734)

Summary:
In ReadOption `async_io` which prefetches the data asynchronously, db_bench and db_stress runs were failing  because wrong data was prefetched which resulted in Error: Checksum mismatched. Wrong data was copied because capacity was less than actual size needed. It has been fixed in this PR.

Since there are two separate methods for async and sync prefetching, these changes are in async prefetching methods and any changes would not effect normal prefetching. I ran the regressions to make sure normal prefetching is fine.

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

Test Plan:
1. CircleCI jobs

2.  Ran db_bench
```
. /db_bench -use_existing_db=true
-db=/tmp/prefix_scan_prefetch_main -benchmarks="seekrandom" -key_size=32
-value_size=512 -num=5000000 -use_direct_reads=true -seek_nexts=327680
-duration=120 -ops_between_duration_checks=1 -async_io=1 -adaptive_readahead=1

```
3. Ran db_stress test
```
export CRASH_TEST_EXT_ARGS=" --async_io=1 --adaptive_readahead=1"
make crash_test -j
```

4. Run regressions for async_io disabled.

Old flow without any async changes:
```
./db_bench -use_existing_db=true -db=/tmp/prefix_scan_prefetch_main -benchmarks="seekrandom" -key_size=32 -value_size=512 -num=5000000 -use_direct_reads=true -seek_nexts=327680 -duration=120 -ops_between_duration_checks=1
Initializing RocksDB Options from the specified file
Initializing RocksDB Options from command-line flags
RocksDB:    version 7.0
Date:       Thu Mar 17 13:11:34 2022
CPU:        24 * Intel Core Processor (Broadwell)
CPUCache:   16384 KB
Keys:       32 bytes each (+ 0 bytes user-defined timestamp)
Values:     512 bytes each (256 bytes after compression)
Entries:    5000000
Prefix:    0 bytes
Keys per prefix:    0
RawSize:    2594.0 MB (estimated)
FileSize:   1373.3 MB (estimated)
Write rate: 0 bytes/second
Read rate: 0 ops/second
Compression: Snappy
Compression sampling rate: 0
Memtablerep: SkipListFactory
Perf Level: 1
------------------------------------------------
DB path: [/tmp/prefix_scan_prefetch_main]
seekrandom   :  483618.390 micros/op 2 ops/sec;  338.9 MB/s (249 of 249 found)
```

With async prefetching changes and async_io disabled to make sure in normal prefetching there is no regression.
 ```
 ./db_bench -use_existing_db=true -db=/tmp/prefix_scan_prefetch_main -benchmarks="seekrandom" -key_size=32 -value_size=512 -num=5000000 -use_direct_reads=true -seek_nexts=327680 -duration=120 -ops_between_duration_checks=1 --async_io=0
Initializing RocksDB Options from the specified file
Initializing RocksDB Options from command-line flags
RocksDB:    version 7.1
Date:       Wed Mar 23 15:56:37 2022
CPU:        24 * Intel Core Processor (Broadwell)
CPUCache:   16384 KB
Keys:       32 bytes each (+ 0 bytes user-defined timestamp)
Values:     512 bytes each (256 bytes after compression)
Entries:    5000000
Prefix:    0 bytes
Keys per prefix:    0
RawSize:    2594.0 MB (estimated)
FileSize:   1373.3 MB (estimated)
Write rate: 0 bytes/second
Read rate: 0 ops/second
Compression: Snappy
Compression sampling rate: 0
Memtablerep: SkipListFactory
Perf Level: 1
------------------------------------------------
DB path: [/tmp/prefix_scan_prefetch_main]
seekrandom   :  481819.816 micros/op 2 ops/sec;  340.2 MB/s (250 of 250 found)
```

Reviewed By: riversand963

Differential Revision: D35058471

Pulled By: akankshamahajan15

fbshipit-source-id: 9233a1e6d97cea0c7a8111bfb9e8ac3251c341ce

.circleci/config.yml

@@ -2,6 +2,7 @@ version: 2.1
 
 orbs:
   win: circleci/windows@2.4.0
+  slack: circleci/slack@3.4.2
 
 aliases:
   - &notify-on-main-failure
@@ -56,6 +57,7 @@ commands:
 
   post-steps:
     steps:
+      - slack/status: *notify-on-main-failure
       - store_test_results: # store test result if there's any
           path: /tmp/test-results
       - store_artifacts: # store LOG for debugging if there's any

HISTORY.md

@@ -1,17 +1,4 @@
 # Rocksdb Change Log
-## 7.1.2 (04/19/2022)
-### Bug Fixes
-* Fixed bug which caused rocksdb failure in the situation when rocksdb was accessible using UNC path
-* Fixed a race condition when 2PC is disabled and WAL tracking in the MANIFEST is enabled. The race condition is between two background flush threads trying to install flush results, causing a WAL deletion not tracked in the MANIFEST. A future DB open may fail.
-* Fixed a heap use-after-free race with DropColumnFamily.
-* Fixed a bug that `rocksdb.read.block.compaction.micros` cannot track compaction stats (#9722).
-* Fixed `file_type`, `relative_filename` and `directory` fields returned by `GetLiveFilesMetaData()`, which were added in inheriting from `FileStorageInfo`.
-* Fixed a bug affecting `track_and_verify_wals_in_manifest`. Without the fix, application may see "open error: Corruption: Missing WAL with log number" while trying to open the db. The corruption is a false alarm but prevents DB open (#9766).
-
-## 7.1.1 (04/07/2022)
-### Bug Fixes
-* Fix segfault in FilePrefetchBuffer with async_io as it doesn't wait for pending jobs to complete on destruction.
-
 ## 7.1.0 (03/23/2022)
 ### New Features
 * Allow WriteBatchWithIndex to index a WriteBatch that includes keys with user-defined timestamps. The index itself does not have timestamp.

Makefile

@@ -2044,8 +2044,8 @@ ROCKSDB_JAVADOCS_JAR = rocksdbjni-$(ROCKSDB_JAVA_VERSION)-javadoc.jar
 ROCKSDB_SOURCES_JAR = rocksdbjni-$(ROCKSDB_JAVA_VERSION)-sources.jar
 SHA256_CMD = sha256sum
 
-ZLIB_VER ?= 1.2.12
-ZLIB_SHA256 ?= 91844808532e5ce316b3c010929493c0244f3d37593afd6de04f71821d5136d9
+ZLIB_VER ?= 1.2.11
+ZLIB_SHA256 ?= c3e5e9fdd5004dcb542feda5ee4f0ff0744628baf8ed2dd5d66f8ca1197cb1a1
 ZLIB_DOWNLOAD_BASE ?= http://zlib.net
 BZIP2_VER ?= 1.0.8
 BZIP2_SHA256 ?= ab5a03176ee106d3f0fa90e381da478ddae405918153cca248e682cd0c4a2269

db/column_family.cc

@@ -1562,6 +1562,20 @@ ColumnFamilyData* ColumnFamilySet::CreateColumnFamily(
   return new_cfd;
 }
 
+// REQUIRES: DB mutex held
+void ColumnFamilySet::FreeDeadColumnFamilies() {
+  autovector<ColumnFamilyData*> to_delete;
+  for (auto cfd = dummy_cfd_->next_; cfd != dummy_cfd_; cfd = cfd->next_) {
+    if (cfd->refs_.load(std::memory_order_relaxed) == 0) {
+      to_delete.push_back(cfd);
+    }
+  }
+  for (auto cfd : to_delete) {
+    // this is very rare, so it's not a problem that we do it under a mutex
+    delete cfd;
+  }
+}
+
 // under a DB mutex AND from a write thread
 void ColumnFamilySet::RemoveColumnFamily(ColumnFamilyData* cfd) {
   auto cfd_iter = column_family_data_.find(cfd->GetID());

db/column_family.h

@@ -520,10 +520,9 @@ class ColumnFamilyData {
   ThreadLocalPtr* TEST_GetLocalSV() { return local_sv_.get(); }
   WriteBufferManager* write_buffer_mgr() { return write_buffer_manager_; }
 
-  static const uint32_t kDummyColumnFamilyDataId;
-
  private:
   friend class ColumnFamilySet;
+  static const uint32_t kDummyColumnFamilyDataId;
   ColumnFamilyData(uint32_t id, const std::string& name,
                    Version* dummy_versions, Cache* table_cache,
                    WriteBufferManager* write_buffer_manager,
@@ -629,8 +628,10 @@ class ColumnFamilyData {
 // held and it needs to be executed from the write thread. SetDropped() also
 // guarantees that it will be called only from single-threaded LogAndApply(),
 // but this condition is not that important.
-// * Iteration -- hold DB mutex. If you want to release the DB mutex in the
-// body of the iteration, wrap in a RefedColumnFamilySet.
+// * Iteration -- hold DB mutex, but you can release it in the body of
+// iteration. If you release DB mutex in body, reference the column
+// family before the mutex and unreference after you unlock, since the column
+// family might get dropped when the DB mutex is released
 // * GetDefault() -- thread safe
 // * GetColumnFamily() -- either inside of DB mutex or from a write thread
 // * GetNextColumnFamilyID(), GetMaxColumnFamily(), UpdateMaxColumnFamily(),
@@ -642,12 +643,17 @@ class ColumnFamilySet {
    public:
     explicit iterator(ColumnFamilyData* cfd)
         : current_(cfd) {}
-    // NOTE: minimum operators for for-loop iteration
     iterator& operator++() {
-      current_ = current_->next_;
+      // dropped column families might still be included in this iteration
+      // (we're only removing them when client drops the last reference to the
+      // column family).
+      // dummy is never dead, so this will never be infinite
+      do {
+        current_ = current_->next_;
+      } while (current_->refs_.load(std::memory_order_relaxed) == 0);
       return *this;
     }
-    bool operator!=(const iterator& other) const {
+    bool operator!=(const iterator& other) {
       return this->current_ != other.current_;
     }
     ColumnFamilyData* operator*() { return current_; }
@@ -686,6 +692,10 @@ class ColumnFamilySet {
   iterator begin() { return iterator(dummy_cfd_->next_); }
   iterator end() { return iterator(dummy_cfd_); }
 
+  // REQUIRES: DB mutex held
+  // Don't call while iterating over ColumnFamilySet
+  void FreeDeadColumnFamilies();
+
   Cache* get_table_cache() { return table_cache_; }
 
   WriteBufferManager* write_buffer_manager() { return write_buffer_manager_; }
@@ -728,55 +738,6 @@ class ColumnFamilySet {
   std::string db_session_id_;
 };
 
-// A wrapper for ColumnFamilySet that supports releasing DB mutex during each
-// iteration over the iterator, because the cfd is Refed and Unrefed during
-// each iteration to prevent concurrent CF drop from destroying it (until
-// Unref).
-class RefedColumnFamilySet {
- public:
-  explicit RefedColumnFamilySet(ColumnFamilySet* cfs) : wrapped_(cfs) {}
-
-  class iterator {
-   public:
-    explicit iterator(ColumnFamilySet::iterator wrapped) : wrapped_(wrapped) {
-      MaybeRef(*wrapped_);
-    }
-    ~iterator() { MaybeUnref(*wrapped_); }
-    inline void MaybeRef(ColumnFamilyData* cfd) {
-      if (cfd->GetID() != ColumnFamilyData::kDummyColumnFamilyDataId) {
-        cfd->Ref();
-      }
-    }
-    inline void MaybeUnref(ColumnFamilyData* cfd) {
-      if (cfd->GetID() != ColumnFamilyData::kDummyColumnFamilyDataId) {
-        cfd->UnrefAndTryDelete();
-      }
-    }
-    // NOTE: minimum operators for for-loop iteration
-    inline iterator& operator++() {
-      ColumnFamilyData* old = *wrapped_;
-      ++wrapped_;
-      // Can only unref & potentially free cfd after accessing its next_
-      MaybeUnref(old);
-      MaybeRef(*wrapped_);
-      return *this;
-    }
-    inline bool operator!=(const iterator& other) const {
-      return this->wrapped_ != other.wrapped_;
-    }
-    inline ColumnFamilyData* operator*() { return *wrapped_; }
-
-   private:
-    ColumnFamilySet::iterator wrapped_;
-  };
-
-  iterator begin() { return iterator(wrapped_->begin()); }
-  iterator end() { return iterator(wrapped_->end()); }
-
- private:
-  ColumnFamilySet* wrapped_;
-};
-
 // We use ColumnFamilyMemTablesImpl to provide WriteBatch a way to access
 // memtables of different column families (specified by ID in the write batch)
 class ColumnFamilyMemTablesImpl : public ColumnFamilyMemTables {

db/db_filesnapshot.cc

@@ -45,15 +45,17 @@ Status DBImpl::FlushForGetLiveFiles() {
     }
     mutex_.Lock();
   } else {
-    for (auto cfd : versions_->GetRefedColumnFamilySet()) {
+    for (auto cfd : *versions_->GetColumnFamilySet()) {
       if (cfd->IsDropped()) {
         continue;
       }
+      cfd->Ref();
       mutex_.Unlock();
       status = FlushMemTable(cfd, FlushOptions(), FlushReason::kGetLiveFiles);
       TEST_SYNC_POINT("DBImpl::GetLiveFiles:1");
       TEST_SYNC_POINT("DBImpl::GetLiveFiles:2");
       mutex_.Lock();
+      cfd->UnrefAndTryDelete();
       if (!status.ok() && !status.IsColumnFamilyDropped()) {
         break;
       } else if (status.IsColumnFamilyDropped()) {
@@ -61,6 +63,7 @@ Status DBImpl::FlushForGetLiveFiles() {
       }
     }
   }
+  versions_->GetColumnFamilySet()->FreeDeadColumnFamilies();
   return status;
 }
 

db/db_flush_test.cc

@@ -2271,7 +2271,7 @@ TEST_P(DBAtomicFlushTest, ManualFlushUnder2PC) {
 
   // The recovered min log number with prepared data should be non-zero.
   // In 2pc mode, MinLogNumberToKeep returns the
-  // VersionSet::min_log_number_to_keep recovered from MANIFEST, if it's 0,
+  // VersionSet::min_log_number_to_keep_2pc recovered from MANIFEST, if it's 0,
   // it means atomic flush didn't write the min_log_number_to_keep to MANIFEST.
   cfs.push_back(kDefaultColumnFamilyName);
   ASSERT_OK(TryReopenWithColumnFamilies(cfs, options));

db/db_impl/db_impl.cc

@@ -375,12 +375,15 @@ Status DBImpl::ResumeImpl(DBRecoverContext context) {
       s = AtomicFlushMemTables(cfds, flush_opts, context.flush_reason);
       mutex_.Lock();
     } else {
-      for (auto cfd : versions_->GetRefedColumnFamilySet()) {
+      for (auto cfd : *versions_->GetColumnFamilySet()) {
         if (cfd->IsDropped()) {
           continue;
         }
-        InstrumentedMutexUnlock u(&mutex_);
+        cfd->Ref();
+        mutex_.Unlock();
         s = FlushMemTable(cfd, flush_opts, context.flush_reason);
+        mutex_.Lock();
+        cfd->UnrefAndTryDelete();
         if (!s.ok()) {
           break;
         }
@@ -492,14 +495,18 @@ void DBImpl::CancelAllBackgroundWork(bool wait) {
       s.PermitUncheckedError();  //**TODO: What to do on error?
       mutex_.Lock();
     } else {
-      for (auto cfd : versions_->GetRefedColumnFamilySet()) {
+      for (auto cfd : *versions_->GetColumnFamilySet()) {
         if (!cfd->IsDropped() && cfd->initialized() && !cfd->mem()->IsEmpty()) {
-          InstrumentedMutexUnlock u(&mutex_);
+          cfd->Ref();
+          mutex_.Unlock();
           Status s = FlushMemTable(cfd, FlushOptions(), FlushReason::kShutDown);
           s.PermitUncheckedError();  //**TODO: What to do on error?
+          mutex_.Lock();
+          cfd->UnrefAndTryDelete();
         }
       }
     }
+    versions_->GetColumnFamilySet()->FreeDeadColumnFamilies();
   }
 
   shutting_down_.store(true, std::memory_order_release);
@@ -962,13 +969,18 @@ void DBImpl::DumpStats() {
   TEST_SYNC_POINT("DBImpl::DumpStats:StartRunning");
   {
     InstrumentedMutexLock l(&mutex_);
-    for (auto cfd : versions_->GetRefedColumnFamilySet()) {
+    for (auto cfd : *versions_->GetColumnFamilySet()) {
       if (cfd->initialized()) {
         // Release DB mutex for gathering cache entry stats. Pass over all
         // column families for this first so that other stats are dumped
         // near-atomically.
-        InstrumentedMutexUnlock u(&mutex_);
-        cfd->internal_stats()->CollectCacheEntryStats(/*foreground=*/false);
+        // Get a ref before unlocking
+        cfd->Ref();
+        {
+          InstrumentedMutexUnlock u(&mutex_);
+          cfd->internal_stats()->CollectCacheEntryStats(/*foreground=*/false);
+        }
+        cfd->UnrefAndTryDelete();
       }
     }
 
@@ -1888,8 +1900,6 @@ Status DBImpl::GetImpl(const ReadOptions& read_options, const Slice& key,
       return s;
     }
   }
-  TEST_SYNC_POINT("DBImpl::GetImpl:PostMemTableGet:0");
-  TEST_SYNC_POINT("DBImpl::GetImpl:PostMemTableGet:1");
   PinnedIteratorsManager pinned_iters_mgr;
   if (!done) {
     PERF_TIMER_GUARD(get_from_output_files_time);
@@ -1907,6 +1917,8 @@ Status DBImpl::GetImpl(const ReadOptions& read_options, const Slice& key,
   {
     PERF_TIMER_GUARD(get_post_process_time);
 
+    ReturnAndCleanupSuperVersion(cfd, sv);
+
     RecordTick(stats_, NUMBER_KEYS_READ);
     size_t size = 0;
     if (s.ok()) {
@@ -1933,8 +1945,6 @@ Status DBImpl::GetImpl(const ReadOptions& read_options, const Slice& key,
       PERF_COUNTER_ADD(get_read_bytes, size);
     }
     RecordInHistogram(stats_, BYTES_PER_READ, size);
-
-    ReturnAndCleanupSuperVersion(cfd, sv);
   }
   return s;
 }
@@ -3480,13 +3490,15 @@ bool DBImpl::GetAggregatedIntProperty(const Slice& property,
     // Needs mutex to protect the list of column families.
     InstrumentedMutexLock l(&mutex_);
     uint64_t value;
-    for (auto* cfd : versions_->GetRefedColumnFamilySet()) {
+    for (auto* cfd : *versions_->GetColumnFamilySet()) {
       if (!cfd->initialized()) {
         continue;
       }
+      cfd->Ref();
       ret = GetIntPropertyInternal(cfd, *property_info, true, &value);
       // GetIntPropertyInternal may release db mutex and re-acquire it.
       mutex_.AssertHeld();
+      cfd->UnrefAndTryDelete();
       if (ret) {
         sum += value;
       } else {
@@ -5077,7 +5089,6 @@ Status DBImpl::VerifyChecksumInternal(const ReadOptions& read_options,
     }
   }
 
-  // TODO: simplify using GetRefedColumnFamilySet?
   std::vector<ColumnFamilyData*> cfd_list;
   {
     InstrumentedMutexLock l(&mutex_);

db/db_impl/db_impl_compaction_flush.cc

@@ -2973,6 +2973,8 @@ void DBImpl::BackgroundCallCompaction(PrepickedCompaction* prepicked_compaction,
       bg_bottom_compaction_scheduled_--;
     }
 
+    versions_->GetColumnFamilySet()->FreeDeadColumnFamilies();
+
     // See if there's more work to be done
     MaybeScheduleFlushOrCompaction();
 

db/db_impl/db_impl_files.cc

@@ -23,7 +23,11 @@
 namespace ROCKSDB_NAMESPACE {
 
 uint64_t DBImpl::MinLogNumberToKeep() {
-  return versions_->min_log_number_to_keep();
+  if (allow_2pc()) {
+    return versions_->min_log_number_to_keep_2pc();
+  } else {
+    return versions_->MinLogNumberWithUnflushedData();
+  }
 }
 
 uint64_t DBImpl::MinObsoleteSstNumberToKeep() {
@@ -220,6 +224,7 @@ void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force,
     }
 
     // Add log files in wal_dir
+
     if (!immutable_db_options_.IsWalDirSameAsDBPath(dbname_)) {
       std::vector<std::string> log_files;
       Status s = env_->GetChildren(immutable_db_options_.wal_dir, &log_files);
@@ -229,7 +234,6 @@ void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force,
             log_file, immutable_db_options_.wal_dir);
       }
     }
-
     // Add info log files in db_log_dir
     if (!immutable_db_options_.db_log_dir.empty() &&
         immutable_db_options_.db_log_dir != dbname_) {

db/db_impl/db_impl_open.cc

@@ -866,11 +866,6 @@ Status DBImpl::RecoverLogFiles(const std::vector<uint64_t>& wal_numbers,
   bool flushed = false;
   uint64_t corrupted_wal_number = kMaxSequenceNumber;
   uint64_t min_wal_number = MinLogNumberToKeep();
-  if (!allow_2pc()) {
-    // In non-2pc mode, we skip WALs that do not back unflushed data.
-    min_wal_number =
-        std::max(min_wal_number, versions_->MinLogNumberWithUnflushedData());
-  }
   for (auto wal_number : wal_numbers) {
     if (wal_number < min_wal_number) {
       ROCKS_LOG_INFO(immutable_db_options_.info_log,
@@ -1275,16 +1270,9 @@ Status DBImpl::RecoverLogFiles(const std::vector<uint64_t>& wal_numbers,
       }
 
       std::unique_ptr<VersionEdit> wal_deletion;
-      if (flushed) {
-        wal_deletion = std::make_unique<VersionEdit>();
-        if (immutable_db_options_.track_and_verify_wals_in_manifest) {
-          wal_deletion->DeleteWalsBefore(max_wal_number + 1);
-        }
-        if (!allow_2pc()) {
-          // In non-2pc mode, flushing the memtables of the column families
-          // means we can advance min_log_number_to_keep.
-          wal_deletion->SetMinLogNumberToKeep(max_wal_number + 1);
-        }
+      if (immutable_db_options_.track_and_verify_wals_in_manifest) {
+        wal_deletion.reset(new VersionEdit);
+        wal_deletion->DeleteWalsBefore(max_wal_number + 1);
         edit_lists.back().push_back(wal_deletion.get());
       }
 
@@ -1363,14 +1351,7 @@ Status DBImpl::RestoreAliveLogFiles(const std::vector<uint64_t>& wal_numbers) {
   // FindObsoleteFiles()
   total_log_size_ = 0;
   log_empty_ = false;
-  uint64_t min_wal_with_unflushed_data =
-      versions_->MinLogNumberWithUnflushedData();
   for (auto wal_number : wal_numbers) {
-    if (!allow_2pc() && wal_number < min_wal_with_unflushed_data) {
-      // In non-2pc mode, the WAL files not backing unflushed data are not
-      // alive, thus should not be added to the alive_log_files_.
-      continue;
-    }
     // We preallocate space for wals, but then after a crash and restart, those
     // preallocated space are not needed anymore. It is likely only the last
     // log has such preallocated space, so we only truncate for the last log.
@@ -1971,10 +1952,10 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
           "DB::Open() failed --- Unable to persist Options file",
           persist_options_status.ToString());
     }
-  }
-  if (!s.ok()) {
+  } else {
     ROCKS_LOG_WARN(impl->immutable_db_options_.info_log,
-                   "DB::Open() failed: %s", s.ToString().c_str());
+                   "Persisting Option File error: %s",
+                   persist_options_status.ToString().c_str());
   }
   if (s.ok()) {
     s = impl->StartPeriodicWorkScheduler();

db/db_merge_operand_test.cc

@@ -47,7 +47,7 @@ class DBMergeOperandTest : public DBTestBase {
       : DBTestBase("db_merge_operand_test", /*env_do_fsync=*/true) {}
 };
 
-TEST_F(DBMergeOperandTest, CacheEvictedMergeOperandReadAfterFreeBug) {
+TEST_F(DBMergeOperandTest, MergeOperandReadAfterFreeBug) {
   // There was a bug of reading merge operands after they are mistakely freed
   // in DB::GetMergeOperands, which is surfaced by cache full.
   // See PR#9507 for more.
@@ -86,42 +86,6 @@ TEST_F(DBMergeOperandTest, CacheEvictedMergeOperandReadAfterFreeBug) {
   ASSERT_EQ(values[3].ToString(), "v4");
 }
 
-TEST_F(DBMergeOperandTest, FlushedMergeOperandReadAfterFreeBug) {
-  // Repro for a bug where a memtable containing a merge operand could be
-  // deleted before the merge operand was saved to the result.
-  auto options = CurrentOptions();
-  options.merge_operator = MergeOperators::CreateStringAppendOperator();
-  Reopen(options);
-
-  ASSERT_OK(Merge("key", "value"));
-
-  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency(
-      {{"DBImpl::GetImpl:PostMemTableGet:0",
-        "DBMergeOperandTest::FlushedMergeOperandReadAfterFreeBug:PreFlush"},
-       {"DBMergeOperandTest::FlushedMergeOperandReadAfterFreeBug:PostFlush",
-        "DBImpl::GetImpl:PostMemTableGet:1"}});
-  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
-
-  auto flush_thread = port::Thread([&]() {
-    TEST_SYNC_POINT(
-        "DBMergeOperandTest::FlushedMergeOperandReadAfterFreeBug:PreFlush");
-    ASSERT_OK(Flush());
-    TEST_SYNC_POINT(
-        "DBMergeOperandTest::FlushedMergeOperandReadAfterFreeBug:PostFlush");
-  });
-
-  PinnableSlice value;
-  GetMergeOperandsOptions merge_operands_info;
-  merge_operands_info.expected_max_number_of_operands = 1;
-  int number_of_operands;
-  ASSERT_OK(db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(),
-                                  "key", &value, &merge_operands_info,
-                                  &number_of_operands));
-  ASSERT_EQ(1, number_of_operands);
-
-  flush_thread.join();
-}
-
 TEST_F(DBMergeOperandTest, GetMergeOperandsBasic) {
   Options options;
   options.create_if_missing = true;

db/db_test.cc

@@ -1078,11 +1078,6 @@ void CheckColumnFamilyMeta(
       ASSERT_LE(file_meta_from_cf.file_creation_time, end_time);
       ASSERT_GE(file_meta_from_cf.oldest_ancester_time, start_time);
       ASSERT_LE(file_meta_from_cf.oldest_ancester_time, end_time);
-      // More from FileStorageInfo
-      ASSERT_EQ(file_meta_from_cf.file_type, kTableFile);
-      ASSERT_EQ(file_meta_from_cf.name,
-                "/" + file_meta_from_cf.relative_filename);
-      ASSERT_EQ(file_meta_from_cf.directory, file_meta_from_cf.db_path);
     }
 
     ASSERT_EQ(level_meta_from_cf.size, level_size);
@@ -1127,11 +1122,6 @@ void CheckLiveFilesMeta(
     ASSERT_EQ(meta.oldest_blob_file_number,
               expected_meta.oldest_blob_file_number);
 
-    // More from FileStorageInfo
-    ASSERT_EQ(meta.file_type, kTableFile);
-    ASSERT_EQ(meta.name, "/" + meta.relative_filename);
-    ASSERT_EQ(meta.directory, meta.db_path);
-
     ++i;
   }
 }

db/db_wal_test.cc

@@ -1491,93 +1491,6 @@ TEST_F(DBWALTest, kPointInTimeRecoveryCFConsistency) {
   ASSERT_NOK(TryReopenWithColumnFamilies({"default", "one", "two"}, options));
 }
 
-TEST_F(DBWALTest, RaceInstallFlushResultsWithWalObsoletion) {
-  Options options = CurrentOptions();
-  options.env = env_;
-  options.track_and_verify_wals_in_manifest = true;
-  // The following make sure there are two bg flush threads.
-  options.max_background_jobs = 8;
-
-  const std::string cf1_name("cf1");
-  CreateAndReopenWithCF({cf1_name}, options);
-  assert(handles_.size() == 2);
-
-  {
-    dbfull()->TEST_LockMutex();
-    ASSERT_LE(2, dbfull()->GetBGJobLimits().max_flushes);
-    dbfull()->TEST_UnlockMutex();
-  }
-
-  ASSERT_OK(dbfull()->PauseBackgroundWork());
-
-  ASSERT_OK(db_->Put(WriteOptions(), handles_[1], "foo", "value"));
-  ASSERT_OK(db_->Put(WriteOptions(), "foo", "value"));
-
-  ASSERT_OK(dbfull()->TEST_FlushMemTable(false, true, handles_[1]));
-
-  ASSERT_OK(db_->Put(WriteOptions(), "foo", "value"));
-  ASSERT_OK(dbfull()->TEST_FlushMemTable(false, true, handles_[0]));
-
-  bool called = false;
-  SyncPoint::GetInstance()->DisableProcessing();
-  SyncPoint::GetInstance()->ClearAllCallBacks();
-  // This callback will be called when the first bg flush thread reaches the
-  // point before entering the MANIFEST write queue after flushing the SST
-  // file.
-  // The purpose of the sync points here is to ensure both bg flush threads
-  // finish computing `min_wal_number_to_keep` before any of them updates the
-  // `log_number` for the column family that's being flushed.
-  SyncPoint::GetInstance()->SetCallBack(
-      "MemTableList::TryInstallMemtableFlushResults:AfterComputeMinWalToKeep",
-      [&](void* /*arg*/) {
-        dbfull()->mutex()->AssertHeld();
-        if (!called) {
-          // We are the first bg flush thread in the MANIFEST write queue.
-          // We set up the dependency between sync points for two threads that
-          // will be executing the same code.
-          // For the interleaving of events, see
-          // https://github.com/facebook/rocksdb/pull/9715.
-          // bg flush thread1 will release the db mutex while in the MANIFEST
-          // write queue. In the meantime, bg flush thread2 locks db mutex and
-          // computes the min_wal_number_to_keep (before thread1 writes to
-          // MANIFEST thus before cf1->log_number is updated). Bg thread2 joins
-          // the MANIFEST write queue afterwards and bg flush thread1 proceeds
-          // with writing to MANIFEST.
-          called = true;
-          SyncPoint::GetInstance()->LoadDependency({
-              {"VersionSet::LogAndApply:WriteManifestStart",
-               "DBWALTest::RaceInstallFlushResultsWithWalObsoletion:BgFlush2"},
-              {"DBWALTest::RaceInstallFlushResultsWithWalObsoletion:BgFlush2",
-               "VersionSet::LogAndApply:WriteManifest"},
-          });
-        } else {
-          // The other bg flush thread has already been in the MANIFEST write
-          // queue, and we are after.
-          TEST_SYNC_POINT(
-              "DBWALTest::RaceInstallFlushResultsWithWalObsoletion:BgFlush2");
-        }
-      });
-  SyncPoint::GetInstance()->EnableProcessing();
-
-  ASSERT_OK(dbfull()->ContinueBackgroundWork());
-
-  ASSERT_OK(dbfull()->TEST_WaitForFlushMemTable(handles_[0]));
-  ASSERT_OK(dbfull()->TEST_WaitForFlushMemTable(handles_[1]));
-
-  ASSERT_TRUE(called);
-
-  Close();
-
-  SyncPoint::GetInstance()->DisableProcessing();
-  SyncPoint::GetInstance()->ClearAllCallBacks();
-
-  DB* db1 = nullptr;
-  Status s = DB::OpenForReadOnly(options, dbname_, &db1);
-  ASSERT_OK(s);
-  assert(db1);
-  delete db1;
-}
-
 // Test scope:
 // - We expect to open data store under all circumstances
 // - We expect only data upto the point where the first error was encountered

db/event_helpers.cc

@@ -95,9 +95,8 @@ void EventHelpers::LogAndNotifyTableFileCreationFinished(
     jwriter << "cf_name" << cf_name << "job" << job_id << "event"
             << "table_file_creation"
             << "file_number" << fd.GetNumber() << "file_size"
-            << fd.GetFileSize() << "file_checksum"
-            << Slice(file_checksum).ToString(true) << "file_checksum_func_name"
-            << file_checksum_func_name;
+            << fd.GetFileSize() << "file_checksum" << file_checksum
+            << "file_checksum_func_name" << file_checksum_func_name;
 
     // table_properties
     {

db/filename_test.cc

@@ -171,69 +171,6 @@ TEST_F(FileNameTest, Construction) {
   ASSERT_EQ(kMetaDatabase, type);
 }
 
-TEST_F(FileNameTest, NormalizePath) {
-  // No leading slash
-  const std::string sep = std::string(1, kFilePathSeparator);
-
-  std::string expected = "FOLDER" + sep + "filename.ext";
-  std::string given = "FOLDER" + sep + "filename.ext";
-
-  ASSERT_EQ(expected, NormalizePath(given));
-
-  // Two chars /a
-
-  expected = sep + "a";
-  given = expected;
-  ASSERT_EQ(expected, NormalizePath(given));
-
-  // Two chars a/
-  expected = "a" + sep;
-  given = expected;
-  ASSERT_EQ(expected, NormalizePath(given));
-
-  // Server only
-  expected = sep + sep + "a";
-  given = expected;
-  ASSERT_EQ(expected, NormalizePath(given));
-
-  // Two slashes after character
-  expected = "a" + sep;
-  given = "a" + sep + sep;
-
-  ASSERT_EQ(expected, NormalizePath(given));
-
-  // slash only   /
-  expected = sep;
-  given = expected;
-  ASSERT_EQ(expected, NormalizePath(given));
-
-  // UNC only   //
-  expected = sep;
-  given = sep + sep;
-
-  ASSERT_EQ(expected, NormalizePath(given));
-
-  // 3 slashesy   //
-  expected = sep + sep;
-  given = sep + sep + sep;
-  ASSERT_EQ(expected, NormalizePath(given));
-
-  // 3 slashes   //
-  expected = sep + sep + "a" + sep;
-  given = sep + sep + sep + "a" + sep;
-  ASSERT_EQ(expected, NormalizePath(given));
-
-  // 2 separators in the middle
-  expected = "a" + sep + "b";
-  given = "a" + sep + sep + "b";
-  ASSERT_EQ(expected, NormalizePath(given));
-
-  // UNC with duplicate slashes
-  expected = sep + sep + "SERVER" + sep + "a" + sep + "b" + sep + "c";
-  given = sep + sep + "SERVER" + sep + "a" + sep + sep + "b" + sep + "c";
-  ASSERT_EQ(expected, NormalizePath(given));
-}
-
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

db/memtable_list.cc

@@ -489,8 +489,8 @@ Status MemTableList::TryInstallMemtableFlushResults(
     // TODO(myabandeh): Not sure how batch_count could be 0 here.
     if (batch_count > 0) {
       uint64_t min_wal_number_to_keep = 0;
-      assert(edit_list.size() > 0);
       if (vset->db_options()->allow_2pc) {
+        assert(edit_list.size() > 0);
         // Note that if mempurge is successful, the edit_list will
         // not be applicable (contains info of new min_log number to keep,
         // and level 0 file path of SST file created during normal flush,
@@ -499,26 +499,24 @@ Status MemTableList::TryInstallMemtableFlushResults(
         min_wal_number_to_keep = PrecomputeMinLogNumberToKeep2PC(
             vset, *cfd, edit_list, memtables_to_flush, prep_tracker);
 
-        // We piggyback the information of earliest log file to keep in the
+        // We piggyback the information of  earliest log file to keep in the
         // manifest entry for the last file flushed.
-      } else {
-        min_wal_number_to_keep =
-            PrecomputeMinLogNumberToKeepNon2PC(vset, *cfd, edit_list);
+        edit_list.back()->SetMinLogNumberToKeep(min_wal_number_to_keep);
       }
 
-      VersionEdit wal_deletion;
-      wal_deletion.SetMinLogNumberToKeep(min_wal_number_to_keep);
+      std::unique_ptr<VersionEdit> wal_deletion;
       if (vset->db_options()->track_and_verify_wals_in_manifest) {
+        if (!vset->db_options()->allow_2pc) {
+          min_wal_number_to_keep =
+              PrecomputeMinLogNumberToKeepNon2PC(vset, *cfd, edit_list);
+        }
         if (min_wal_number_to_keep >
             vset->GetWalSet().GetMinWalNumberToKeep()) {
-          wal_deletion.DeleteWalsBefore(min_wal_number_to_keep);
+          wal_deletion.reset(new VersionEdit);
+          wal_deletion->DeleteWalsBefore(min_wal_number_to_keep);
+          edit_list.push_back(wal_deletion.get());
         }
-        TEST_SYNC_POINT_CALLBACK(
-            "MemTableList::TryInstallMemtableFlushResults:"
-            "AfterComputeMinWalToKeep",
-            nullptr);
       }
-      edit_list.push_back(&wal_deletion);
 
       const auto manifest_write_cb = [this, cfd, batch_count, log_buffer,
                                       to_delete, mu](const Status& status) {
@@ -800,19 +798,22 @@ Status InstallMemtableAtomicFlushResults(
   if (vset->db_options()->allow_2pc) {
     min_wal_number_to_keep = PrecomputeMinLogNumberToKeep2PC(
         vset, cfds, edit_lists, mems_list, prep_tracker);
-  } else {
-    min_wal_number_to_keep =
-        PrecomputeMinLogNumberToKeepNon2PC(vset, cfds, edit_lists);
+    edit_lists.back().back()->SetMinLogNumberToKeep(min_wal_number_to_keep);
   }
 
-  VersionEdit wal_deletion;
-  wal_deletion.SetMinLogNumberToKeep(min_wal_number_to_keep);
-  if (vset->db_options()->track_and_verify_wals_in_manifest &&
-      min_wal_number_to_keep > vset->GetWalSet().GetMinWalNumberToKeep()) {
-    wal_deletion.DeleteWalsBefore(min_wal_number_to_keep);
+  std::unique_ptr<VersionEdit> wal_deletion;
+  if (vset->db_options()->track_and_verify_wals_in_manifest) {
+    if (!vset->db_options()->allow_2pc) {
+      min_wal_number_to_keep =
+          PrecomputeMinLogNumberToKeepNon2PC(vset, cfds, edit_lists);
+    }
+    if (min_wal_number_to_keep > vset->GetWalSet().GetMinWalNumberToKeep()) {
+      wal_deletion.reset(new VersionEdit);
+      wal_deletion->DeleteWalsBefore(min_wal_number_to_keep);
+      edit_lists.back().push_back(wal_deletion.get());
+      ++num_entries;
+    }
   }
-  edit_lists.back().push_back(&wal_deletion);
-  ++num_entries;
 
   // Mark the version edits as an atomic group if the number of version edits
   // exceeds 1.

db/version_edit.cc

@@ -120,9 +120,6 @@ bool VersionEdit::EncodeTo(std::string* dst) const {
   if (has_max_column_family_) {
     PutVarint32Varint32(dst, kMaxColumnFamily, max_column_family_);
   }
-  if (has_min_log_number_to_keep_) {
-    PutVarint32Varint64(dst, kMinLogNumberToKeep, min_log_number_to_keep_);
-  }
   if (has_last_sequence_) {
     PutVarint32Varint64(dst, kLastSequence, last_sequence_);
   }

db/version_edit_handler.cc

@@ -394,7 +394,7 @@ void VersionEditHandler::CheckIterationResult(const log::Reader& reader,
   if (s->ok()) {
     version_set_->GetColumnFamilySet()->UpdateMaxColumnFamily(
         version_edit_params_.max_column_family_);
-    version_set_->MarkMinLogNumberToKeep(
+    version_set_->MarkMinLogNumberToKeep2PC(
         version_edit_params_.min_log_number_to_keep_);
     version_set_->MarkFileNumberUsed(version_edit_params_.prev_log_number_);
     version_set_->MarkFileNumberUsed(version_edit_params_.log_number_);
@@ -970,11 +970,12 @@ void DumpManifestHandler::CheckIterationResult(const log::Reader& reader,
   fprintf(stdout,
           "next_file_number %" PRIu64 " last_sequence %" PRIu64
           "  prev_log_number %" PRIu64 " max_column_family %" PRIu32
-          " min_log_number_to_keep %" PRIu64 "\n",
+          " min_log_number_to_keep "
+          "%" PRIu64 "\n",
           version_set_->current_next_file_number(),
           version_set_->LastSequence(), version_set_->prev_log_number(),
           version_set_->column_family_set_->GetMaxColumnFamily(),
-          version_set_->min_log_number_to_keep());
+          version_set_->min_log_number_to_keep_2pc());
 }
 
 }  // namespace ROCKSDB_NAMESPACE

db/version_set.cc

@@ -1475,7 +1475,7 @@ void Version::GetColumnFamilyMetaData(ColumnFamilyMetaData* cf_meta) {
       const uint64_t file_number = file->fd.GetNumber();
       files.emplace_back(
           MakeTableFileName("", file_number), file_number, file_path,
-          file->fd.GetFileSize(), file->fd.smallest_seqno,
+          static_cast<size_t>(file->fd.GetFileSize()), file->fd.smallest_seqno,
           file->fd.largest_seqno, file->smallest.user_key().ToString(),
           file->largest.user_key().ToString(),
           file->stats.num_reads_sampled.load(std::memory_order_relaxed),
@@ -4141,7 +4141,7 @@ void VersionSet::Reset() {
   }
   db_id_.clear();
   next_file_number_.store(2);
-  min_log_number_to_keep_.store(0);
+  min_log_number_to_keep_2pc_.store(0);
   manifest_file_number_ = 0;
   options_file_number_ = 0;
   pending_manifest_file_number_ = 0;
@@ -4610,7 +4610,8 @@ Status VersionSet::ProcessManifestWrites(
       }
 
       if (last_min_log_number_to_keep != 0) {
-        MarkMinLogNumberToKeep(last_min_log_number_to_keep);
+        // Should only be set in 2PC mode.
+        MarkMinLogNumberToKeep2PC(last_min_log_number_to_keep);
       }
 
       for (int i = 0; i < static_cast<int>(versions.size()); ++i) {
@@ -4964,7 +4965,7 @@ Status VersionSet::Recover(
         ",min_log_number_to_keep is %" PRIu64 "\n",
         manifest_path.c_str(), manifest_file_number_, next_file_number_.load(),
         last_sequence_.load(), log_number, prev_log_number_,
-        column_family_set_->GetMaxColumnFamily(), min_log_number_to_keep());
+        column_family_set_->GetMaxColumnFamily(), min_log_number_to_keep_2pc());
 
     for (auto cfd : *column_family_set_) {
       if (cfd->IsDropped()) {
@@ -5377,9 +5378,9 @@ void VersionSet::MarkFileNumberUsed(uint64_t number) {
 }
 // Called only either from ::LogAndApply which is protected by mutex or during
 // recovery which is single-threaded.
-void VersionSet::MarkMinLogNumberToKeep(uint64_t number) {
-  if (min_log_number_to_keep_.load(std::memory_order_relaxed) < number) {
-    min_log_number_to_keep_.store(number, std::memory_order_relaxed);
+void VersionSet::MarkMinLogNumberToKeep2PC(uint64_t number) {
+  if (min_log_number_to_keep_2pc_.load(std::memory_order_relaxed) < number) {
+    min_log_number_to_keep_2pc_.store(number, std::memory_order_relaxed);
   }
 }
 
@@ -5505,7 +5506,7 @@ Status VersionSet::WriteCurrentStateToManifest(
         // min_log_number_to_keep is for the whole db, not for specific column family.
         // So it does not need to be set for every column family, just need to be set once.
         // Since default CF can never be dropped, we set the min_log to the default CF here.
-        uint64_t min_log = min_log_number_to_keep();
+        uint64_t min_log = min_log_number_to_keep_2pc();
         if (min_log != 0) {
           edit.SetMinLogNumberToKeep(min_log);
         }
@@ -5893,13 +5894,11 @@ void VersionSet::GetLiveFilesMetaData(std::vector<LiveFileMetaData>* metadata) {
           assert(!cfd->ioptions()->cf_paths.empty());
           filemetadata.db_path = cfd->ioptions()->cf_paths.back().path;
         }
-        filemetadata.directory = filemetadata.db_path;
         const uint64_t file_number = file->fd.GetNumber();
         filemetadata.name = MakeTableFileName("", file_number);
-        filemetadata.relative_filename = filemetadata.name.substr(1);
         filemetadata.file_number = file_number;
         filemetadata.level = level;
-        filemetadata.size = file->fd.GetFileSize();
+        filemetadata.size = static_cast<size_t>(file->fd.GetFileSize());
         filemetadata.smallestkey = file->smallest.user_key().ToString();
         filemetadata.largestkey = file->largest.user_key().ToString();
         filemetadata.smallest_seqno = file->fd.smallest_seqno;

db/version_set.h

@@ -1128,8 +1128,8 @@ class VersionSet {
 
   uint64_t current_next_file_number() const { return next_file_number_.load(); }
 
-  uint64_t min_log_number_to_keep() const {
-    return min_log_number_to_keep_.load();
+  uint64_t min_log_number_to_keep_2pc() const {
+    return min_log_number_to_keep_2pc_.load();
   }
 
   // Allocate and return a new file number
@@ -1187,7 +1187,7 @@ class VersionSet {
   // Mark the specified log number as deleted
   // REQUIRED: this is only called during single-threaded recovery or repair, or
   // from ::LogAndApply where the global mutex is held.
-  void MarkMinLogNumberToKeep(uint64_t number);
+  void MarkMinLogNumberToKeep2PC(uint64_t number);
 
   // Return the log file number for the log file that is currently
   // being compacted, or zero if there is no such log file.
@@ -1196,12 +1196,10 @@ class VersionSet {
   // Returns the minimum log number which still has data not flushed to any SST
   // file.
   // In non-2PC mode, all the log numbers smaller than this number can be safely
-  // deleted, although we still use `min_log_number_to_keep_` to determine when
-  // to delete a WAL file.
+  // deleted.
   uint64_t MinLogNumberWithUnflushedData() const {
     return PreComputeMinLogNumberWithUnflushedData(nullptr);
   }
-
   // Returns the minimum log number which still has data not flushed to any SST
   // file.
   // Empty column families' log number is considered to be
@@ -1299,10 +1297,6 @@ class VersionSet {
                         uint64_t min_pending_output);
 
   ColumnFamilySet* GetColumnFamilySet() { return column_family_set_.get(); }
-  RefedColumnFamilySet GetRefedColumnFamilySet() {
-    return RefedColumnFamilySet(GetColumnFamilySet());
-  }
-
   const FileOptions& file_options() { return file_options_; }
   void ChangeFileOptions(const MutableDBOptions& new_options) {
     file_options_.writable_file_max_buffer_size =
@@ -1405,8 +1399,9 @@ class VersionSet {
   const ImmutableDBOptions* const db_options_;
   std::atomic<uint64_t> next_file_number_;
   // Any WAL number smaller than this should be ignored during recovery,
-  // and is qualified for being deleted.
-  std::atomic<uint64_t> min_log_number_to_keep_ = {0};
+  // and is qualified for being deleted in 2PC mode. In non-2PC mode, this
+  // number is ignored.
+  std::atomic<uint64_t> min_log_number_to_keep_2pc_ = {0};
   uint64_t manifest_file_number_;
   uint64_t options_file_number_;
   uint64_t options_file_size_;

db/version_set_test.cc

@@ -3424,7 +3424,6 @@ TEST_F(VersionSetTestMissingFiles, NoFileMissing) {
 }
 
 TEST_F(VersionSetTestMissingFiles, MinLogNumberToKeep2PC) {
-  db_options_.allow_2pc = true;
   NewDB();
 
   SstInfo sst(100, kDefaultColumnFamilyName, "a");
@@ -3436,12 +3435,12 @@ TEST_F(VersionSetTestMissingFiles, MinLogNumberToKeep2PC) {
   edit.AddFile(0, file_metas[0]);
   edit.SetMinLogNumberToKeep(kMinWalNumberToKeep2PC);
   ASSERT_OK(LogAndApplyToDefaultCF(edit));
-  ASSERT_EQ(versions_->min_log_number_to_keep(), kMinWalNumberToKeep2PC);
+  ASSERT_EQ(versions_->min_log_number_to_keep_2pc(), kMinWalNumberToKeep2PC);
 
   for (int i = 0; i < 3; i++) {
     CreateNewManifest();
     ReopenDB();
-    ASSERT_EQ(versions_->min_log_number_to_keep(), kMinWalNumberToKeep2PC);
+    ASSERT_EQ(versions_->min_log_number_to_keep_2pc(), kMinWalNumberToKeep2PC);
   }
 }
 

env/env_test.cc

@@ -85,8 +85,6 @@ struct Deleter {
   void (*fn_)(void*);
 };
 
-extern "C" bool RocksDbIOUringEnable() { return true; }
-
 std::unique_ptr<char, Deleter> NewAligned(const size_t size, const char ch) {
   char* ptr = nullptr;
 #ifdef OS_WIN
@@ -1258,7 +1256,7 @@ TEST_P(EnvPosixTestWithParam, MultiRead) {
     // Random Read
     Random rnd(301 + attempt);
     ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-        "UpdateResults::io_uring_result", [&](void* arg) {
+        "UpdateResults:io_uring_result", [&](void* arg) {
           if (attempt > 0) {
             // No failure in the first attempt.
             size_t& bytes_read = *static_cast<size_t*>(arg);
@@ -1328,7 +1326,7 @@ TEST_F(EnvPosixTest, MultiReadNonAlignedLargeNum) {
     const int num_reads = rnd.Uniform(512) + 1;
 
     ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-        "UpdateResults::io_uring_result", [&](void* arg) {
+        "UpdateResults:io_uring_result", [&](void* arg) {
           if (attempt > 5) {
             // Improve partial result rates in second half of the run to
             // cover the case of repeated partial results.
@@ -3310,6 +3308,7 @@ TEST_F(TestAsyncRead, ReadAsync) {
     }
   }
 }
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

env/fs_posix.cc

@@ -1045,12 +1045,9 @@ class PosixFileSystem : public FileSystem {
 
   // EXPERIMENTAL
   //
-  // TODO akankshamahajan:
-  // 1. Update Poll API to take into account min_completions
+  // TODO akankshamahajan: Update Poll API to take into account min_completions
   // and returns if number of handles in io_handles (any order) completed is
   // equal to atleast min_completions.
-  // 2. Currently in case of direct_io, Read API is called because of which call
-  // to Poll API fails as it expects IOHandle to be populated.
   virtual IOStatus Poll(std::vector<void*>& io_handles,
                         size_t /*min_completions*/) override {
 #if defined(ROCKSDB_IOURING_PRESENT)
@@ -1097,14 +1094,12 @@ class PosixFileSystem : public FileSystem {
         req.offset = posix_handle->offset;
         req.len = posix_handle->len;
         size_t finished_len = 0;
-        size_t bytes_read = 0;
         UpdateResult(cqe, "", req.len, posix_handle->iov.iov_len,
-                     true /*async_read*/, finished_len, &req, bytes_read);
+                     true /*async_read*/, finished_len, &req);
         posix_handle->is_finished = true;
         io_uring_cqe_seen(iu, cqe);
         posix_handle->cb(req, posix_handle->cb_arg);
         (void)finished_len;
-        (void)bytes_read;
 
         if (static_cast<Posix_IOHandle*>(io_handles[i]) == posix_handle) {
           break;

env/io_posix.cc

@@ -744,36 +744,31 @@ IOStatus PosixRandomAccessFile::MultiRead(FSReadRequest* reqs,
       wrap_cache.erase(wrap_check);
 
       FSReadRequest* req = req_wrap->req;
-      size_t bytes_read = 0;
       UpdateResult(cqe, filename_, req->len, req_wrap->iov.iov_len,
-                   false /*async_read*/, req_wrap->finished_len, req,
-                   bytes_read);
+                   false /*async_read*/, req_wrap->finished_len, req);
       int32_t res = cqe->res;
-      if (res >= 0) {
-        if (bytes_read == 0) {
-          /// cqe->res == 0 can means EOF, or can mean partial results. See
-          // comment
-          // https://github.com/facebook/rocksdb/pull/6441#issuecomment-589843435
-          // Fall back to pread in this case.
-          if (use_direct_io() &&
-              !IsSectorAligned(req_wrap->finished_len,
-                               GetRequiredBufferAlignment())) {
-            // Bytes reads don't fill sectors. Should only happen at the end
-            // of the file.
-            req->result = Slice(req->scratch, req_wrap->finished_len);
-            req->status = IOStatus::OK();
-          } else {
-            Slice tmp_slice;
-            req->status =
-                Read(req->offset + req_wrap->finished_len,
-                     req->len - req_wrap->finished_len, options, &tmp_slice,
-                     req->scratch + req_wrap->finished_len, dbg);
-            req->result =
-                Slice(req->scratch, req_wrap->finished_len + tmp_slice.size());
-          }
-        } else if (bytes_read < req_wrap->iov.iov_len) {
-          incomplete_rq_list.push_back(req_wrap);
+      if (res == 0) {
+        /// cqe->res == 0 can means EOF, or can mean partial results. See
+        // comment
+        // https://github.com/facebook/rocksdb/pull/6441#issuecomment-589843435
+        // Fall back to pread in this case.
+        if (use_direct_io() && !IsSectorAligned(req_wrap->finished_len,
+                                                GetRequiredBufferAlignment())) {
+          // Bytes reads don't fill sectors. Should only happen at the end
+          // of the file.
+          req->result = Slice(req->scratch, req_wrap->finished_len);
+          req->status = IOStatus::OK();
+        } else {
+          Slice tmp_slice;
+          req->status =
+              Read(req->offset + req_wrap->finished_len,
+                   req->len - req_wrap->finished_len, options, &tmp_slice,
+                   req->scratch + req_wrap->finished_len, dbg);
+          req->result =
+              Slice(req->scratch, req_wrap->finished_len + tmp_slice.size());
         }
+      } else if (res > 0 && res < static_cast<int32_t>(req_wrap->iov.iov_len)) {
+        incomplete_rq_list.push_back(req_wrap);
       }
       io_uring_cqe_seen(iu, cqe);
     }
@@ -901,8 +896,8 @@ IOStatus PosixRandomAccessFile::ReadAsync(
 
   // Initialize Posix_IOHandle.
   posix_handle->iu = iu;
-  posix_handle->iov.iov_base = req.scratch;
-  posix_handle->iov.iov_len = req.len;
+  posix_handle->iov.iov_base = posix_handle->scratch;
+  posix_handle->iov.iov_len = posix_handle->len;
   posix_handle->cb = cb;
   posix_handle->cb_arg = cb_arg;
   posix_handle->offset = req.offset;

env/io_posix.h

@@ -66,13 +66,12 @@ struct Posix_IOHandle {
 
 inline void UpdateResult(struct io_uring_cqe* cqe, const std::string& file_name,
                          size_t len, size_t iov_len, bool async_read,
-                         size_t& finished_len, FSReadRequest* req,
-                         size_t& bytes_read) {
+                         size_t& finished_len, FSReadRequest* req) {
   if (cqe->res < 0) {
     req->result = Slice(req->scratch, 0);
     req->status = IOError("Req failed", file_name, cqe->res);
   } else {
-    bytes_read = static_cast<size_t>(cqe->res);
+    size_t bytes_read = static_cast<size_t>(cqe->res);
     TEST_SYNC_POINT_CALLBACK("UpdateResults::io_uring_result", &bytes_read);
     if (bytes_read == iov_len) {
       req->result = Slice(req->scratch, req->len);

file/file_prefetch_buffer.cc

@@ -111,6 +111,13 @@ Status FilePrefetchBuffer::ReadAsync(const IOOptions& opts,
                                      Env::IOPriority rate_limiter_priority,
                                      uint64_t read_len, uint64_t chunk_len,
                                      uint64_t rounddown_start, uint32_t index) {
+  // Reset io_handle.
+  if (io_handle_ != nullptr && del_fn_ != nullptr) {
+    del_fn_(io_handle_);
+    io_handle_ = nullptr;
+    del_fn_ = nullptr;
+  }
+
   // callback for async read request.
   auto fp = std::bind(&FilePrefetchBuffer::PrefetchAsyncCallback, this,
                       std::placeholders::_1, std::placeholders::_2);
@@ -122,7 +129,6 @@ Status FilePrefetchBuffer::ReadAsync(const IOOptions& opts,
   req.scratch = bufs_[index].buffer_.BufferStart() + chunk_len;
   Status s = reader->ReadAsync(req, opts, fp, nullptr /*cb_arg*/, &io_handle_,
                                &del_fn_, rate_limiter_priority);
-  req.status.PermitUncheckedError();
   if (s.ok()) {
     async_read_in_progress_ = true;
   }
@@ -215,34 +221,24 @@ void FilePrefetchBuffer::CopyDataToBuffer(uint32_t src, uint64_t& offset,
 //        the asynchronous prefetching in second buffer.
 Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
                                          RandomAccessFileReader* reader,
-                                         uint64_t offset, size_t length,
-                                         size_t readahead_size,
+                                         FileSystem* fs, uint64_t offset,
+                                         size_t length, size_t readahead_size,
                                          Env::IOPriority rate_limiter_priority,
                                          bool& copy_to_third_buffer) {
   if (!enable_) {
     return Status::OK();
   }
-  if (async_read_in_progress_ && fs_ != nullptr) {
+  if (async_read_in_progress_ && fs != nullptr) {
     // Wait for prefetch data to complete.
     // No mutex is needed as PrefetchAsyncCallback updates the result in second
     // buffer and FilePrefetchBuffer should wait for Poll before accessing the
     // second buffer.
     std::vector<void*> handles;
     handles.emplace_back(io_handle_);
-    fs_->Poll(handles, 1).PermitUncheckedError();
+    fs->Poll(handles, 1).PermitUncheckedError();
   }
 
-  // Reset and Release io_handle_ after the Poll API as request has been
-  // completed.
-  async_read_in_progress_ = false;
-  if (io_handle_ != nullptr && del_fn_ != nullptr) {
-    del_fn_(io_handle_);
-    io_handle_ = nullptr;
-    del_fn_ = nullptr;
-  }
-
-  // TODO akanksha: Update TEST_SYNC_POINT after Async APIs are merged with
-  // normal prefetching.
+  // TODO akanksha: Update TEST_SYNC_POINT after new tests are added.
   TEST_SYNC_POINT("FilePrefetchBuffer::Prefetch:Start");
   Status s;
   size_t prefetch_size = length + readahead_size;
@@ -442,8 +438,8 @@ bool FilePrefetchBuffer::TryReadFromCache(const IOOptions& opts,
 bool FilePrefetchBuffer::TryReadFromCacheAsync(
     const IOOptions& opts, RandomAccessFileReader* reader, uint64_t offset,
     size_t n, Slice* result, Status* status,
-    Env::IOPriority rate_limiter_priority, bool for_compaction /* = false */
-) {
+    Env::IOPriority rate_limiter_priority, bool for_compaction /* = false */,
+    FileSystem* fs) {
   if (track_min_offset_ && offset < min_offset_read_) {
     min_offset_read_ = static_cast<size_t>(offset);
   }
@@ -478,7 +474,7 @@ bool FilePrefetchBuffer::TryReadFromCacheAsync(
         if (implicit_auto_readahead_ && async_io_) {
           // Prefetch n + readahead_size_/2 synchronously as remaining
           // readahead_size_/2 will be prefetched asynchronously.
-          s = PrefetchAsync(opts, reader, offset, n, readahead_size_ / 2,
+          s = PrefetchAsync(opts, reader, fs, offset, n, readahead_size_ / 2,
                             rate_limiter_priority, copy_to_third_buffer);
         } else {
           s = Prefetch(opts, reader, offset, n + readahead_size_,
@@ -515,6 +511,7 @@ bool FilePrefetchBuffer::TryReadFromCacheAsync(
 
 void FilePrefetchBuffer::PrefetchAsyncCallback(const FSReadRequest& req,
                                                void* /*cb_arg*/) {
+  async_read_in_progress_ = false;
   uint32_t index = curr_ ^ 1;
   if (req.status.ok()) {
     if (req.offset + req.result.size() <=
@@ -530,5 +527,12 @@ void FilePrefetchBuffer::PrefetchAsyncCallback(const FSReadRequest& req,
     size_t current_size = bufs_[index].buffer_.CurrentSize();
     bufs_[index].buffer_.Size(current_size + req.result.size());
   }
+
+  // Release io_handle_.
+  if (io_handle_ != nullptr && del_fn_ != nullptr) {
+    del_fn_(io_handle_);
+    io_handle_ = nullptr;
+    del_fn_ = nullptr;
+  }
 }
 }  // namespace ROCKSDB_NAMESPACE

file/file_prefetch_buffer.h

@@ -65,7 +65,7 @@ class FilePrefetchBuffer {
   FilePrefetchBuffer(size_t readahead_size = 0, size_t max_readahead_size = 0,
                      bool enable = true, bool track_min_offset = false,
                      bool implicit_auto_readahead = false,
-                     bool async_io = false, FileSystem* fs = nullptr)
+                     bool async_io = false)
       : curr_(0),
         readahead_size_(readahead_size),
         max_readahead_size_(max_readahead_size),
@@ -79,29 +79,13 @@ class FilePrefetchBuffer {
         io_handle_(nullptr),
         del_fn_(nullptr),
         async_read_in_progress_(false),
-        async_io_(async_io),
-        fs_(fs) {
+        async_io_(async_io) {
     // If async_io_ is enabled, data is asynchronously filled in second buffer
     // while curr_ is being consumed. If data is overlapping in two buffers,
     // data is copied to third buffer to return continuous buffer.
     bufs_.resize(3);
   }
 
-  ~FilePrefetchBuffer() {
-    // Wait for any pending async job before destroying the class object.
-    if (async_read_in_progress_ && fs_ != nullptr) {
-      std::vector<void*> handles;
-      handles.emplace_back(io_handle_);
-      fs_->Poll(handles, 1).PermitUncheckedError();
-    }
-    // Release io_handle_.
-    if (io_handle_ != nullptr && del_fn_ != nullptr) {
-      del_fn_(io_handle_);
-      io_handle_ = nullptr;
-      del_fn_ = nullptr;
-    }
-  }
-
   // Load data into the buffer from a file.
   // reader                : the file reader.
   // offset                : the file offset to start reading from.
@@ -116,7 +100,8 @@ class FilePrefetchBuffer {
                   Env::IOPriority rate_limiter_priority);
 
   Status PrefetchAsync(const IOOptions& opts, RandomAccessFileReader* reader,
-                       uint64_t offset, size_t length, size_t readahead_size,
+                       FileSystem* fs, uint64_t offset, size_t length,
+                       size_t readahead_size,
                        Env::IOPriority rate_limiter_priority,
                        bool& copy_to_third_buffer);
 
@@ -144,7 +129,7 @@ class FilePrefetchBuffer {
                              RandomAccessFileReader* reader, uint64_t offset,
                              size_t n, Slice* result, Status* status,
                              Env::IOPriority rate_limiter_priority,
-                             bool for_compaction /* = false */);
+                             bool for_compaction /* = false */, FileSystem* fs);
 
   // The minimum `offset` ever passed to TryReadFromCache(). This will nly be
   // tracked if track_min_offset = true.
@@ -271,6 +256,5 @@ class FilePrefetchBuffer {
   IOHandleDeleter del_fn_;
   bool async_read_in_progress_;
   bool async_io_;
-  FileSystem* fs_;
 };
 }  // namespace ROCKSDB_NAMESPACE

file/filename.cc

@@ -491,12 +491,6 @@ Status GetInfoLogFiles(const std::shared_ptr<FileSystem>& fs,
 
 std::string NormalizePath(const std::string& path) {
   std::string dst;
-
-  if (path.length() > 2 && path[0] == kFilePathSeparator &&
-      path[1] == kFilePathSeparator) {  // Handle UNC names
-    dst.append(2, kFilePathSeparator);
-  }
-
   for (auto c : path) {
     if (!dst.empty() && (c == kFilePathSeparator || c == '/') &&
         (dst.back() == kFilePathSeparator || dst.back() == '/')) {

file/prefetch_test.cc

@@ -694,10 +694,8 @@ TEST_P(PrefetchTest1, DBIterLevelReadAhead) {
   options.write_buffer_size = 1024;
   options.create_if_missing = true;
   options.compression = kNoCompression;
-  options.statistics = CreateDBStatistics();
   options.env = env.get();
-  bool use_direct_io = std::get<0>(GetParam());
-  if (use_direct_io) {
+  if (std::get<0>(GetParam())) {
     options.use_direct_reads = true;
     options.use_direct_io_for_flush_and_compaction = true;
   }
@@ -710,7 +708,8 @@ TEST_P(PrefetchTest1, DBIterLevelReadAhead) {
   options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
   Status s = TryReopen(options);
-  if (use_direct_io && (s.IsNotSupported() || s.IsInvalidArgument())) {
+  if (std::get<0>(GetParam()) &&
+      (s.IsNotSupported() || s.IsInvalidArgument())) {
     // If direct IO is not supported, skip the test
     return;
   } else {
@@ -767,8 +766,6 @@ TEST_P(PrefetchTest1, DBIterLevelReadAhead) {
       // TODO akanksha: Remove after adding new units.
       ro.async_io = true;
     }
-
-    ASSERT_OK(options.statistics->Reset());
     auto iter = std::unique_ptr<Iterator>(db_->NewIterator(ro));
     int num_keys = 0;
     for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
@@ -776,25 +773,15 @@ TEST_P(PrefetchTest1, DBIterLevelReadAhead) {
       num_keys++;
     }
     ASSERT_EQ(num_keys, total_keys);
+
     ASSERT_GT(buff_prefetch_count, 0);
+    buff_prefetch_count = 0;
     // For index and data blocks.
     if (is_adaptive_readahead) {
       ASSERT_EQ(readahead_carry_over_count, 2 * (num_sst_files - 1));
     } else {
       ASSERT_EQ(readahead_carry_over_count, 0);
     }
-
-    // Check stats to make sure async prefetch is done.
-    {
-      HistogramData async_read_bytes;
-      options.statistics->histogramData(ASYNC_READ_BYTES, &async_read_bytes);
-      if (ro.async_io && !use_direct_io) {
-        ASSERT_GT(async_read_bytes.count, 0);
-      } else {
-        ASSERT_EQ(async_read_bytes.count, 0);
-      }
-    }
-
     SyncPoint::GetInstance()->DisableProcessing();
     SyncPoint::GetInstance()->ClearAllCallBacks();
   }
@@ -915,8 +902,6 @@ TEST_P(PrefetchTest2, DecreaseReadAheadIfInCache) {
     options.use_direct_reads = true;
     options.use_direct_io_for_flush_and_compaction = true;
   }
-
-  options.statistics = CreateDBStatistics();
   BlockBasedTableOptions table_options;
   std::shared_ptr<Cache> cache = NewLRUCache(4 * 1024 * 1024, 2);  // 8MB
   table_options.block_cache = cache;
@@ -963,6 +948,7 @@ TEST_P(PrefetchTest2, DecreaseReadAheadIfInCache) {
   SyncPoint::GetInstance()->EnableProcessing();
   ReadOptions ro;
   ro.adaptive_readahead = true;
+  // TODO akanksha: Remove after adding new units.
   ro.async_io = true;
   {
     /*
@@ -978,9 +964,7 @@ TEST_P(PrefetchTest2, DecreaseReadAheadIfInCache) {
     iter->Seek(BuildKey(1019));
     buff_prefetch_count = 0;
   }
-
   {
-    ASSERT_OK(options.statistics->Reset());
     // After caching, blocks will be read from cache (Sequential blocks)
     auto iter = std::unique_ptr<Iterator>(db_->NewIterator(ro));
     iter->Seek(BuildKey(0));
@@ -1024,118 +1008,11 @@ TEST_P(PrefetchTest2, DecreaseReadAheadIfInCache) {
     ASSERT_TRUE(iter->Valid());
     ASSERT_EQ(current_readahead_size, expected_current_readahead_size);
     ASSERT_EQ(buff_prefetch_count, 2);
-
-    // Check stats to make sure async prefetch is done.
-    {
-      HistogramData async_read_bytes;
-      options.statistics->histogramData(ASYNC_READ_BYTES, &async_read_bytes);
-      if (GetParam()) {
-        ASSERT_EQ(async_read_bytes.count, 0);
-      } else {
-        ASSERT_GT(async_read_bytes.count, 0);
-      }
-    }
-
     buff_prefetch_count = 0;
   }
   Close();
 }
 
-extern "C" bool RocksDbIOUringEnable() { return true; }
-
-// Tests the default implementation of ReadAsync API with PosixFileSystem.
-TEST_F(PrefetchTest2, ReadAsyncWithPosixFS) {
-  if (mem_env_ || encrypted_env_) {
-    ROCKSDB_GTEST_SKIP("Test requires non-mem or non-encrypted environment");
-    return;
-  }
-
-  const int kNumKeys = 1000;
-  std::shared_ptr<MockFS> fs = std::make_shared<MockFS>(
-      FileSystem::Default(), /*support_prefetch=*/false);
-  std::unique_ptr<Env> env(new CompositeEnvWrapper(env_, fs));
-
-  bool use_direct_io = false;
-  Options options = CurrentOptions();
-  options.write_buffer_size = 1024;
-  options.create_if_missing = true;
-  options.compression = kNoCompression;
-  options.env = env.get();
-  options.statistics = CreateDBStatistics();
-  if (use_direct_io) {
-    options.use_direct_reads = true;
-    options.use_direct_io_for_flush_and_compaction = true;
-  }
-  BlockBasedTableOptions table_options;
-  table_options.no_block_cache = true;
-  table_options.cache_index_and_filter_blocks = false;
-  table_options.metadata_block_size = 1024;
-  table_options.index_type =
-      BlockBasedTableOptions::IndexType::kTwoLevelIndexSearch;
-  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
-
-  Status s = TryReopen(options);
-  if (use_direct_io && (s.IsNotSupported() || s.IsInvalidArgument())) {
-    // If direct IO is not supported, skip the test
-    return;
-  } else {
-    ASSERT_OK(s);
-  }
-
-  int total_keys = 0;
-  // Write the keys.
-  {
-    WriteBatch batch;
-    Random rnd(309);
-    for (int j = 0; j < 5; j++) {
-      for (int i = j * kNumKeys; i < (j + 1) * kNumKeys; i++) {
-        ASSERT_OK(batch.Put(BuildKey(i), rnd.RandomString(1000)));
-        total_keys++;
-      }
-      ASSERT_OK(db_->Write(WriteOptions(), &batch));
-      ASSERT_OK(Flush());
-    }
-    MoveFilesToLevel(2);
-  }
-
-  int buff_prefetch_count = 0;
-  SyncPoint::GetInstance()->SetCallBack("FilePrefetchBuffer::Prefetch:Start",
-                                        [&](void*) { buff_prefetch_count++; });
-  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
-
-  // Read the keys.
-  {
-    ReadOptions ro;
-    ro.adaptive_readahead = true;
-    ro.async_io = true;
-
-    ASSERT_OK(options.statistics->Reset());
-    auto iter = std::unique_ptr<Iterator>(db_->NewIterator(ro));
-    int num_keys = 0;
-    for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
-      ASSERT_OK(iter->status());
-      num_keys++;
-    }
-    ASSERT_EQ(num_keys, total_keys);
-    ASSERT_GT(buff_prefetch_count, 0);
-
-    // Check stats to make sure async prefetch is done.
-    {
-      HistogramData async_read_bytes;
-      options.statistics->histogramData(ASYNC_READ_BYTES, &async_read_bytes);
-#if defined(ROCKSDB_IOURING_PRESENT)
-      ASSERT_GT(async_read_bytes.count, 0);
-#else
-      ASSERT_EQ(async_read_bytes.count, 0);
-#endif
-    }
-  }
-
-  SyncPoint::GetInstance()->DisableProcessing();
-  SyncPoint::GetInstance()->ClearAllCallBacks();
-
-  Close();
-}
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

file/random_access_file_reader.cc

@@ -425,75 +425,19 @@ IOStatus RandomAccessFileReader::PrepareIOOptions(const ReadOptions& ro,
   }
 }
 
-// TODO akanksha:
-// 1. Handle use_direct_io case which currently calls Read API.
+// TODO akanksha: Add perf_times etc.
 IOStatus RandomAccessFileReader::ReadAsync(
     FSReadRequest& req, const IOOptions& opts,
     std::function<void(const FSReadRequest&, void*)> cb, void* cb_arg,
     void** io_handle, IOHandleDeleter* del_fn,
     Env::IOPriority rate_limiter_priority) {
   if (use_direct_io()) {
-    // For direct_io, it calls Read API.
     req.status = Read(opts, req.offset, req.len, &(req.result), req.scratch,
                       nullptr /*dbg*/, rate_limiter_priority);
     cb(req, cb_arg);
     return IOStatus::OK();
   }
-
-  // Create a callback and populate info.
-  auto read_async_callback =
-      std::bind(&RandomAccessFileReader::ReadAsyncCallback, this,
-                std::placeholders::_1, std::placeholders::_2);
-  ReadAsyncInfo* read_async_info = new ReadAsyncInfo;
-  read_async_info->cb_ = cb;
-  read_async_info->cb_arg_ = cb_arg;
-  read_async_info->start_time_ = clock_->NowMicros();
-
-#ifndef ROCKSDB_LITE
-  if (ShouldNotifyListeners()) {
-    read_async_info->fs_start_ts_ = FileOperationInfo::StartNow();
-  }
-#endif
-
-  IOStatus s = file_->ReadAsync(req, opts, read_async_callback, read_async_info,
-                                io_handle, del_fn, nullptr /*dbg*/);
-  if (!s.ok()) {
-    delete read_async_info;
-  }
-  return s;
-}
-
-void RandomAccessFileReader::ReadAsyncCallback(const FSReadRequest& req,
-                                               void* cb_arg) {
-  ReadAsyncInfo* read_async_info = static_cast<ReadAsyncInfo*>(cb_arg);
-  assert(read_async_info);
-  assert(read_async_info->cb_);
-
-  read_async_info->cb_(req, read_async_info->cb_arg_);
-
-  // Update stats and notify listeners.
-  if (stats_ != nullptr && file_read_hist_ != nullptr) {
-    // elapsed doesn't take into account delay and overwrite as StopWatch does
-    // in Read.
-    uint64_t elapsed = clock_->NowMicros() - read_async_info->start_time_;
-    file_read_hist_->Add(elapsed);
-  }
-  if (req.status.ok()) {
-    RecordInHistogram(stats_, ASYNC_READ_BYTES, req.result.size());
-  }
-#ifndef ROCKSDB_LITE
-  if (ShouldNotifyListeners()) {
-    auto finish_ts = FileOperationInfo::FinishNow();
-    NotifyOnFileReadFinish(req.offset, req.result.size(),
-                           read_async_info->fs_start_ts_, finish_ts,
-                           req.status);
-  }
-  if (!req.status.ok()) {
-    NotifyOnIOError(req.status, FileOperationType::kRead, file_name(),
-                    req.result.size(), req.offset);
-  }
-#endif
-  RecordIOStats(stats_, file_temperature_, is_last_level_, req.result.size());
-  delete read_async_info;
+  return file_->ReadAsync(req, opts, cb, cb_arg, io_handle, del_fn,
+                          nullptr /*dbg*/);
 }
 }  // namespace ROCKSDB_NAMESPACE

file/random_access_file_reader.h

@@ -92,15 +92,6 @@ class RandomAccessFileReader {
   const Temperature file_temperature_;
   const bool is_last_level_;
 
-  struct ReadAsyncInfo {
-#ifndef ROCKSDB_LITE
-    FileOperationInfo::StartTimePoint fs_start_ts_;
-#endif
-    uint64_t start_time_;
-    std::function<void(const FSReadRequest&, void*)> cb_;
-    void* cb_arg_;
-  };
-
  public:
   explicit RandomAccessFileReader(
       std::unique_ptr<FSRandomAccessFile>&& raf, const std::string& _file_name,
@@ -188,7 +179,5 @@ class RandomAccessFileReader {
                      std::function<void(const FSReadRequest&, void*)> cb,
                      void* cb_arg, void** io_handle, IOHandleDeleter* del_fn,
                      Env::IOPriority rate_limiter_priority);
-
-  void ReadAsyncCallback(const FSReadRequest& req, void* cb_arg);
 };
 }  // namespace ROCKSDB_NAMESPACE

file/writable_file_writer.h

@@ -160,9 +160,7 @@ class WritableFileWriter {
   bool perform_data_verification_;
   uint32_t buffered_data_crc32c_checksum_;
   bool buffered_data_with_checksum_;
-#ifndef ROCKSDB_LITE
   Temperature temperature_;
-#endif  // ROCKSDB_LITE
 
  public:
   WritableFileWriter(
@@ -193,10 +191,8 @@ class WritableFileWriter {
         checksum_finalized_(false),
         perform_data_verification_(perform_data_verification),
         buffered_data_crc32c_checksum_(0),
-        buffered_data_with_checksum_(buffered_data_with_checksum) {
-#ifndef ROCKSDB_LITE
-    temperature_ = options.temperature;
-#endif  // ROCKSDB_LITE
+        buffered_data_with_checksum_(buffered_data_with_checksum),
+        temperature_(options.temperature) {
     assert(!use_direct_io() || max_buffer_size_ > 0);
     TEST_SYNC_POINT_CALLBACK("WritableFileWriter::WritableFileWriter:0",
                              reinterpret_cast<void*>(max_buffer_size_));

include/rocksdb/metadata.h

@@ -72,10 +72,10 @@ struct LiveFileStorageInfo : public FileStorageInfo {
 // The metadata that describes an SST file. (Does not need to extend
 // LiveFileStorageInfo because SST files are always immutable.)
 struct SstFileMetaData : public FileStorageInfo {
-  SstFileMetaData() { file_type = kTableFile; }
+  SstFileMetaData() {}
 
   SstFileMetaData(const std::string& _file_name, uint64_t _file_number,
-                  const std::string& _directory, uint64_t _size,
+                  const std::string& _directory, size_t _size,
                   SequenceNumber _smallest_seqno, SequenceNumber _largest_seqno,
                   const std::string& _smallestkey,
                   const std::string& _largestkey, uint64_t _num_reads_sampled,

include/rocksdb/statistics.h

@@ -534,8 +534,6 @@ enum Histograms : uint32_t {
   // Error handler statistics
   ERROR_HANDLER_AUTORESUME_RETRY_COUNT,
 
-  ASYNC_READ_BYTES,
-
   HISTOGRAM_ENUM_MAX,
 };
 

include/rocksdb/version.h

@@ -11,7 +11,7 @@
 
 #define ROCKSDB_MAJOR 7
 #define ROCKSDB_MINOR 1
-#define ROCKSDB_PATCH 2
+#define ROCKSDB_PATCH 0
 
 // 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

java/rocksjni/portal.h

@@ -5551,9 +5551,7 @@ class HistogramTypeJni {
       case ROCKSDB_NAMESPACE::Histograms::NUM_SST_READ_PER_LEVEL:
         return 0x31;
       case ROCKSDB_NAMESPACE::Histograms::ERROR_HANDLER_AUTORESUME_RETRY_COUNT:
-        return 0x32;
-      case ROCKSDB_NAMESPACE::Histograms::ASYNC_READ_BYTES:
-        return 0x33;
+        return 0x31;
       case ROCKSDB_NAMESPACE::Histograms::HISTOGRAM_ENUM_MAX:
         // 0x1F for backwards compatibility on current minor version.
         return 0x1F;
@@ -5671,8 +5669,6 @@ class HistogramTypeJni {
       case 0x32:
         return ROCKSDB_NAMESPACE::Histograms::
             ERROR_HANDLER_AUTORESUME_RETRY_COUNT;
-      case 0x33:
-        return ROCKSDB_NAMESPACE::Histograms::ASYNC_READ_BYTES;
       case 0x1F:
         // 0x1F for backwards compatibility on current minor version.
         return ROCKSDB_NAMESPACE::Histograms::HISTOGRAM_ENUM_MAX;

java/src/main/java/org/rocksdb/HistogramType.java

@@ -180,8 +180,6 @@ public enum HistogramType {
    */
   ERROR_HANDLER_AUTORESUME_RETRY_COUNT((byte) 0x32),
 
-  ASYNC_READ_BYTES((byte) 0x33),
-
   // 0x1F for backwards compatibility on current minor version.
   HISTOGRAM_ENUM_MAX((byte) 0x1F);
 

monitoring/instrumented_mutex.h

@@ -54,9 +54,10 @@ class InstrumentedMutex {
 class ALIGN_AS(CACHE_LINE_SIZE) CacheAlignedInstrumentedMutex
     : public InstrumentedMutex {
   using InstrumentedMutex::InstrumentedMutex;
+  char padding[(CACHE_LINE_SIZE - sizeof(InstrumentedMutex) % CACHE_LINE_SIZE) %
+               CACHE_LINE_SIZE] ROCKSDB_FIELD_UNUSED;
 };
-static_assert(alignof(CacheAlignedInstrumentedMutex) != CACHE_LINE_SIZE ||
-              sizeof(CacheAlignedInstrumentedMutex) % CACHE_LINE_SIZE == 0);
+static_assert(sizeof(CacheAlignedInstrumentedMutex) % CACHE_LINE_SIZE == 0);
 
 // RAII wrapper for InstrumentedMutex
 class InstrumentedMutexLock {

monitoring/statistics.cc

@@ -283,7 +283,6 @@ const std::vector<std::pair<Histograms, std::string>> HistogramsNameMap = {
     {NUM_SST_READ_PER_LEVEL, "rocksdb.num.sst.read.per.level"},
     {ERROR_HANDLER_AUTORESUME_RETRY_COUNT,
      "rocksdb.error.handler.autoresume.retry.count"},
-    {ASYNC_READ_BYTES, "rocksdb.async.read.bytes"},
 };
 
 std::shared_ptr<Statistics> CreateDBStatistics() {

table/block_based/block_based_table_reader.cc

@@ -1465,10 +1465,9 @@ template <typename TBlocklike>
 Status BlockBasedTable::MaybeReadBlockAndLoadToCache(
     FilePrefetchBuffer* prefetch_buffer, const ReadOptions& ro,
     const BlockHandle& handle, const UncompressionDict& uncompression_dict,
-    const bool wait, const bool for_compaction,
-    CachableEntry<TBlocklike>* block_entry, BlockType block_type,
-    GetContext* get_context, BlockCacheLookupContext* lookup_context,
-    BlockContents* contents) const {
+    const bool wait, CachableEntry<TBlocklike>* block_entry,
+    BlockType block_type, GetContext* get_context,
+    BlockCacheLookupContext* lookup_context, BlockContents* contents) const {
   assert(block_entry != nullptr);
   const bool no_io = (ro.read_tier == kBlockCacheTier);
   Cache* block_cache = rep_->table_options.block_cache.get();
@@ -1521,9 +1520,7 @@ Status BlockBasedTable::MaybeReadBlockAndLoadToCache(
       CompressionType raw_block_comp_type;
       BlockContents raw_block_contents;
       if (!contents) {
-        Histograms histogram = for_compaction ? READ_BLOCK_COMPACTION_MICROS
-                                              : READ_BLOCK_GET_MICROS;
-        StopWatch sw(rep_->ioptions.clock, statistics, histogram);
+        StopWatch sw(rep_->ioptions.clock, statistics, READ_BLOCK_GET_MICROS);
         BlockFetcher block_fetcher(
             rep_->file.get(), prefetch_buffer, rep_->footer, ro, handle,
             &raw_block_contents, rep_->ioptions, do_uncompress,
@@ -1881,9 +1878,8 @@ void BlockBasedTable::RetrieveMultipleBlocks(
         // avoid looking up the block cache
         s = MaybeReadBlockAndLoadToCache(
             nullptr, options, handle, uncompression_dict, /*wait=*/true,
-            /*for_compaction=*/false, block_entry, BlockType::kData,
-            mget_iter->get_context, &lookup_data_block_context,
-            &raw_block_contents);
+            block_entry, BlockType::kData, mget_iter->get_context,
+            &lookup_data_block_context, &raw_block_contents);
 
         // block_entry value could be null if no block cache is present, i.e
         // BlockBasedTableOptions::no_block_cache is true and no compressed
@@ -1937,7 +1933,7 @@ Status BlockBasedTable::RetrieveBlock(
   if (use_cache) {
     s = MaybeReadBlockAndLoadToCache(
         prefetch_buffer, ro, handle, uncompression_dict, wait_for_cache,
-        for_compaction, block_entry, block_type, get_context, lookup_context,
+        block_entry, block_type, get_context, lookup_context,
         /*contents=*/nullptr);
 
     if (!s.ok()) {
@@ -1966,9 +1962,8 @@ Status BlockBasedTable::RetrieveBlock(
   std::unique_ptr<TBlocklike> block;
 
   {
-    Histograms histogram =
-        for_compaction ? READ_BLOCK_COMPACTION_MICROS : READ_BLOCK_GET_MICROS;
-    StopWatch sw(rep_->ioptions.clock, rep_->ioptions.stats, histogram);
+    StopWatch sw(rep_->ioptions.clock, rep_->ioptions.stats,
+                 READ_BLOCK_GET_MICROS);
     s = ReadBlockFromFile(
         rep_->file.get(), prefetch_buffer, rep_->footer, ro, handle, &block,
         rep_->ioptions, do_uncompress, maybe_compressed, block_type,

table/block_based/block_based_table_reader.h

@@ -343,10 +343,9 @@ class BlockBasedTable : public TableReader {
   Status MaybeReadBlockAndLoadToCache(
       FilePrefetchBuffer* prefetch_buffer, const ReadOptions& ro,
       const BlockHandle& handle, const UncompressionDict& uncompression_dict,
-      const bool wait, const bool for_compaction,
-      CachableEntry<TBlocklike>* block_entry, BlockType block_type,
-      GetContext* get_context, BlockCacheLookupContext* lookup_context,
-      BlockContents* contents) const;
+      const bool wait, CachableEntry<TBlocklike>* block_entry,
+      BlockType block_type, GetContext* get_context,
+      BlockCacheLookupContext* lookup_context, BlockContents* contents) const;
 
   // Similar to the above, with one crucial difference: it will retrieve the
   // block from the file even if there are no caches configured (assuming the
@@ -655,10 +654,10 @@ struct BlockBasedTable::Rep {
                                 std::unique_ptr<FilePrefetchBuffer>* fpb,
                                 bool implicit_auto_readahead,
                                 bool async_io) const {
-    fpb->reset(new FilePrefetchBuffer(
-        readahead_size, max_readahead_size,
-        !ioptions.allow_mmap_reads /* enable */, false /* track_min_offset */,
-        implicit_auto_readahead, async_io, ioptions.fs.get()));
+    fpb->reset(new FilePrefetchBuffer(readahead_size, max_readahead_size,
+                                      !ioptions.allow_mmap_reads /* enable */,
+                                      false /* track_min_offset */,
+                                      implicit_auto_readahead, async_io));
   }
 
   void CreateFilePrefetchBufferIfNotExists(

table/block_based/partitioned_filter_block.cc

@@ -520,8 +520,8 @@ Status PartitionedFilterBlockReader::CacheDependencies(const ReadOptions& ro,
     // filter blocks
     s = table()->MaybeReadBlockAndLoadToCache(
         prefetch_buffer.get(), ro, handle, UncompressionDict::GetEmptyDict(),
-        /* wait */ true, /* for_compaction */ false, &block, BlockType::kFilter,
-        nullptr /* get_context */, &lookup_context, nullptr /* contents */);
+        /* wait */ true, &block, BlockType::kFilter, nullptr /* get_context */,
+        &lookup_context, nullptr /* contents */);
     if (!s.ok()) {
       return s;
     }

table/block_based/partitioned_index_reader.cc

@@ -182,8 +182,8 @@ Status PartitionIndexReader::CacheDependencies(const ReadOptions& ro,
     // filter blocks
     Status s = table()->MaybeReadBlockAndLoadToCache(
         prefetch_buffer.get(), ro, handle, UncompressionDict::GetEmptyDict(),
-        /*wait=*/true, /*for_compaction=*/false, &block, BlockType::kIndex,
-        /*get_context=*/nullptr, &lookup_context, /*contents=*/nullptr);
+        /*wait=*/true, &block, BlockType::kIndex, /*get_context=*/nullptr,
+        &lookup_context, /*contents=*/nullptr);
 
     if (!s.ok()) {
       return s;

table/block_fetcher.cc

@@ -74,7 +74,8 @@ inline bool BlockFetcher::TryGetFromPrefetchBuffer() {
       if (read_options_.async_io) {
         read_from_prefetch_buffer = prefetch_buffer_->TryReadFromCacheAsync(
             opts, file_, handle_.offset(), block_size_with_trailer_, &slice_,
-            &io_s, read_options_.rate_limiter_priority, for_compaction_);
+            &io_s, read_options_.rate_limiter_priority, for_compaction_,
+            ioptions_.fs.get());
       } else {
         read_from_prefetch_buffer = prefetch_buffer_->TryReadFromCache(
             opts, file_, handle_.offset(), block_size_with_trailer_, &slice_,

utilities/blob_db/blob_db_impl_filesnapshot.cc

@@ -93,7 +93,7 @@ void BlobDBImpl::GetLiveFilesMetaData(std::vector<LiveFileMetaData>* metadata) {
   for (auto bfile_pair : blob_files_) {
     auto blob_file = bfile_pair.second;
     LiveFileMetaData filemetadata;
-    filemetadata.size = blob_file->GetFileSize();
+    filemetadata.size = static_cast<size_t>(blob_file->GetFileSize());
     const uint64_t file_number = blob_file->BlobFileNumber();
     // Path should be relative to db_name, but begin with slash.
     filemetadata.name = BlobFileName("", bdb_options_.blob_dir, file_number);