diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index cf7259341b0..131c191aabf 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -1508,6 +1508,14 @@ Status DBImpl::StartPeriodicTaskScheduler() { return s; } +Status DBImpl::RegisterTriggerCompactionTask() { + options_mutex_.AssertHeld(); + return periodic_task_scheduler_.Register( + PeriodicTaskType::kTriggerCompaction, + periodic_task_functions_.at(PeriodicTaskType::kTriggerCompaction), + ComputeTriggerCompactionPeriod(), /*run_immediately=*/false); +} + Status DBImpl::RegisterRecordSeqnoTimeWorker() { options_mutex_.AssertHeld(); @@ -2005,6 +2013,8 @@ Status DBImpl::SetOptions( s = RegisterRecordSeqnoTimeWorker(); } } + // Earlier CFs can have changed even if a later CF's SetOptions failed. + s.UpdateIfOk(RegisterTriggerCompactionTask()); ROCKS_LOG_INFO(immutable_db_options_.info_log, "SetOptions() on [%zu] column families, inputs:", @@ -2209,6 +2219,9 @@ Status DBImpl::SetDBOptions( persist_options_status.PermitUncheckedError(); } } + if (s.ok()) { + s = RegisterTriggerCompactionTask(); + } ROCKS_LOG_INFO(immutable_db_options_.info_log, "SetDBOptions(), inputs:"); for (const auto& o : options_map) { ROCKS_LOG_INFO(immutable_db_options_.info_log, "%s: %s\n", o.first.c_str(), @@ -4508,6 +4521,7 @@ Status DBImpl::WrapUpCreateColumnFamilies( if (register_worker) { s.UpdateIfOk(RegisterRecordSeqnoTimeWorker()); } + s.UpdateIfOk(RegisterTriggerCompactionTask()); return s; } @@ -4968,6 +4982,9 @@ Status DBImpl::DropColumnFamilyWithOptionsFile( if (preserve_info.IsEnabled()) { s = RegisterRecordSeqnoTimeWorker(); } + if (cfd->IsDropped()) { + s.UpdateIfOk(RegisterTriggerCompactionTask()); + } if (s.ok()) { // Note that here we erase the associated cf_info of the to-be-dropped diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h index f85b78a1484..fccee662b9b 100644 --- a/db/db_impl/db_impl.h +++ b/db/db_impl/db_impl.h @@ -3136,11 +3136,14 @@ class DBImpl : public DB // Compute the repeat period for the kTriggerCompaction task, which ensures // compactions not dependent on writes (flushes) are eventually triggered when - // there are no writes (flushes). NOT thread safe; only called during DB open - // (StartPeriodicTaskScheduler). KNOWN LIMITATION: doesn't get updated with - // dynamic option updates. (Probably not worth the extra complexity.) + // there are no writes (flushes). Call only during DB open or while holding + // options_mutex_, which serializes changes to the options and CF set. uint64_t ComputeTriggerCompactionPeriod(); + // Refresh the task after a change to its DB or CF options, or the CF set. + // Requires options_mutex_. + Status RegisterTriggerCompactionTask(); + // Cancel scheduled periodic tasks Status CancelPeriodicTaskScheduler(); diff --git a/db/periodic_task_scheduler_test.cc b/db/periodic_task_scheduler_test.cc index 73d4928eead..7994c887e8d 100644 --- a/db/periodic_task_scheduler_test.cc +++ b/db/periodic_task_scheduler_test.cc @@ -98,6 +98,8 @@ TEST_F(PeriodicTaskSchedulerTest, Basic) { ASSERT_EQ(3, dump_st_counter); ASSERT_EQ(3, pst_st_counter); ASSERT_EQ(3, flush_info_log_counter); + const int trigger_compaction_count_before_disable = + trigger_compaction_counter; // Disable scheduler with SetOption ASSERT_OK(dbfull()->SetDBOptions( @@ -111,6 +113,8 @@ TEST_F(PeriodicTaskSchedulerTest, Basic) { ASSERT_EQ(3, dump_st_counter); ASSERT_EQ(3, pst_st_counter); ASSERT_EQ(4, flush_info_log_counter); + ASSERT_EQ(trigger_compaction_count_before_disable, + trigger_compaction_counter); ASSERT_EQ(2u, scheduler.TEST_GetValidTaskNum()); @@ -127,19 +131,16 @@ TEST_F(PeriodicTaskSchedulerTest, Basic) { ASSERT_EQ(3, pst_st_counter); ASSERT_EQ(5, flush_info_log_counter); - // With the new ComputeTriggerCompactionPeriod() logic, since - // stats_dump_period_sec = stats_persist_period_sec = kPeriodSec, the trigger - // compaction period is also kPeriodSec. Unlike other periodic tasks, - // kTriggerCompaction uses run_immediately=false, so it doesn't fire on - // initial registration. It fires after each period elapses. - // By now we've had: initial 9s + 10s + 10s + 10s + 10s = 49s elapsed. - // kTriggerCompaction fires at t=10, 20, 30, 40 = 4 times. - ASSERT_EQ(4, trigger_compaction_counter); + // Disabling the stats tasks also lengthens the compaction trigger period. + // Re-enabling the dump task shortens it again, starting a new schedule. + ASSERT_GT(trigger_compaction_counter, + trigger_compaction_count_before_disable); + const int trigger_compaction_count = trigger_compaction_counter; - // Sleep one more period and verify it increments + // Sleep one more period and verify the new schedule keeps running. dbfull()->TEST_WaitForPeriodicTaskRun( [&] { mock_clock_->MockSleepForSeconds(static_cast(kPeriodSec)); }); - ASSERT_EQ(5, trigger_compaction_counter); + ASSERT_GT(trigger_compaction_counter, trigger_compaction_count); Close(); } @@ -322,6 +323,98 @@ TEST_F(PeriodicTaskSchedulerTest, TriggerCompactionPeriodComputation) { } } +TEST_F(PeriodicTaskSchedulerTest, TriggerCompactionPeriodFollowsOptionChanges) { + Close(); + uint64_t trigger_compaction_period = 0; + SyncPoint::GetInstance()->SetCallBack( + "PeriodicTaskScheduler::Register:TaskRegistered", [&](void* arg) { + auto* task_info = + static_cast*>(arg); + if (task_info->first == PeriodicTaskType::kTriggerCompaction) { + trigger_compaction_period = task_info->second; + } + }); + SyncPoint::GetInstance()->EnableProcessing(); + + Options options; + options.create_if_missing = true; + options.env = mock_env_.get(); + options.stats_dump_period_sec = 0; + options.stats_persist_period_sec = 0; + Reopen(options); + ASSERT_EQ(12 * 60 * 60, trigger_compaction_period); + + ASSERT_OK(dbfull()->SetOptions({{"ttl", "100"}})); + ASSERT_EQ(20, trigger_compaction_period); + ASSERT_OK(dbfull()->SetOptions({{"ttl", "0"}})); + ASSERT_EQ(12 * 60 * 60, trigger_compaction_period); + + ASSERT_OK(dbfull()->SetDBOptions( + {{"max_compaction_trigger_wakeup_seconds", "30"}})); + ASSERT_EQ(30, trigger_compaction_period); + ASSERT_OK(dbfull()->SetDBOptions({{"stats_dump_period_sec", "10"}})); + ASSERT_EQ(10, trigger_compaction_period); + ASSERT_OK(dbfull()->SetDBOptions({{"stats_dump_period_sec", "0"}})); + ASSERT_EQ(30, trigger_compaction_period); + + ColumnFamilyOptions cf_options; + cf_options.ttl = 50; + ColumnFamilyHandle* cf_handle = nullptr; + ASSERT_OK(db_->CreateColumnFamily(cf_options, "ttl_cf", &cf_handle)); + ASSERT_EQ(10, trigger_compaction_period); + ASSERT_OK( + db_->SetOptions(cf_handle, {{"periodic_compaction_seconds", "20"}})); + ASSERT_EQ(4, trigger_compaction_period); + ASSERT_OK(db_->SetOptions(cf_handle, {{"periodic_compaction_seconds", "0"}})); + ASSERT_EQ(10, trigger_compaction_period); + ASSERT_OK(db_->DropColumnFamily(cf_handle)); + ASSERT_EQ(30, trigger_compaction_period); + delete cf_handle; + Close(); +} + +TEST_F(PeriodicTaskSchedulerTest, TriggerCompactionPeriodAfterPartialUpdate) { + Close(); + uint64_t trigger_compaction_period = 0; + SyncPoint::GetInstance()->SetCallBack( + "PeriodicTaskScheduler::Register:TaskRegistered", [&](void* arg) { + auto* task_info = + static_cast*>(arg); + if (task_info->first == PeriodicTaskType::kTriggerCompaction) { + trigger_compaction_period = task_info->second; + } + }); + SyncPoint::GetInstance()->EnableProcessing(); + + Options options; + options.create_if_missing = true; + options.env = mock_env_.get(); + options.stats_dump_period_sec = 0; + options.stats_persist_period_sec = 0; + Reopen(options); + ASSERT_EQ(12 * 60 * 60, trigger_compaction_period); + + ColumnFamilyHandle* cf_handle = nullptr; + ASSERT_OK(db_->CreateColumnFamily(options, "other_cf", &cf_handle)); + std::unordered_map> + cf_options; + cf_options.emplace(db_->DefaultColumnFamily(), + std::unordered_map{}); + cf_options.emplace(cf_handle, std::unordered_map{}); + // Assign by iteration order so the valid change is applied before the + // invalid one, regardless of how the handles hash on this platform. + auto it = cf_options.begin(); + it->second = {{"ttl", "100"}}; + (++it)->second = {{"unknown_option", "1"}}; + ASSERT_NOK(db_->SetOptions(cf_options)); + ASSERT_EQ(20, trigger_compaction_period); + + ASSERT_OK(db_->DropColumnFamily(cf_handle)); + delete cf_handle; + Close(); +} + // Test that TriggerPeriodicCompaction() properly considers CFs for compaction // based on all time-based compaction options (not just // periodic_compaction_seconds). This is a regression test for the bug where diff --git a/unreleased_history/bug_fixes/dynamic_time_based_compaction_trigger.md b/unreleased_history/bug_fixes/dynamic_time_based_compaction_trigger.md new file mode 100644 index 00000000000..800fb137294 --- /dev/null +++ b/unreleased_history/bug_fixes/dynamic_time_based_compaction_trigger.md @@ -0,0 +1 @@ +Fixed time-based compactions not waking up on idle DBs after their intervals are changed with SetOptions or SetDBOptions, or after column families are added or dropped.