rocksdb/file/random_access_file_reader.cc

//  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).
//
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.

#include "file/random_access_file_reader.h"

#include <algorithm>
#include <mutex>

#include "file/file_util.h"
#include "monitoring/histogram.h"
#include "monitoring/iostats_context_imp.h"
#include "port/port.h"
#include "rocksdb/io_status.h"
#include "table/format.h"
#include "test_util/sync_point.h"
#include "util/random.h"
#include "util/rate_limiter_impl.h"

namespace ROCKSDB_NAMESPACE {
inline Histograms GetFileReadHistograms(Statistics* stats,
                                        Env::IOActivity io_activity) {
  switch (io_activity) {
    case Env::IOActivity::kFlush:
      return Histograms::FILE_READ_FLUSH_MICROS;
    case Env::IOActivity::kCompaction:
      return Histograms::FILE_READ_COMPACTION_MICROS;
    case Env::IOActivity::kDBOpen:
      return Histograms::FILE_READ_DB_OPEN_MICROS;
    default:
      break;
  }

  if (stats && stats->get_stats_level() > StatsLevel::kExceptDetailedTimers) {
    switch (io_activity) {
      case Env::IOActivity::kGet:
        return Histograms::FILE_READ_GET_MICROS;
      case Env::IOActivity::kMultiGet:
        return Histograms::FILE_READ_MULTIGET_MICROS;
      case Env::IOActivity::kDBIterator:
        return Histograms::FILE_READ_DB_ITERATOR_MICROS;
      case Env::IOActivity::kVerifyDBChecksum:
        return Histograms::FILE_READ_VERIFY_DB_CHECKSUM_MICROS;
      case Env::IOActivity::kVerifyFileChecksums:
        return Histograms::FILE_READ_VERIFY_FILE_CHECKSUMS_MICROS;
      default:
        break;
    }
  }
  return Histograms::HISTOGRAM_ENUM_MAX;
}
inline void RecordIOStats(Statistics* stats, Temperature file_temperature,
                          bool is_last_level, size_t size) {
  IOSTATS_ADD(bytes_read, size);
  // record for last/non-last level
  if (is_last_level) {
    RecordTick(stats, LAST_LEVEL_READ_BYTES, size);
    RecordTick(stats, LAST_LEVEL_READ_COUNT, 1);
  } else {
    RecordTick(stats, NON_LAST_LEVEL_READ_BYTES, size);
    RecordTick(stats, NON_LAST_LEVEL_READ_COUNT, 1);
  }

  // record for temperature file
  switch (file_temperature) {
    case Temperature::kHot:
      IOSTATS_ADD(file_io_stats_by_temperature.hot_file_bytes_read, size);
      IOSTATS_ADD(file_io_stats_by_temperature.hot_file_read_count, 1);
      RecordTick(stats, HOT_FILE_READ_BYTES, size);
      RecordTick(stats, HOT_FILE_READ_COUNT, 1);
      break;
    case Temperature::kWarm:
      IOSTATS_ADD(file_io_stats_by_temperature.warm_file_bytes_read, size);
      IOSTATS_ADD(file_io_stats_by_temperature.warm_file_read_count, 1);
      RecordTick(stats, WARM_FILE_READ_BYTES, size);
      RecordTick(stats, WARM_FILE_READ_COUNT, 1);
      break;
    case Temperature::kCool:
      IOSTATS_ADD(file_io_stats_by_temperature.cool_file_bytes_read, size);
      IOSTATS_ADD(file_io_stats_by_temperature.cool_file_read_count, 1);
      RecordTick(stats, COOL_FILE_READ_BYTES, size);
      RecordTick(stats, COOL_FILE_READ_COUNT, 1);
      break;
    case Temperature::kCold:
      IOSTATS_ADD(file_io_stats_by_temperature.cold_file_bytes_read, size);
      IOSTATS_ADD(file_io_stats_by_temperature.cold_file_read_count, 1);
      RecordTick(stats, COLD_FILE_READ_BYTES, size);
      RecordTick(stats, COLD_FILE_READ_COUNT, 1);
      break;
    case Temperature::kIce:
      IOSTATS_ADD(file_io_stats_by_temperature.ice_file_bytes_read, size);
      IOSTATS_ADD(file_io_stats_by_temperature.ice_file_read_count, 1);
      RecordTick(stats, ICE_FILE_READ_BYTES, size);
      RecordTick(stats, ICE_FILE_READ_COUNT, 1);
      break;
    case Temperature::kUnknown:
      if (is_last_level) {
        IOSTATS_ADD(file_io_stats_by_temperature.unknown_last_level_bytes_read,
                    size);
        IOSTATS_ADD(file_io_stats_by_temperature.unknown_last_level_read_count,
                    1);
      } else {
        IOSTATS_ADD(
            file_io_stats_by_temperature.unknown_non_last_level_bytes_read,
            size);
        IOSTATS_ADD(
            file_io_stats_by_temperature.unknown_non_last_level_read_count, 1);
      }
      break;
    default:
      break;
  }
}

IOStatus RandomAccessFileReader::Create(
    const std::shared_ptr<FileSystem>& fs, const std::string& fname,
    const FileOptions& file_opts,
    std::unique_ptr<RandomAccessFileReader>* reader, IODebugContext* dbg) {
  std::unique_ptr<FSRandomAccessFile> file;
  IOStatus io_s = fs->NewRandomAccessFile(fname, file_opts, &file, dbg);
  if (io_s.ok()) {
    reader->reset(new RandomAccessFileReader(std::move(file), fname));
  }
  return io_s;
}

IOStatus RandomAccessFileReader::Read(
    const IOOptions& opts, uint64_t offset, size_t n, Slice* result,
    char* scratch, AlignedBufferAllocationContext* direct_io_buffer_context,
    IODebugContext* dbg) const {
  AlignedBuffer* direct_io_buffer = direct_io_buffer_context != nullptr
                                        ? direct_io_buffer_context->buffer
                                        : nullptr;
  const AlignedBuffer::Allocator* direct_io_allocator =
      direct_io_buffer_context != nullptr ? direct_io_buffer_context->allocator
                                          : nullptr;
  assert(direct_io_buffer_context == nullptr || direct_io_buffer != nullptr);
  const Env::IOPriority rate_limiter_priority = opts.rate_limiter_priority;

  TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::Read", nullptr);
  TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::Read:IODebugContext",
                           const_cast<void*>(static_cast<void*>(dbg)));

