Merge pull request #862 from ceph/wip-env

implement EnvMirror
main
Igor Canadi 9 years ago
commit 64fa43843b
  1. 1
      CMakeLists.txt
  2. 3
      Makefile
  3. 1
      include/rocksdb/env.h
  4. 166
      include/rocksdb/utilities/env_mirror.h
  5. 1
      src.mk
  6. 266
      utilities/env_mirror.cc
  7. 222
      utilities/env_mirror_test.cc

@ -244,6 +244,7 @@ set(SOURCES
utilities/document/document_db.cc
utilities/document/json_document.cc
utilities/document/json_document_builder.cc
utilities/env_mirror.cc
utilities/flashcache/flashcache.cc
utilities/geodb/geodb_impl.cc
utilities/leveldb_options/leveldb_options.cc

@ -805,6 +805,9 @@ json_document_test: utilities/document/json_document_test.o $(LIBOBJECTS) $(TEST
spatial_db_test: utilities/spatialdb/spatial_db_test.o $(LIBOBJECTS) $(TESTHARNESS)
$(AM_LINK)
env_mirror_test: utilities/env_mirror_test.o $(LIBOBJECTS) $(TESTHARNESS)
$(AM_LINK)
ttl_test: utilities/ttl/ttl_test.o $(LIBOBJECTS) $(TESTHARNESS)
$(AM_LINK)

@ -615,6 +615,7 @@ class WritableFile {
protected:
friend class WritableFileWrapper;
friend class WritableFileMirror;
Env::IOPriority io_priority_;
};

@ -0,0 +1,166 @@
// Copyright (c) 2015, Red Hat, Inc. All rights reserved.
// This source code is licensed under the BSD-style license found in the
// LICENSE file in the root directory of this source tree. An additional grant
// of patent rights can be found in the PATENTS file in the same 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.
//
// MirrorEnv is an Env implementation that mirrors all file-related
// operations to two backing Env's (provided at construction time).
// Writes are mirrored. For read operations, we do the read from both
// backends and assert that the results match.
//
// This is useful when implementing a new Env and ensuring that the
// semantics and behavior are correct (in that they match that of an
// existing, stable Env, like the default POSIX one).
#ifndef ROCKSDB_LITE
#ifndef STORAGE_ROCKSDB_INCLUDE_UTILITIES_ENVMIRROR_H_
#define STORAGE_ROCKSDB_INCLUDE_UTLIITIES_ENVMIRROR_H_
#include <iostream>
#include <algorithm>
#include <vector>
#include "rocksdb/env.h"
namespace rocksdb {
class SequentialFileMirror;
class RandomAccessFileMirror;
class WritableFileMirror;
class EnvMirror : public EnvWrapper {
Env* a_, *b_;
public:
EnvMirror(Env* a, Env* b) : EnvWrapper(a), a_(a), b_(b) {}
Status NewSequentialFile(const std::string& f, unique_ptr<SequentialFile>* r,
const EnvOptions& options) override;
Status NewRandomAccessFile(const std::string& f,
unique_ptr<RandomAccessFile>* r,
const EnvOptions& options) override;
Status NewWritableFile(const std::string& f, unique_ptr<WritableFile>* r,
const EnvOptions& options) override;
Status ReuseWritableFile(const std::string& fname,
const std::string& old_fname,
unique_ptr<WritableFile>* r,
const EnvOptions& options) override;
virtual Status NewDirectory(const std::string& name,
unique_ptr<Directory>* result) override {
unique_ptr<Directory> br;
Status as = a_->NewDirectory(name, result);
Status bs = b_->NewDirectory(name, &br);
assert(as == bs);
return as;
}
Status FileExists(const std::string& f) override {
Status as = a_->FileExists(f);
Status bs = b_->FileExists(f);
assert(as == bs);
return as;
}
Status GetChildren(const std::string& dir,
std::vector<std::string>* r) override {
std::vector<std::string> ar, br;
Status as = a_->GetChildren(dir, &ar);
Status bs = b_->GetChildren(dir, &br);
assert(as == bs);
std::sort(ar.begin(), ar.end());
std::sort(br.begin(), br.end());
if (!as.ok() || ar != br) {
assert(0 == "getchildren results don't match");
}
*r = ar;
return as;
}
Status DeleteFile(const std::string& f) override {
Status as = a_->DeleteFile(f);
Status bs = b_->DeleteFile(f);
assert(as == bs);
return as;
}
Status CreateDir(const std::string& d) override {
Status as = a_->CreateDir(d);
Status bs = b_->CreateDir(d);
assert(as == bs);
return as;
}
Status CreateDirIfMissing(const std::string& d) override {
Status as = a_->CreateDirIfMissing(d);
Status bs = b_->CreateDirIfMissing(d);
assert(as == bs);
return as;
}
Status DeleteDir(const std::string& d) override {
Status as = a_->DeleteDir(d);
Status bs = b_->DeleteDir(d);
assert(as == bs);
return as;
}
Status GetFileSize(const std::string& f, uint64_t* s) override {
uint64_t asize, bsize;
Status as = a_->GetFileSize(f, &asize);
Status bs = b_->GetFileSize(f, &bsize);
assert(as == bs);
assert(!as.ok() || asize == bsize);
*s = asize;
return as;
}
Status GetFileModificationTime(const std::string& fname,
uint64_t* file_mtime) override {
uint64_t amtime, bmtime;
Status as = a_->GetFileModificationTime(fname, &amtime);
Status bs = b_->GetFileModificationTime(fname, &bmtime);
assert(as == bs);
assert(!as.ok() || amtime - bmtime < 10000 || bmtime - amtime < 10000);
*file_mtime = amtime;
return as;
}
Status RenameFile(const std::string& s, const std::string& t) override {
Status as = a_->RenameFile(s, t);
Status bs = b_->RenameFile(s, t);
assert(as == bs);
return as;
}
Status LinkFile(const std::string& s, const std::string& t) override {
Status as = a_->LinkFile(s, t);
Status bs = b_->LinkFile(s, t);
assert(as == bs);
return as;
}
class FileLockMirror : public FileLock {
public:
FileLock* a_, *b_;
FileLockMirror(FileLock* a, FileLock* b) : a_(a), b_(b) {}
};
Status LockFile(const std::string& f, FileLock** l) override {
FileLock* al, *bl;
Status as = a_->LockFile(f, &al);
Status bs = b_->LockFile(f, &bl);
assert(as == bs);
if (as.ok()) *l = new FileLockMirror(al, bl);
return as;
}
Status UnlockFile(FileLock* l) override {
FileLockMirror* ml = static_cast<FileLockMirror*>(l);
Status as = a_->UnlockFile(ml->a_);
Status bs = b_->UnlockFile(ml->b_);
assert(as == bs);
return as;
}
};
} // namespace rocksdb
#endif // STORAGE_ROCKSDB_INCLUDE_UTILITIES_ENVMIRROR_H_
#endif // ROCKSDB_LITE

@ -111,6 +111,7 @@ LIB_SOURCES = \
utilities/document/document_db.cc \
utilities/document/json_document_builder.cc \
utilities/document/json_document.cc \
utilities/env_mirror.cc \
utilities/flashcache/flashcache.cc \
utilities/geodb/geodb_impl.cc \
utilities/leveldb_options/leveldb_options.cc \

@ -0,0 +1,266 @@
// Copyright (c) 2015, Red Hat, Inc. All rights reserved.
// This source code is licensed under the BSD-style license found in the
// LICENSE file in the root directory of this source tree. An additional grant
// of patent rights can be found in the PATENTS file in the same 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.
#ifndef ROCKSDB_LITE
#include "rocksdb/utilities/env_mirror.h"
namespace rocksdb {
// An implementaiton of Env that mirrors all work over two backend
// Env's. This is useful for debugging purposes.
class SequentialFileMirror : public SequentialFile {
public:
unique_ptr<SequentialFile> a_, b_;
std::string fname;
SequentialFileMirror(std::string f) : fname(f) {}
Status Read(size_t n, Slice* result, char* scratch) {
Slice aslice;
Status as = a_->Read(n, &aslice, scratch);
if (as == Status::OK()) {
char* bscratch = new char[n];
Slice bslice;
size_t off = 0;
size_t left = aslice.size();
while (left) {
Status bs = b_->Read(left, &bslice, bscratch);
assert(as == bs);
assert(memcmp(bscratch, scratch + off, bslice.size()) == 0);
off += bslice.size();
left -= bslice.size();
}
delete[] bscratch;
*result = aslice;
} else {
Status bs = b_->Read(n, result, scratch);
assert(as == bs);
}
return as;
}
Status Skip(uint64_t n) {
Status as = a_->Skip(n);
Status bs = b_->Skip(n);
assert(as == bs);
return as;
}
Status InvalidateCache(size_t offset, size_t length) {
Status as = a_->InvalidateCache(offset, length);
Status bs = b_->InvalidateCache(offset, length);
assert(as == bs);
return as;
};
};
class RandomAccessFileMirror : public RandomAccessFile {
public:
unique_ptr<RandomAccessFile> a_, b_;
std::string fname;
RandomAccessFileMirror(std::string f) : fname(f) {}
Status Read(uint64_t offset, size_t n, Slice* result, char* scratch) const {
Status as = a_->Read(offset, n, result, scratch);
if (as == Status::OK()) {
char* bscratch = new char[n];
Slice bslice;
size_t off = 0;
size_t left = result->size();
while (left) {
Status bs = b_->Read(offset + off, left, &bslice, bscratch);
assert(as == bs);
assert(memcmp(bscratch, scratch + off, bslice.size()) == 0);
off += bslice.size();
left -= bslice.size();
}
delete[] bscratch;
} else {
Status bs = b_->Read(offset, n, result, scratch);
assert(as == bs);
}
return as;
}
bool ShouldForwardRawRequest() const {
// NOTE: not verified
return a_->ShouldForwardRawRequest();
}
size_t GetUniqueId(char* id, size_t max_size) const {
// NOTE: not verified
return a_->GetUniqueId(id, max_size);
}
};
class WritableFileMirror : public WritableFile {
public:
unique_ptr<WritableFile> a_, b_;
std::string fname;
WritableFileMirror(std::string f) : fname(f) {}
Status Append(const Slice& data) override {
Status as = a_->Append(data);
Status bs = b_->Append(data);
assert(as == bs);
return as;
}
Status PositionedAppend(const Slice& data, uint64_t offset) override {
Status as = a_->PositionedAppend(data, offset);
Status bs = b_->PositionedAppend(data, offset);
assert(as == bs);
return as;
}
Status Truncate(uint64_t size) override {
Status as = a_->Truncate(size);
Status bs = b_->Truncate(size);
assert(as == bs);
return as;
}
Status Close() override {
Status as = a_->Close();
Status bs = b_->Close();
assert(as == bs);
return as;
}
Status Flush() override {
Status as = a_->Flush();
Status bs = b_->Flush();
assert(as == bs);
return as;
}
Status Sync() override {
Status as = a_->Sync();
Status bs = b_->Sync();
assert(as == bs);
return as;
}
Status Fsync() override {
Status as = a_->Fsync();
Status bs = b_->Fsync();
assert(as == bs);
return as;
}
bool IsSyncThreadSafe() const override {
bool as = a_->IsSyncThreadSafe();
bool bs = b_->IsSyncThreadSafe();
assert(as == bs);
return as;
}
void SetIOPriority(Env::IOPriority pri) override {
a_->SetIOPriority(pri);
b_->SetIOPriority(pri);
}
Env::IOPriority GetIOPriority() override {
// NOTE: we don't verify this one
return a_->GetIOPriority();
}
uint64_t GetFileSize() override {
uint64_t as = a_->GetFileSize();
uint64_t bs = b_->GetFileSize();
assert(as == bs);
return as;
}
void GetPreallocationStatus(size_t* block_size,
size_t* last_allocated_block) override {
// NOTE: we don't verify this one
return a_->GetPreallocationStatus(block_size, last_allocated_block);
}
size_t GetUniqueId(char* id, size_t max_size) const override {
// NOTE: we don't verify this one
return a_->GetUniqueId(id, max_size);
}
Status InvalidateCache(size_t offset, size_t length) override {
Status as = a_->InvalidateCache(offset, length);
Status bs = b_->InvalidateCache(offset, length);
assert(as == bs);
return as;
}
protected:
Status Allocate(uint64_t offset, uint64_t length) override {
Status as = a_->Allocate(offset, length);
Status bs = b_->Allocate(offset, length);
assert(as == bs);
return as;
}
Status RangeSync(uint64_t offset, uint64_t nbytes) override {
Status as = a_->RangeSync(offset, nbytes);
Status bs = b_->RangeSync(offset, nbytes);
assert(as == bs);
return as;
}
};
Status EnvMirror::NewSequentialFile(const std::string& f,
unique_ptr<SequentialFile>* r,
const EnvOptions& options) {
if (f.find("/proc/") == 0) {
return a_->NewSequentialFile(f, r, options);
}
SequentialFileMirror* mf = new SequentialFileMirror(f);
Status as = a_->NewSequentialFile(f, &mf->a_, options);
Status bs = b_->NewSequentialFile(f, &mf->b_, options);
assert(as == bs);
if (as.ok())
r->reset(mf);
else
delete mf;
return as;
}
Status EnvMirror::NewRandomAccessFile(const std::string& f,
unique_ptr<RandomAccessFile>* r,
const EnvOptions& options) {
if (f.find("/proc/") == 0) {
return a_->NewRandomAccessFile(f, r, options);
}
RandomAccessFileMirror* mf = new RandomAccessFileMirror(f);
Status as = a_->NewRandomAccessFile(f, &mf->a_, options);
Status bs = b_->NewRandomAccessFile(f, &mf->b_, options);
assert(as == bs);
if (as.ok())
r->reset(mf);
else
delete mf;
return as;
}
Status EnvMirror::NewWritableFile(const std::string& f,
unique_ptr<WritableFile>* r,
const EnvOptions& options) {
if (f.find("/proc/") == 0) return a_->NewWritableFile(f, r, options);
WritableFileMirror* mf = new WritableFileMirror(f);
Status as = a_->NewWritableFile(f, &mf->a_, options);
Status bs = b_->NewWritableFile(f, &mf->b_, options);
assert(as == bs);
if (as.ok())
r->reset(mf);
else
delete mf;
return as;
}
Status EnvMirror::ReuseWritableFile(const std::string& fname,
const std::string& old_fname,
unique_ptr<WritableFile>* r,
const EnvOptions& options) {
if (fname.find("/proc/") == 0)
return a_->ReuseWritableFile(fname, old_fname, r, options);
WritableFileMirror* mf = new WritableFileMirror(fname);
Status as = a_->ReuseWritableFile(fname, old_fname, &mf->a_, options);
Status bs = b_->ReuseWritableFile(fname, old_fname, &mf->b_, options);
assert(as == bs);
if (as.ok())
r->reset(mf);
else
delete mf;
return as;
}
} // namespace rocksdb
#endif

@ -0,0 +1,222 @@
// Copyright (c) 2015, Red Hat, Inc. All rights reserved.
// This source code is licensed under the BSD-style license found in the
// LICENSE file in the root directory of this source tree. An additional grant
// of patent rights can be found in the PATENTS file in the same directory.
#ifndef ROCKSDB_LITE
#include "rocksdb/utilities/env_mirror.h"
#include "util/mock_env.h"
#include "util/testharness.h"
namespace rocksdb {
class EnvMirrorTest : public testing::Test {
public:
Env* default_;
MockEnv* a_, *b_;
EnvMirror* env_;
const EnvOptions soptions_;
EnvMirrorTest()
: default_(Env::Default()),
a_(new MockEnv(default_)),
b_(new MockEnv(default_)),
env_(new EnvMirror(a_, b_)) {}
~EnvMirrorTest() {
delete env_;
delete a_;
delete b_;
}
};
TEST_F(EnvMirrorTest, Basics) {
uint64_t file_size;
unique_ptr<WritableFile> writable_file;
std::vector<std::string> children;
ASSERT_OK(env_->CreateDir("/dir"));
// Check that the directory is empty.
ASSERT_EQ(Status::NotFound(), env_->FileExists("/dir/non_existent"));
ASSERT_TRUE(!env_->GetFileSize("/dir/non_existent", &file_size).ok());
ASSERT_OK(env_->GetChildren("/dir", &children));
ASSERT_EQ(0U, children.size());
// Create a file.
ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
writable_file.reset();
// Check that the file exists.
ASSERT_OK(env_->FileExists("/dir/f"));
ASSERT_OK(a_->FileExists("/dir/f"));
ASSERT_OK(b_->FileExists("/dir/f"));
ASSERT_OK(env_->GetFileSize("/dir/f", &file_size));
ASSERT_EQ(0U, file_size);
ASSERT_OK(env_->GetChildren("/dir", &children));
ASSERT_EQ(1U, children.size());
ASSERT_EQ("f", children[0]);
ASSERT_OK(a_->GetChildren("/dir", &children));
ASSERT_EQ(1U, children.size());
ASSERT_EQ("f", children[0]);
ASSERT_OK(b_->GetChildren("/dir", &children));
ASSERT_EQ(1U, children.size());
ASSERT_EQ("f", children[0]);
// Write to the file.
ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
ASSERT_OK(writable_file->Append("abc"));
writable_file.reset();
// Check for expected size.
ASSERT_OK(env_->GetFileSize("/dir/f", &file_size));
ASSERT_EQ(3U, file_size);
ASSERT_OK(a_->GetFileSize("/dir/f", &file_size));
ASSERT_EQ(3U, file_size);
ASSERT_OK(b_->GetFileSize("/dir/f", &file_size));
ASSERT_EQ(3U, file_size);
// Check that renaming works.
ASSERT_TRUE(!env_->RenameFile("/dir/non_existent", "/dir/g").ok());
ASSERT_OK(env_->RenameFile("/dir/f", "/dir/g"));
ASSERT_EQ(Status::NotFound(), env_->FileExists("/dir/f"));
ASSERT_OK(env_->FileExists("/dir/g"));
ASSERT_OK(env_->GetFileSize("/dir/g", &file_size));
ASSERT_EQ(3U, file_size);
ASSERT_OK(a_->FileExists("/dir/g"));
ASSERT_OK(a_->GetFileSize("/dir/g", &file_size));
ASSERT_EQ(3U, file_size);
ASSERT_OK(b_->FileExists("/dir/g"));
ASSERT_OK(b_->GetFileSize("/dir/g", &file_size));
ASSERT_EQ(3U, file_size);
// Check that opening non-existent file fails.
unique_ptr<SequentialFile> seq_file;
unique_ptr<RandomAccessFile> rand_file;
ASSERT_TRUE(
!env_->NewSequentialFile("/dir/non_existent", &seq_file, soptions_).ok());
ASSERT_TRUE(!seq_file);
ASSERT_TRUE(!env_->NewRandomAccessFile("/dir/non_existent", &rand_file,
soptions_).ok());
ASSERT_TRUE(!rand_file);
// Check that deleting works.
ASSERT_TRUE(!env_->DeleteFile("/dir/non_existent").ok());
ASSERT_OK(env_->DeleteFile("/dir/g"));
ASSERT_EQ(Status::NotFound(), env_->FileExists("/dir/g"));
ASSERT_OK(env_->GetChildren("/dir", &children));
ASSERT_EQ(0U, children.size());
ASSERT_OK(env_->DeleteDir("/dir"));
}
TEST_F(EnvMirrorTest, ReadWrite) {
unique_ptr<WritableFile> writable_file;
unique_ptr<SequentialFile> seq_file;
unique_ptr<RandomAccessFile> rand_file;
Slice result;
char scratch[100];
ASSERT_OK(env_->CreateDir("/dir"));
ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
ASSERT_OK(writable_file->Append("hello "));
ASSERT_OK(writable_file->Append("world"));
writable_file.reset();
// Read sequentially.
ASSERT_OK(env_->NewSequentialFile("/dir/f", &seq_file, soptions_));
ASSERT_OK(seq_file->Read(5, &result, scratch)); // Read "hello".
ASSERT_EQ(0, result.compare("hello"));
ASSERT_OK(seq_file->Skip(1));
ASSERT_OK(seq_file->Read(1000, &result, scratch)); // Read "world".
ASSERT_EQ(0, result.compare("world"));
ASSERT_OK(seq_file->Read(1000, &result, scratch)); // Try reading past EOF.
ASSERT_EQ(0U, result.size());
ASSERT_OK(seq_file->Skip(100)); // Try to skip past end of file.
ASSERT_OK(seq_file->Read(1000, &result, scratch));
ASSERT_EQ(0U, result.size());
// Random reads.
ASSERT_OK(env_->NewRandomAccessFile("/dir/f", &rand_file, soptions_));
ASSERT_OK(rand_file->Read(6, 5, &result, scratch)); // Read "world".
ASSERT_EQ(0, result.compare("world"));
ASSERT_OK(rand_file->Read(0, 5, &result, scratch)); // Read "hello".
ASSERT_EQ(0, result.compare("hello"));
ASSERT_OK(rand_file->Read(10, 100, &result, scratch)); // Read "d".
ASSERT_EQ(0, result.compare("d"));
// Too high offset.
ASSERT_TRUE(!rand_file->Read(1000, 5, &result, scratch).ok());
}
TEST_F(EnvMirrorTest, Locks) {
FileLock* lock;
// These are no-ops, but we test they return success.
ASSERT_OK(env_->LockFile("some file", &lock));
ASSERT_OK(env_->UnlockFile(lock));
}
TEST_F(EnvMirrorTest, Misc) {
std::string test_dir;
ASSERT_OK(env_->GetTestDirectory(&test_dir));
ASSERT_TRUE(!test_dir.empty());
unique_ptr<WritableFile> writable_file;
ASSERT_OK(env_->NewWritableFile("/a/b", &writable_file, soptions_));
// These are no-ops, but we test they return success.
ASSERT_OK(writable_file->Sync());
ASSERT_OK(writable_file->Flush());
ASSERT_OK(writable_file->Close());
writable_file.reset();
}
TEST_F(EnvMirrorTest, LargeWrite) {
const size_t kWriteSize = 300 * 1024;
char* scratch = new char[kWriteSize * 2];
std::string write_data;
for (size_t i = 0; i < kWriteSize; ++i) {
write_data.append(1, static_cast<char>(i));
}
unique_ptr<WritableFile> writable_file;
ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
ASSERT_OK(writable_file->Append("foo"));
ASSERT_OK(writable_file->Append(write_data));
writable_file.reset();
unique_ptr<SequentialFile> seq_file;
Slice result;
ASSERT_OK(env_->NewSequentialFile("/dir/f", &seq_file, soptions_));
ASSERT_OK(seq_file->Read(3, &result, scratch)); // Read "foo".
ASSERT_EQ(0, result.compare("foo"));
size_t read = 0;
std::string read_data;
while (read < kWriteSize) {
ASSERT_OK(seq_file->Read(kWriteSize - read, &result, scratch));
read_data.append(result.data(), result.size());
read += result.size();
}
ASSERT_TRUE(write_data == read_data);
delete[] scratch;
}
} // namespace rocksdb
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
#else
#include <stdio.h>
int main(int argc, char** argv) {
fprintf(stderr, "SKIPPED as EnvMirror is not supported in ROCKSDB_LITE\n");
return 0;
}
#endif // !ROCKSDB_LITE
Loading…
Cancel
Save