diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 2041aeac9870..a318c11b2ff2 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -5928,12 +5928,54 @@ Status DBImpl::IngestExternalFiles( } // Run ingestion jobs. if (status.ok()) { + if (allow_write) { + // Briefly stop writes while reserving sequence numbers for ingestion. + write_thread_.EnterUnbatched(&w, &mutex_); + if (two_write_queues_) { + nonmem_write_thread_.EnterUnbatched(&nonmem_w, &mutex_); + } + WaitForPendingWrites(); + } + + SequenceNumber last_seqno = versions_->LastSequence(); + SequenceNumber reserved_seqno_count = 0; + if (allow_write) { + // Each file consumes at most one sequence number. Jobs for different + // column families share the same sequence range, so reserve the maximum + // file count. Unused sequence numbers are harmless gaps. + for (size_t i = 0; i != num_cfs; ++i) { + reserved_seqno_count = + std::max(reserved_seqno_count, + static_cast( + ingestion_jobs[i].files_to_ingest().size())); + } + assert(reserved_seqno_count > 0); + const SequenceNumber reserved_last_seqno = + last_seqno + reserved_seqno_count; + versions_->SetLastAllocatedSequence(reserved_last_seqno); + versions_->SetLastPublishedSequence(reserved_last_seqno); + versions_->SetLastSequence(reserved_last_seqno); + // Resume writes + if (two_write_queues_) { + nonmem_write_thread_.ExitUnbatched(&nonmem_w); + } + write_thread_.ExitUnbatched(&w); + + // The reservation cannot be rolled back if ingestion fails because a + // foreground write may have already consumed a later sequence number. + TEST_SYNC_POINT("DBImpl::IngestExternalFiles:AfterReserveSeqno"); + } + for (size_t i = 0; i != num_cfs; ++i) { mutex_.AssertHeld(); - status = ingestion_jobs[i].Run(); + status = ingestion_jobs[i].Run(last_seqno); if (!status.ok()) { break; } + assert(!allow_write || + static_cast( + ingestion_jobs[i].ConsumedSequenceNumbersCount()) <= + reserved_seqno_count); ingestion_jobs[i].RegisterRange(); } } @@ -5964,29 +6006,37 @@ Status DBImpl::IngestExternalFiles( } assert(0 == num_entries); } + // With allow_write, a concurrent flush may persist a higher last sequence + // before this ingestion edit is applied. LogAndApply updates the edit as + // needed to keep VersionEdit::last_sequence non-decreasing in the + // MANIFEST. status = versions_->LogAndApply(cfds_to_commit, mutable_cf_options_list, read_options, edit_lists, &mutex_, directories_.GetDbDir()); - // It is safe to update VersionSet last seqno here after LogAndApply since - // LogAndApply persists last sequence number from VersionEdits, - // which are from file's largest seqno and not from VersionSet. - // - // It is necessary to update last seqno here since LogAndApply releases - // mutex when persisting MANIFEST file, and the snapshots taken during - // that period will not be stable if VersionSet last seqno is updated - // before LogAndApply. - int consumed_seqno_count = - ingestion_jobs[0].ConsumedSequenceNumbersCount(); - for (size_t i = 1; i != num_cfs; ++i) { - consumed_seqno_count = - std::max(consumed_seqno_count, - ingestion_jobs[i].ConsumedSequenceNumbersCount()); - } - if (consumed_seqno_count > 0) { - const SequenceNumber last_seqno = versions_->LastSequence(); - versions_->SetLastAllocatedSequence(last_seqno + consumed_seqno_count); - versions_->SetLastPublishedSequence(last_seqno + consumed_seqno_count); - versions_->SetLastSequence(last_seqno + consumed_seqno_count); + if (!allow_write) { + // It is safe to update VersionSet last seqno here after LogAndApply + // since LogAndApply persists last sequence number from VersionEdits, + // which are from file's largest seqno and not from VersionSet. + // + // It is necessary to update last seqno here since LogAndApply releases + // mutex when persisting MANIFEST file, and the snapshots taken during + // that period will not be stable if VersionSet last seqno is updated + // before LogAndApply. + int consumed_seqno_count = + ingestion_jobs[0].ConsumedSequenceNumbersCount(); + for (size_t i = 1; i != num_cfs; ++i) { + consumed_seqno_count = + std::max(consumed_seqno_count, + ingestion_jobs[i].ConsumedSequenceNumbersCount()); + } + if (consumed_seqno_count > 0) { + const SequenceNumber last_seqno = versions_->LastSequence(); + versions_->SetLastAllocatedSequence(last_seqno + + consumed_seqno_count); + versions_->SetLastPublishedSequence(last_seqno + + consumed_seqno_count); + versions_->SetLastSequence(last_seqno + consumed_seqno_count); + } } } diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h index c75c8c33a774..fed4cef867c1 100644 --- a/db/db_impl/db_impl.h +++ b/db/db_impl/db_impl.h @@ -2045,9 +2045,10 @@ class DBImpl : public DB { return; } - // Wait for the ones who already wrote to the WAL to finish their - // memtable write. + // Wait for writers that have allocated sequence numbers to finish their + // memtable writes and publish their sequences. if (pending_memtable_writes_.load() != 0) { + TEST_SYNC_POINT("DBImpl::WaitForPendingWrites:PendingWrites"); std::unique_lock guard(switch_mutex_); switch_cv_.wait(guard, [&] { return pending_memtable_writes_.load() == 0; }); @@ -2719,7 +2720,7 @@ class DBImpl : public DB { std::condition_variable switch_cv_; // The mutex used by switch_cv_. mutex_ should be acquired beforehand. std::mutex switch_mutex_; - // Number of threads intending to write to memtable + // Number of writers with pending memtable writes or sequence publication. std::atomic pending_memtable_writes_ = {}; // A flag indicating whether the current rocksdb database has any diff --git a/db/db_impl/db_impl_write.cc b/db/db_impl/db_impl_write.cc index c74ec8dba933..762462a8b425 100644 --- a/db/db_impl/db_impl_write.cc +++ b/db/db_impl/db_impl_write.cc @@ -366,13 +366,14 @@ Status DBImpl::MultiBatchWriteImpl(const WriteOptions& write_options, const ReadOptions read_options; writer.status = ApplyWALToManifest(read_options, &synced_wals); } - if (writer.status.ok()) { - pending_memtable_writes_ += memtable_write_cnt; - } else { + if (!writer.status.ok()) { // The `pending_wb_cnt` must be reset to avoid other writers helping // the front writer write its WBs after it failed to write the WAL. writer.ResetPendingWBCnt(); } + // Every writer in the commit queue calls MultiBatchWriteCommit, including + // when the write fails before its memtable write. + pending_memtable_writes_ += memtable_write_cnt; write_thread_.ExitAsBatchGroupLeader(wal_write_group, writer.status); } diff --git a/db/external_sst_file_ingestion_job.cc b/db/external_sst_file_ingestion_job.cc index 3487de8b3f95..2abbf4f09228 100644 --- a/db/external_sst_file_ingestion_job.cc +++ b/db/external_sst_file_ingestion_job.cc @@ -372,9 +372,7 @@ Status ExternalSstFileIngestionJob::NeedsFlush(bool* flush_needed, return status; } -// REQUIRES: we have become the only writer by entering both write_thread_ and -// nonmem_write_thread_ -Status ExternalSstFileIngestionJob::Run() { +Status ExternalSstFileIngestionJob::Run(SequenceNumber last_seqno) { Status status; SuperVersion* super_version = cfd_->GetSuperVersion(); #ifndef NDEBUG @@ -398,9 +396,6 @@ Status ExternalSstFileIngestionJob::Run() { // if the don't overlap with any ranges since we have snapshots force_global_seqno = true; } - // It is safe to use this instead of LastAllocatedSequence since we are - // the only active writer, and hence they are equal - SequenceNumber last_seqno = versions_->LastSequence(); edit_.SetColumnFamily(cfd_->GetID()); // The levels that the files will be ingested into diff --git a/db/external_sst_file_ingestion_job.h b/db/external_sst_file_ingestion_job.h index 49bb1e31e59f..d915ed25d566 100644 --- a/db/external_sst_file_ingestion_job.h +++ b/db/external_sst_file_ingestion_job.h @@ -132,7 +132,7 @@ class ExternalSstFileIngestionJob { // Will execute the ingestion job and prepare edit() to be applied. // REQUIRES: Mutex held - Status Run(); + Status Run(SequenceNumber last_seqno); // Register key range involved in this ingestion job // to prevent key range conflict with other ongoing compaction/file ingestion diff --git a/db/external_sst_file_test.cc b/db/external_sst_file_test.cc index f86af51e2a41..411dc678c325 100644 --- a/db/external_sst_file_test.cc +++ b/db/external_sst_file_test.cc @@ -2394,28 +2394,272 @@ TEST_P(ExternalSSTFileTest, IngestBehind) { TEST_P(ExternalSSTFileTest, WriteDuringIngest) { SyncPoint::GetInstance()->DisableProcessing(); - // Set callback to simulate concurrent write during ingestion - SyncPoint::GetInstance()->SetCallBack( - "DBImpl::IngestExternalFile:AfterAllowWriteCheck", [&](void*) { - // Write a non-overlapping key - ASSERT_OK(Put("foo", "v1")); - }); - Options options = CurrentOptions(); + options.enable_multi_batch_write = true; DestroyAndReopen(options); + std::vector external_files; + for (const std::string& key : {"foo1", "foo2"}) { + std::string file_path = sst_files_dir_ + env_->GenerateUniqueId(); + SstFileWriter writer(EnvOptions(), options); + ASSERT_OK(writer.Open(file_path)); + ASSERT_OK(writer.Put(key, "v1")); + ASSERT_OK(writer.Finish()); + external_files.push_back(std::move(file_path)); + } + + const SequenceNumber last_seqno = db_->GetLatestSequenceNumber(); + const Snapshot* snapshot = db_->GetSnapshot(); + std::vector assigned_seqnos; + Status write_status; + std::unique_ptr write_thread; + + // Start a foreground write after the ingestion sequence numbers have been + // reserved. The callback runs with the DB mutex held, so run Put in another + // thread and join it after ingestion to avoid a mutex deadlock. + SyncPoint::GetInstance()->SetCallBack( + "DBImpl::IngestExternalFiles:AfterReserveSeqno", [&](void*) { + ASSERT_EQ(last_seqno + external_files.size(), + db_->GetLatestSequenceNumber()); + write_thread = std::make_unique( + [&] { write_status = Put("bar", "v1"); }); + }); + SyncPoint::GetInstance()->SetCallBack( + "ExternalSstFileIngestionJob::Run", [&](void* arg) { + ASSERT_NE(arg, nullptr); + assigned_seqnos.push_back(*static_cast(arg)); + }); + SyncPoint::GetInstance()->EnableProcessing(); - ASSERT_OK(GenerateAndAddExternalFile(options, {{"foo", "v1"}}, -1, true, - false, true, false, false, - true /* allow_write */)); - ASSERT_OK(Put("bar", "v1")); - ASSERT_EQ(Get("foo"), "v1"); + IngestExternalFileOptions ifo; + ifo.allow_global_seqno = true; + ifo.write_global_seqno = std::get<0>(GetParam()); + ifo.verify_checksums_before_ingest = std::get<1>(GetParam()); + ifo.allow_write = true; + Status ingest_status = db_->IngestExternalFile(external_files, ifo); + + ASSERT_NE(write_thread, nullptr); + write_thread->join(); + ASSERT_OK(ingest_status); + ASSERT_OK(write_status); + + ASSERT_EQ((std::vector{last_seqno + 1, last_seqno + 2}), + assigned_seqnos); + ASSERT_EQ(last_seqno + 3, db_->GetLatestSequenceNumber()); + ASSERT_EQ(Get("foo1"), "v1"); + ASSERT_EQ(Get("foo2"), "v1"); ASSERT_EQ(Get("bar"), "v1"); + ReadOptions read_options; + read_options.snapshot = snapshot; + std::string value; + ASSERT_TRUE(db_->Get(read_options, "foo1", &value).IsNotFound()); + ASSERT_TRUE(db_->Get(read_options, "foo2", &value).IsNotFound()); + db_->ReleaseSnapshot(snapshot); + SyncPoint::GetInstance()->DisableProcessing(); SyncPoint::GetInstance()->ClearAllCallBacks(); } +TEST_P(ExternalSSTFileTest, AllowWriteIngestWaitsForActiveWriter) { + constexpr auto kReleaseWriter = + "ExternalSSTFileTest::AllowWriteIngestWaitsForActiveWriter:" + "ReleaseWriter"; + constexpr auto kWriterPrepared = + "ExternalSSTFileTest::AllowWriteIngestWaitsForActiveWriter:" + "WriterPrepared"; + constexpr auto kStartIngest = + "ExternalSSTFileTest::AllowWriteIngestWaitsForActiveWriter:" + "StartIngest"; + constexpr auto kContinueWriter = + "ExternalSSTFileTest::AllowWriteIngestWaitsForActiveWriter:" + "ContinueWriter"; + + auto* sync_point = SyncPoint::GetInstance(); + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); + sync_point->LoadDependency( + {{kWriterPrepared, kStartIngest}, {kReleaseWriter, kContinueWriter}}); + + sync_point->SetCallBack("DBImpl::WriteImpl:BeforeLeaderEnters", [&](void*) { + TEST_SYNC_POINT(kWriterPrepared); + TEST_SYNC_POINT(kContinueWriter); + }); + + const auto release_writer = [&](void*) { TEST_SYNC_POINT(kReleaseWriter); }; + sync_point->SetCallBack("WriteThread::EnterUnbatched:Wait", release_writer); + + SequenceNumber assigned_seqno = 0; + sync_point->SetCallBack("ExternalSstFileIngestionJob::Run", [&](void* arg) { + // Release the writer here as a fallback so that a failure to wait in + // EnterUnbatched causes an assertion failure instead of a deadlock. + release_writer(nullptr); + assigned_seqno = *static_cast(arg); + }); + + Options options = CurrentOptions(); + options.enable_multi_batch_write = false; + DestroyAndReopen(options); + + const SequenceNumber last_seqno = db_->GetLatestSequenceNumber(); + const Snapshot* snapshot = db_->GetSnapshot(); + Status write_status; + + sync_point->EnableProcessing(); + port::Thread writer([&]() { write_status = Put("bar", "v1"); }); + + TEST_SYNC_POINT(kStartIngest); + ASSERT_OK(GenerateAndAddExternalFile( + options, {{"foo", "v"}}, -1, true, std::get<0>(GetParam()), + std::get<1>(GetParam()), false, false, true /* allow_write */)); + writer.join(); + + ASSERT_OK(write_status); + ASSERT_EQ(last_seqno + 2, assigned_seqno); + ASSERT_EQ(last_seqno + 2, db_->GetLatestSequenceNumber()); + ASSERT_EQ("v1", Get("bar")); + ASSERT_EQ("v", Get("foo")); + db_->ReleaseSnapshot(snapshot); + + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); +} + +TEST_F(ExternalSSTFileTest, AllowWriteIngestWaitsForPendingMultiBatchWrite) { + constexpr auto kStartIngest = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForPendingMultiBatchWrite:StartIngest"; + constexpr auto kReleaseWriter = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForPendingMultiBatchWrite:ReleaseWriter"; + + auto* sync_point = SyncPoint::GetInstance(); + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); + sync_point->LoadDependency( + {{"DBImpl::WriteImpl:CommitAfterWriteWAL", kStartIngest}, + {kReleaseWriter, "DBImpl::WriteImpl:BeforePipelineWriteMemtable"}}); + + const auto release_writer = [&](void*) { TEST_SYNC_POINT(kReleaseWriter); }; + sync_point->SetCallBack("DBImpl::WaitForPendingWrites:BeforeBlock", + release_writer); + + SequenceNumber assigned_seqno = 0; + sync_point->SetCallBack("ExternalSstFileIngestionJob::Run", [&](void* arg) { + assigned_seqno = *static_cast(arg); + }); + + Options options = CurrentOptions(); + options.enable_pipelined_write = false; + options.unordered_write = false; + options.enable_multi_batch_write = true; + DestroyAndReopen(options); + + const SequenceNumber last_seqno = db_->GetLatestSequenceNumber(); + // Force the ingested file to consume a global sequence number. + const Snapshot* snapshot = db_->GetSnapshot(); + Status write_status; + + sync_point->EnableProcessing(); + port::Thread writer([&]() { write_status = Put("bar", "v1"); }); + + // The writer has allocated a sequence and released write_thread_, but has + // not inserted into the memtable or published the sequence yet. + TEST_SYNC_POINT(kStartIngest); + Status ingest_status = + GenerateAndAddExternalFile(options, {{"foo", "v"}}, -1, true, false, true, + false, false, true /* allow_write */); + + // If ingestion does not wait for the pending writer, release it here so the + // sequence assertions fail instead of hanging in writer.join(). + release_writer(nullptr); + writer.join(); + + ASSERT_OK(ingest_status); + ASSERT_OK(write_status); + ASSERT_EQ(last_seqno + 2, assigned_seqno); + ASSERT_EQ(last_seqno + 2, db_->GetLatestSequenceNumber()); + ASSERT_EQ("v1", Get("bar")); + ASSERT_EQ("v", Get("foo")); + db_->ReleaseSnapshot(snapshot); + + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); +} + +TEST_F(ExternalSSTFileTest, AllowWriteIngestWaitsForFailedMultiBatchWrite) { + constexpr auto kWriterPrepared = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForFailedMultiBatchWrite:WriterPrepared"; + constexpr auto kStartIngest = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForFailedMultiBatchWrite:StartIngest"; + constexpr auto kReleaseWriter = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForFailedMultiBatchWrite:ReleaseWriter"; + constexpr auto kContinueWriter = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForFailedMultiBatchWrite:ContinueWriter"; + + auto* sync_point = SyncPoint::GetInstance(); + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); + sync_point->LoadDependency( + {{kWriterPrepared, kStartIngest}, {kReleaseWriter, kContinueWriter}}); + + sync_point->SetCallBack("DBImpl::WriteImpl:CommitAfterWriteWAL", [&](void*) { + TEST_SYNC_POINT(kWriterPrepared); + TEST_SYNC_POINT(kContinueWriter); + }); + + const auto release_writer = [&](void*) { TEST_SYNC_POINT(kReleaseWriter); }; + sync_point->SetCallBack("DBImpl::WaitForPendingWrites:PendingWrites", + release_writer); + + SequenceNumber assigned_seqno = 0; + sync_point->SetCallBack("ExternalSstFileIngestionJob::Run", [&](void* arg) { + // Release the writer here as a fallback so that a failure to wait for it + // causes the sequence assertions to fail instead of a deadlock. + release_writer(nullptr); + assigned_seqno = *static_cast(arg); + }); + + Options options = CurrentOptions(); + options.enable_pipelined_write = false; + options.unordered_write = false; + options.enable_multi_batch_write = true; + // Keep the DB usable after the injected WAL error so ingestion can proceed. + options.paranoid_checks = false; + DestroyAndReopen(options); + + const SequenceNumber last_seqno = db_->GetLatestSequenceNumber(); + // Force the ingested file to consume a global sequence number. + const Snapshot* snapshot = db_->GetSnapshot(); + Status write_status; + + env_->log_write_error_.store(true, std::memory_order_release); + sync_point->EnableProcessing(); + port::Thread writer([&]() { write_status = Put("bar", "v1"); }); + + TEST_SYNC_POINT(kStartIngest); + env_->log_write_error_.store(false, std::memory_order_release); + Status ingest_status = + GenerateAndAddExternalFile(options, {{"foo", "v"}}, -1, true, false, true, + false, false, true /* allow_write */); + writer.join(); + + ASSERT_OK(ingest_status); + ASSERT_NOK(write_status); + ASSERT_EQ(last_seqno + 2, assigned_seqno); + ASSERT_EQ(last_seqno + 2, db_->GetLatestSequenceNumber()); + ASSERT_EQ("NOT_FOUND", Get("bar")); + ASSERT_EQ("v", Get("foo")); + db_->ReleaseSnapshot(snapshot); + + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); +} + TEST_P(ExternalSSTFileTest, InconsistentAllowWriteArguments) { Options options = CurrentOptions(); CreateAndReopenWithCF({"koko"}, options); diff --git a/include/rocksdb/options.h b/include/rocksdb/options.h index 312c65fc2566..0c35cb9424cd 100644 --- a/include/rocksdb/options.h +++ b/include/rocksdb/options.h @@ -2101,7 +2101,12 @@ struct IngestExternalFileOptions { // ingest_behind takes precedence over fail_if_not_bottommost_level. bool fail_if_not_bottommost_level = false; // Set to TRUE if user wants to allow writes to the DB during ingestion. - // User must ensure no writes overlap with the ingested data. + // User must ensure that concurrent writes do not overlap the ingested key + // ranges. + // Snapshot consistency is not guaranteed for snapshots created during + // ingestion because ingestion sequence numbers are published before the + // ingested files become visible. Such snapshots must not be used to read the + // ingested key ranges. bool allow_write = false; };