diff --git a/db/db_flush_test.cc b/db/db_flush_test.cc index af89883ac0..7fbb89c65d 100644 --- a/db/db_flush_test.cc +++ b/db/db_flush_test.cc @@ -3725,6 +3725,108 @@ TEST_F(DBFlushTest, NonAtomicNormalFlushAbortWhenBGError) { SyncPoint::GetInstance()->DisableProcessing(); } +TEST_F(DBFlushTest, NonRecoveryFlushDoesNotStarveErrorRecovery) { + constexpr uint64_t kRecoveryTimeoutMicros = 15 * 1000 * 1000; + constexpr int kLateWriteCount = 32; + + Options opts = CurrentOptions(); + opts.atomic_flush = true; + opts.memtable_factory.reset(test::NewSpecialSkipListFactory(1)); + opts.max_write_buffer_number = 64; + opts.max_background_flushes = 1; + DestroyAndReopen(opts); + env_->SetBackgroundThreads(1, Env::HIGH); + + std::atomic_int flush_write_table_count{0}; + port::Mutex wait_mutex; + port::CondVar wait_cv(&wait_mutex); + bool recovery_wait_started = false; + bool recovery_succeeded = false; + SyncPoint::GetInstance()->ClearAllCallBacks(); + SyncPoint::GetInstance()->DisableProcessing(); + SyncPoint::GetInstance()->SetCallBack( + "FlushJob::WriteLevel0Table:s", [&](void* s_ptr) { + if (flush_write_table_count.fetch_add(1) == 0) { + Status* s = static_cast(s_ptr); + IOStatus io_error = IOStatus::IOError("injected foobar"); + io_error.SetRetryable(true); + *s = io_error; + TEST_SYNC_POINT( + "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:" + "FirstFlushFailed"); + TEST_SYNC_POINT( + "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:" + "ContinueFirstFlush"); + } + }); + SyncPoint::GetInstance()->SetCallBack( + "DBImpl::ResumeImpl:BeforeWaitForBackgroundWork", [&](void*) { + MutexLock l(&wait_mutex); + recovery_wait_started = true; + wait_cv.SignalAll(); + }); + SyncPoint::GetInstance()->SetCallBack( + "RecoverFromRetryableBGIOError:RecoverSuccess", [&](void*) { + MutexLock l(&wait_mutex); + recovery_succeeded = true; + wait_cv.SignalAll(); + }); + SyncPoint::GetInstance()->LoadDependency( + {{"DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:" + "FirstFlushFailed", + "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:" + "FailureObserved"}, + {"DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:" + "ReleaseFirstFlush", + "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:" + "ContinueFirstFlush"}}); + SyncPoint::GetInstance()->EnableProcessing(); + + auto wait_until = [&](const auto& predicate) { + const uint64_t abs_time = + SystemClock::Default()->NowMicros() + kRecoveryTimeoutMicros; + MutexLock l(&wait_mutex); + while (!predicate()) { + if (wait_cv.TimedWait(abs_time)) { + return false; + } + } + return true; + }; + + ASSERT_OK(Put(Key(1), "val1")); + ASSERT_OK(Put(Key(2), "val2")); + TEST_SYNC_POINT( + "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:" + "FailureObserved"); + TEST_SYNC_POINT( + "DBFlushTest::NonRecoveryFlushDoesNotStarveErrorRecovery:" + "ReleaseFirstFlush"); + + const bool recovery_wait_reached = + wait_until([&] { return recovery_wait_started; }); + Status late_write_status; + if (recovery_wait_reached) { + for (int key = 3; key < 3 + kLateWriteCount && late_write_status.ok(); + ++key) { + late_write_status = Put(Key(key), "val"); + } + } + const bool recovered = wait_until([&] { return recovery_succeeded; }); + + if (!recovered) { + IOStatus cleanup_error = IOStatus::IOError("stop stuck error recovery"); + dbfull()->TEST_SetBGError(cleanup_error, BackgroundErrorReason::kFlush); + dbfull()->TEST_SignalAllBgCv(); + } + + SyncPoint::GetInstance()->ClearAllCallBacks(); + SyncPoint::GetInstance()->DisableProcessing(); + ASSERT_TRUE(recovery_wait_reached); + ASSERT_OK(late_write_status); + ASSERT_TRUE(recovered); +} + TEST_F(DBFlushTest, DBStuckAfterAtomicFlushError) { // Test for a bug with atomic flush where DB can become stuck // after a flush error. A repro timeline: diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 3335d63953..c827ae395d 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -347,6 +347,22 @@ Status DBImpl::ResumeImpl(DBRecoverContext context, const ReadOptions read_options(io_activity); const WriteOptions write_options(io_activity); + assert(static_cast(unscheduled_flushes_) <= flush_queue_.size()); + // Recovery rebuilds flush requests for every column family with pending + // immutable data after scheduled workers exit, so every existing request is + // redundant and safe to remove. A worker can exit on the background error + // without popping a request, leaving it out of unscheduled_flushes_. + while (!flush_queue_.empty()) { + FlushRequest flush_req = PopFirstFromFlushQueue(); + for (const auto& item : flush_req.cfd_to_max_mem_id_to_persist) { + ColumnFamilyData* cfd = item.first; + assert(cfd); + cfd->UnrefAndTryDelete(); + } + } + unscheduled_flushes_ = 0; + + TEST_SYNC_POINT("DBImpl::ResumeImpl:BeforeWaitForBackgroundWork"); WaitForBackgroundWork(); TEST_SYNC_POINT("DBImpl::ResumeImpl:Start"); @@ -474,9 +490,8 @@ Status DBImpl::ResumeImpl(DBRecoverContext context, s = Status::ShutdownInProgress(); } if (s.ok() && context.flush_after_recovery) { - // Since we drop all non-recovery flush requests during recovery, - // and new memtable may fill up during recovery, - // schedule one more round of flush. + // Normal flush requests are discarded during recovery, and a new memtable + // may fill while recovery releases the DB mutex. Schedule a catch-up flush. Status status = RetryFlushesForErrorRecovery( FlushReason::kCatchUpAfterErrorRecovery, false /* wait */); if (!status.ok()) { diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h index 7a19eae399..9f92abb11d 100644 --- a/db/db_impl/db_impl.h +++ b/db/db_impl/db_impl.h @@ -2785,7 +2785,11 @@ class DBImpl : public DB // equal to this per-column-family specified value, this flush request is // considered to have completed its work of flushing this column family. // After completing the work for all column families in this request, this - // flush is considered complete. + // flush is considered complete. EnqueuePendingFlush() acquires one + // reference for each CFD when it successfully queues this request. + // PopFirstFromFlushQueue() transfers responsibility for those references + // to its caller, which must release each one with UnrefAndTryDelete() after + // processing or discarding the request. std::unordered_map cfd_to_max_mem_id_to_persist; diff --git a/db/db_impl/db_impl_compaction_flush.cc b/db/db_impl/db_impl_compaction_flush.cc index 4d7f613309..de470245a0 100644 --- a/db/db_impl/db_impl_compaction_flush.cc +++ b/db/db_impl/db_impl_compaction_flush.cc @@ -3512,14 +3512,9 @@ void DBImpl::MaybeScheduleFlushOrCompaction() { return; } else if (error_handler_.IsBGWorkStopped() && !error_handler_.IsRecoveryInProgress()) { - // There has been a hard error and this call is not part of the recovery - // sequence. Bail out here so we don't get into an endless loop of - // scheduling BG work which will again call this function - // - // Note that a non-recovery flush can still be scheduled if - // error_handler_.IsRecoveryInProgress() returns true. We rely on - // BackgroundCallFlush() to check flush reason and drop non-recovery - // flushes. + // Recovery must be able to schedule its flushes while background work is + // stopped. New non-recovery requests are rejected by EnqueuePendingFlush, + // and BackgroundFlush drops any request queued before the error. return; } else if (shutting_down_.load(std::memory_order_acquire)) { // DB is being deleted; no more background compactions @@ -3817,6 +3812,10 @@ bool DBImpl::EnqueuePendingFlush(const FlushRequest& flush_req) { if (reject_new_background_jobs_) { return enqueued; } + if (error_handler_.IsBGWorkStopped() && + !IsRecoveryFlush(flush_req.flush_reason)) { + return enqueued; + } if (flush_req.cfd_to_max_mem_id_to_persist.empty()) { return enqueued; } @@ -3973,8 +3972,8 @@ Status DBImpl::BackgroundFlush(bool* made_progress, JobContext* job_context, Status status; *reason = FlushReason::kOthers; - // If BG work is stopped due to an error, but a recovery is in progress, - // that means this flush is part of the recovery. So allow it to go through + // During recovery, allow flush workers to inspect the queue. Non-recovery + // requests are dropped below. if (!error_handler_.IsBGWorkStopped()) { if (shutting_down_.load(std::memory_order_acquire)) { status = Status::ShutdownInProgress(); @@ -3996,12 +3995,9 @@ Status DBImpl::BackgroundFlush(bool* made_progress, JobContext* job_context, FlushRequest flush_req = PopFirstFromFlushQueue(); FlushReason flush_reason = flush_req.flush_reason; if (!error_handler_.GetBGError().ok() && error_handler_.IsBGWorkStopped() && - flush_reason != FlushReason::kErrorRecovery && - flush_reason != FlushReason::kErrorRecoveryRetryFlush) { - // Stop non-recovery flush when bg work is stopped - // Note that we drop the flush request here. - // Recovery thread should schedule further flushes after bg error - // is cleared. + !IsRecoveryFlush(flush_reason)) { + // A request queued before the error can reach a worker before recovery + // clears the queue. Drop it; recovery rebuilds flush work for all CFs. status = error_handler_.GetBGError(); assert(!status.ok()); ROCKS_LOG_BUFFER(log_buffer, diff --git a/db/error_handler.cc b/db/error_handler.cc index 7b7d875f40..ca50d31a2a 100644 --- a/db/error_handler.cc +++ b/db/error_handler.cc @@ -637,10 +637,6 @@ Status ErrorHandler::RecoverFromBGError(bool is_manual) { return Status::Busy("Recovery already in progress"); } recovery_in_prog_ = true; - - // In manual resume, we allow the bg work to run. If it is a auto resume, - // the bg work should follow this tag. - soft_error_no_bg_work_ = false; } // kErrorRecovery can be the default context for a generic soft error that @@ -659,7 +655,9 @@ Status ErrorHandler::RecoverFromBGError(bool is_manual) { // can generate background errors should be the flush operations recovery_error_ = IOStatus::OK(); recovery_error_.PermitUncheckedError(); - Status s = db_->ResumeImpl(recover_context_, Env::IOActivity::kFlush); + DBRecoverContext context = recover_context_; + context.flush_after_recovery = true; + Status s = db_->ResumeImpl(context, Env::IOActivity::kFlush); if (s.ok()) { soft_error_no_bg_work_ = false; } else { diff --git a/unreleased_history/bug_fixes/background_error_recovery_flush_starvation.md b/unreleased_history/bug_fixes/background_error_recovery_flush_starvation.md new file mode 100644 index 0000000000..dfdc55e939 --- /dev/null +++ b/unreleased_history/bug_fixes/background_error_recovery_flush_starvation.md @@ -0,0 +1 @@ +Fixed a liveness bug where RocksDB could remain indefinitely in recovery from a soft background error when concurrent writes continuously generated normal flush requests.