  // To be paranoid: modify scratch a little bit, so in case underlying
  // FileSystem doesn't fill the buffer but return success and `scratch` returns
  // contains a previous block, returned value will not pass checksum.
  if (n > 0 && scratch != nullptr) {
    // This byte might not change anything for direct I/O case, but it's OK.
    scratch[0]++;
  }

  IOStatus io_s;
  uint64_t elapsed = 0;
  size_t alignment = file_->GetRequiredBufferAlignment();
  bool is_aligned = false;
  if (direct_io_buffer == nullptr && scratch != nullptr) {
    // Check if offset, length and buffer are aligned.
    is_aligned = (offset & (alignment - 1)) == 0 &&
                 (n & (alignment - 1)) == 0 &&
                 (uintptr_t(scratch) & (alignment - 1)) == 0;
  }

  {
    StopWatch sw(clock_, stats_, hist_type_,
                 GetFileReadHistograms(stats_, opts.io_activity),
                 (stats_ != nullptr) ? &elapsed : nullptr, true /*overwrite*/,
                 true /*delay_enabled*/);
    auto prev_perf_level = GetPerfLevel();
    IOSTATS_TIMER_GUARD(read_nanos);
    if (use_direct_io() && is_aligned == false) {
      size_t aligned_offset =
          TruncateToPageBoundary(alignment, static_cast<size_t>(offset));
      size_t offset_advance = static_cast<size_t>(offset) - aligned_offset;
      size_t read_size =
          Roundup(static_cast<size_t>(offset + n), alignment) - aligned_offset;
      AlignedBuffer buf;
      AlignedBuffer* aligned_read_buffer =
          direct_io_buffer != nullptr ? direct_io_buffer : &buf;
      aligned_read_buffer->Alignment(alignment);
      Status allocate_status = aligned_read_buffer->AllocateNewBuffer(
          read_size, direct_io_allocator);
      if (!allocate_status.ok()) {
        io_s = status_to_io_status(std::move(allocate_status));
      }
      char* aligned_scratch = aligned_read_buffer->BufferStart();
      size_t current_size = 0;
      while (io_s.ok() && current_size < read_size) {
        size_t allowed;
        if (rate_limiter_priority != Env::IO_TOTAL &&
            rate_limiter_ != nullptr) {
          allowed = rate_limiter_->RequestToken(
              read_size - current_size, alignment, rate_limiter_priority,
              stats_, RateLimiter::OpType::kRead);
        } else {
          assert(current_size == 0);
          allowed = read_size;
        }
        Slice tmp;

        FileOperationInfo::StartTimePoint start_ts;
        uint64_t orig_offset = 0;
        if (ShouldNotifyListeners()) {
          start_ts = FileOperationInfo::StartNow();
          orig_offset = aligned_offset + current_size;
        }

        {
          IOSTATS_CPU_TIMER_GUARD(cpu_read_nanos, clock_);
          // Only user reads are expected to specify a timeout. And user reads
          // are not subjected to rate_limiter and should go through only
          // one iteration of this loop, so we don't need to check and adjust
          // the opts.timeout before calling file_->Read
          assert(!opts.timeout.count() || allowed == read_size);
          io_s = file_->Read(aligned_offset + current_size, allowed, opts, &tmp,
                             aligned_scratch + current_size, dbg);
        }
        if (ShouldNotifyListeners()) {
          auto finish_ts = FileOperationInfo::FinishNow();
          NotifyOnFileReadFinish(orig_offset, tmp.size(), start_ts, finish_ts,
                                 io_s);
          if (!io_s.ok()) {
            NotifyOnIOError(io_s, FileOperationType::kRead, file_name(),
                            tmp.size(), orig_offset);
          }
        }

        current_size += tmp.size();
        if (direct_io_buffer == nullptr) {
          buf.Size(current_size);
        }
        if (!io_s.ok() || tmp.size() < allowed) {
          break;
        }
      }
      size_t res_len = 0;
      if (io_s.ok() && offset_advance < current_size) {
        res_len = std::min(current_size - offset_advance, n);
        if (direct_io_buffer != nullptr) {
          scratch = aligned_scratch + offset_advance;
        } else {
          assert(scratch != nullptr);
          buf.Read(scratch, offset_advance, res_len);
        }
      }
      *result = Slice(scratch, res_len);
    } else {
      size_t pos = 0;
      const char* res_scratch = nullptr;
      while (pos < n) {
        size_t allowed;
        if (rate_limiter_priority != Env::IO_TOTAL &&
            rate_limiter_ != nullptr) {
          if (rate_limiter_->IsRateLimited(RateLimiter::OpType::kRead)) {
            sw.DelayStart();
          }
          allowed = rate_limiter_->RequestToken(
              n - pos, (use_direct_io() ? alignment : 0), rate_limiter_priority,
              stats_, RateLimiter::OpType::kRead);
          if (rate_limiter_->IsRateLimited(RateLimiter::OpType::kRead)) {
            sw.DelayStop();
          }
        } else {
          allowed = n;
        }
        Slice tmp_result;

        FileOperationInfo::StartTimePoint start_ts;
        if (ShouldNotifyListeners()) {
          start_ts = FileOperationInfo::StartNow();
        }

        {
          IOSTATS_CPU_TIMER_GUARD(cpu_read_nanos, clock_);
          // Only user reads are expected to specify a timeout. And user reads
          // are not subjected to rate_limiter and should go through only
          // one iteration of this loop, so we don't need to check and adjust
          // the opts.timeout before calling file_->Read
          assert(!opts.timeout.count() || allowed == n);
          io_s = file_->Read(offset + pos, allowed, opts, &tmp_result,
                             scratch + pos, dbg);
        }
        if (ShouldNotifyListeners()) {
          auto finish_ts = FileOperationInfo::FinishNow();
          NotifyOnFileReadFinish(offset + pos, tmp_result.size(), start_ts,
                                 finish_ts, io_s);

          if (!io_s.ok()) {
            NotifyOnIOError(io_s, FileOperationType::kRead, file_name(),
                            tmp_result.size(), offset + pos);
          }
        }
        if (res_scratch == nullptr) {
          // we can't simply use `scratch` because reads of mmap'd files return
          // data in a different buffer.
          res_scratch = tmp_result.data();
        } else {
          // make sure chunks are inserted contiguously into `res_scratch`.
          assert(tmp_result.data() == res_scratch + pos);
        }
        pos += tmp_result.size();
        if (!io_s.ok() || tmp_result.size() < allowed) {
          break;
        }
      }
      *result = Slice(res_scratch, io_s.ok() ? pos : 0);
    }
    RecordIOStats(stats_, file_temperature_, is_last_level_, result->size());
    SetPerfLevel(prev_perf_level);
  }
  if (stats_ != nullptr && file_read_hist_ != nullptr) {
    file_read_hist_->Add(elapsed);
  }

#ifndef NDEBUG
  auto pair = std::make_pair(&file_name_, &io_s);
  if (offset == 0) {
    TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::Read::BeforeReturn",
                             &pair);
  }
  TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::Read::AnyOffset", &pair);
