diff --git a/db/blob/blob_file_builder.cc b/db/blob/blob_file_builder.cc index 9e5a5d0d67..6c158b66a4 100644 --- a/db/blob/blob_file_builder.cc +++ b/db/blob/blob_file_builder.cc @@ -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); diff --git a/db/blob/blob_file_cache.cc b/db/blob/blob_file_cache.cc index 95eebea003..58a8959ee5 100644 --- a/db/blob/blob_file_cache.cc +++ b/db/blob/blob_file_cache.cc @@ -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_, diff --git a/db/blob/blob_file_partition_manager.cc b/db/blob/blob_file_partition_manager.cc index e1e1f071ff..e83a922ca1 100644 --- a/db/blob/blob_file_partition_manager.cc +++ b/db/blob/blob_file_partition_manager.cc @@ -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 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( - 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); diff --git a/db/blob/blob_file_reader.cc b/db/blob/blob_file_reader.cc index 66be37301e..041d4a1500 100644 --- a/db/blob/blob_file_reader.cc +++ b/db/blob/blob_file_reader.cc @@ -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 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); } diff --git a/db/blob/blob_file_reader_test.cc b/db/blob/blob_file_reader_test.cc index 7377770be6..535ae193bb 100644 --- a/db/blob/blob_file_reader_test.cc +++ b/db/blob/blob_file_reader_test.cc @@ -7,10 +7,12 @@ #include #include +#include #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 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& 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&& 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&& 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& 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* result, + IODebugContext* dbg) override { + std::unique_ptr 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 WrapTargetFile( + std::unique_ptr&& 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& target, + std::string target_path) + : OpenFileSizeFileSystem(target, std::move(target_path)) {} + + const char* Name() const override { return kClassName(); } + + private: + std::unique_ptr WrapTargetFile( + std::unique_ptr&& file) override { + return std::unique_ptr( + 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& target, + std::string target_path) + : OpenFileSizeFileSystem(target, std::move(target_path)) {} + + const char* Name() const override { return kClassName(); } + + private: + std::unique_ptr WrapTargetFile( + std::unique_ptr&& file) override { + return std::unique_ptr( + 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& 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( + 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 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 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( + 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 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 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( + 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 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( + 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 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{ - "BlobFileReader::OpenFile:GetFileSize", "BlobFileReader::OpenFile:NewRandomAccessFile", "BlobFileReader::ReadHeader:ReadFromFile", "BlobFileReader::ReadFooter:ReadFromFile", diff --git a/db/blob/db_blob_direct_write_test.cc b/db/blob/db_blob_direct_write_test.cc index bc908d5374..0d4c05794e 100644 --- a/db/blob/db_blob_direct_write_test.cc +++ b/db/blob/db_blob_direct_write_test.cc @@ -10,11 +10,14 @@ #include #include +#include #include #include #include #include #include +#include +#include #include #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 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&& target, + std::function 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 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&& target, bool expose_current_size, + std::shared_ptr 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 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& 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* result, + IODebugContext* dbg) override { + std::unique_ptr 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 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* result, + IODebugContext* dbg) override { + std::unique_ptr 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 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 lock(mu_); + return saw_no_reopen_for_write_writer_contract_; + } + + bool SawNoReadersWhileOpenForWriteWriterContract() const { + std::lock_guard lock(mu_); + return saw_no_readers_while_open_for_write_writer_contract_; + } + + bool SawNoReadersWhileOpenForWriteReaderContract() const { + std::lock_guard lock(mu_); + return saw_no_readers_while_open_for_write_reader_contract_; + } + + private: + void DeactivateBlobPath(const std::string& fname) { + std::lock_guard lock(mu_); + active_blob_paths_.erase(fname); + } + + bool IsActiveBlobPath(const std::string& fname) const { + std::lock_guard 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 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(env_->GetFileSystem()); + std::unique_ptr 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; diff --git a/db/builder.cc b/db/builder.cc index 26a0e139f9..7040f06f74 100644 --- a/db/builder.cc +++ b/db/builder.cc @@ -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()); diff --git a/db/compaction/compaction_job.cc b/db/compaction/compaction_job.cc index 98b33e6f9c..aa23fcd119 100644 --- a/db/compaction/compaction_job.cc +++ b/db/compaction/compaction_job.cc @@ -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; diff --git a/db/fault_injection_test.cc b/db/fault_injection_test.cc index 46ca3d1527..e29d955379 100644 --- a/db/fault_injection_test.cc +++ b/db/fault_injection_test.cc @@ -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 fault_fs = + std::make_shared(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 writer; + ASSERT_OK(fault_fs->NewWritableFile(fname, writer_options, &writer, nullptr)); + + std::unique_ptr 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 fault_fs = + std::make_shared(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 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 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) { diff --git a/file/read_write_util.cc b/file/read_write_util.cc index 3617a35e33..a3d99619ab 100644 --- a/file/read_write_util.cc +++ b/file/read_write_util.cc @@ -9,6 +9,7 @@ #include "file/read_write_util.h" +#include #include #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; diff --git a/file/read_write_util.h b/file/read_write_util.h index edcddd5557..2215853590 100644 --- a/file/read_write_util.h +++ b/file/read_write_util.h @@ -25,6 +25,18 @@ IOStatus NewWritableFile(FileSystem* fs, const std::string& fname, std::unique_ptr* 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 diff --git a/include/rocksdb/file_system.h b/include/rocksdb/file_system.h index 314c3718b5..f3080ac240 100644 --- a/include/rocksdb/file_system.h +++ b/include/rocksdb/file_system.h @@ -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 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(static_cast(lhs) | + static_cast(rhs)); +} + +inline FileOpenContract& operator|=(FileOpenContract& lhs, + FileOpenContract rhs) { + lhs = lhs | rhs; + return lhs; +} + +constexpr bool HasFileOpenContract(FileOpenContract contracts, + FileOpenContract contract) { + return (static_cast(contracts) & static_cast(contract)) == + static_cast(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), diff --git a/table/format.cc b/table/format.cc index 30b2030b87..9c7d0fa1a8 100644 --- a/table/format.cc +++ b/table/format.cc @@ -12,9 +12,11 @@ #include #include #include +#include #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) { diff --git a/table/sst_file_writer.cc b/table/sst_file_writer.cc index 2d958ade08..6f6aa2750b 100644 --- a/table/sst_file_writer.cc +++ b/table/sst_file_writer.cc @@ -350,6 +350,8 @@ Status SstFileWriter::Open(const std::string& file_path, Temperature temp) { std::unique_ptr 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()) { diff --git a/utilities/fault_injection_fs.cc b/utilities/fault_injection_fs.cc index e5aa2f8655..25f3303bab 100644 --- a/utilities/fault_injection_fs.cc +++ b/utilities/fault_injection_fs.cc @@ -174,18 +174,27 @@ IOStatus TestFSDirectory::FsyncWithDirOptions( TestFSWritableFile::TestFSWritableFile(const std::string& fname, const FileOptions& file_opts, std::unique_ptr&& 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* 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* 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( diff --git a/utilities/fault_injection_fs.h b/utilities/fault_injection_fs.h index 575d0db376..c861c9c4db 100644 --- a/utilities/fault_injection_fs.h +++ b/utilities/fault_injection_fs.h @@ -313,7 +313,8 @@ class TestFSWritableFile : public FSWritableFile { explicit TestFSWritableFile(const std::string& fname, const FileOptions& file_opts, std::unique_ptr&& 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 db_file_state_; - std::set open_managed_files_; + std::map open_managed_files_; + std::map 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