Support async IO for MultiScan (#13932)

Summary:
add option MultiScanArgs::use_async_io option and implementation for using ReadAsync() for multiscan. Read requests are submitted during Prepare() and polled during actual scanning.

Pull Request resolved: #13932

Test Plan:
- updated existing unit test to use async_io.
- crash test: `python3 -u ./tools/db_crashtest.py whitebox --iterpercent=60 --prefix_size=-1 --prefixpercent=0 --readpercent=0 --test_batches_snapshots=0 --use_multiscan=1 --read_fault_one_in=0 --kill_random_test=88888 --interval=60 --multiscan_use_async_io=1 --mmap_read=0`

Benchmark:
- Default multiscan benchmark:
```
Set up: /db_bench --benchmarks="fillseq,compact" --disable_wal=1 --threads=1 --num_levels=1 --compaction_style=2 --fifo_compaction_max_table_files_size_mb=1000 --write_buffer_size=268435456

Without async IO:
./db_bench --db="/tmp/rocksdbtest-543376/dbbench" --use_existing_db=1 --benchmarks=multiscan --disable_auto_compactions=1 --seek_nexts=100 --threads=32 --duration=10 --statistics=1 --use_direct_reads=1 --multiscan_use_async_io=0

multiscan    :     415.569 micros/op 75805 ops/sec 10.355 seconds 784968 operations; (multscans:24999)
rocksdb.read.async.micros COUNT : 0

With asycn IO:
./db_bench --db="/tmp/rocksdbtest-543376/dbbench" --use_existing_db=1 --benchmarks=multiscan --disable_auto_compactions=1 --seek_nexts=100 --threads=32 --duration=10 --statistics=1 --use_direct_reads=1 --multiscan_use_async_io=1

multiscan    :     413.236 micros/op 76044 ops/sec 10.375 seconds 788968 operations; (multscans:24999)
rocksdb.read.async.micros COUNT : 3916499

Similar performance.
```

- Larger scan, more scans per multiscan, do not coalesce IO so that async IO can progress while scanning, and use one thread:
```
multiscan_stride = 1000
multiscan_size = 100
seek_nexts = 1000

./db_bench --db="/tmp/rocksdbtest-543376/dbbench" --use_existing_db=1 --benchmarks=multiscan --disable_auto_compactions=1 --threads=1 --duration=10 --statistics=0 --use_direct_reads=1  --cache_size=2097152 --multiscan_size=100 --multiscan_stride=1000 --seek_nexts=1000 --seed=1 --multiscan_coalesce_threshold=0  --multiscan_use_async_io=0

Without async IO:
multiscan    :   20495.205 micros/op 48 ops/sec 10.002 seconds 488 operations; (multscans:488)

With async IO:
multiscan    :   18337.883 micros/op 54 ops/sec 10.013 seconds 546 operations; (multscans:546)

~10% improvement in throughput
```

Reviewed By: xingbowang

Differential Revision: D82077818

Pulled By: cbi42

fbshipit-source-id: 66e32cf4039183c4841827409286dfbaa6dfbcd8

Author: cbi42

db/version_set.cc

@@ -1165,6 +1165,7 @@ class LevelIterator final : public InternalIterator {
     // Propagate io colaescing threshold
     for (auto& file_to_arg : *file_to_scan_opts_) {
       file_to_arg.second.io_coalesce_threshold = so->io_coalesce_threshold;
+      file_to_arg.second.use_async_io = so->use_async_io;
     }
   }
 

db_stress_tool/db_stress_common.h

@@ -432,6 +432,7 @@ DECLARE_uint32(memtable_avg_op_scan_flush_trigger);
 DECLARE_uint32(ingest_wbwi_one_in);
 DECLARE_bool(universal_reduce_file_locking);
 DECLARE_bool(use_multiscan);
+DECLARE_bool(multiscan_use_async_io);
 
 // Compaction deletion trigger declarations for stress testing
 DECLARE_bool(enable_compaction_on_deletion_trigger);

db_stress_tool/db_stress_gflags.cc

@@ -1535,4 +1535,7 @@ DEFINE_bool(
 DEFINE_bool(use_multiscan, false,
             "If set, use the batched MultiScan API for scans.");
 
+DEFINE_bool(multiscan_use_async_io, false,
+            "If set, enable async_io for MultiScan operations.");
+
 #endif  // GFLAGS

db_stress_tool/db_stress_test_base.cc

@@ -1693,6 +1693,7 @@ Status StressTest::TestMultiScan(ThreadState* thread,
   std::vector<std::string> start_key_strs;
   std::vector<std::string> end_key_strs;
   MultiScanArgs scan_opts;
+  scan_opts.use_async_io = FLAGS_multiscan_use_async_io;
   start_key_strs.reserve(num_scans);
   end_key_strs.reserve(num_scans);
 

env/fs_posix.cc

@@ -1153,7 +1153,7 @@ class PosixFileSystem : public FileSystem {
     return IOStatus::OK();
 #else
     (void)io_handles;
-    return IOStatus::NotSupported("Poll");
+    return IOStatus::NotSupported("Poll not implemented");
 #endif
   }
 

file/random_access_file_reader.cc

@@ -497,6 +497,13 @@ IOStatus RandomAccessFileReader::PrepareIOOptions(const ReadOptions& ro,
   }
 }
 
+// Notes for when direct_io is enabled:
+// Unless req.offset, req.len, req.scratch are all already aligned,
+// RandomAccessFileReader will creats aligned requests and aligned buffer for
+// the request. User should only provide either req.scratch or aligned_buf. If
+// only req.scratch is provided, result will be copied from allocated aligned
+// buffer to req.scratch. If only alignd_buf is provided, it will be set to
+// the ailgned buf allocated by RandomAccessFileReader and saves a copy.
 IOStatus RandomAccessFileReader::ReadAsync(
     FSReadRequest& req, const IOOptions& opts,
     std::function<void(FSReadRequest&, void*)> cb, void* cb_arg,

include/rocksdb/options.h

@@ -1791,10 +1791,12 @@ class MultiScanArgs {
     original_ranges_ = other.original_ranges_;
     io_coalesce_threshold = other.io_coalesce_threshold;
     max_prefetch_size = other.max_prefetch_size;
+    use_async_io = other.use_async_io;
   }
   MultiScanArgs(MultiScanArgs&& other) noexcept
       : io_coalesce_threshold(other.io_coalesce_threshold),
         max_prefetch_size(other.max_prefetch_size),
+        use_async_io(other.use_async_io),
         comp_(other.comp_),
         original_ranges_(std::move(other.original_ranges_)) {}
 
@@ -1803,6 +1805,7 @@ class MultiScanArgs {
     original_ranges_ = other.original_ranges_;
     io_coalesce_threshold = other.io_coalesce_threshold;
     max_prefetch_size = other.max_prefetch_size;
+    use_async_io = other.use_async_io;
     return *this;
   }
 
@@ -1812,6 +1815,7 @@ class MultiScanArgs {
       original_ranges_ = std::move(other.original_ranges_);
       io_coalesce_threshold = other.io_coalesce_threshold;
       max_prefetch_size = other.max_prefetch_size;
+      use_async_io = other.use_async_io;
     }
     return *this;
   }
@@ -1865,6 +1869,11 @@ class MultiScanArgs {
   // Note that this limit is per file and applies to compressed block size.
   uint64_t max_prefetch_size = 0;
 
+  // Enable async I/O for multi-scan operations
+  // When true, BlockBasedTableIterator will use ReadAsync() for reading blocks
+  // When false, it will use synchronous MultiRead().
+  bool use_async_io = false;
+
  private:
   // The comparator used for ordering ranges
   const Comparator* comp_;