#endif
  return io_s;
}

size_t End(const FSReadRequest& r) {
  return static_cast<size_t>(r.offset) + r.len;
}

FSReadRequest Align(const FSReadRequest& r, size_t alignment) {
  FSReadRequest req;
  req.offset = static_cast<uint64_t>(
      TruncateToPageBoundary(alignment, static_cast<size_t>(r.offset)));
  req.len = Roundup(End(r), alignment) - req.offset;
  req.scratch = nullptr;
  return req;
}

bool TryMerge(FSReadRequest* dest, const FSReadRequest& src) {
  size_t dest_offset = static_cast<size_t>(dest->offset);
  size_t src_offset = static_cast<size_t>(src.offset);
  size_t dest_end = End(*dest);
  size_t src_end = End(src);
  if (std::max(dest_offset, src_offset) > std::min(dest_end, src_end)) {
    return false;
  }
  dest->offset = static_cast<uint64_t>(std::min(dest_offset, src_offset));
  dest->len = std::max(dest_end, src_end) - dest->offset;
  return true;
}

IOStatus RandomAccessFileReader::MultiRead(
    const IOOptions& opts, FSReadRequest* read_reqs, size_t num_reqs,
    AlignedBufferAllocationContext* direct_io_buffer_context,
    IODebugContext* dbg) const {
  assert(num_reqs > 0);
  AlignedBuffer* direct_io_buffer = direct_io_buffer_context != nullptr
                                        ? direct_io_buffer_context->buffer
                                        : nullptr;
  const AlignedBuffer::Allocator* direct_io_allocator =
      direct_io_buffer_context != nullptr ? direct_io_buffer_context->allocator
                                          : nullptr;
  assert(direct_io_buffer != nullptr);

#ifndef NDEBUG
  for (size_t i = 0; i < num_reqs - 1; ++i) {
    assert(read_reqs[i].offset <= read_reqs[i + 1].offset);
  }
#endif  // !NDEBUG
  const Env::IOPriority rate_limiter_priority = opts.rate_limiter_priority;

  // To be paranoid modify scratch a little bit, so in case underlying
  // FileSystem doesn't fill the buffer but return success and `scratch` returns
  // contains a previous block, returned value will not pass checksum.
  // This byte might not change anything for direct I/O case, but it's OK.
  for (size_t i = 0; i < num_reqs; i++) {
    FSReadRequest& r = read_reqs[i];
    if (r.len > 0 && r.scratch != nullptr) {
      r.scratch[0]++;
    }
  }

  IOStatus io_s;
  uint64_t elapsed = 0;
  {
    StopWatch sw(clock_, stats_, hist_type_,
                 GetFileReadHistograms(stats_, opts.io_activity),
                 (stats_ != nullptr) ? &elapsed : nullptr, true /*overwrite*/,
                 true /*delay_enabled*/);
    auto prev_perf_level = GetPerfLevel();
    IOSTATS_TIMER_GUARD(read_nanos);

    FSReadRequest* fs_reqs = read_reqs;
    size_t num_fs_reqs = num_reqs;
    std::vector<FSReadRequest> aligned_reqs;
    if (use_direct_io()) {
      // num_reqs is the max possible size,
      // this can reduce std::vecector's internal resize operations.
      aligned_reqs.reserve(num_reqs);
      // Align and merge the read requests.
      size_t alignment = file_->GetRequiredBufferAlignment();
      for (size_t i = 0; i < num_reqs; i++) {
        FSReadRequest r = Align(read_reqs[i], alignment);
        if (i == 0) {
          // head
          aligned_reqs.push_back(std::move(r));

        } else if (!TryMerge(&aligned_reqs.back(), r)) {
          // head + n
          aligned_reqs.push_back(std::move(r));

        } else {
          // unused
          r.status.PermitUncheckedError();
        }
      }
      TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::MultiRead:AlignedReqs",
                               &aligned_reqs);

      // Allocate aligned buffer and let scratch buffers point to it.
      size_t total_len = 0;
      for (const auto& r : aligned_reqs) {
        total_len += r.len;
      }
      direct_io_buffer->Alignment(alignment);
      Status allocate_status =
          direct_io_buffer->AllocateNewBuffer(total_len, direct_io_allocator);
      if (!allocate_status.ok()) {
        io_s = status_to_io_status(std::move(allocate_status));
      }
      if (io_s.ok()) {
        char* scratch = direct_io_buffer->BufferStart();
        for (auto& r : aligned_reqs) {
          r.scratch = scratch;
          scratch += r.len;
        }
      }

      fs_reqs = aligned_reqs.data();
      num_fs_reqs = aligned_reqs.size();
    }

    FileOperationInfo::StartTimePoint start_ts;
    if (ShouldNotifyListeners()) {
      start_ts = FileOperationInfo::StartNow();
    }

    if (io_s.ok()) {
      IOSTATS_CPU_TIMER_GUARD(cpu_read_nanos, clock_);
      if (rate_limiter_priority != Env::IO_TOTAL && rate_limiter_ != nullptr) {
        // TODO: ideally we should call `RateLimiter::RequestToken()` for
        // allowed bytes to multi-read and then consume those bytes by
        // satisfying as many requests in `MultiRead()` as possible, instead of
        // what we do here, which can cause burst when the
        // `total_multi_read_size` is big.
        size_t total_multi_read_size = 0;
        assert(fs_reqs != nullptr);
        for (size_t i = 0; i < num_fs_reqs; ++i) {
          FSReadRequest& req = fs_reqs[i];
          total_multi_read_size += req.len;
        }
        size_t remaining_bytes = total_multi_read_size;
        size_t request_bytes = 0;
        while (remaining_bytes > 0) {
          request_bytes = std::min(
              static_cast<size_t>(rate_limiter_->GetSingleBurstBytes()),
              remaining_bytes);
          rate_limiter_->Request(request_bytes, rate_limiter_priority,
                                 nullptr /* stats */,
                                 RateLimiter::OpType::kRead);
          remaining_bytes -= request_bytes;
        }
      }
      TEST_SYNC_POINT_CALLBACK(
          "RandomAccessFileReader::MultiRead:IODebugContext",
          const_cast<void*>(static_cast<void*>(dbg)));
      io_s = file_->MultiRead(fs_reqs, num_fs_reqs, opts, dbg);
      RecordInHistogram(stats_, MULTIGET_IO_BATCH_SIZE, num_fs_reqs);
    }

    if (use_direct_io() && io_s.ok()) {
      // Populate results in the unaligned read requests.
      size_t aligned_i = 0;
      for (size_t i = 0; i < num_reqs; i++) {
        auto& r = read_reqs[i];
        if (static_cast<size_t>(r.offset) > End(aligned_reqs[aligned_i])) {
          aligned_i++;
        }
        const auto& fs_r = fs_reqs[aligned_i];
        r.status = fs_r.status;
        if (r.status.ok()) {
          uint64_t offset = r.offset - fs_r.offset;
          if (fs_r.result.size() <= offset) {
            // No byte in the read range is returned.
            r.result = Slice();
          } else {
            size_t len = std::min(
                r.len, static_cast<size_t>(fs_r.result.size() - offset));
            r.result = Slice(fs_r.scratch + offset, len);
          }
        } else {
          r.result = Slice();
        }
      }
    }

    const bool overall_io_ok = io_s.ok();
    for (size_t i = 0; i < num_reqs; ++i) {
      const IOStatus& req_status = overall_io_ok ? read_reqs[i].status : io_s;
      const size_t result_size = overall_io_ok ? read_reqs[i].result.size() : 0;
      if (ShouldNotifyListeners()) {
        auto finish_ts = FileOperationInfo::FinishNow();
        NotifyOnFileReadFinish(read_reqs[i].offset, result_size, start_ts,
                               finish_ts, req_status);
      }
      if (!req_status.ok()) {
        NotifyOnIOError(req_status, FileOperationType::kRead, file_name(),
                        result_size, read_reqs[i].offset);
      }
      RecordIOStats(stats_, file_temperature_, is_last_level_, result_size);
    }
    SetPerfLevel(prev_perf_level);
  }
  if (stats_ != nullptr && file_read_hist_ != nullptr) {
    file_read_hist_->Add(elapsed);
  }

  return io_s;
}

