ceph/src/rocksdb/file/sequence_file_reader.h

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under both the GPLv2 (found in the
   3 //  COPYING file in the root directory) and Apache 2.0 License
   4 //  (found in the LICENSE.Apache file in the root directory).
   5 //
   6 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   7 // Use of this source code is governed by a BSD-style license that can be
   8 // found in the LICENSE file. See the AUTHORS file for names of contributors.
   9
  10 #pragma once
  11 #include <atomic>
  12 #include <string>
  13
  14 #include "env/file_system_tracer.h"
  15 #include "port/port.h"
  16 #include "rocksdb/env.h"
  17 #include "rocksdb/file_system.h"
  18
  19 namespace ROCKSDB_NAMESPACE {
  20
  21 // SequentialFileReader is a wrapper on top of Env::SequentialFile. It handles
  22 // Buffered (i.e when page cache is enabled) and Direct (with O_DIRECT / page
  23 // cache disabled) reads appropriately, and also updates the IO stats.
  24 class SequentialFileReader {
  25  private:
  26 #ifndef ROCKSDB_LITE
  27   void NotifyOnFileReadFinish(
  28       uint64_t offset, size_t length,
  29       const FileOperationInfo::StartTimePoint& start_ts,
  30       const FileOperationInfo::FinishTimePoint& finish_ts,
  31       const Status& status) const {
  32     FileOperationInfo info(FileOperationType::kRead, file_name_, start_ts,
  33                            finish_ts, status);
  34     info.offset = offset;
  35     info.length = length;
  36
  37     for (auto& listener : listeners_) {
  38       listener->OnFileReadFinish(info);
  39     }
  40     info.status.PermitUncheckedError();
  41   }
  42
  43   void AddFileIOListeners(
  44       const std::vector<std::shared_ptr<EventListener>>& listeners) {
  45     std::for_each(listeners.begin(), listeners.end(),
  46                   [this](const std::shared_ptr<EventListener>& e) {
  47                     if (e->ShouldBeNotifiedOnFileIO()) {
  48                       listeners_.emplace_back(e);
  49                     }
  50                   });
  51   }
  52 #endif  // ROCKSDB_LITE
  53
  54   bool ShouldNotifyListeners() const { return !listeners_.empty(); }
  55
  56   std::string file_name_;
  57   FSSequentialFilePtr file_;
  58   std::atomic<size_t> offset_{0};  // read offset
  59   std::vector<std::shared_ptr<EventListener>> listeners_{};
  60   RateLimiter* rate_limiter_;
  61
  62  public:
  63   explicit SequentialFileReader(
  64       std::unique_ptr<FSSequentialFile>&& _file, const std::string& _file_name,
  65       const std::shared_ptr<IOTracer>& io_tracer = nullptr,
  66       const std::vector<std::shared_ptr<EventListener>>& listeners = {},
  67       RateLimiter* rate_limiter =
  68           nullptr)  // TODO: migrate call sites to provide rate limiter
  69       : file_name_(_file_name),
  70         file_(std::move(_file), io_tracer, _file_name),
  71         listeners_(),
  72         rate_limiter_(rate_limiter) {
  73 #ifndef ROCKSDB_LITE
  74     AddFileIOListeners(listeners);
  75 #else
  76     (void)listeners;
  77 #endif
  78   }
  79
  80   explicit SequentialFileReader(
  81       std::unique_ptr<FSSequentialFile>&& _file, const std::string& _file_name,
  82       size_t _readahead_size,
  83       const std::shared_ptr<IOTracer>& io_tracer = nullptr,
  84       const std::vector<std::shared_ptr<EventListener>>& listeners = {},
  85       RateLimiter* rate_limiter =
  86           nullptr)  // TODO: migrate call sites to provide rate limiter
  87       : file_name_(_file_name),
  88         file_(NewReadaheadSequentialFile(std::move(_file), _readahead_size),
  89               io_tracer, _file_name),
  90         listeners_(),
  91         rate_limiter_(rate_limiter) {
  92 #ifndef ROCKSDB_LITE
  93     AddFileIOListeners(listeners);
  94 #else
  95     (void)listeners;
  96 #endif
  97   }
  98   static IOStatus Create(const std::shared_ptr<FileSystem>& fs,
  99                          const std::string& fname, const FileOptions& file_opts,
 100                          std::unique_ptr<SequentialFileReader>* reader,
 101                          IODebugContext* dbg, RateLimiter* rate_limiter);
 102
 103   SequentialFileReader(const SequentialFileReader&) = delete;
 104   SequentialFileReader& operator=(const SequentialFileReader&) = delete;
 105
 106   // `rate_limiter_priority` is used to charge the internal rate limiter when
 107   // enabled. The special value `Env::IO_TOTAL` makes this operation bypass the
 108   // rate limiter. The amount charged to the internal rate limiter is n, even
 109   // when less than n bytes are actually read (e.g. at end of file). To avoid
 110   // overcharging the rate limiter, the caller can use file size to cap n to
 111   // read until end of file.
 112   IOStatus Read(size_t n, Slice* result, char* scratch,
 113                 Env::IOPriority rate_limiter_priority);
 114
 115   IOStatus Skip(uint64_t n);
 116
 117   FSSequentialFile* file() { return file_.get(); }
 118
 119   std::string file_name() { return file_name_; }
 120
 121   bool use_direct_io() const { return file_->use_direct_io(); }
 122
 123  private:
 124   // NewReadaheadSequentialFile provides a wrapper over SequentialFile to
 125   // always prefetch additional data with every read.
 126   static std::unique_ptr<FSSequentialFile> NewReadaheadSequentialFile(
 127       std::unique_ptr<FSSequentialFile>&& file, size_t readahead_size);
 128 };
 129 }  // namespace ROCKSDB_NAMESPACE