Sync dir containing CURRENT after RenameFile on CURRENT as much as possible (#10573)

Summary:
**Context:**
Below crash test revealed a bug that directory containing CURRENT file (short for `dir_contains_current_file` below) was not always get synced after a new CURRENT is created and being called with `RenameFile` as part of the creation.

This bug exposes a risk that such un-synced directory containing the updated CURRENT can’t survive a host crash (e.g, power loss) hence get corrupted. This then will be followed by a recovery from a corrupted CURRENT that we don't want.

The root-cause is that a nullptr `FSDirectory* dir_contains_current_file` sometimes gets passed-down to `SetCurrentFile()` hence in those case `dir_contains_current_file->FSDirectory::FsyncWithDirOptions()` will be skipped  (which otherwise will internally call`Env/FS::SyncDic()` )
```
./db_stress --acquire_snapshot_one_in=10000 --adaptive_readahead=1 --allow_data_in_errors=True --avoid_unnecessary_blocking_io=0 --backup_max_size=104857600 --backup_one_in=100000 --batch_protection_bytes_per_key=8 --block_size=16384 --bloom_bits=134.8015470676662 --bottommost_compression_type=disable --cache_size=8388608 --checkpoint_one_in=1000000 --checksum_type=kCRC32c --clear_column_family_one_in=0 --compact_files_one_in=1000000 --compact_range_one_in=1000000 --compaction_pri=2 --compaction_ttl=100 --compression_max_dict_buffer_bytes=511 --compression_max_dict_bytes=16384 --compression_type=zstd --compression_use_zstd_dict_trainer=1 --compression_zstd_max_train_bytes=65536 --continuous_verification_interval=0 --data_block_index_type=0 --db=$db --db_write_buffer_size=1048576 --delpercent=5 --delrangepercent=0 --destroy_db_initially=0 --disable_wal=0 --enable_compaction_filter=0 --enable_pipelined_write=1 --expected_values_dir=$exp --fail_if_options_file_error=1 --file_checksum_impl=none --flush_one_in=1000000 --get_current_wal_file_one_in=0 --get_live_files_one_in=1000000 --get_property_one_in=1000000 --get_sorted_wal_files_one_in=0 --index_block_restart_interval=4 --ingest_external_file_one_in=0 --iterpercent=10 --key_len_percent_dist=1,30,69 --level_compaction_dynamic_level_bytes=True --mark_for_compaction_one_file_in=10 --max_background_compactions=20 --max_bytes_for_level_base=10485760 --max_key=10000 --max_key_len=3 --max_manifest_file_size=16384 --max_write_batch_group_size_bytes=64 --max_write_buffer_number=3 --max_write_buffer_size_to_maintain=0 --memtable_prefix_bloom_size_ratio=0.001 --memtable_protection_bytes_per_key=1 --memtable_whole_key_filtering=1 --mmap_read=1 --nooverwritepercent=1 --open_metadata_write_fault_one_in=0 --open_read_fault_one_in=0 --open_write_fault_one_in=0 --ops_per_thread=100000000 --optimize_filters_for_memory=1 --paranoid_file_checks=1 --partition_pinning=2 --pause_background_one_in=1000000 --periodic_compaction_seconds=0 --prefix_size=5 --prefixpercent=5 --prepopulate_block_cache=1 --progress_reports=0 --read_fault_one_in=1000 --readpercent=45 --recycle_log_file_num=0 --reopen=0 --ribbon_starting_level=999 --secondary_cache_fault_one_in=32 --secondary_cache_uri=compressed_secondary_cache://capacity=8388608 --set_options_one_in=10000 --snapshot_hold_ops=100000 --sst_file_manager_bytes_per_sec=0 --sst_file_manager_bytes_per_truncate=0 --subcompactions=3 --sync_fault_injection=1 --target_file_size_base=2097 --target_file_size_multiplier=2 --test_batches_snapshots=1 --top_level_index_pinning=1 --use_full_merge_v1=1 --use_merge=1 --value_size_mult=32 --verify_checksum=1 --verify_checksum_one_in=1000000 --verify_db_one_in=100000 --verify_sst_unique_id_in_manifest=1 --wal_bytes_per_sync=524288 --write_buffer_size=4194 --writepercent=35
```

```
stderr:
WARNING: prefix_size is non-zero but memtablerep != prefix_hash
db_stress: utilities/fault_injection_fs.cc:748: virtual rocksdb::IOStatus rocksdb::FaultInjectionTestFS::RenameFile(const std::string &, const std::string &, const rocksdb::IOOptions &, rocksdb::IODebugContext *): Assertion `tlist.find(tdn.second) == tlist.end()' failed.`
```

**Summary:**
The PR ensured the non-test path pass down a non-null dir containing CURRENT (which is by current RocksDB assumption just db_dir) by doing the following:
- Renamed `directory_to_fsync` as `dir_contains_current_file` in `SetCurrentFile()` to tighten the association between this directory and CURRENT file
- Changed `SetCurrentFile()` API to require `dir_contains_current_file` being passed-in, instead of making it by default nullptr.
    -  Because `SetCurrentFile()`'s `dir_contains_current_file` is passed down from `VersionSet::LogAndApply()` then `VersionSet::ProcessManifestWrites()` (i.e, think about this as a chain of 3 functions related to MANIFEST update), these 2 functions also got refactored to require `dir_contains_current_file`
- Updated the non-test-path callers of these 3 functions to obtain and pass in non-nullptr `dir_contains_current_file`, which by current assumption of RocksDB, is the `FSDirectory* db_dir`.
    - `db_impl` path will obtain `DBImpl::directories_.getDbDir()` while others with no access to such `directories_` are obtained on the fly by creating such object `FileSystem::NewDirectory(..)` and manage it by unique pointers to ensure short life time.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/10573

Test Plan:
- `make check`
- Passed the repro db_stress command
- For future improvement, since we currently don't assert dir containing CURRENT to be non-nullptr due to https://github.com/facebook/rocksdb/pull/10573#pullrequestreview-1087698899, there is still chances that future developers mistakenly pass down nullptr dir containing CURRENT thus resulting skipped sync dir and cause the bug again. Therefore a smarter test (e.g, such as quoted from ajkr  "(make) unsynced data loss to be dropping files corresponding to unsynced directory entries") is still needed.

Reviewed By: ajkr

Differential Revision: D39005886

Pulled By: hx235

fbshipit-source-id: 336fb9090d0cfa6ca3dd580db86268007dde7f5a
main
Hui Xiao 2 years ago committed by Facebook GitHub Bot
parent 7818560194
commit e484b81eee
  1. 1
      HISTORY.md
  2. 2
      db/compaction/compaction_job_test.cc
  3. 6
      db/db_impl/db_impl.cc
  4. 2
      db/db_impl/db_impl_compaction_flush.cc
  5. 3
      db/db_impl/db_impl_write.cc
  6. 6
      db/experimental.cc
  7. 21
      db/repair.cc
  8. 13
      db/version_set.cc
  9. 25
      db/version_set.h
  10. 18
      db/version_set_test.cc
  11. 5
      db/version_util.h
  12. 6
      file/filename.cc
  13. 6
      file/filename.h
  14. 7
      tools/ldb_cmd.cc

@ -5,6 +5,7 @@
* Fixed bug where `FlushWAL(true /* sync */)` (used by `GetLiveFilesStorageInfo()`, which is used by checkpoint and backup) could cause parallel writes at the tail of a WAL file to never be synced. * Fixed bug where `FlushWAL(true /* sync */)` (used by `GetLiveFilesStorageInfo()`, which is used by checkpoint and backup) could cause parallel writes at the tail of a WAL file to never be synced.
* Fix periodic_task unable to re-register the same task type, which may cause `SetOptions()` fail to update periodical_task time like: `stats_dump_period_sec`, `stats_persist_period_sec`. * Fix periodic_task unable to re-register the same task type, which may cause `SetOptions()` fail to update periodical_task time like: `stats_dump_period_sec`, `stats_persist_period_sec`.
* Fixed a bug in the rocksdb.prefetched.bytes.discarded stat. It was counting the prefetch buffer size, rather than the actual number of bytes discarded from the buffer. * Fixed a bug in the rocksdb.prefetched.bytes.discarded stat. It was counting the prefetch buffer size, rather than the actual number of bytes discarded from the buffer.
* Fix bug where the directory containing CURRENT can left unsynced after CURRENT is updated to point to the latest MANIFEST, which leads to risk of unsync data loss of CURRENT.
### Public API changes ### Public API changes
* Add `rocksdb_column_family_handle_get_id`, `rocksdb_column_family_handle_get_name` to get name, id of column family in C API * Add `rocksdb_column_family_handle_get_id`, `rocksdb_column_family_handle_get_name` to get name, id of column family in C API

@ -377,7 +377,7 @@ class CompactionJobTestBase : public testing::Test {
mutex_.Lock(); mutex_.Lock();
EXPECT_OK( EXPECT_OK(
versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(), versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(),
mutable_cf_options_, &edit, &mutex_)); mutable_cf_options_, &edit, &mutex_, nullptr));
mutex_.Unlock(); mutex_.Unlock();
} }

@ -1534,8 +1534,8 @@ Status DBImpl::SyncWAL() {
Status DBImpl::ApplyWALToManifest(VersionEdit* synced_wals) { Status DBImpl::ApplyWALToManifest(VersionEdit* synced_wals) {
// not empty, write to MANIFEST. // not empty, write to MANIFEST.
mutex_.AssertHeld(); mutex_.AssertHeld();
Status status = Status status = versions_->LogAndApplyToDefaultColumnFamily(
versions_->LogAndApplyToDefaultColumnFamily(synced_wals, &mutex_); synced_wals, &mutex_, directories_.GetDbDir());
if (!status.ok() && versions_->io_status().IsIOError()) { if (!status.ok() && versions_->io_status().IsIOError()) {
status = error_handler_.SetBGError(versions_->io_status(), status = error_handler_.SetBGError(versions_->io_status(),
BackgroundErrorReason::kManifestWrite); BackgroundErrorReason::kManifestWrite);
@ -3149,7 +3149,7 @@ Status DBImpl::DropColumnFamilyImpl(ColumnFamilyHandle* column_family) {
WriteThread::Writer w; WriteThread::Writer w;
write_thread_.EnterUnbatched(&w, &mutex_); write_thread_.EnterUnbatched(&w, &mutex_);
s = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(), &edit, s = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(), &edit,
&mutex_); &mutex_, directories_.GetDbDir());
write_thread_.ExitUnbatched(&w); write_thread_.ExitUnbatched(&w);
} }
if (s.ok()) { if (s.ok()) {

@ -988,7 +988,7 @@ Status DBImpl::IncreaseFullHistoryTsLowImpl(ColumnFamilyData* cfd,
} }
Status s = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(), Status s = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
&edit, &mutex_); &edit, &mutex_, directories_.GetDbDir());
if (!s.ok()) { if (!s.ok()) {
return s; return s;
} }

@ -2163,7 +2163,8 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
VersionEdit wal_deletion; VersionEdit wal_deletion;
wal_deletion.DeleteWalsBefore(min_wal_number_to_keep); wal_deletion.DeleteWalsBefore(min_wal_number_to_keep);
s = versions_->LogAndApplyToDefaultColumnFamily(&wal_deletion, &mutex_); s = versions_->LogAndApplyToDefaultColumnFamily(&wal_deletion, &mutex_,
directories_.GetDbDir());
if (!s.ok() && versions_->io_status().IsIOError()) { if (!s.ok() && versions_->io_status().IsIOError()) {
s = error_handler_.SetBGError(versions_->io_status(), s = error_handler_.SetBGError(versions_->io_status(),
BackgroundErrorReason::kManifestWrite); BackgroundErrorReason::kManifestWrite);

@ -122,7 +122,11 @@ Status UpdateManifestForFilesState(
} }
if (s.ok() && edit.NumEntries() > 0) { if (s.ok() && edit.NumEntries() > 0) {
s = w.LogAndApply(cfd, &edit); std::unique_ptr<FSDirectory> db_dir;
s = fs->NewDirectory(db_name, IOOptions(), &db_dir, nullptr);
if (s.ok()) {
s = w.LogAndApply(cfd, &edit, db_dir.get());
}
if (s.ok()) { if (s.ok()) {
++cfs_updated; ++cfs_updated;
} }

@ -162,9 +162,13 @@ class Repairer {
edit.AddColumnFamily(cf_name); edit.AddColumnFamily(cf_name);
mutex_.Lock(); mutex_.Lock();
Status status = vset_.LogAndApply(cfd, mut_cf_opts, &edit, &mutex_, std::unique_ptr<FSDirectory> db_dir;
nullptr /* db_directory */, Status status = env_->GetFileSystem()->NewDirectory(dbname_, IOOptions(),
false /* new_descriptor_log */, cf_opts); &db_dir, nullptr);
if (status.ok()) {
status = vset_.LogAndApply(cfd, mut_cf_opts, &edit, &mutex_, db_dir.get(),
false /* new_descriptor_log */, cf_opts);
}
mutex_.Unlock(); mutex_.Unlock();
return status; return status;
} }
@ -656,9 +660,14 @@ class Repairer {
assert(next_file_number_ > 0); assert(next_file_number_ > 0);
vset_.MarkFileNumberUsed(next_file_number_ - 1); vset_.MarkFileNumberUsed(next_file_number_ - 1);
mutex_.Lock(); mutex_.Lock();
Status status = vset_.LogAndApply( std::unique_ptr<FSDirectory> db_dir;
cfd, *cfd->GetLatestMutableCFOptions(), &edit, &mutex_, Status status = env_->GetFileSystem()->NewDirectory(dbname_, IOOptions(),
nullptr /* db_directory */, false /* new_descriptor_log */); &db_dir, nullptr);
if (status.ok()) {
status = vset_.LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
&edit, &mutex_, db_dir.get(),
false /* new_descriptor_log */);
}
mutex_.Unlock(); mutex_.Unlock();
if (!status.ok()) { if (!status.ok()) {
return status; return status;

@ -4562,7 +4562,7 @@ void VersionSet::AppendVersion(ColumnFamilyData* column_family_data,
Status VersionSet::ProcessManifestWrites( Status VersionSet::ProcessManifestWrites(
std::deque<ManifestWriter>& writers, InstrumentedMutex* mu, std::deque<ManifestWriter>& writers, InstrumentedMutex* mu,
FSDirectory* db_directory, bool new_descriptor_log, FSDirectory* dir_contains_current_file, bool new_descriptor_log,
const ColumnFamilyOptions* new_cf_options) { const ColumnFamilyOptions* new_cf_options) {
mu->AssertHeld(); mu->AssertHeld();
assert(!writers.empty()); assert(!writers.empty());
@ -4893,7 +4893,7 @@ Status VersionSet::ProcessManifestWrites(
} }
if (s.ok() && new_descriptor_log) { if (s.ok() && new_descriptor_log) {
io_s = SetCurrentFile(fs_.get(), dbname_, pending_manifest_file_number_, io_s = SetCurrentFile(fs_.get(), dbname_, pending_manifest_file_number_,
db_directory); dir_contains_current_file);
if (!io_s.ok()) { if (!io_s.ok()) {
s = io_s; s = io_s;
} }
@ -5120,8 +5120,8 @@ Status VersionSet::LogAndApply(
const autovector<ColumnFamilyData*>& column_family_datas, const autovector<ColumnFamilyData*>& column_family_datas,
const autovector<const MutableCFOptions*>& mutable_cf_options_list, const autovector<const MutableCFOptions*>& mutable_cf_options_list,
const autovector<autovector<VersionEdit*>>& edit_lists, const autovector<autovector<VersionEdit*>>& edit_lists,
InstrumentedMutex* mu, FSDirectory* db_directory, bool new_descriptor_log, InstrumentedMutex* mu, FSDirectory* dir_contains_current_file,
const ColumnFamilyOptions* new_cf_options, bool new_descriptor_log, const ColumnFamilyOptions* new_cf_options,
const std::vector<std::function<void(const Status&)>>& manifest_wcbs) { const std::vector<std::function<void(const Status&)>>& manifest_wcbs) {
mu->AssertHeld(); mu->AssertHeld();
int num_edits = 0; int num_edits = 0;
@ -5195,9 +5195,8 @@ Status VersionSet::LogAndApply(
} }
return Status::ColumnFamilyDropped(); return Status::ColumnFamilyDropped();
} }
return ProcessManifestWrites(writers, mu, dir_contains_current_file,
return ProcessManifestWrites(writers, mu, db_directory, new_descriptor_log, new_descriptor_log, new_cf_options);
new_cf_options);
} }
void VersionSet::LogAndApplyCFHelper(VersionEdit* edit, void VersionSet::LogAndApplyCFHelper(VersionEdit* edit,

@ -1099,13 +1099,14 @@ class VersionSet {
Status LogAndApplyToDefaultColumnFamily( Status LogAndApplyToDefaultColumnFamily(
VersionEdit* edit, InstrumentedMutex* mu, VersionEdit* edit, InstrumentedMutex* mu,
FSDirectory* db_directory = nullptr, bool new_descriptor_log = false, FSDirectory* dir_contains_current_file, bool new_descriptor_log = false,
const ColumnFamilyOptions* column_family_options = nullptr) { const ColumnFamilyOptions* column_family_options = nullptr) {
ColumnFamilyData* default_cf = GetColumnFamilySet()->GetDefault(); ColumnFamilyData* default_cf = GetColumnFamilySet()->GetDefault();
const MutableCFOptions* cf_options = const MutableCFOptions* cf_options =
default_cf->GetLatestMutableCFOptions(); default_cf->GetLatestMutableCFOptions();
return LogAndApply(default_cf, *cf_options, edit, mu, db_directory, return LogAndApply(default_cf, *cf_options, edit, mu,
new_descriptor_log, column_family_options); dir_contains_current_file, new_descriptor_log,
column_family_options);
} }
// Apply *edit to the current version to form a new descriptor that // Apply *edit to the current version to form a new descriptor that
@ -1117,7 +1118,7 @@ class VersionSet {
Status LogAndApply( Status LogAndApply(
ColumnFamilyData* column_family_data, ColumnFamilyData* column_family_data,
const MutableCFOptions& mutable_cf_options, VersionEdit* edit, const MutableCFOptions& mutable_cf_options, VersionEdit* edit,
InstrumentedMutex* mu, FSDirectory* db_directory = nullptr, InstrumentedMutex* mu, FSDirectory* dir_contains_current_file,
bool new_descriptor_log = false, bool new_descriptor_log = false,
const ColumnFamilyOptions* column_family_options = nullptr) { const ColumnFamilyOptions* column_family_options = nullptr) {
autovector<ColumnFamilyData*> cfds; autovector<ColumnFamilyData*> cfds;
@ -1129,7 +1130,8 @@ class VersionSet {
edit_list.emplace_back(edit); edit_list.emplace_back(edit);
edit_lists.emplace_back(edit_list); edit_lists.emplace_back(edit_list);
return LogAndApply(cfds, mutable_cf_options_list, edit_lists, mu, return LogAndApply(cfds, mutable_cf_options_list, edit_lists, mu,
db_directory, new_descriptor_log, column_family_options); dir_contains_current_file, new_descriptor_log,
column_family_options);
} }
// The batch version. If edit_list.size() > 1, caller must ensure that // The batch version. If edit_list.size() > 1, caller must ensure that
// no edit in the list column family add or drop // no edit in the list column family add or drop
@ -1137,7 +1139,7 @@ class VersionSet {
ColumnFamilyData* column_family_data, ColumnFamilyData* column_family_data,
const MutableCFOptions& mutable_cf_options, const MutableCFOptions& mutable_cf_options,
const autovector<VersionEdit*>& edit_list, InstrumentedMutex* mu, const autovector<VersionEdit*>& edit_list, InstrumentedMutex* mu,
FSDirectory* db_directory = nullptr, bool new_descriptor_log = false, FSDirectory* dir_contains_current_file, bool new_descriptor_log = false,
const ColumnFamilyOptions* column_family_options = nullptr, const ColumnFamilyOptions* column_family_options = nullptr,
const std::function<void(const Status&)>& manifest_wcb = {}) { const std::function<void(const Status&)>& manifest_wcb = {}) {
autovector<ColumnFamilyData*> cfds; autovector<ColumnFamilyData*> cfds;
@ -1147,8 +1149,8 @@ class VersionSet {
autovector<autovector<VersionEdit*>> edit_lists; autovector<autovector<VersionEdit*>> edit_lists;
edit_lists.emplace_back(edit_list); edit_lists.emplace_back(edit_list);
return LogAndApply(cfds, mutable_cf_options_list, edit_lists, mu, return LogAndApply(cfds, mutable_cf_options_list, edit_lists, mu,
db_directory, new_descriptor_log, column_family_options, dir_contains_current_file, new_descriptor_log,
{manifest_wcb}); column_family_options, {manifest_wcb});
} }
// The across-multi-cf batch version. If edit_lists contain more than // The across-multi-cf batch version. If edit_lists contain more than
@ -1158,7 +1160,7 @@ class VersionSet {
const autovector<ColumnFamilyData*>& cfds, const autovector<ColumnFamilyData*>& cfds,
const autovector<const MutableCFOptions*>& mutable_cf_options_list, const autovector<const MutableCFOptions*>& mutable_cf_options_list,
const autovector<autovector<VersionEdit*>>& edit_lists, const autovector<autovector<VersionEdit*>>& edit_lists,
InstrumentedMutex* mu, FSDirectory* db_directory = nullptr, InstrumentedMutex* mu, FSDirectory* dir_contains_current_file,
bool new_descriptor_log = false, bool new_descriptor_log = false,
const ColumnFamilyOptions* new_cf_options = nullptr, const ColumnFamilyOptions* new_cf_options = nullptr,
const std::vector<std::function<void(const Status&)>>& manifest_wcbs = const std::vector<std::function<void(const Status&)>>& manifest_wcbs =
@ -1574,7 +1576,8 @@ class VersionSet {
private: private:
// REQUIRES db mutex at beginning. may release and re-acquire db mutex // REQUIRES db mutex at beginning. may release and re-acquire db mutex
Status ProcessManifestWrites(std::deque<ManifestWriter>& writers, Status ProcessManifestWrites(std::deque<ManifestWriter>& writers,
InstrumentedMutex* mu, FSDirectory* db_directory, InstrumentedMutex* mu,
FSDirectory* dir_contains_current_file,
bool new_descriptor_log, bool new_descriptor_log,
const ColumnFamilyOptions* new_cf_options); const ColumnFamilyOptions* new_cf_options);
@ -1636,7 +1639,7 @@ class ReactiveVersionSet : public VersionSet {
const autovector<ColumnFamilyData*>& /*cfds*/, const autovector<ColumnFamilyData*>& /*cfds*/,
const autovector<const MutableCFOptions*>& /*mutable_cf_options_list*/, const autovector<const MutableCFOptions*>& /*mutable_cf_options_list*/,
const autovector<autovector<VersionEdit*>>& /*edit_lists*/, const autovector<autovector<VersionEdit*>>& /*edit_lists*/,
InstrumentedMutex* /*mu*/, FSDirectory* /*db_directory*/, InstrumentedMutex* /*mu*/, FSDirectory* /*dir_contains_current_file*/,
bool /*new_descriptor_log*/, const ColumnFamilyOptions* /*new_cf_option*/, bool /*new_descriptor_log*/, const ColumnFamilyOptions* /*new_cf_option*/,
const std::vector<std::function<void(const Status&)>>& /*manifest_wcbs*/) const std::vector<std::function<void(const Status&)>>& /*manifest_wcbs*/)
override { override {

@ -1272,7 +1272,7 @@ class VersionSetTestBase {
mutex_.Lock(); mutex_.Lock();
Status s = Status s =
versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(), versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(),
mutable_cf_options_, &edit, &mutex_); mutable_cf_options_, &edit, &mutex_, nullptr);
mutex_.Unlock(); mutex_.Unlock();
return s; return s;
} }
@ -1286,7 +1286,7 @@ class VersionSetTestBase {
mutex_.Lock(); mutex_.Lock();
Status s = Status s =
versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(), versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(),
mutable_cf_options_, vedits, &mutex_); mutable_cf_options_, vedits, &mutex_, nullptr);
mutex_.Unlock(); mutex_.Unlock();
return s; return s;
} }
@ -1384,8 +1384,8 @@ TEST_F(VersionSetTest, SameColumnFamilyGroupCommit) {
}); });
SyncPoint::GetInstance()->EnableProcessing(); SyncPoint::GetInstance()->EnableProcessing();
mutex_.Lock(); mutex_.Lock();
Status s = Status s = versions_->LogAndApply(cfds, all_mutable_cf_options, edit_lists,
versions_->LogAndApply(cfds, all_mutable_cf_options, edit_lists, &mutex_); &mutex_, nullptr);
mutex_.Unlock(); mutex_.Unlock();
EXPECT_OK(s); EXPECT_OK(s);
EXPECT_EQ(kGroupSize - 1, count); EXPECT_EQ(kGroupSize - 1, count);
@ -1587,7 +1587,7 @@ TEST_F(VersionSetTest, ObsoleteBlobFile) {
mutex_.Lock(); mutex_.Lock();
Status s = Status s =
versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(), versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(),
mutable_cf_options_, &edit, &mutex_); mutable_cf_options_, &edit, &mutex_, nullptr);
mutex_.Unlock(); mutex_.Unlock();
ASSERT_OK(s); ASSERT_OK(s);
@ -2194,7 +2194,7 @@ class VersionSetWithTimestampTest : public VersionSetTest {
Status s; Status s;
mutex_.Lock(); mutex_.Lock();
s = versions_->LogAndApply(cfd_, *(cfd_->GetLatestMutableCFOptions()), s = versions_->LogAndApply(cfd_, *(cfd_->GetLatestMutableCFOptions()),
edits_, &mutex_); edits_, &mutex_, nullptr);
mutex_.Unlock(); mutex_.Unlock();
ASSERT_OK(s); ASSERT_OK(s);
VerifyFullHistoryTsLow(*std::max_element(ts_lbs.begin(), ts_lbs.end())); VerifyFullHistoryTsLow(*std::max_element(ts_lbs.begin(), ts_lbs.end()));
@ -2661,7 +2661,7 @@ TEST_P(VersionSetTestDropOneCF, HandleDroppedColumnFamilyInAtomicGroup) {
mutex_.Lock(); mutex_.Lock();
s = versions_->LogAndApply(cfd_to_drop, s = versions_->LogAndApply(cfd_to_drop,
*cfd_to_drop->GetLatestMutableCFOptions(), *cfd_to_drop->GetLatestMutableCFOptions(),
&drop_cf_edit, &mutex_); &drop_cf_edit, &mutex_, nullptr);
mutex_.Unlock(); mutex_.Unlock();
ASSERT_OK(s); ASSERT_OK(s);
@ -2710,8 +2710,8 @@ TEST_P(VersionSetTestDropOneCF, HandleDroppedColumnFamilyInAtomicGroup) {
}); });
SyncPoint::GetInstance()->EnableProcessing(); SyncPoint::GetInstance()->EnableProcessing();
mutex_.Lock(); mutex_.Lock();
s = versions_->LogAndApply(cfds, mutable_cf_options_list, edit_lists, s = versions_->LogAndApply(cfds, mutable_cf_options_list, edit_lists, &mutex_,
&mutex_); nullptr);
mutex_.Unlock(); mutex_.Unlock();
ASSERT_OK(s); ASSERT_OK(s);
ASSERT_EQ(1, called); ASSERT_EQ(1, called);

@ -31,12 +31,13 @@ class OfflineManifestWriter {
return versions_.Recover(column_families); return versions_.Recover(column_families);
} }
Status LogAndApply(ColumnFamilyData* cfd, VersionEdit* edit) { Status LogAndApply(ColumnFamilyData* cfd, VersionEdit* edit,
FSDirectory* dir_contains_current_file) {
// Use `mutex` to imitate a locked DB mutex when calling `LogAndApply()`. // Use `mutex` to imitate a locked DB mutex when calling `LogAndApply()`.
InstrumentedMutex mutex; InstrumentedMutex mutex;
mutex.Lock(); mutex.Lock();
Status s = versions_.LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(), Status s = versions_.LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
edit, &mutex, nullptr /* db_directory */, edit, &mutex, dir_contains_current_file,
false /* new_descriptor_log */); false /* new_descriptor_log */);
mutex.Unlock(); mutex.Unlock();
return s; return s;

@ -388,7 +388,7 @@ bool ParseFileName(const std::string& fname, uint64_t* number,
IOStatus SetCurrentFile(FileSystem* fs, const std::string& dbname, IOStatus SetCurrentFile(FileSystem* fs, const std::string& dbname,
uint64_t descriptor_number, uint64_t descriptor_number,
FSDirectory* directory_to_fsync) { FSDirectory* dir_contains_current_file) {
// Remove leading "dbname/" and add newline to manifest file name // Remove leading "dbname/" and add newline to manifest file name
std::string manifest = DescriptorFileName(dbname, descriptor_number); std::string manifest = DescriptorFileName(dbname, descriptor_number);
Slice contents = manifest; Slice contents = manifest;
@ -404,8 +404,8 @@ IOStatus SetCurrentFile(FileSystem* fs, const std::string& dbname,
TEST_SYNC_POINT_CALLBACK("SetCurrentFile:AfterRename", &s); TEST_SYNC_POINT_CALLBACK("SetCurrentFile:AfterRename", &s);
} }
if (s.ok()) { if (s.ok()) {
if (directory_to_fsync != nullptr) { if (dir_contains_current_file != nullptr) {
s = directory_to_fsync->FsyncWithDirOptions( s = dir_contains_current_file->FsyncWithDirOptions(
IOOptions(), nullptr, DirFsyncOptions(CurrentFileName(dbname))); IOOptions(), nullptr, DirFsyncOptions(CurrentFileName(dbname)));
} }
} else { } else {

@ -160,10 +160,12 @@ extern bool ParseFileName(const std::string& filename, uint64_t* number,
FileType* type, WalFileType* log_type = nullptr); FileType* type, WalFileType* log_type = nullptr);
// Make the CURRENT file point to the descriptor file with the // Make the CURRENT file point to the descriptor file with the
// specified number. // specified number. On its success and when dir_contains_current_file is not
// nullptr, the function will fsync the directory containing the CURRENT file
// when
extern IOStatus SetCurrentFile(FileSystem* fs, const std::string& dbname, extern IOStatus SetCurrentFile(FileSystem* fs, const std::string& dbname,
uint64_t descriptor_number, uint64_t descriptor_number,
FSDirectory* directory_to_fsync); FSDirectory* dir_contains_current_file);
// Make the IDENTITY file for the db // Make the IDENTITY file for the db
extern Status SetIdentityFile(Env* env, const std::string& dbname, extern Status SetIdentityFile(Env* env, const std::string& dbname,

@ -4196,7 +4196,12 @@ void UnsafeRemoveSstFileCommand::DoCommand() {
VersionEdit edit; VersionEdit edit;
edit.SetColumnFamily(cfd->GetID()); edit.SetColumnFamily(cfd->GetID());
edit.DeleteFile(level, sst_file_number_); edit.DeleteFile(level, sst_file_number_);
s = w.LogAndApply(cfd, &edit); std::unique_ptr<FSDirectory> db_dir;
s = options_.env->GetFileSystem()->NewDirectory(db_path_, IOOptions(),
&db_dir, nullptr);
if (s.ok()) {
s = w.LogAndApply(cfd, &edit, db_dir.get());
}
} }
if (!s.ok()) { if (!s.ok()) {

Loading…
Cancel
Save