IOStatus RandomAccessFileReader::PrepareIOOptions(const ReadOptions& ro,
                                                  IOOptions& opts,
                                                  IODebugContext* dbg) const {
  if (clock_ != nullptr) {
    return PrepareIOFromReadOptions(ro, clock_, opts, dbg);
  } else {
    return PrepareIOFromReadOptions(ro, SystemClock::Default().get(), opts,
                                    dbg);
  }
}

// Notes for when direct_io is enabled:
// Unless req.offset, req.len, req.scratch are all already aligned,
// RandomAccessFileReader creates an aligned request and aligned buffer for the
// request. If direct_io_buffer_context is provided, its buffer owns the aligned
// backing storage and its optional allocator is used only for this allocation.
// Otherwise, callers should provide either req.scratch or aligned_buf. If only
// req.scratch is provided, the result is copied from the allocated aligned
// buffer to req.scratch. If only aligned_buf is provided, it is set to the
// aligned buffer allocated by RandomAccessFileReader and saves a copy.
IOStatus RandomAccessFileReader::ReadAsync(
    FSReadRequest& req, const IOOptions& opts,
    std::function<void(FSReadRequest&, void*)> cb, void* cb_arg,
    void** io_handle, IOHandleDeleter* del_fn, AlignedBuf* aligned_buf,
    IODebugContext* dbg,
    AlignedBufferAllocationContext* direct_io_buffer_context) {
  AlignedBuffer* direct_io_buffer = direct_io_buffer_context != nullptr
                                        ? direct_io_buffer_context->buffer
                                        : nullptr;
  const AlignedBuffer::Allocator* direct_io_allocator =
      direct_io_buffer_context != nullptr ? direct_io_buffer_context->allocator
                                          : nullptr;
  assert(direct_io_buffer_context == nullptr || direct_io_buffer != nullptr);
  IOStatus s;
  TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::ReadAsync:InjectStatus",
                           &s);
  if (!s.ok()) {
    return s;
  }
  // 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(
      cb, cb_arg, (clock_ != nullptr ? clock_->NowMicros() : 0));

  if (ShouldNotifyListeners()) {
    read_async_info->fs_start_ts_ = FileOperationInfo::StartNow();
  }

  size_t alignment = file_->GetRequiredBufferAlignment();
  // The "already aligned" fast path forwards the caller's request (including
  // its scratch pointer) straight to the FileSystem. It is only safe when the
  // caller actually supplied a scratch buffer to read into. A null scratch
  // combined with an `aligned_buf` out-parameter means the caller expects this
  // reader to allocate the backing buffer (returned via `aligned_buf`), so we
  // must take the allocating path below. Otherwise a null buffer would be
  // submitted to the async read (e.g. a null iovec base to io_uring, which
  // fails with EFAULT).
  const bool caller_needs_allocation =
      req.scratch == nullptr && aligned_buf != nullptr;
  bool is_aligned = direct_io_buffer == nullptr && !caller_needs_allocation &&
                    (req.offset & (alignment - 1)) == 0 &&
                    (req.len & (alignment - 1)) == 0 &&
                    (uintptr_t(req.scratch) & (alignment - 1)) == 0;
  read_async_info->is_aligned_ = is_aligned;

  uint64_t elapsed = 0;
  if (use_direct_io() && is_aligned == false) {
    FSReadRequest aligned_req = Align(req, alignment);
    aligned_req.status.PermitUncheckedError();

    AlignedBuffer* aligned_read_buffer =
        direct_io_buffer != nullptr ? direct_io_buffer : &read_async_info->buf_;
    aligned_read_buffer->Alignment(alignment);
    Status allocate_status = aligned_read_buffer->AllocateNewBuffer(
        aligned_req.len, direct_io_allocator);
    if (!allocate_status.ok()) {
      delete read_async_info;
      return status_to_io_status(std::move(allocate_status));
    }
    aligned_req.scratch = aligned_read_buffer->BufferStart();

    // Set user provided fields to populate back in callback.
    read_async_info->user_scratch_ = req.scratch;
    read_async_info->user_aligned_buf_ = aligned_buf;
    read_async_info->direct_io_buffer_ = direct_io_buffer;
    read_async_info->user_len_ = req.len;
    read_async_info->user_offset_ = req.offset;
    read_async_info->user_result_ = req.result;

    assert(direct_io_buffer != nullptr ||
           read_async_info->buf_.CurrentSize() == 0);

    StopWatch sw(clock_, stats_, hist_type_,
                 GetFileReadHistograms(stats_, opts.io_activity),
                 (stats_ != nullptr) ? &elapsed : nullptr, true /*overwrite*/,
                 true /*delay_enabled*/);
    s = file_->ReadAsync(aligned_req, opts, read_async_callback,
                         read_async_info, io_handle, del_fn, dbg);
  } else {
    StopWatch sw(clock_, stats_, hist_type_,
                 GetFileReadHistograms(stats_, opts.io_activity),
                 (stats_ != nullptr) ? &elapsed : nullptr, true /*overwrite*/,
                 true /*delay_enabled*/);
    s = file_->ReadAsync(req, opts, read_async_callback, read_async_info,
                         io_handle, del_fn, dbg);
  }
  RecordTick(stats_, READ_ASYNC_MICROS, elapsed);

