ceph/src/rocksdb/utilities/env_mirror.cc

   1 // Copyright (c) 2015, Red Hat, 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 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   6 // Use of this source code is governed by a BSD-style license that can be
   7 // found in the LICENSE file. See the AUTHORS file for names of contributors.
   8
   9 #ifndef ROCKSDB_LITE
  10
  11 #include "rocksdb/utilities/env_mirror.h"
  12
  13 namespace rocksdb {
  14
  15 // An implementation of Env that mirrors all work over two backend
  16 // Env's.  This is useful for debugging purposes.
  17 class SequentialFileMirror : public SequentialFile {
  18  public:
  19   std::unique_ptr<SequentialFile> a_, b_;
  20   std::string fname;
  21   explicit SequentialFileMirror(std::string f) : fname(f) {}
  22
  23   Status Read(size_t n, Slice* result, char* scratch) override {
  24     Slice aslice;
  25     Status as = a_->Read(n, &aslice, scratch);
  26     if (as == Status::OK()) {
  27       char* bscratch = new char[n];
  28       Slice bslice;
  29       size_t off = 0;
  30       size_t left = aslice.size();
  31       while (left) {
  32         Status bs = b_->Read(left, &bslice, bscratch);
  33         assert(as == bs);
  34         assert(memcmp(bscratch, scratch + off, bslice.size()) == 0);
  35         off += bslice.size();
  36         left -= bslice.size();
  37       }
  38       delete[] bscratch;
  39       *result = aslice;
  40     } else {
  41       Status bs = b_->Read(n, result, scratch);
  42       assert(as == bs);
  43     }
  44     return as;
  45   }
  46
  47   Status Skip(uint64_t n) override {
  48     Status as = a_->Skip(n);
  49     Status bs = b_->Skip(n);
  50     assert(as == bs);
  51     return as;
  52   }
  53   Status InvalidateCache(size_t offset, size_t length) override {
  54     Status as = a_->InvalidateCache(offset, length);
  55     Status bs = b_->InvalidateCache(offset, length);
  56     assert(as == bs);
  57     return as;
  58   };
  59 };
  60
  61 class RandomAccessFileMirror : public RandomAccessFile {
  62  public:
  63   std::unique_ptr<RandomAccessFile> a_, b_;
  64   std::string fname;
  65   explicit RandomAccessFileMirror(std::string f) : fname(f) {}
  66
  67   Status Read(uint64_t offset, size_t n, Slice* result,
  68               char* scratch) const override {
  69     Status as = a_->Read(offset, n, result, scratch);
  70     if (as == Status::OK()) {
  71       char* bscratch = new char[n];
  72       Slice bslice;
  73       size_t off = 0;
  74       size_t left = result->size();
  75       while (left) {
  76         Status bs = b_->Read(offset + off, left, &bslice, bscratch);
  77         assert(as == bs);
  78         assert(memcmp(bscratch, scratch + off, bslice.size()) == 0);
  79         off += bslice.size();
  80         left -= bslice.size();
  81       }
  82       delete[] bscratch;
  83     } else {
  84       Status bs = b_->Read(offset, n, result, scratch);
  85       assert(as == bs);
  86     }
  87     return as;
  88   }
  89
  90   size_t GetUniqueId(char* id, size_t max_size) const override {
  91     // NOTE: not verified
  92     return a_->GetUniqueId(id, max_size);
  93   }
  94 };
  95
  96 class WritableFileMirror : public WritableFile {
  97  public:
  98   std::unique_ptr<WritableFile> a_, b_;
  99   std::string fname;
 100   explicit WritableFileMirror(std::string f) : fname(f) {}
 101
 102   Status Append(const Slice& data) override {
 103     Status as = a_->Append(data);
 104     Status bs = b_->Append(data);
 105     assert(as == bs);
 106     return as;
 107   }
 108   Status PositionedAppend(const Slice& data, uint64_t offset) override {
 109     Status as = a_->PositionedAppend(data, offset);
 110     Status bs = b_->PositionedAppend(data, offset);
 111     assert(as == bs);
 112     return as;
 113   }
 114   Status Truncate(uint64_t size) override {
 115     Status as = a_->Truncate(size);
 116     Status bs = b_->Truncate(size);
 117     assert(as == bs);
 118     return as;
 119   }
 120   Status Close() override {
 121     Status as = a_->Close();
 122     Status bs = b_->Close();
 123     assert(as == bs);
 124     return as;
 125   }
 126   Status Flush() override {
 127     Status as = a_->Flush();
 128     Status bs = b_->Flush();
 129     assert(as == bs);
 130     return as;
 131   }
 132   Status Sync() override {
 133     Status as = a_->Sync();
 134     Status bs = b_->Sync();
 135     assert(as == bs);
 136     return as;
 137   }
 138   Status Fsync() override {
 139     Status as = a_->Fsync();
 140     Status bs = b_->Fsync();
 141     assert(as == bs);
 142     return as;
 143   }
 144   bool IsSyncThreadSafe() const override {
 145     bool as = a_->IsSyncThreadSafe();
 146     assert(as == b_->IsSyncThreadSafe());
 147     return as;
 148   }
 149   void SetIOPriority(Env::IOPriority pri) override {
 150     a_->SetIOPriority(pri);
 151     b_->SetIOPriority(pri);
 152   }
 153   Env::IOPriority GetIOPriority() override {
 154     // NOTE: we don't verify this one
 155     return a_->GetIOPriority();
 156   }
 157   uint64_t GetFileSize() override {
 158     uint64_t as = a_->GetFileSize();
 159     assert(as == b_->GetFileSize());
 160     return as;
 161   }
 162   void GetPreallocationStatus(size_t* block_size,
 163                               size_t* last_allocated_block) override {
 164     // NOTE: we don't verify this one
 165     return a_->GetPreallocationStatus(block_size, last_allocated_block);
 166   }
 167   size_t GetUniqueId(char* id, size_t max_size) const override {
 168     // NOTE: we don't verify this one
 169     return a_->GetUniqueId(id, max_size);
 170   }
 171   Status InvalidateCache(size_t offset, size_t length) override {
 172     Status as = a_->InvalidateCache(offset, length);
 173     Status bs = b_->InvalidateCache(offset, length);
 174     assert(as == bs);
 175     return as;
 176   }
 177
 178  protected:
 179   Status Allocate(uint64_t offset, uint64_t length) override {
 180     Status as = a_->Allocate(offset, length);
 181     Status bs = b_->Allocate(offset, length);
 182     assert(as == bs);
 183     return as;
 184   }
 185   Status RangeSync(uint64_t offset, uint64_t nbytes) override {
 186     Status as = a_->RangeSync(offset, nbytes);
 187     Status bs = b_->RangeSync(offset, nbytes);
 188     assert(as == bs);
 189     return as;
 190   }
 191 };
 192
 193 Status EnvMirror::NewSequentialFile(const std::string& f,
 194                                     std::unique_ptr<SequentialFile>* r,
 195                                     const EnvOptions& options) {
 196   if (f.find("/proc/") == 0) {
 197     return a_->NewSequentialFile(f, r, options);
 198   }
 199   SequentialFileMirror* mf = new SequentialFileMirror(f);
 200   Status as = a_->NewSequentialFile(f, &mf->a_, options);
 201   Status bs = b_->NewSequentialFile(f, &mf->b_, options);
 202   assert(as == bs);
 203   if (as.ok())
 204     r->reset(mf);
 205   else
 206     delete mf;
 207   return as;
 208 }
 209
 210 Status EnvMirror::NewRandomAccessFile(const std::string& f,
 211                                       std::unique_ptr<RandomAccessFile>* r,
 212                                       const EnvOptions& options) {
 213   if (f.find("/proc/") == 0) {
 214     return a_->NewRandomAccessFile(f, r, options);
 215   }
 216   RandomAccessFileMirror* mf = new RandomAccessFileMirror(f);
 217   Status as = a_->NewRandomAccessFile(f, &mf->a_, options);
 218   Status bs = b_->NewRandomAccessFile(f, &mf->b_, options);
 219   assert(as == bs);
 220   if (as.ok())
 221     r->reset(mf);
 222   else
 223     delete mf;
 224   return as;
 225 }
 226
 227 Status EnvMirror::NewWritableFile(const std::string& f,
 228                                   std::unique_ptr<WritableFile>* r,
 229                                   const EnvOptions& options) {
 230   if (f.find("/proc/") == 0) return a_->NewWritableFile(f, r, options);
 231   WritableFileMirror* mf = new WritableFileMirror(f);
 232   Status as = a_->NewWritableFile(f, &mf->a_, options);
 233   Status bs = b_->NewWritableFile(f, &mf->b_, options);
 234   assert(as == bs);
 235   if (as.ok())
 236     r->reset(mf);
 237   else
 238     delete mf;
 239   return as;
 240 }
 241
 242 Status EnvMirror::ReuseWritableFile(const std::string& fname,
 243                                     const std::string& old_fname,
 244                                     std::unique_ptr<WritableFile>* r,
 245                                     const EnvOptions& options) {
 246   if (fname.find("/proc/") == 0)
 247     return a_->ReuseWritableFile(fname, old_fname, r, options);
 248   WritableFileMirror* mf = new WritableFileMirror(fname);
 249   Status as = a_->ReuseWritableFile(fname, old_fname, &mf->a_, options);
 250   Status bs = b_->ReuseWritableFile(fname, old_fname, &mf->b_, options);
 251   assert(as == bs);
 252   if (as.ok())
 253     r->reset(mf);
 254   else
 255     delete mf;
 256   return as;
 257 }
 258
 259 }  // namespace rocksdb
 260 #endif