// Copyright (c) 2019-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). // #include "env/composite_env_wrapper.h" #include "rocksdb/utilities/options_type.h" #include "util/string_util.h" namespace ROCKSDB_NAMESPACE { namespace { // The CompositeEnvWrapper class provides an interface that is compatible // with the old monolithic Env API, and an implementation that wraps around // the new Env that provides threading and other OS related functionality, and // the new FileSystem API that provides storage functionality. By // providing the old Env interface, it allows the rest of RocksDB code to // be agnostic of whether the underlying Env implementation is a monolithic // Env or an Env + FileSystem. In the former case, the user will specify // Options::env only, whereas in the latter case, the user will specify // Options::env and Options::file_system. class CompositeSequentialFileWrapper : public SequentialFile { public: explicit CompositeSequentialFileWrapper( std::unique_ptr& target) : target_(std::move(target)) {} Status Read(size_t n, Slice* result, char* scratch) override { IOOptions io_opts; IODebugContext dbg; return target_->Read(n, io_opts, result, scratch, &dbg); } Status Skip(uint64_t n) override { return target_->Skip(n); } bool use_direct_io() const override { return target_->use_direct_io(); } size_t GetRequiredBufferAlignment() const override { return target_->GetRequiredBufferAlignment(); } Status InvalidateCache(size_t offset, size_t length) override { return target_->InvalidateCache(offset, length); } Status PositionedRead(uint64_t offset, size_t n, Slice* result, char* scratch) override { IOOptions io_opts; IODebugContext dbg; return target_->PositionedRead(offset, n, io_opts, result, scratch, &dbg); } private: std::unique_ptr target_; }; class CompositeRandomAccessFileWrapper : public RandomAccessFile { public: explicit CompositeRandomAccessFileWrapper( std::unique_ptr& target) : target_(std::move(target)) {} Status Read(uint64_t offset, size_t n, Slice* result, char* scratch) const override { IOOptions io_opts; IODebugContext dbg; return target_->Read(offset, n, io_opts, result, scratch, &dbg); } Status MultiRead(ReadRequest* reqs, size_t num_reqs) override { IOOptions io_opts; IODebugContext dbg; std::vector fs_reqs; Status status; fs_reqs.resize(num_reqs); for (size_t i = 0; i < num_reqs; ++i) { fs_reqs[i].offset = reqs[i].offset; fs_reqs[i].len = reqs[i].len; fs_reqs[i].scratch = reqs[i].scratch; fs_reqs[i].status = IOStatus::OK(); } status = target_->MultiRead(fs_reqs.data(), num_reqs, io_opts, &dbg); for (size_t i = 0; i < num_reqs; ++i) { reqs[i].result = fs_reqs[i].result; reqs[i].status = fs_reqs[i].status; } return status; } Status Prefetch(uint64_t offset, size_t n) override { IOOptions io_opts; IODebugContext dbg; return target_->Prefetch(offset, n, io_opts, &dbg); } size_t GetUniqueId(char* id, size_t max_size) const override { return target_->GetUniqueId(id, max_size); } void Hint(AccessPattern pattern) override { target_->Hint((FSRandomAccessFile::AccessPattern)pattern); } bool use_direct_io() const override { return target_->use_direct_io(); } size_t GetRequiredBufferAlignment() const override { return target_->GetRequiredBufferAlignment(); } Status InvalidateCache(size_t offset, size_t length) override { return target_->InvalidateCache(offset, length); } private: std::unique_ptr target_; }; class CompositeWritableFileWrapper : public WritableFile { public: explicit CompositeWritableFileWrapper(std::unique_ptr& t) : target_(std::move(t)) {} Status Append(const Slice& data) override { IOOptions io_opts; IODebugContext dbg; return target_->Append(data, io_opts, &dbg); } Status Append(const Slice& data, const DataVerificationInfo& verification_info) override { IOOptions io_opts; IODebugContext dbg; return target_->Append(data, io_opts, verification_info, &dbg); } Status PositionedAppend(const Slice& data, uint64_t offset) override { IOOptions io_opts; IODebugContext dbg; return target_->PositionedAppend(data, offset, io_opts, &dbg); } Status PositionedAppend( const Slice& data, uint64_t offset, const DataVerificationInfo& verification_info) override { IOOptions io_opts; IODebugContext dbg; return target_->PositionedAppend(data, offset, io_opts, verification_info, &dbg); } Status Truncate(uint64_t size) override { IOOptions io_opts; IODebugContext dbg; return target_->Truncate(size, io_opts, &dbg); } Status Close() override { IOOptions io_opts; IODebugContext dbg; return target_->Close(io_opts, &dbg); } Status Flush() override { IOOptions io_opts; IODebugContext dbg; return target_->Flush(io_opts, &dbg); } Status Sync() override { IOOptions io_opts; IODebugContext dbg; return target_->Sync(io_opts, &dbg); } Status Fsync() override { IOOptions io_opts; IODebugContext dbg; return target_->Fsync(io_opts, &dbg); } bool IsSyncThreadSafe() const override { return target_->IsSyncThreadSafe(); } bool use_direct_io() const override { return target_->use_direct_io(); } size_t GetRequiredBufferAlignment() const override { return target_->GetRequiredBufferAlignment(); } void SetWriteLifeTimeHint(Env::WriteLifeTimeHint hint) override { target_->SetWriteLifeTimeHint(hint); } Env::WriteLifeTimeHint GetWriteLifeTimeHint() override { return target_->GetWriteLifeTimeHint(); } uint64_t GetFileSize() override { IOOptions io_opts; IODebugContext dbg; return target_->GetFileSize(io_opts, &dbg); } void SetPreallocationBlockSize(size_t size) override { target_->SetPreallocationBlockSize(size); } void GetPreallocationStatus(size_t* block_size, size_t* last_allocated_block) override { target_->GetPreallocationStatus(block_size, last_allocated_block); } size_t GetUniqueId(char* id, size_t max_size) const override { return target_->GetUniqueId(id, max_size); } Status InvalidateCache(size_t offset, size_t length) override { return target_->InvalidateCache(offset, length); } Status RangeSync(uint64_t offset, uint64_t nbytes) override { IOOptions io_opts; IODebugContext dbg; return target_->RangeSync(offset, nbytes, io_opts, &dbg); } void PrepareWrite(size_t offset, size_t len) override { IOOptions io_opts; IODebugContext dbg; target_->PrepareWrite(offset, len, io_opts, &dbg); } Status Allocate(uint64_t offset, uint64_t len) override { IOOptions io_opts; IODebugContext dbg; return target_->Allocate(offset, len, io_opts, &dbg); } std::unique_ptr* target() { return &target_; } private: std::unique_ptr target_; }; class CompositeRandomRWFileWrapper : public RandomRWFile { public: explicit CompositeRandomRWFileWrapper(std::unique_ptr& target) : target_(std::move(target)) {} bool use_direct_io() const override { return target_->use_direct_io(); } size_t GetRequiredBufferAlignment() const override { return target_->GetRequiredBufferAlignment(); } Status Write(uint64_t offset, const Slice& data) override { IOOptions io_opts; IODebugContext dbg; return target_->Write(offset, data, io_opts, &dbg); } Status Read(uint64_t offset, size_t n, Slice* result, char* scratch) const override { IOOptions io_opts; IODebugContext dbg; return target_->Read(offset, n, io_opts, result, scratch, &dbg); } Status Flush() override { IOOptions io_opts; IODebugContext dbg; return target_->Flush(io_opts, &dbg); } Status Sync() override { IOOptions io_opts; IODebugContext dbg; return target_->Sync(io_opts, &dbg); } Status Fsync() override { IOOptions io_opts; IODebugContext dbg; return target_->Fsync(io_opts, &dbg); } Status Close() override { IOOptions io_opts; IODebugContext dbg; return target_->Close(io_opts, &dbg); } private: std::unique_ptr target_; }; class CompositeDirectoryWrapper : public Directory { public: explicit CompositeDirectoryWrapper(std::unique_ptr& target) : target_(std::move(target)) {} Status Fsync() override { IOOptions io_opts; IODebugContext dbg; return target_->FsyncWithDirOptions(io_opts, &dbg, DirFsyncOptions()); } Status Close() override { IOOptions io_opts; IODebugContext dbg; return target_->Close(io_opts, &dbg); } size_t GetUniqueId(char* id, size_t max_size) const override { return target_->GetUniqueId(id, max_size); } private: std::unique_ptr target_; }; } // namespace Status CompositeEnv::NewSequentialFile(const std::string& f, std::unique_ptr* r, const EnvOptions& options) { IODebugContext dbg; std::unique_ptr file; Status status; status = file_system_->NewSequentialFile(f, FileOptions(options), &file, &dbg); if (status.ok()) { r->reset(new CompositeSequentialFileWrapper(file)); } return status; } Status CompositeEnv::NewRandomAccessFile(const std::string& f, std::unique_ptr* r, const EnvOptions& options) { IODebugContext dbg; std::unique_ptr file; Status status; status = file_system_->NewRandomAccessFile(f, FileOptions(options), &file, &dbg); if (status.ok()) { r->reset(new CompositeRandomAccessFileWrapper(file)); } return status; } Status CompositeEnv::NewWritableFile(const std::string& f, std::unique_ptr* r, const EnvOptions& options) { IODebugContext dbg; std::unique_ptr file; Status status; status = file_system_->NewWritableFile(f, FileOptions(options), &file, &dbg); if (status.ok()) { r->reset(new CompositeWritableFileWrapper(file)); } return status; } Status CompositeEnv::ReopenWritableFile(const std::string& fname, std::unique_ptr* result, const EnvOptions& options) { IODebugContext dbg; Status status; std::unique_ptr file; status = file_system_->ReopenWritableFile(fname, FileOptions(options), &file, &dbg); if (status.ok()) { result->reset(new CompositeWritableFileWrapper(file)); } return status; } Status CompositeEnv::ReuseWritableFile(const std::string& fname, const std::string& old_fname, std::unique_ptr* r, const EnvOptions& options) { IODebugContext dbg; Status status; std::unique_ptr file; status = file_system_->ReuseWritableFile(fname, old_fname, FileOptions(options), &file, &dbg); if (status.ok()) { r->reset(new CompositeWritableFileWrapper(file)); } return status; } Status CompositeEnv::NewRandomRWFile(const std::string& fname, std::unique_ptr* result, const EnvOptions& options) { IODebugContext dbg; std::unique_ptr file; Status status; status = file_system_->NewRandomRWFile(fname, FileOptions(options), &file, &dbg); if (status.ok()) { result->reset(new CompositeRandomRWFileWrapper(file)); } return status; } Status CompositeEnv::NewDirectory(const std::string& name, std::unique_ptr* result) { IOOptions io_opts; IODebugContext dbg; std::unique_ptr dir; Status status; status = file_system_->NewDirectory(name, io_opts, &dir, &dbg); if (status.ok()) { result->reset(new CompositeDirectoryWrapper(dir)); } return status; } namespace { static std::unordered_map env_wrapper_type_info = { {"target", OptionTypeInfo(0, OptionType::kUnknown, OptionVerificationType::kByName, OptionTypeFlags::kDontSerialize) .SetParseFunc([](const ConfigOptions& opts, const std::string& /*name*/, const std::string& value, void* addr) { auto target = static_cast(addr); return Env::CreateFromString(opts, value, &(target->env), &(target->guard)); }) .SetEqualsFunc([](const ConfigOptions& opts, const std::string& /*name*/, const void* addr1, const void* addr2, std::string* mismatch) { const auto target1 = static_cast(addr1); const auto target2 = static_cast(addr2); if (target1->env != nullptr) { return target1->env->AreEquivalent(opts, target2->env, mismatch); } else { return (target2->env == nullptr); } }) .SetPrepareFunc([](const ConfigOptions& opts, const std::string& /*name*/, void* addr) { auto target = static_cast(addr); if (target->guard.get() != nullptr) { target->env = target->guard.get(); } else if (target->env == nullptr) { target->env = Env::Default(); } return target->env->PrepareOptions(opts); }) .SetValidateFunc([](const DBOptions& db_opts, const ColumnFamilyOptions& cf_opts, const std::string& /*name*/, const void* addr) { const auto target = static_cast(addr); if (target->env == nullptr) { return Status::InvalidArgument("Target Env not specified"); } else { return target->env->ValidateOptions(db_opts, cf_opts); } })}, }; static std::unordered_map composite_fs_wrapper_type_info = { {"file_system", OptionTypeInfo::AsCustomSharedPtr( 0, OptionVerificationType::kByName, OptionTypeFlags::kNone)}, }; static std::unordered_map composite_clock_wrapper_type_info = { {"clock", OptionTypeInfo::AsCustomSharedPtr( 0, OptionVerificationType::kByName, OptionTypeFlags::kNone)}, }; } // namespace std::unique_ptr NewCompositeEnv(const std::shared_ptr& fs) { return std::unique_ptr(new CompositeEnvWrapper(Env::Default(), fs)); } CompositeEnvWrapper::CompositeEnvWrapper(Env* env, const std::shared_ptr& fs, const std::shared_ptr& sc) : CompositeEnv(fs, sc), target_(env) { RegisterOptions("", &target_, &env_wrapper_type_info); RegisterOptions("", &file_system_, &composite_fs_wrapper_type_info); RegisterOptions("", &system_clock_, &composite_clock_wrapper_type_info); } CompositeEnvWrapper::CompositeEnvWrapper(const std::shared_ptr& env, const std::shared_ptr& fs, const std::shared_ptr& sc) : CompositeEnv(fs, sc), target_(env) { RegisterOptions("", &target_, &env_wrapper_type_info); RegisterOptions("", &file_system_, &composite_fs_wrapper_type_info); RegisterOptions("", &system_clock_, &composite_clock_wrapper_type_info); } Status CompositeEnvWrapper::PrepareOptions(const ConfigOptions& options) { target_.Prepare(); if (file_system_ == nullptr) { file_system_ = target_.env->GetFileSystem(); } if (system_clock_ == nullptr) { system_clock_ = target_.env->GetSystemClock(); } return Env::PrepareOptions(options); } std::string CompositeEnvWrapper::SerializeOptions( const ConfigOptions& config_options, const std::string& header) const { auto options = CompositeEnv::SerializeOptions(config_options, header); if (target_.env != nullptr && target_.env != Env::Default()) { options.append("target="); options.append(target_.env->ToString(config_options)); } return options; } EnvWrapper::EnvWrapper(Env* t) : target_(t) { RegisterOptions("", &target_, &env_wrapper_type_info); } EnvWrapper::EnvWrapper(std::unique_ptr&& t) : target_(std::move(t)) { RegisterOptions("", &target_, &env_wrapper_type_info); } EnvWrapper::EnvWrapper(const std::shared_ptr& t) : target_(t) { RegisterOptions("", &target_, &env_wrapper_type_info); } EnvWrapper::~EnvWrapper() = default; Status EnvWrapper::PrepareOptions(const ConfigOptions& options) { target_.Prepare(); return Env::PrepareOptions(options); } std::string EnvWrapper::SerializeOptions(const ConfigOptions& config_options, const std::string& header) const { auto parent = Env::SerializeOptions(config_options, ""); if (config_options.IsShallow() || target_.env == nullptr || target_.env == Env::Default()) { return parent; } else { std::string result = header; if (!StartsWith(parent, OptionTypeInfo::kIdPropName())) { result.append(OptionTypeInfo::kIdPropName()).append("="); } result.append(parent); if (!EndsWith(result, config_options.delimiter)) { result.append(config_options.delimiter); } result.append("target=").append(target_.env->ToString(config_options)); return result; } } } // namespace ROCKSDB_NAMESPACE