// Suppress false positive clang analyzer warnings.
// Memory is not released if file_->ReadAsync returns !s.ok(), because
// ReadAsyncCallback is never called in that case. If ReadAsyncCallback is
// called then ReadAsync should always return IOStatus::OK().
#ifndef __clang_analyzer__
  if (!s.ok()) {
    delete read_async_info;
  }
#endif  // __clang_analyzer__

  return s;
}

void RandomAccessFileReader::ReadAsyncCallback(FSReadRequest& req,
                                               void* cb_arg) {
  ReadAsyncInfo* read_async_info = static_cast<ReadAsyncInfo*>(cb_arg);
  assert(read_async_info);
  assert(read_async_info->cb_);

  if (use_direct_io() && read_async_info->is_aligned_ == false) {
    // Create FSReadRequest with user provided fields.
    FSReadRequest user_req;
    user_req.scratch = read_async_info->user_scratch_;
    user_req.offset = read_async_info->user_offset_;
    user_req.len = read_async_info->user_len_;

    // Update results in user_req.
    user_req.result = req.result;
    user_req.status = req.status;

    size_t current_size = req.result.size();
    if (read_async_info->direct_io_buffer_ == nullptr) {
      read_async_info->buf_.Size(read_async_info->buf_.CurrentSize() +
                                 req.result.size());
      current_size = read_async_info->buf_.CurrentSize();
    }

    size_t offset_advance_len = static_cast<size_t>(
        /*offset_passed_by_user=*/read_async_info->user_offset_ -
        /*aligned_offset=*/req.offset);

    size_t res_len = 0;
    if (req.status.ok() && offset_advance_len < current_size) {
      res_len = std::min(current_size - offset_advance_len,
                         read_async_info->user_len_);
      if (read_async_info->direct_io_buffer_ != nullptr) {
        user_req.scratch = read_async_info->direct_io_buffer_->BufferStart() +
                           offset_advance_len;
      } else if (read_async_info->user_aligned_buf_ == nullptr) {
        // Copy the data into user's scratch.
// Clang analyzer assumes that it will take use_direct_io() == false in
// ReadAsync and use_direct_io() == true in Callback which cannot be true.
#ifndef __clang_analyzer__
        read_async_info->buf_.Read(user_req.scratch, offset_advance_len,
                                   res_len);
#endif  // __clang_analyzer__
      } else {
        // Set aligned_buf provided by user without additional copy.
        user_req.scratch =
            read_async_info->buf_.BufferStart() + offset_advance_len;
        *read_async_info->user_aligned_buf_ = read_async_info->buf_.Release();
      }
      user_req.result = Slice(user_req.scratch, res_len);
    } else {
      // Either req.status is not ok or data was not read.
      user_req.result = Slice();
    }
    read_async_info->cb_(user_req, read_async_info->cb_arg_);
  } else {
    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());
  } else if (!req.status.IsAborted()) {
    RecordTick(stats_, ASYNC_READ_ERROR_COUNT, 1);
  }
  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);
  }
  RecordIOStats(stats_, file_temperature_, is_last_level_, req.result.size());
  delete read_async_info;
}
}  // namespace ROCKSDB_NAMESPACE