ceph/src/rocksdb/util/file_reader_writer.h

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under the BSD-style license found in the
   3 //  LICENSE file in the root directory of this source tree. An additional grant
   4 //  of patent rights can be found in the PATENTS file in the same 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 #pragma once
  10 #include <atomic>
  11 #include <string>
  12 #include "port/port.h"
  13 #include "rocksdb/env.h"
  14 #include "util/aligned_buffer.h"
  15
  16 namespace rocksdb {
  17
  18 class Statistics;
  19 class HistogramImpl;
  20
  21 std::unique_ptr<RandomAccessFile> NewReadaheadRandomAccessFile(
  22   std::unique_ptr<RandomAccessFile>&& file, size_t readahead_size);
  23
  24 class SequentialFileReader {
  25  private:
  26   std::unique_ptr<SequentialFile> file_;
  27   std::atomic<size_t> offset_;  // read offset
  28
  29  public:
  30   explicit SequentialFileReader(std::unique_ptr<SequentialFile>&& _file)
  31       : file_(std::move(_file)), offset_(0) {}
  32
  33   SequentialFileReader(SequentialFileReader&& o) ROCKSDB_NOEXCEPT {
  34     *this = std::move(o);
  35   }
  36
  37   SequentialFileReader& operator=(SequentialFileReader&& o) ROCKSDB_NOEXCEPT {
  38     file_ = std::move(o.file_);
  39     return *this;
  40   }
  41
  42   SequentialFileReader(const SequentialFileReader&) = delete;
  43   SequentialFileReader& operator=(const SequentialFileReader&) = delete;
  44
  45   Status Read(size_t n, Slice* result, char* scratch);
  46
  47   Status Skip(uint64_t n);
  48
  49   SequentialFile* file() { return file_.get(); }
  50
  51   bool use_direct_io() const { return file_->use_direct_io(); }
  52
  53  protected:
  54   Status DirectRead(size_t n, Slice* result, char* scratch);
  55 };
  56
  57 class RandomAccessFileReader {
  58  private:
  59   std::unique_ptr<RandomAccessFile> file_;
  60   Env*            env_;
  61   Statistics*     stats_;
  62   uint32_t        hist_type_;
  63   HistogramImpl*  file_read_hist_;
  64
  65  public:
  66   explicit RandomAccessFileReader(std::unique_ptr<RandomAccessFile>&& raf,
  67                                   Env* env = nullptr,
  68                                   Statistics* stats = nullptr,
  69                                   uint32_t hist_type = 0,
  70                                   HistogramImpl* file_read_hist = nullptr)
  71       : file_(std::move(raf)),
  72         env_(env),
  73         stats_(stats),
  74         hist_type_(hist_type),
  75         file_read_hist_(file_read_hist) {}
  76
  77   RandomAccessFileReader(RandomAccessFileReader&& o) ROCKSDB_NOEXCEPT {
  78     *this = std::move(o);
  79   }
  80
  81   RandomAccessFileReader& operator=(RandomAccessFileReader&& o) ROCKSDB_NOEXCEPT{
  82     file_ = std::move(o.file_);
  83     env_ = std::move(o.env_);
  84     stats_ = std::move(o.stats_);
  85     hist_type_ = std::move(o.hist_type_);
  86     file_read_hist_ = std::move(o.file_read_hist_);
  87     return *this;
  88   }
  89
  90   RandomAccessFileReader(const RandomAccessFileReader&) = delete;
  91   RandomAccessFileReader& operator=(const RandomAccessFileReader&) = delete;
  92
  93   Status Read(uint64_t offset, size_t n, Slice* result, char* scratch) const;
  94
  95   Status Prefetch(uint64_t offset, size_t n) const {
  96     return file_->Prefetch(offset, n);
  97   }
  98
  99   RandomAccessFile* file() { return file_.get(); }
 100
 101   bool use_direct_io() const { return file_->use_direct_io(); }
 102
 103  protected:
 104   Status DirectRead(uint64_t offset, size_t n, Slice* result,
 105                     char* scratch) const;
 106 };
 107
 108 // Use posix write to write data to a file.
 109 class WritableFileWriter {
 110  private:
 111   std::unique_ptr<WritableFile> writable_file_;
 112   AlignedBuffer           buf_;
 113   size_t                  max_buffer_size_;
 114   // Actually written data size can be used for truncate
 115   // not counting padding data
 116   uint64_t                filesize_;
 117   // This is necessary when we use unbuffered access
 118   // and writes must happen on aligned offsets
 119   // so we need to go back and write that page again
 120   uint64_t                next_write_offset_;
 121   bool                    pending_sync_;
 122   uint64_t                last_sync_size_;
 123   uint64_t                bytes_per_sync_;
 124   RateLimiter*            rate_limiter_;
 125   Statistics* stats_;
 126
 127  public:
 128   WritableFileWriter(std::unique_ptr<WritableFile>&& file,
 129                      const EnvOptions& options, Statistics* stats = nullptr)
 130       : writable_file_(std::move(file)),
 131         buf_(),
 132         max_buffer_size_(options.writable_file_max_buffer_size),
 133         filesize_(0),
 134         next_write_offset_(0),
 135         pending_sync_(false),
 136         last_sync_size_(0),
 137         bytes_per_sync_(options.bytes_per_sync),
 138         rate_limiter_(options.rate_limiter),
 139         stats_(stats) {
 140     buf_.Alignment(writable_file_->GetRequiredBufferAlignment());
 141     buf_.AllocateNewBuffer(use_direct_io()
 142                                ? max_buffer_size_
 143                                : std::min((size_t)65536, max_buffer_size_));
 144   }
 145
 146   WritableFileWriter(const WritableFileWriter&) = delete;
 147
 148   WritableFileWriter& operator=(const WritableFileWriter&) = delete;
 149
 150   ~WritableFileWriter() { Close(); }
 151
 152   Status Append(const Slice& data);
 153
 154   Status Flush();
 155
 156   Status Close();
 157
 158   Status Sync(bool use_fsync);
 159
 160   // Sync only the data that was already Flush()ed. Safe to call concurrently
 161   // with Append() and Flush(). If !writable_file_->IsSyncThreadSafe(),
 162   // returns NotSupported status.
 163   Status SyncWithoutFlush(bool use_fsync);
 164
 165   uint64_t GetFileSize() { return filesize_; }
 166
 167   Status InvalidateCache(size_t offset, size_t length) {
 168     return writable_file_->InvalidateCache(offset, length);
 169   }
 170
 171   WritableFile* writable_file() const { return writable_file_.get(); }
 172
 173   bool use_direct_io() { return writable_file_->use_direct_io(); }
 174
 175  private:
 176   // Used when os buffering is OFF and we are writing
 177   // DMA such as in Direct I/O mode
 178 #ifndef ROCKSDB_LITE
 179   Status WriteDirect();
 180 #endif  // !ROCKSDB_LITE
 181   // Normal write
 182   Status WriteBuffered(const char* data, size_t size);
 183   Status RangeSync(uint64_t offset, uint64_t nbytes);
 184   size_t RequestToken(size_t bytes, bool align);
 185   Status SyncInternal(bool use_fsync);
 186 };
 187
 188 extern Status NewWritableFile(Env* env, const std::string& fname,
 189                               unique_ptr<WritableFile>* result,
 190                               const EnvOptions& options);
 191 }  // namespace rocksdb