mirror of
https://github.com/facebook/rocksdb.git
synced 2026-10-23 08:09:02 +03:00
Support remote file system for blob direct write file (#14585)
Summary: Pull Request resolved: https://github.com/facebook/rocksdb/pull/14585 Adds typed file-open contracts and open-file size handling needed for blob direct-write files on remote or SHM-backed file systems. This diff: - Adds typed `FileOpenContract` semantics to RocksDB `FileOptions`, including `kNoReopenForWrite`, `kNoReadersWhileOpenForWrite`, bitwise helpers, and constructor propagation from `EnvOptions` to `FileOptions`. - Marks SST/table output creation paths with both `kNoReopenForWrite` and `kNoReadersWhileOpenForWrite`, including builder output, compaction output, and external `SstFileWriter` output. - Marks blob write paths with the appropriate contracts: regular blob file builders use both `kNoReopenForWrite` and `kNoReadersWhileOpenForWrite`, while blob direct-write partition output uses `kNoReopenForWrite` so active readers can still observe SHM-backed data through the read path. - Adds `GetFileSizeFromOpenFileOrPath` so readers can prefer an already-open `FSRandomAccessFile::GetFileSize` result and fall back to path-level `FileSystem::GetFileSize` according to the caller's fallback policy. - Updates blob file reader and table footer reader size checks to use the new open-file-or-path size helper, preserving malformed-file validation while supporting file systems where path-level size is not the only valid source. - Extends `FaultInjectionTestFS` to track file-open contracts, reject readers while a no-readers writer is open, reject reopened data writes for `kNoReopenForWrite`, allow `SyncFile` to use a reopened handle, and preserve contract state across create, reopen, reuse, rename, link, reset, and delete paths. - Keeps `IOStatus` return paths simple by returning status locals directly in the file-size helper and fault-injection contract validation paths. - Adds clang-format-compliant tests covering file-open contract enforcement, `SyncFile` being allowed while reopened data writes are rejected, blob direct-write behavior, blob reader behavior, and the new file-size fallback helper behavior. - Leaves WAL and MANIFEST contract assignment for follow-up work because live WAL/MANIFEST readers and `reuse_manifest_on_open` need separate handling before RocksDB can safely claim stronger file-open contracts for those file classes. Reviewed By: anand1976, pdillinger Differential Revision: D100002686 fbshipit-source-id: 26b4021bc13333ede1077f0ff8cbd71335c5f294
This commit is contained in:
committed by
meta-codesync[bot]
parent
b5922eab11
commit
2d68f30425
@@ -22,6 +22,7 @@
|
||||
#include "logging/logging.h"
|
||||
#include "options/cf_options.h"
|
||||
#include "options/options_helper.h"
|
||||
#include "rocksdb/file_system.h"
|
||||
#include "rocksdb/slice.h"
|
||||
#include "rocksdb/status.h"
|
||||
#include "test_util/sync_point.h"
|
||||
@@ -200,6 +201,8 @@ Status BlobFileBuilder::OpenBlobFileIfNeeded() {
|
||||
{
|
||||
assert(file_options_);
|
||||
fo_copy = *file_options_;
|
||||
fo_copy.open_contract = FileOpenContract::kNoReopenForWrite |
|
||||
FileOpenContract::kNoReadersWhileOpenForWrite;
|
||||
fo_copy.write_hint = write_hint_;
|
||||
Status s = NewWritableFile(fs_, blob_file_path, &file, fo_copy);
|
||||
|
||||
|
||||
@@ -123,6 +123,7 @@ Status BlobFileCache::OpenBlobFileReaderUncached(
|
||||
Statistics* const statistics = immutable_options_->stats;
|
||||
RecordTick(statistics, NO_FILE_OPENS);
|
||||
|
||||
assert(file_options_);
|
||||
Status s = BlobFileReader::Create(
|
||||
*immutable_options_, read_options, *file_options_, column_family_id_,
|
||||
blob_file_read_hist_, blob_file_number, io_tracer_,
|
||||
|
||||
@@ -24,6 +24,7 @@
|
||||
#include "file/read_write_util.h"
|
||||
#include "file/writable_file_writer.h"
|
||||
#include "logging/logging.h"
|
||||
#include "rocksdb/file_system.h"
|
||||
#include "util/aligned_buffer.h"
|
||||
#include "util/compression.h"
|
||||
#include "util/mutexlock.h"
|
||||
@@ -199,7 +200,9 @@ Status BlobFilePartitionManager::OpenNewBlobFile(Partition* partition,
|
||||
}
|
||||
|
||||
std::unique_ptr<FSWritableFile> file;
|
||||
Status s = NewWritableFile(fs_, blob_file_path, &file, file_options_);
|
||||
FileOptions writer_file_options = file_options_;
|
||||
writer_file_options.open_contract = FileOpenContract::kNoReopenForWrite;
|
||||
Status s = NewWritableFile(fs_, blob_file_path, &file, writer_file_options);
|
||||
if (!s.ok()) {
|
||||
RemoveFilePartitionMapping(blob_file_number);
|
||||
return s;
|
||||
@@ -208,7 +211,7 @@ Status BlobFilePartitionManager::OpenNewBlobFile(Partition* partition,
|
||||
const bool perform_data_verification =
|
||||
checksum_handoff_file_types_.Contains(FileType::kBlobFile);
|
||||
auto file_writer = std::make_unique<WritableFileWriter>(
|
||||
std::move(file), blob_file_path, file_options_, clock_, io_tracer_,
|
||||
std::move(file), blob_file_path, writer_file_options, clock_, io_tracer_,
|
||||
statistics_, Histograms::BLOB_DB_BLOB_FILE_WRITE_MICROS, listeners_,
|
||||
file_checksum_gen_factory_, perform_data_verification);
|
||||
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
#include "db/blob/blob_log_format.h"
|
||||
#include "file/file_prefetch_buffer.h"
|
||||
#include "file/filename.h"
|
||||
#include "file/read_write_util.h"
|
||||
#include "monitoring/statistics_impl.h"
|
||||
#include "options/cf_options.h"
|
||||
#include "rocksdb/file_system.h"
|
||||
@@ -105,24 +106,6 @@ Status BlobFileReader::OpenFile(
|
||||
|
||||
constexpr IODebugContext* dbg = nullptr;
|
||||
|
||||
{
|
||||
TEST_SYNC_POINT("BlobFileReader::OpenFile:GetFileSize");
|
||||
|
||||
const Status s =
|
||||
fs->GetFileSize(blob_file_path, IOOptions(), file_size, dbg);
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
}
|
||||
|
||||
if (!skip_footer_size_check &&
|
||||
*file_size < BlobLogHeader::kSize + BlobLogFooter::kSize) {
|
||||
return Status::Corruption("Malformed blob file");
|
||||
}
|
||||
if (skip_footer_size_check && *file_size < BlobLogHeader::kSize) {
|
||||
return Status::Corruption("Malformed blob file");
|
||||
}
|
||||
|
||||
std::unique_ptr<FSRandomAccessFile> file;
|
||||
FileOptions reader_file_opts = file_opts;
|
||||
|
||||
@@ -142,6 +125,24 @@ Status BlobFileReader::OpenFile(
|
||||
|
||||
assert(file);
|
||||
|
||||
{
|
||||
Status s = GetFileSizeFromOpenFileOrPath(
|
||||
file.get(), fs, blob_file_path, file_size, dbg,
|
||||
FileSizeFallback::kNotSupportedOnly,
|
||||
[]() { TEST_SYNC_POINT("BlobFileReader::OpenFile:GetFileSize"); });
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
}
|
||||
|
||||
if (!skip_footer_size_check &&
|
||||
*file_size < BlobLogHeader::kSize + BlobLogFooter::kSize) {
|
||||
return Status::Corruption("Malformed blob file");
|
||||
}
|
||||
if (skip_footer_size_check && *file_size < BlobLogHeader::kSize) {
|
||||
return Status::Corruption("Malformed blob file");
|
||||
}
|
||||
|
||||
if (immutable_options.advise_random_on_open) {
|
||||
file->Hint(FSRandomAccessFile::kRandom);
|
||||
}
|
||||
|
||||
@@ -7,10 +7,12 @@
|
||||
|
||||
#include <cassert>
|
||||
#include <string>
|
||||
#include <utility>
|
||||
|
||||
#include "db/blob/blob_contents.h"
|
||||
#include "db/blob/blob_log_format.h"
|
||||
#include "db/blob/blob_log_writer.h"
|
||||
#include "env/composite_env_wrapper.h"
|
||||
#include "env/mock_env.h"
|
||||
#include "file/filename.h"
|
||||
#include "file/read_write_util.h"
|
||||
@@ -136,6 +138,167 @@ class BlobFileReaderTest : public testing::Test {
|
||||
std::unique_ptr<Env> mock_env_;
|
||||
};
|
||||
|
||||
namespace {
|
||||
|
||||
// Returns a stale path-level size for one target blob file while leaving the
|
||||
// opened file handle untouched. This lets the test verify
|
||||
// BlobFileReader::Create prefers the opened handle's size when the path stat
|
||||
// lags behind.
|
||||
class StalePathSizeFileSystem : public FileSystemWrapper {
|
||||
public:
|
||||
static const char* kClassName() { return "StalePathSizeFileSystem"; }
|
||||
|
||||
StalePathSizeFileSystem(const std::shared_ptr<FileSystem>& target,
|
||||
std::string stale_path)
|
||||
: FileSystemWrapper(target), stale_path_(std::move(stale_path)) {}
|
||||
|
||||
const char* Name() const override { return kClassName(); }
|
||||
|
||||
IOStatus GetFileSize(const std::string& fname, const IOOptions& options,
|
||||
uint64_t* file_size, IODebugContext* dbg) override {
|
||||
if (fname == stale_path_) {
|
||||
*file_size = 0;
|
||||
return IOStatus::OK();
|
||||
}
|
||||
return FileSystemWrapper::GetFileSize(fname, options, file_size, dbg);
|
||||
}
|
||||
|
||||
private:
|
||||
const std::string stale_path_;
|
||||
};
|
||||
|
||||
// Makes opened blob file handles fail GetFileSize() with a hard error so the
|
||||
// reader path can verify that such errors are propagated directly.
|
||||
class FailingOpenFileSizeRandomAccessFile
|
||||
: public FSRandomAccessFileOwnerWrapper {
|
||||
public:
|
||||
explicit FailingOpenFileSizeRandomAccessFile(
|
||||
std::unique_ptr<FSRandomAccessFile>&& target)
|
||||
: FSRandomAccessFileOwnerWrapper(std::move(target)) {}
|
||||
|
||||
IOStatus GetFileSize(uint64_t* /*result*/) override {
|
||||
return IOStatus::IOError("open file size failed");
|
||||
}
|
||||
};
|
||||
|
||||
// Makes opened blob file handles report GetFileSize() as unsupported so the
|
||||
// reader path exercises its path-level fallback logic.
|
||||
class NotSupportedOpenFileSizeRandomAccessFile
|
||||
: public FSRandomAccessFileOwnerWrapper {
|
||||
public:
|
||||
explicit NotSupportedOpenFileSizeRandomAccessFile(
|
||||
std::unique_ptr<FSRandomAccessFile>&& target)
|
||||
: FSRandomAccessFileOwnerWrapper(std::move(target)) {}
|
||||
|
||||
IOStatus GetFileSize(uint64_t* /*result*/) override {
|
||||
return IOStatus::NotSupported("open file size not supported");
|
||||
}
|
||||
};
|
||||
|
||||
class OpenFileSizeFileSystem : public FileSystemWrapper {
|
||||
public:
|
||||
OpenFileSizeFileSystem(const std::shared_ptr<FileSystem>& target,
|
||||
std::string target_path)
|
||||
: FileSystemWrapper(target), target_path_(std::move(target_path)) {}
|
||||
|
||||
IOStatus NewRandomAccessFile(const std::string& fname,
|
||||
const FileOptions& options,
|
||||
std::unique_ptr<FSRandomAccessFile>* result,
|
||||
IODebugContext* dbg) override {
|
||||
std::unique_ptr<FSRandomAccessFile> file;
|
||||
IOStatus s =
|
||||
FileSystemWrapper::NewRandomAccessFile(fname, options, &file, dbg);
|
||||
if (!s.ok()) {
|
||||
return IOStatus(std::move(s));
|
||||
}
|
||||
if (fname == target_path_) {
|
||||
*result = WrapTargetFile(std::move(file));
|
||||
} else {
|
||||
*result = std::move(file);
|
||||
}
|
||||
return IOStatus::OK();
|
||||
}
|
||||
|
||||
protected:
|
||||
virtual std::unique_ptr<FSRandomAccessFile> WrapTargetFile(
|
||||
std::unique_ptr<FSRandomAccessFile>&& file) = 0;
|
||||
|
||||
private:
|
||||
const std::string target_path_;
|
||||
};
|
||||
|
||||
// Wraps the target blob file in FailingOpenFileSizeRandomAccessFile while
|
||||
// leaving all other files alone.
|
||||
class FailingOpenFileSizeFileSystem : public OpenFileSizeFileSystem {
|
||||
public:
|
||||
static const char* kClassName() { return "FailingOpenFileSizeFileSystem"; }
|
||||
|
||||
FailingOpenFileSizeFileSystem(const std::shared_ptr<FileSystem>& target,
|
||||
std::string target_path)
|
||||
: OpenFileSizeFileSystem(target, std::move(target_path)) {}
|
||||
|
||||
const char* Name() const override { return kClassName(); }
|
||||
|
||||
private:
|
||||
std::unique_ptr<FSRandomAccessFile> WrapTargetFile(
|
||||
std::unique_ptr<FSRandomAccessFile>&& file) override {
|
||||
return std::unique_ptr<FSRandomAccessFile>(
|
||||
new FailingOpenFileSizeRandomAccessFile(std::move(file)));
|
||||
}
|
||||
};
|
||||
|
||||
// Wraps the target blob file in NotSupportedOpenFileSizeRandomAccessFile while
|
||||
// leaving all other files alone.
|
||||
class NotSupportedOpenFileSizeFileSystem : public OpenFileSizeFileSystem {
|
||||
public:
|
||||
static const char* kClassName() {
|
||||
return "NotSupportedOpenFileSizeFileSystem";
|
||||
}
|
||||
|
||||
NotSupportedOpenFileSizeFileSystem(const std::shared_ptr<FileSystem>& target,
|
||||
std::string target_path)
|
||||
: OpenFileSizeFileSystem(target, std::move(target_path)) {}
|
||||
|
||||
const char* Name() const override { return kClassName(); }
|
||||
|
||||
private:
|
||||
std::unique_ptr<FSRandomAccessFile> WrapTargetFile(
|
||||
std::unique_ptr<FSRandomAccessFile>&& file) override {
|
||||
return std::unique_ptr<FSRandomAccessFile>(
|
||||
new NotSupportedOpenFileSizeRandomAccessFile(std::move(file)));
|
||||
}
|
||||
};
|
||||
|
||||
// Forces the target blob file onto the NotSupported open-handle-size branch and
|
||||
// then fails the fallback path-level GetFileSize() call.
|
||||
class FailingFallbackPathSizeFileSystem
|
||||
: public NotSupportedOpenFileSizeFileSystem {
|
||||
public:
|
||||
static const char* kClassName() {
|
||||
return "FailingFallbackPathSizeFileSystem";
|
||||
}
|
||||
|
||||
FailingFallbackPathSizeFileSystem(const std::shared_ptr<FileSystem>& target,
|
||||
std::string target_path)
|
||||
: NotSupportedOpenFileSizeFileSystem(target, target_path),
|
||||
target_path_(std::move(target_path)) {}
|
||||
|
||||
const char* Name() const override { return kClassName(); }
|
||||
|
||||
IOStatus GetFileSize(const std::string& fname, const IOOptions& options,
|
||||
uint64_t* file_size, IODebugContext* dbg) override {
|
||||
if (fname == target_path_) {
|
||||
return IOStatus::IOError("fallback path size failed");
|
||||
}
|
||||
return FileSystemWrapper::GetFileSize(fname, options, file_size, dbg);
|
||||
}
|
||||
|
||||
private:
|
||||
const std::string target_path_;
|
||||
};
|
||||
|
||||
} // namespace
|
||||
|
||||
TEST_F(BlobFileReaderTest, CreateReaderAndGetBlob) {
|
||||
Options options;
|
||||
options.env = mock_env_.get();
|
||||
@@ -428,6 +591,211 @@ TEST_F(BlobFileReaderTest, CreateReaderAndGetBlob) {
|
||||
}
|
||||
}
|
||||
|
||||
// Verifies Create() prefers the file size reported by the opened handle over a
|
||||
// stale path-level size query by returning a readable reader even when the path
|
||||
// stat intentionally reports 0 for the blob file.
|
||||
TEST_F(BlobFileReaderTest, CreateReaderUsesOpenedFileSizeWhenPathSizeIsStale) {
|
||||
const std::string db_path = test::PerThreadDBPath(
|
||||
mock_env_.get(),
|
||||
"BlobFileReaderTest_CreateReaderUsesOpenedFileSizeWhenPathSizeIsStale");
|
||||
|
||||
constexpr uint64_t blob_file_number = 1;
|
||||
const std::string blob_file_path = BlobFileName(db_path, blob_file_number);
|
||||
|
||||
auto stale_size_fs = std::make_shared<StalePathSizeFileSystem>(
|
||||
mock_env_->GetFileSystem(), blob_file_path);
|
||||
CompositeEnvWrapper stale_size_env(mock_env_.get(), stale_size_fs);
|
||||
|
||||
Options options;
|
||||
options.env = &stale_size_env;
|
||||
options.cf_paths.emplace_back(db_path, 0);
|
||||
options.enable_blob_files = true;
|
||||
|
||||
ImmutableOptions immutable_options(options);
|
||||
|
||||
constexpr uint32_t column_family_id = 1;
|
||||
constexpr bool has_ttl = false;
|
||||
constexpr ExpirationRange expiration_range;
|
||||
constexpr char key[] = "key";
|
||||
constexpr char blob[] = "blob";
|
||||
|
||||
uint64_t blob_offset = 0;
|
||||
uint64_t blob_size = 0;
|
||||
WriteBlobFile(immutable_options, column_family_id, has_ttl, expiration_range,
|
||||
expiration_range, blob_file_number, key, blob, kNoCompression,
|
||||
&blob_offset, &blob_size);
|
||||
|
||||
std::unique_ptr<BlobFileReader> reader;
|
||||
ReadOptions read_options;
|
||||
read_options.verify_checksums = false;
|
||||
ASSERT_OK(BlobFileReader::Create(
|
||||
immutable_options, read_options, FileOptions(), column_family_id,
|
||||
/*blob_file_read_hist=*/nullptr, blob_file_number, nullptr /*IOTracer*/,
|
||||
&reader));
|
||||
ASSERT_NE(reader, nullptr);
|
||||
|
||||
std::unique_ptr<BlobContents> value;
|
||||
uint64_t bytes_read = 0;
|
||||
ASSERT_OK(reader->GetBlob(read_options, key, blob_offset, blob_size,
|
||||
kNoCompression,
|
||||
/*prefetch_buffer=*/nullptr,
|
||||
/*allocator=*/nullptr, &value, &bytes_read));
|
||||
ASSERT_NE(value, nullptr);
|
||||
ASSERT_EQ(value->data(), Slice(blob));
|
||||
ASSERT_EQ(bytes_read, blob_size);
|
||||
}
|
||||
|
||||
// Verifies Create() falls back to the path-level GetFileSize() call when the
|
||||
// opened file handle reports that size queries are unsupported.
|
||||
TEST_F(BlobFileReaderTest,
|
||||
CreateReaderFallsBackToPathSizeWhenOpenedFileSizeNotSupported) {
|
||||
const std::string db_path = test::PerThreadDBPath(
|
||||
mock_env_.get(),
|
||||
"BlobFileReaderTest_"
|
||||
"CreateReaderFallsBackToPathSizeWhenOpenedFileSizeNotSupported");
|
||||
|
||||
constexpr uint64_t blob_file_number = 1;
|
||||
const std::string blob_file_path = BlobFileName(db_path, blob_file_number);
|
||||
|
||||
auto not_supported_open_file_size_fs =
|
||||
std::make_shared<NotSupportedOpenFileSizeFileSystem>(
|
||||
mock_env_->GetFileSystem(), blob_file_path);
|
||||
CompositeEnvWrapper not_supported_open_file_size_env(
|
||||
mock_env_.get(), not_supported_open_file_size_fs);
|
||||
|
||||
Options options;
|
||||
options.env = ¬_supported_open_file_size_env;
|
||||
options.cf_paths.emplace_back(db_path, 0);
|
||||
options.enable_blob_files = true;
|
||||
|
||||
ImmutableOptions immutable_options(options);
|
||||
|
||||
constexpr uint32_t column_family_id = 1;
|
||||
constexpr bool has_ttl = false;
|
||||
constexpr ExpirationRange expiration_range;
|
||||
constexpr char key[] = "key";
|
||||
constexpr char blob[] = "blob";
|
||||
|
||||
uint64_t blob_offset = 0;
|
||||
uint64_t blob_size = 0;
|
||||
WriteBlobFile(immutable_options, column_family_id, has_ttl, expiration_range,
|
||||
expiration_range, blob_file_number, key, blob, kNoCompression,
|
||||
&blob_offset, &blob_size);
|
||||
|
||||
std::unique_ptr<BlobFileReader> reader;
|
||||
ReadOptions read_options;
|
||||
read_options.verify_checksums = false;
|
||||
ASSERT_OK(BlobFileReader::Create(
|
||||
immutable_options, read_options, FileOptions(), column_family_id,
|
||||
/*blob_file_read_hist=*/nullptr, blob_file_number, nullptr /*IOTracer*/,
|
||||
&reader));
|
||||
ASSERT_NE(reader, nullptr);
|
||||
|
||||
std::unique_ptr<BlobContents> value;
|
||||
uint64_t bytes_read = 0;
|
||||
ASSERT_OK(reader->GetBlob(read_options, key, blob_offset, blob_size,
|
||||
kNoCompression,
|
||||
/*prefetch_buffer=*/nullptr,
|
||||
/*allocator=*/nullptr, &value, &bytes_read));
|
||||
ASSERT_NE(value, nullptr);
|
||||
ASSERT_EQ(value->data(), Slice(blob));
|
||||
ASSERT_EQ(bytes_read, blob_size);
|
||||
}
|
||||
|
||||
// Verifies Create() propagates a real error from the opened file handle's
|
||||
// GetFileSize() rather than masking it with any path-level fallback.
|
||||
TEST_F(BlobFileReaderTest, CreateReaderPropagatesOpenedFileSizeError) {
|
||||
const std::string db_path = test::PerThreadDBPath(
|
||||
mock_env_.get(),
|
||||
"BlobFileReaderTest_CreateReaderPropagatesOpenedFileSizeError");
|
||||
|
||||
constexpr uint64_t blob_file_number = 1;
|
||||
const std::string blob_file_path = BlobFileName(db_path, blob_file_number);
|
||||
|
||||
auto failing_open_file_size_fs =
|
||||
std::make_shared<FailingOpenFileSizeFileSystem>(
|
||||
mock_env_->GetFileSystem(), blob_file_path);
|
||||
CompositeEnvWrapper failing_open_file_size_env(mock_env_.get(),
|
||||
failing_open_file_size_fs);
|
||||
|
||||
Options options;
|
||||
options.env = &failing_open_file_size_env;
|
||||
options.cf_paths.emplace_back(db_path, 0);
|
||||
options.enable_blob_files = true;
|
||||
|
||||
ImmutableOptions immutable_options(options);
|
||||
|
||||
constexpr uint32_t column_family_id = 1;
|
||||
constexpr bool has_ttl = false;
|
||||
constexpr ExpirationRange expiration_range;
|
||||
constexpr char key[] = "key";
|
||||
constexpr char blob[] = "blob";
|
||||
|
||||
uint64_t blob_offset = 0;
|
||||
uint64_t blob_size = 0;
|
||||
WriteBlobFile(immutable_options, column_family_id, has_ttl, expiration_range,
|
||||
expiration_range, blob_file_number, key, blob, kNoCompression,
|
||||
&blob_offset, &blob_size);
|
||||
|
||||
std::unique_ptr<BlobFileReader> reader;
|
||||
ReadOptions read_options;
|
||||
Status s = BlobFileReader::Create(
|
||||
immutable_options, read_options, FileOptions(), column_family_id,
|
||||
/*blob_file_read_hist=*/nullptr, blob_file_number, nullptr /*IOTracer*/,
|
||||
&reader);
|
||||
ASSERT_TRUE(s.IsIOError());
|
||||
ASSERT_EQ(reader, nullptr);
|
||||
}
|
||||
|
||||
// Verifies Create() also propagates errors from the path-level fallback size
|
||||
// query when the opened file handle only reports NotSupported for GetFileSize.
|
||||
TEST_F(
|
||||
BlobFileReaderTest,
|
||||
CreateReaderPropagatesFallbackPathSizeErrorWhenOpenedFileSizeNotSupported) {
|
||||
const std::string db_path =
|
||||
test::PerThreadDBPath(mock_env_.get(),
|
||||
"BlobFileReaderTest_"
|
||||
"CreateReaderPropagatesFallbackPathSizeErrorWhenOpe"
|
||||
"nedFileSizeNotSupported");
|
||||
|
||||
constexpr uint64_t blob_file_number = 1;
|
||||
const std::string blob_file_path = BlobFileName(db_path, blob_file_number);
|
||||
|
||||
auto failing_fallback_path_size_fs =
|
||||
std::make_shared<FailingFallbackPathSizeFileSystem>(
|
||||
mock_env_->GetFileSystem(), blob_file_path);
|
||||
CompositeEnvWrapper failing_fallback_path_size_env(
|
||||
mock_env_.get(), failing_fallback_path_size_fs);
|
||||
|
||||
Options options;
|
||||
options.env = &failing_fallback_path_size_env;
|
||||
options.cf_paths.emplace_back(db_path, 0);
|
||||
options.enable_blob_files = true;
|
||||
|
||||
ImmutableOptions immutable_options(options);
|
||||
|
||||
constexpr uint32_t column_family_id = 1;
|
||||
constexpr bool has_ttl = false;
|
||||
constexpr ExpirationRange expiration_range;
|
||||
constexpr char key[] = "key";
|
||||
constexpr char blob[] = "blob";
|
||||
|
||||
uint64_t blob_offset = 0;
|
||||
uint64_t blob_size = 0;
|
||||
WriteBlobFile(immutable_options, column_family_id, has_ttl, expiration_range,
|
||||
expiration_range, blob_file_number, key, blob, kNoCompression,
|
||||
&blob_offset, &blob_size);
|
||||
|
||||
std::unique_ptr<BlobFileReader> reader;
|
||||
ReadOptions read_options;
|
||||
Status s = BlobFileReader::Create(
|
||||
immutable_options, read_options, FileOptions(), column_family_id,
|
||||
/*blob_file_read_hist=*/nullptr, blob_file_number, nullptr /*IOTracer*/,
|
||||
&reader);
|
||||
ASSERT_TRUE(s.IsIOError());
|
||||
ASSERT_EQ(reader, nullptr);
|
||||
}
|
||||
|
||||
TEST_F(BlobFileReaderTest, Malformed) {
|
||||
// Write a blob file consisting of nothing but a header, and make sure we
|
||||
// detect the error when we open it for reading
|
||||
@@ -849,7 +1217,6 @@ class BlobFileReaderIOErrorTest
|
||||
|
||||
INSTANTIATE_TEST_CASE_P(BlobFileReaderTest, BlobFileReaderIOErrorTest,
|
||||
::testing::ValuesIn(std::vector<std::string>{
|
||||
"BlobFileReader::OpenFile:GetFileSize",
|
||||
"BlobFileReader::OpenFile:NewRandomAccessFile",
|
||||
"BlobFileReader::ReadHeader:ReadFromFile",
|
||||
"BlobFileReader::ReadFooter:ReadFromFile",
|
||||
|
||||
@@ -10,11 +10,14 @@
|
||||
|
||||
#include <algorithm>
|
||||
#include <array>
|
||||
#include <functional>
|
||||
#include <limits>
|
||||
#include <mutex>
|
||||
#include <set>
|
||||
#include <sstream>
|
||||
#include <string>
|
||||
#include <unordered_set>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
#include "cache/compressed_secondary_cache.h"
|
||||
@@ -26,10 +29,12 @@
|
||||
#include "db/db_test_util.h"
|
||||
#include "db/db_with_timestamp_test_util.h"
|
||||
#include "db/wide/wide_column_test_util.h"
|
||||
#include "env/composite_env_wrapper.h"
|
||||
#include "file/filename.h"
|
||||
#include "file/random_access_file_reader.h"
|
||||
#include "port/stack_trace.h"
|
||||
#include "rocksdb/convenience.h"
|
||||
#include "rocksdb/file_system.h"
|
||||
#include "rocksdb/trace_reader_writer.h"
|
||||
#include "rocksdb/trace_record.h"
|
||||
#include "rocksdb/utilities/replayer.h"
|
||||
@@ -101,6 +106,206 @@ class RecordingBlobDirectWritePartitionStrategy
|
||||
std::vector<Call> calls_;
|
||||
};
|
||||
|
||||
namespace {
|
||||
|
||||
// Matches BlobDB file names so the test filesystem can special-case only active
|
||||
// blob files without changing unrelated file opens.
|
||||
bool IsBlobFilePath(const std::string& fname) {
|
||||
const size_t basename_pos = fname.find_last_of("/\\");
|
||||
const std::string basename = basename_pos == std::string::npos
|
||||
? fname
|
||||
: fname.substr(basename_pos + 1);
|
||||
uint64_t file_number = 0;
|
||||
FileType file_type = kWalFile;
|
||||
return ParseFileName(basename, &file_number, &file_type) &&
|
||||
file_type == kBlobFile;
|
||||
}
|
||||
|
||||
// Keeps track of when an active blob file stops being writer-visible so the
|
||||
// test filesystem can model current-size visibility only while the file remains
|
||||
// active.
|
||||
class ActiveBlobVisibilityWritableFile : public FSWritableFileOwnerWrapper {
|
||||
public:
|
||||
ActiveBlobVisibilityWritableFile(std::unique_ptr<FSWritableFile>&& target,
|
||||
std::function<void()> on_close)
|
||||
: FSWritableFileOwnerWrapper(std::move(target)),
|
||||
on_close_(std::move(on_close)) {}
|
||||
|
||||
~ActiveBlobVisibilityWritableFile() override { DeactivateOnce(); }
|
||||
|
||||
IOStatus Close(const IOOptions& options, IODebugContext* dbg) override {
|
||||
IOStatus s = target()->Close(options, dbg);
|
||||
DeactivateOnce();
|
||||
return s;
|
||||
}
|
||||
|
||||
private:
|
||||
void DeactivateOnce() {
|
||||
if (on_close_) {
|
||||
on_close_();
|
||||
on_close_ = nullptr;
|
||||
}
|
||||
}
|
||||
|
||||
std::function<void()> on_close_;
|
||||
};
|
||||
|
||||
// Simulates a remote readable file whose GetFileSize() either exposes the live
|
||||
// on-disk size for default-contract active blobs or hides it for
|
||||
// append-only-no-readers files.
|
||||
class ActiveBlobVisibilityRandomAccessFile
|
||||
: public FSRandomAccessFileOwnerWrapper {
|
||||
public:
|
||||
ActiveBlobVisibilityRandomAccessFile(
|
||||
std::unique_ptr<FSRandomAccessFile>&& target, bool expose_current_size,
|
||||
std::shared_ptr<FileSystem> underlying_fs, std::string fname)
|
||||
: FSRandomAccessFileOwnerWrapper(std::move(target)),
|
||||
expose_current_size_(expose_current_size),
|
||||
underlying_fs_(std::move(underlying_fs)),
|
||||
fname_(std::move(fname)) {}
|
||||
|
||||
IOStatus GetFileSize(uint64_t* result) override {
|
||||
if (!expose_current_size_) {
|
||||
*result = 0;
|
||||
return IOStatus::OK();
|
||||
}
|
||||
// Delegate to the underlying FS path-level GetFileSize (bypassing the
|
||||
// wrapper that reports 0 for active blobs) to simulate a remote FS that
|
||||
// exposes the current file size through the open handle.
|
||||
return underlying_fs_->GetFileSize(fname_, IOOptions(), result,
|
||||
/*dbg=*/nullptr);
|
||||
}
|
||||
|
||||
private:
|
||||
const bool expose_current_size_;
|
||||
std::shared_ptr<FileSystem> underlying_fs_;
|
||||
std::string fname_;
|
||||
};
|
||||
|
||||
// Models a remote filesystem where active direct-write blobs are only
|
||||
// reader-visible when the writer does not promise no concurrent readers.
|
||||
class RemoteBlobVisibilityFileSystem : public FileSystemWrapper {
|
||||
public:
|
||||
explicit RemoteBlobVisibilityFileSystem(const std::shared_ptr<FileSystem>& fs)
|
||||
: FileSystemWrapper(fs) {}
|
||||
|
||||
static const char* kClassName() { return "RemoteBlobVisibilityFileSystem"; }
|
||||
const char* Name() const override { return kClassName(); }
|
||||
|
||||
IOStatus NewWritableFile(const std::string& fname, const FileOptions& opts,
|
||||
std::unique_ptr<FSWritableFile>* result,
|
||||
IODebugContext* dbg) override {
|
||||
std::unique_ptr<FSWritableFile> file;
|
||||
IOStatus s = target()->NewWritableFile(fname, opts, &file, dbg);
|
||||
if (!s.ok() || !IsBlobFilePath(fname)) {
|
||||
*result = std::move(file);
|
||||
return IOStatus(std::move(s));
|
||||
}
|
||||
|
||||
const bool has_no_reopen_for_write_contract = HasFileOpenContract(
|
||||
opts.open_contract, FileOpenContract::kNoReopenForWrite);
|
||||
const bool has_no_readers_while_open_for_write_contract =
|
||||
HasFileOpenContract(opts.open_contract,
|
||||
FileOpenContract::kNoReadersWhileOpenForWrite);
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mu_);
|
||||
saw_no_reopen_for_write_writer_contract_ |=
|
||||
has_no_reopen_for_write_contract;
|
||||
saw_no_readers_while_open_for_write_writer_contract_ |=
|
||||
has_no_readers_while_open_for_write_contract;
|
||||
if (!has_no_readers_while_open_for_write_contract) {
|
||||
active_blob_paths_.insert(fname);
|
||||
}
|
||||
}
|
||||
|
||||
if (has_no_readers_while_open_for_write_contract) {
|
||||
*result = std::move(file);
|
||||
return IOStatus::OK();
|
||||
}
|
||||
|
||||
result->reset(new ActiveBlobVisibilityWritableFile(
|
||||
std::move(file), [this, tracked_path = std::string(fname)]() {
|
||||
DeactivateBlobPath(tracked_path);
|
||||
}));
|
||||
return IOStatus::OK();
|
||||
}
|
||||
|
||||
IOStatus NewRandomAccessFile(const std::string& fname,
|
||||
const FileOptions& opts,
|
||||
std::unique_ptr<FSRandomAccessFile>* result,
|
||||
IODebugContext* dbg) override {
|
||||
std::unique_ptr<FSRandomAccessFile> file;
|
||||
IOStatus s = target()->NewRandomAccessFile(fname, opts, &file, dbg);
|
||||
if (!s.ok() || !IsBlobFilePath(fname)) {
|
||||
*result = std::move(file);
|
||||
return IOStatus(std::move(s));
|
||||
}
|
||||
|
||||
const bool has_no_readers_while_open_for_write_contract =
|
||||
HasFileOpenContract(opts.open_contract,
|
||||
FileOpenContract::kNoReadersWhileOpenForWrite);
|
||||
const bool active_blob = IsActiveBlobPath(fname);
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mu_);
|
||||
saw_no_readers_while_open_for_write_reader_contract_ |=
|
||||
has_no_readers_while_open_for_write_contract;
|
||||
}
|
||||
|
||||
if (!active_blob) {
|
||||
*result = std::move(file);
|
||||
return IOStatus::OK();
|
||||
}
|
||||
|
||||
result->reset(new ActiveBlobVisibilityRandomAccessFile(
|
||||
std::move(file), !has_no_readers_while_open_for_write_contract, target_,
|
||||
fname));
|
||||
return IOStatus::OK();
|
||||
}
|
||||
|
||||
IOStatus GetFileSize(const std::string& fname, const IOOptions& options,
|
||||
uint64_t* file_size, IODebugContext* dbg) override {
|
||||
if (IsActiveBlobPath(fname)) {
|
||||
*file_size = 0;
|
||||
return IOStatus::OK();
|
||||
}
|
||||
return target()->GetFileSize(fname, options, file_size, dbg);
|
||||
}
|
||||
|
||||
bool SawNoReopenForWriteWriterContract() const {
|
||||
std::lock_guard<std::mutex> lock(mu_);
|
||||
return saw_no_reopen_for_write_writer_contract_;
|
||||
}
|
||||
|
||||
bool SawNoReadersWhileOpenForWriteWriterContract() const {
|
||||
std::lock_guard<std::mutex> lock(mu_);
|
||||
return saw_no_readers_while_open_for_write_writer_contract_;
|
||||
}
|
||||
|
||||
bool SawNoReadersWhileOpenForWriteReaderContract() const {
|
||||
std::lock_guard<std::mutex> lock(mu_);
|
||||
return saw_no_readers_while_open_for_write_reader_contract_;
|
||||
}
|
||||
|
||||
private:
|
||||
void DeactivateBlobPath(const std::string& fname) {
|
||||
std::lock_guard<std::mutex> lock(mu_);
|
||||
active_blob_paths_.erase(fname);
|
||||
}
|
||||
|
||||
bool IsActiveBlobPath(const std::string& fname) const {
|
||||
std::lock_guard<std::mutex> lock(mu_);
|
||||
return active_blob_paths_.find(fname) != active_blob_paths_.end();
|
||||
}
|
||||
|
||||
mutable std::mutex mu_;
|
||||
bool saw_no_reopen_for_write_writer_contract_ = false;
|
||||
bool saw_no_readers_while_open_for_write_writer_contract_ = false;
|
||||
bool saw_no_readers_while_open_for_write_reader_contract_ = false;
|
||||
std::unordered_set<std::string> active_blob_paths_;
|
||||
};
|
||||
|
||||
} // namespace
|
||||
|
||||
class DBBlobDirectWriteTest : public DBTestBase {
|
||||
protected:
|
||||
DBBlobDirectWriteTest()
|
||||
@@ -816,6 +1021,30 @@ TEST_F(DBBlobDirectWriteTest, DirectWriteImmutableMemtableRead) {
|
||||
verify_reads();
|
||||
}
|
||||
|
||||
// Verifies active blob direct-write files promise no write reopen without also
|
||||
// promising no concurrent readers.
|
||||
TEST_F(DBBlobDirectWriteTest, DirectWriteUsesReadableContractForRemoteFile) {
|
||||
auto remote_fs =
|
||||
std::make_shared<RemoteBlobVisibilityFileSystem>(env_->GetFileSystem());
|
||||
std::unique_ptr<Env> remote_env(new CompositeEnvWrapper(env_, remote_fs));
|
||||
|
||||
Options options = GetDirectWriteOptions();
|
||||
options.env = remote_env.get();
|
||||
Reopen(options);
|
||||
|
||||
const std::string key = "remote_blob_key";
|
||||
const std::string value(128, 'w');
|
||||
|
||||
ASSERT_OK(Put(key, value));
|
||||
ASSERT_TRUE(remote_fs->SawNoReopenForWriteWriterContract());
|
||||
ASSERT_FALSE(remote_fs->SawNoReadersWhileOpenForWriteWriterContract());
|
||||
|
||||
ASSERT_EQ(Get(key), value);
|
||||
ASSERT_FALSE(remote_fs->SawNoReadersWhileOpenForWriteReaderContract());
|
||||
|
||||
Close();
|
||||
}
|
||||
|
||||
TEST_F(DBBlobDirectWriteTest, DirectWriteRefreshesReaderAfterFlush) {
|
||||
Options options = GetDirectWriteOptions();
|
||||
options.blob_direct_write_partitions = 1;
|
||||
|
||||
@@ -179,6 +179,8 @@ Status BuildTable(
|
||||
TEST_SYNC_POINT_CALLBACK("BuildTable:create_file", &use_direct_writes);
|
||||
#endif // !NDEBUG
|
||||
FileOptions fo_copy = file_options;
|
||||
fo_copy.open_contract = FileOpenContract::kNoReopenForWrite |
|
||||
FileOpenContract::kNoReadersWhileOpenForWrite;
|
||||
fo_copy.write_hint = write_hint;
|
||||
IOStatus io_s = NewWritableFile(fs, fname, &file, fo_copy);
|
||||
assert(s.ok());
|
||||
|
||||
@@ -2470,6 +2470,8 @@ Status CompactionJob::OpenCompactionOutputFile(SubcompactionState* sub_compact,
|
||||
auto temperature =
|
||||
sub_compact->compaction->GetOutputTemperature(outputs.IsProximalLevel());
|
||||
fo_copy.temperature = temperature;
|
||||
fo_copy.open_contract = FileOpenContract::kNoReopenForWrite |
|
||||
FileOpenContract::kNoReadersWhileOpenForWrite;
|
||||
fo_copy.write_hint = write_hint_;
|
||||
|
||||
Status s;
|
||||
|
||||
@@ -646,6 +646,54 @@ TEST(FaultInjectionFSTest, ReadUnsyncedData) {
|
||||
ASSERT_EQ(0, sl.compare(Slice(data.data() + first_read_len, sl.size())));
|
||||
}
|
||||
}
|
||||
|
||||
TEST(FaultInjectionFSTest, FileOpenContractRejectsReadersWhileOpenForWrite) {
|
||||
std::shared_ptr<FaultInjectionTestFS> fault_fs =
|
||||
std::make_shared<FaultInjectionTestFS>(FileSystem::Default());
|
||||
const std::string fname = test::PerThreadDBPath(
|
||||
"file_open_contract_no_readers_while_open_for_write");
|
||||
|
||||
FileOptions writer_options;
|
||||
writer_options.open_contract = FileOpenContract::kNoReadersWhileOpenForWrite;
|
||||
|
||||
std::unique_ptr<FSWritableFile> writer;
|
||||
ASSERT_OK(fault_fs->NewWritableFile(fname, writer_options, &writer, nullptr));
|
||||
|
||||
std::unique_ptr<FSRandomAccessFile> reader;
|
||||
IOStatus s =
|
||||
fault_fs->NewRandomAccessFile(fname, FileOptions(), &reader, nullptr);
|
||||
ASSERT_TRUE(s.IsNotSupported());
|
||||
|
||||
ASSERT_OK(writer->Close(IOOptions(), nullptr));
|
||||
ASSERT_OK(
|
||||
fault_fs->NewRandomAccessFile(fname, FileOptions(), &reader, nullptr));
|
||||
}
|
||||
|
||||
TEST(FaultInjectionFSTest,
|
||||
FileOpenContractAllowsSyncFileButRejectsReopenedWrite) {
|
||||
std::shared_ptr<FaultInjectionTestFS> fault_fs =
|
||||
std::make_shared<FaultInjectionTestFS>(FileSystem::Default());
|
||||
const std::string fname =
|
||||
test::PerThreadDBPath("file_open_contract_no_reopen_for_write");
|
||||
|
||||
FileOptions writer_options;
|
||||
writer_options.open_contract = FileOpenContract::kNoReopenForWrite;
|
||||
|
||||
std::unique_ptr<FSWritableFile> writer;
|
||||
ASSERT_OK(fault_fs->NewWritableFile(fname, writer_options, &writer, nullptr));
|
||||
ASSERT_OK(writer->Append("abc", IOOptions(), nullptr));
|
||||
ASSERT_OK(writer->Close(IOOptions(), nullptr));
|
||||
|
||||
ASSERT_OK(fault_fs->SyncFile(fname, FileOptions(), IOOptions(),
|
||||
/*use_fsync=*/false, nullptr));
|
||||
|
||||
std::unique_ptr<FSWritableFile> reopened_writer;
|
||||
ASSERT_OK(fault_fs->ReopenWritableFile(fname, FileOptions(), &reopened_writer,
|
||||
nullptr));
|
||||
IOStatus s = reopened_writer->Append("d", IOOptions(), nullptr);
|
||||
ASSERT_TRUE(s.IsNotSupported());
|
||||
ASSERT_OK(reopened_writer->Close(IOOptions(), nullptr));
|
||||
}
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
|
||||
@@ -9,6 +9,7 @@
|
||||
|
||||
#include "file/read_write_util.h"
|
||||
|
||||
#include <cassert>
|
||||
#include <sstream>
|
||||
|
||||
#include "test_util/sync_point.h"
|
||||
@@ -25,6 +26,29 @@ IOStatus NewWritableFile(FileSystem* fs, const std::string& fname,
|
||||
return s;
|
||||
}
|
||||
|
||||
IOStatus GetFileSizeFromOpenFileOrPath(
|
||||
FSRandomAccessFile* file, FileSystem* fs, const std::string& fname,
|
||||
uint64_t* file_size, IODebugContext* dbg, FileSizeFallback fallback,
|
||||
BeforePathFileSizeFallback before_path_fallback) {
|
||||
assert(file != nullptr);
|
||||
assert(fs != nullptr);
|
||||
assert(file_size != nullptr);
|
||||
|
||||
IOStatus s = file->GetFileSize(file_size);
|
||||
if (s.ok()) {
|
||||
return s;
|
||||
}
|
||||
|
||||
if (fallback == FileSizeFallback::kAnyOpenFileError || s.IsNotSupported()) {
|
||||
if (before_path_fallback != nullptr) {
|
||||
before_path_fallback();
|
||||
}
|
||||
return fs->GetFileSize(fname, IOOptions(), file_size, dbg);
|
||||
}
|
||||
|
||||
return s;
|
||||
}
|
||||
|
||||
#ifndef NDEBUG
|
||||
bool IsFileSectorAligned(const size_t off, size_t sector_size) {
|
||||
return off % sector_size == 0;
|
||||
|
||||
@@ -25,6 +25,18 @@ IOStatus NewWritableFile(FileSystem* fs, const std::string& fname,
|
||||
std::unique_ptr<FSWritableFile>* result,
|
||||
const FileOptions& options);
|
||||
|
||||
enum class FileSizeFallback {
|
||||
kNotSupportedOnly,
|
||||
kAnyOpenFileError,
|
||||
};
|
||||
|
||||
using BeforePathFileSizeFallback = void (*)();
|
||||
|
||||
IOStatus GetFileSizeFromOpenFileOrPath(
|
||||
FSRandomAccessFile* file, FileSystem* fs, const std::string& fname,
|
||||
uint64_t* file_size, IODebugContext* dbg, FileSizeFallback fallback,
|
||||
BeforePathFileSizeFallback before_path_fallback = nullptr);
|
||||
|
||||
#ifndef NDEBUG
|
||||
bool IsFileSectorAligned(const size_t off, size_t sector_size);
|
||||
#endif // NDEBUG
|
||||
|
||||
@@ -119,7 +119,8 @@ struct IOOptions {
|
||||
// custom contract between a FileSystem user and the provider. This is only
|
||||
// useful in cases where a RocksDB user directly uses the FileSystem or file
|
||||
// object for their own purposes, and wants to pass extra options to APIs
|
||||
// such as NewRandomAccessFile and NewWritableFile.
|
||||
// such as NewRandomAccessFile and NewWritableFile. Prefer typed fields on
|
||||
// IOOptions/FileOptions when a standardized semantic exists.
|
||||
std::unordered_map<std::string, std::string> property_bag;
|
||||
|
||||
// Force directory fsync, some file systems like btrfs may skip directory
|
||||
@@ -176,9 +177,37 @@ struct DirFsyncOptions {
|
||||
explicit DirFsyncOptions(FsyncReason fsync_reason);
|
||||
};
|
||||
|
||||
// File scope options that control how a file is opened/created and accessed
|
||||
// while its open. We may add more options here in the future such as
|
||||
// redundancy level, media to use etc.
|
||||
// File-scope contracts that describe how RocksDB will open/create and access a
|
||||
// file. A FileSystem that detects a contract violation must return a non-OK
|
||||
// status or preserve its normal valid semantics; it must not expose undefined
|
||||
// or corrupt data.
|
||||
enum class FileOpenContract : uint8_t {
|
||||
kDefault = 0,
|
||||
// RocksDB will not call ReopenWritableFile for this file to overwrite,
|
||||
// append, or truncate data. SyncFile is a separate durability API.
|
||||
kNoReopenForWrite = 1 << 0,
|
||||
// RocksDB will not open readers while this file is open for write.
|
||||
kNoReadersWhileOpenForWrite = 1 << 1,
|
||||
};
|
||||
|
||||
constexpr FileOpenContract operator|(FileOpenContract lhs,
|
||||
FileOpenContract rhs) {
|
||||
return static_cast<FileOpenContract>(static_cast<uint8_t>(lhs) |
|
||||
static_cast<uint8_t>(rhs));
|
||||
}
|
||||
|
||||
inline FileOpenContract& operator|=(FileOpenContract& lhs,
|
||||
FileOpenContract rhs) {
|
||||
lhs = lhs | rhs;
|
||||
return lhs;
|
||||
}
|
||||
|
||||
constexpr bool HasFileOpenContract(FileOpenContract contracts,
|
||||
FileOpenContract contract) {
|
||||
return (static_cast<uint8_t>(contracts) & static_cast<uint8_t>(contract)) ==
|
||||
static_cast<uint8_t>(contract);
|
||||
}
|
||||
|
||||
struct FileOptions : EnvOptions {
|
||||
// Embedded IOOptions to control the parameters for any IOs that need
|
||||
// to be issued for the file open/creation
|
||||
@@ -191,6 +220,9 @@ struct FileOptions : EnvOptions {
|
||||
// coding.
|
||||
Temperature temperature = Temperature::kUnknown;
|
||||
|
||||
// File-open contract. kDefault uses the provider's standard semantics.
|
||||
FileOpenContract open_contract = FileOpenContract::kDefault;
|
||||
|
||||
// The checksum type that is used to calculate the checksum value for
|
||||
// handoff during file writes.
|
||||
ChecksumType handoff_checksum_type;
|
||||
@@ -236,6 +268,7 @@ struct FileOptions : EnvOptions {
|
||||
: EnvOptions(opts),
|
||||
io_options(opts.io_options),
|
||||
temperature(opts.temperature),
|
||||
open_contract(opts.open_contract),
|
||||
handoff_checksum_type(opts.handoff_checksum_type),
|
||||
write_hint(opts.write_hint),
|
||||
file_checksum(opts.file_checksum),
|
||||
|
||||
@@ -12,9 +12,11 @@
|
||||
#include <cinttypes>
|
||||
#include <cstdint>
|
||||
#include <string>
|
||||
#include <utility>
|
||||
|
||||
#include "block_fetcher.h"
|
||||
#include "file/random_access_file_reader.h"
|
||||
#include "file/read_write_util.h"
|
||||
#include "memory/memory_allocator_impl.h"
|
||||
#include "monitoring/perf_context_imp.h"
|
||||
#include "monitoring/statistics_impl.h"
|
||||
@@ -498,16 +500,11 @@ static Status ReadFooterFromFileInternal(
|
||||
FilePrefetchBuffer* prefetch_buffer, uint64_t expected_file_size,
|
||||
Footer* footer, uint64_t enforce_table_magic_number) {
|
||||
uint64_t file_size_from_file_system = 0;
|
||||
Status s;
|
||||
// Prefer the more efficient FSRandomAccessFile::GetFileSize when available
|
||||
s = file->file()->GetFileSize(&file_size_from_file_system);
|
||||
Status s = GetFileSizeFromOpenFileOrPath(file->file(), &fs, file->file_name(),
|
||||
&file_size_from_file_system, nullptr,
|
||||
FileSizeFallback::kAnyOpenFileError);
|
||||
if (!s.ok()) {
|
||||
// Fall back on FileSystem::GetFileSize on failure
|
||||
s = fs.GetFileSize(file->file_name(), IOOptions(),
|
||||
&file_size_from_file_system, nullptr);
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
return s;
|
||||
}
|
||||
|
||||
if (expected_file_size != file_size_from_file_system) {
|
||||
|
||||
@@ -350,6 +350,8 @@ Status SstFileWriter::Open(const std::string& file_path, Temperature temp) {
|
||||
std::unique_ptr<FSWritableFile> sst_file;
|
||||
FileOptions cur_file_opts(r->env_options);
|
||||
cur_file_opts.temperature = temp;
|
||||
cur_file_opts.open_contract = FileOpenContract::kNoReopenForWrite |
|
||||
FileOpenContract::kNoReadersWhileOpenForWrite;
|
||||
s = r->ioptions.env->GetFileSystem()->NewWritableFile(
|
||||
file_path, cur_file_opts, &sst_file, nullptr);
|
||||
if (!s.ok()) {
|
||||
|
||||
@@ -174,18 +174,27 @@ IOStatus TestFSDirectory::FsyncWithDirOptions(
|
||||
TestFSWritableFile::TestFSWritableFile(const std::string& fname,
|
||||
const FileOptions& file_opts,
|
||||
std::unique_ptr<FSWritableFile>&& f,
|
||||
FaultInjectionTestFS* fs)
|
||||
FaultInjectionTestFS* fs,
|
||||
bool reopened_for_write)
|
||||
: state_(fname),
|
||||
file_opts_(file_opts),
|
||||
target_(std::move(f)),
|
||||
should_forward_close_(true),
|
||||
fs_(fs),
|
||||
unsync_data_loss_(fs_->InjectUnsyncedDataLoss()) {
|
||||
unsync_data_loss_(fs_->InjectUnsyncedDataLoss()),
|
||||
reopened_for_write_(reopened_for_write) {
|
||||
assert(target_ != nullptr);
|
||||
assert(state_.pos_at_last_append_ == 0);
|
||||
assert(state_.pos_at_last_sync_ == 0);
|
||||
}
|
||||
|
||||
IOStatus TestFSWritableFile::ValidateWrite(const char* op_name) {
|
||||
if (!reopened_for_write_) {
|
||||
return IOStatus::OK();
|
||||
}
|
||||
return fs_->ValidateReopenedWrite(state_.filename_, op_name);
|
||||
}
|
||||
|
||||
TestFSWritableFile::~TestFSWritableFile() {
|
||||
const ThreadStatus::OperationType thread_op =
|
||||
ThreadStatusUtil::GetThreadOperation();
|
||||
@@ -205,8 +214,12 @@ IOStatus TestFSWritableFile::Append(const Slice& data, const IOOptions& options,
|
||||
if (!fs_->IsFilesystemActive()) {
|
||||
return fs_->GetError();
|
||||
}
|
||||
IOStatus s = ValidateWrite("Append");
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
|
||||
IOStatus s = fs_->MaybeInjectThreadLocalError(
|
||||
s = fs_->MaybeInjectThreadLocalError(
|
||||
FaultInjectionIOType::kWrite, options, "Append", state_.filename_,
|
||||
fault_injection_detail::SizeAndHead(data),
|
||||
FaultInjectionTestFS::ErrorOperation::kAppend);
|
||||
@@ -243,7 +256,12 @@ IOStatus TestFSWritableFile::Append(
|
||||
return IOStatus::Corruption("Data is corrupted!");
|
||||
}
|
||||
|
||||
IOStatus s = fs_->MaybeInjectThreadLocalError(
|
||||
IOStatus s = ValidateWrite("Append");
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
|
||||
s = fs_->MaybeInjectThreadLocalError(
|
||||
FaultInjectionIOType::kWrite, options, "Append", state_.filename_,
|
||||
fault_injection_detail::SizeAndHead(data),
|
||||
FaultInjectionTestFS::ErrorOperation::kAppend);
|
||||
@@ -284,9 +302,13 @@ IOStatus TestFSWritableFile::Truncate(uint64_t size, const IOOptions& options,
|
||||
if (!fs_->IsFilesystemActive()) {
|
||||
return fs_->GetError();
|
||||
}
|
||||
IOStatus s = fs_->MaybeInjectThreadLocalError(
|
||||
FaultInjectionIOType::kWrite, options, "Truncate", state_.filename_,
|
||||
fault_injection_detail::Size(size));
|
||||
IOStatus s = ValidateWrite("Truncate");
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
s = fs_->MaybeInjectThreadLocalError(FaultInjectionIOType::kWrite, options,
|
||||
"Truncate", state_.filename_,
|
||||
fault_injection_detail::Size(size));
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
@@ -309,7 +331,11 @@ IOStatus TestFSWritableFile::PositionedAppend(const Slice& data,
|
||||
if (fs_->ShouldDataCorruptionBeforeWrite()) {
|
||||
return IOStatus::Corruption("Data is corrupted!");
|
||||
}
|
||||
IOStatus s = fs_->MaybeInjectThreadLocalError(
|
||||
IOStatus s = ValidateWrite("PositionedAppend");
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
s = fs_->MaybeInjectThreadLocalError(
|
||||
FaultInjectionIOType::kWrite, options, "PositionedAppend",
|
||||
state_.filename_, fault_injection_detail::OffsetSizeAndHead(offset, data),
|
||||
FaultInjectionTestFS::ErrorOperation::kPositionedAppend);
|
||||
@@ -337,7 +363,11 @@ IOStatus TestFSWritableFile::PositionedAppend(
|
||||
if (fs_->ShouldDataCorruptionBeforeWrite()) {
|
||||
return IOStatus::Corruption("Data is corrupted!");
|
||||
}
|
||||
IOStatus s = fs_->MaybeInjectThreadLocalError(
|
||||
IOStatus s = ValidateWrite("PositionedAppend");
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
s = fs_->MaybeInjectThreadLocalError(
|
||||
FaultInjectionIOType::kWrite, options, "PositionedAppend",
|
||||
state_.filename_, fault_injection_detail::OffsetSizeAndHead(offset, data),
|
||||
FaultInjectionTestFS::ErrorOperation::kPositionedAppend);
|
||||
@@ -915,6 +945,50 @@ IOStatus FaultInjectionTestFS::GetChildrenFileAttributes(
|
||||
return io_s;
|
||||
}
|
||||
|
||||
void FaultInjectionTestFS::WritableFileOpened(const std::string& fname,
|
||||
FileOpenContract open_contract) {
|
||||
MutexLock l(&mutex_);
|
||||
open_managed_files_[fname] = open_contract;
|
||||
file_open_contracts_[fname] = open_contract;
|
||||
}
|
||||
|
||||
IOStatus FaultInjectionTestFS::ValidateReadOpen(const std::string& fname,
|
||||
const char* op_name) {
|
||||
MutexLock l(&mutex_);
|
||||
auto it = open_managed_files_.find(fname);
|
||||
if (it != open_managed_files_.end() &&
|
||||
HasFileOpenContract(it->second,
|
||||
FileOpenContract::kNoReadersWhileOpenForWrite)) {
|
||||
return IOStatus::NotSupported(
|
||||
std::string(op_name) +
|
||||
" violates no-readers-while-open-for-write contract: " + fname);
|
||||
}
|
||||
return IOStatus::OK();
|
||||
}
|
||||
|
||||
IOStatus FaultInjectionTestFS::ValidateWriteOpen(const std::string& fname,
|
||||
const char* op_name) {
|
||||
return ValidateNoReopenForWrite(fname, op_name);
|
||||
}
|
||||
|
||||
IOStatus FaultInjectionTestFS::ValidateReopenedWrite(const std::string& fname,
|
||||
const char* op_name) {
|
||||
return ValidateNoReopenForWrite(fname, op_name);
|
||||
}
|
||||
|
||||
IOStatus FaultInjectionTestFS::ValidateNoReopenForWrite(
|
||||
const std::string& fname, const char* op_name) {
|
||||
MutexLock l(&mutex_);
|
||||
auto it = file_open_contracts_.find(fname);
|
||||
if (it != file_open_contracts_.end() &&
|
||||
HasFileOpenContract(it->second, FileOpenContract::kNoReopenForWrite)) {
|
||||
return IOStatus::NotSupported(
|
||||
std::string(op_name) +
|
||||
" violates no-reopen-for-write contract: " + fname);
|
||||
}
|
||||
return IOStatus::OK();
|
||||
}
|
||||
|
||||
IOStatus FaultInjectionTestFS::NewWritableFile(
|
||||
const std::string& fname, const FileOptions& file_opts,
|
||||
std::unique_ptr<FSWritableFile>* result, IODebugContext* dbg) {
|
||||
@@ -933,6 +1007,11 @@ IOStatus FaultInjectionTestFS::NewWritableFile(
|
||||
return io_s;
|
||||
}
|
||||
|
||||
io_s = ValidateWriteOpen(fname, "NewWritableFile");
|
||||
if (!io_s.ok()) {
|
||||
return io_s;
|
||||
}
|
||||
|
||||
io_s = target()->NewWritableFile(fname, file_opts, result, dbg);
|
||||
if (io_s.ok()) {
|
||||
result->reset(
|
||||
@@ -940,9 +1019,9 @@ IOStatus FaultInjectionTestFS::NewWritableFile(
|
||||
// WritableFileWriter* file is opened
|
||||
// again then it will be truncated - so forget our saved state.
|
||||
UntrackFile(fname);
|
||||
WritableFileOpened(fname, file_opts.open_contract);
|
||||
{
|
||||
MutexLock l(&mutex_);
|
||||
open_managed_files_.insert(fname);
|
||||
auto dir_and_name = TestFSGetDirAndName(fname);
|
||||
auto& list = dir_to_new_files_since_last_sync_[dir_and_name.first];
|
||||
// The new file could overwrite an old one. Here we simplify
|
||||
@@ -986,6 +1065,8 @@ IOStatus FaultInjectionTestFS::ReopenWritableFile(
|
||||
return io_s;
|
||||
}
|
||||
|
||||
// Reopen is allowed so SyncFile can use the reopened handle. Reopened handles
|
||||
// carrying kNoReopenForWrite are rejected only if they mutate file data.
|
||||
io_s = target()->ReopenWritableFile(fname, file_opts, result, dbg);
|
||||
|
||||
// Only track files we created. Files created outside of this
|
||||
@@ -1004,7 +1085,8 @@ IOStatus FaultInjectionTestFS::ReopenWritableFile(
|
||||
} else if (!exists) {
|
||||
// It was created by this `FileSystem` just now.
|
||||
should_track = true;
|
||||
open_managed_files_.insert(fname);
|
||||
open_managed_files_[fname] = file_opts.open_contract;
|
||||
file_open_contracts_[fname] = file_opts.open_contract;
|
||||
auto dir_and_name = TestFSGetDirAndName(fname);
|
||||
auto& list = dir_to_new_files_since_last_sync_[dir_and_name.first];
|
||||
list[dir_and_name.second] = kNewFileNoOverwrite;
|
||||
@@ -1013,8 +1095,9 @@ IOStatus FaultInjectionTestFS::ReopenWritableFile(
|
||||
}
|
||||
}
|
||||
if (should_track) {
|
||||
result->reset(
|
||||
new TestFSWritableFile(fname, file_opts, std::move(*result), this));
|
||||
result->reset(new TestFSWritableFile(fname, file_opts, std::move(*result),
|
||||
this,
|
||||
/*reopened_for_write=*/exists));
|
||||
}
|
||||
}
|
||||
return io_s;
|
||||
@@ -1034,6 +1117,10 @@ IOStatus FaultInjectionTestFS::ReuseWritableFile(
|
||||
const std::string& fname, const std::string& old_fname,
|
||||
const FileOptions& file_opts, std::unique_ptr<FSWritableFile>* result,
|
||||
IODebugContext* dbg) {
|
||||
IOStatus contract_s = ValidateWriteOpen(old_fname, "ReuseWritableFile");
|
||||
if (!contract_s.ok()) {
|
||||
return contract_s;
|
||||
}
|
||||
IOStatus s = RenameFile(old_fname, fname, file_opts.io_options, dbg);
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
@@ -1057,6 +1144,11 @@ IOStatus FaultInjectionTestFS::NewRandomRWFile(
|
||||
return io_s;
|
||||
}
|
||||
|
||||
io_s = ValidateWriteOpen(fname, "NewRandomRWFile");
|
||||
if (!io_s.ok()) {
|
||||
return io_s;
|
||||
}
|
||||
|
||||
io_s = target()->NewRandomRWFile(fname, file_opts, result, dbg);
|
||||
|
||||
if (io_s.ok()) {
|
||||
@@ -1064,9 +1156,9 @@ IOStatus FaultInjectionTestFS::NewRandomRWFile(
|
||||
// WritableFileWriter* file is opened
|
||||
// again then it will be truncated - so forget our saved state.
|
||||
UntrackFile(fname);
|
||||
WritableFileOpened(fname, file_opts.open_contract);
|
||||
{
|
||||
MutexLock l(&mutex_);
|
||||
open_managed_files_.insert(fname);
|
||||
auto dir_and_name = TestFSGetDirAndName(fname);
|
||||
auto& list = dir_to_new_files_since_last_sync_[dir_and_name.first];
|
||||
// It could be overwriting an old file, but we simplify the
|
||||
@@ -1092,6 +1184,11 @@ IOStatus FaultInjectionTestFS::NewRandomAccessFile(
|
||||
return io_s;
|
||||
}
|
||||
|
||||
io_s = ValidateReadOpen(fname, "NewRandomAccessFile");
|
||||
if (!io_s.ok()) {
|
||||
return io_s;
|
||||
}
|
||||
|
||||
io_s = target()->NewRandomAccessFile(fname, file_opts, result, dbg);
|
||||
|
||||
if (io_s.ok()) {
|
||||
@@ -1115,6 +1212,11 @@ IOStatus FaultInjectionTestFS::NewSequentialFile(
|
||||
return io_s;
|
||||
}
|
||||
|
||||
io_s = ValidateReadOpen(fname, "NewSequentialFile");
|
||||
if (!io_s.ok()) {
|
||||
return io_s;
|
||||
}
|
||||
|
||||
io_s = target()->NewSequentialFile(fname, file_opts, result, dbg);
|
||||
|
||||
if (io_s.ok()) {
|
||||
@@ -1228,6 +1330,16 @@ IOStatus FaultInjectionTestFS::RenameFile(const std::string& s,
|
||||
db_file_state_[t] = db_file_state_[s];
|
||||
db_file_state_.erase(s);
|
||||
}
|
||||
auto open_contract_it = open_managed_files_.find(s);
|
||||
if (open_contract_it != open_managed_files_.end()) {
|
||||
open_managed_files_[t] = open_contract_it->second;
|
||||
open_managed_files_.erase(open_contract_it);
|
||||
}
|
||||
auto contract_it = file_open_contracts_.find(s);
|
||||
if (contract_it != file_open_contracts_.end()) {
|
||||
file_open_contracts_[t] = contract_it->second;
|
||||
file_open_contracts_.erase(contract_it);
|
||||
}
|
||||
|
||||
auto sdn = TestFSGetDirAndName(s);
|
||||
auto tdn = TestFSGetDirAndName(t);
|
||||
@@ -1272,6 +1384,10 @@ IOStatus FaultInjectionTestFS::LinkFile(const std::string& s,
|
||||
if (db_file_state_.find(s) != db_file_state_.end()) {
|
||||
db_file_state_[t] = db_file_state_[s];
|
||||
}
|
||||
auto contract_it = file_open_contracts_.find(s);
|
||||
if (contract_it != file_open_contracts_.end()) {
|
||||
file_open_contracts_[t] = contract_it->second;
|
||||
}
|
||||
|
||||
auto sdn = TestFSGetDirAndName(s);
|
||||
auto tdn = TestFSGetDirAndName(t);
|
||||
@@ -1461,6 +1577,8 @@ IOStatus FaultInjectionTestFS::DeleteFilesCreatedAfterLastDirSync(
|
||||
void FaultInjectionTestFS::ResetState() {
|
||||
MutexLock l(&mutex_);
|
||||
db_file_state_.clear();
|
||||
open_managed_files_.clear();
|
||||
file_open_contracts_.clear();
|
||||
dir_to_new_files_since_last_sync_.clear();
|
||||
SetFilesystemActiveNoLock(true);
|
||||
}
|
||||
@@ -1472,6 +1590,7 @@ void FaultInjectionTestFS::UntrackFile(const std::string& f) {
|
||||
dir_and_name.second);
|
||||
db_file_state_.erase(f);
|
||||
open_managed_files_.erase(f);
|
||||
file_open_contracts_.erase(f);
|
||||
}
|
||||
|
||||
IOStatus FaultInjectionTestFS::MaybeInjectThreadLocalReadError(
|
||||
|
||||
@@ -313,7 +313,8 @@ class TestFSWritableFile : public FSWritableFile {
|
||||
explicit TestFSWritableFile(const std::string& fname,
|
||||
const FileOptions& file_opts,
|
||||
std::unique_ptr<FSWritableFile>&& f,
|
||||
FaultInjectionTestFS* fs);
|
||||
FaultInjectionTestFS* fs,
|
||||
bool reopened_for_write = false);
|
||||
virtual ~TestFSWritableFile();
|
||||
IOStatus Append(const Slice& data, const IOOptions&,
|
||||
IODebugContext*) override;
|
||||
@@ -347,6 +348,7 @@ class TestFSWritableFile : public FSWritableFile {
|
||||
|
||||
private:
|
||||
IOStatus CloseImpl(const IOOptions& options, IODebugContext* dbg);
|
||||
IOStatus ValidateWrite(const char* op_name);
|
||||
|
||||
FSFileState state_; // Need protection by mutex_
|
||||
FileOptions file_opts_;
|
||||
@@ -359,6 +361,7 @@ class TestFSWritableFile : public FSWritableFile {
|
||||
FaultInjectionTestFS* fs_;
|
||||
port::Mutex mutex_;
|
||||
const bool unsync_data_loss_;
|
||||
const bool reopened_for_write_;
|
||||
};
|
||||
|
||||
// A per-file fault-injection wrapper around FSRandomRWFile that reports
|
||||
@@ -621,6 +624,15 @@ class FaultInjectionTestFS : public FileSystemWrapper {
|
||||
|
||||
void RandomRWFileClosed(const std::string& fname);
|
||||
|
||||
void WritableFileOpened(const std::string& fname,
|
||||
FileOpenContract open_contract);
|
||||
|
||||
IOStatus ValidateReadOpen(const std::string& fname, const char* op_name);
|
||||
|
||||
IOStatus ValidateWriteOpen(const std::string& fname, const char* op_name);
|
||||
|
||||
IOStatus ValidateReopenedWrite(const std::string& fname, const char* op_name);
|
||||
|
||||
IOStatus DropUnsyncedFileData();
|
||||
|
||||
IOStatus DropRandomUnsyncedFileData(Random* rnd);
|
||||
@@ -902,9 +914,14 @@ class FaultInjectionTestFS : public FileSystemWrapper {
|
||||
private:
|
||||
inline static const std::string kFailedToWriteToWAL =
|
||||
"failed to write to WAL";
|
||||
|
||||
IOStatus ValidateNoReopenForWrite(const std::string& fname,
|
||||
const char* op_name);
|
||||
|
||||
port::Mutex mutex_;
|
||||
std::map<std::string, FSFileState> db_file_state_;
|
||||
std::set<std::string> open_managed_files_;
|
||||
std::map<std::string, FileOpenContract> open_managed_files_;
|
||||
std::map<std::string, FileOpenContract> file_open_contracts_;
|
||||
// directory -> (file name -> file contents to recover)
|
||||
// When data is recovered from unsyned parent directory, the files with
|
||||
// empty file contents to recover is deleted. Those with non-empty ones
|
||||
|
||||
Reference in New Issue
Block a user