Adjust thread pool sizes when setting max_background_jobs dynamically (#6300)

Summary:
https://github.com/facebook/rocksdb/pull/2205 introduced a new
configuration option called `max_background_jobs`, superseding the
earlier options `max_background_flushes` and
`max_background_compactions`. However, unlike
`max_background_compactions`, setting `max_background_jobs` dynamically
through the `SetDBOptions` interface does not adjust the size of the
thread pools (see https://github.com/facebook/rocksdb/issues/6298). The
patch fixes this.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/6300

Test Plan: Extended unit test.

Differential Revision: D19430899

Pulled By: ltamasi

fbshipit-source-id: 704006605b3c13c3d1b997ccc0831ee369721074

HISTORY.md

@@ -19,6 +19,7 @@
 * Fix a bug on fractional cascading index when multiple files at the same level contain the same smallest user key, and those user keys are for merge operands. In this case, Get() the exact key may miss some merge operands.
 * Delcare kHashSearch index type feature-incompatible with index_block_restart_interval larger than 1.
 * Fix incorrect results while block-based table uses kHashSearch, together with Prev()/SeekForPrev().
+* Fixed an issue where the thread pools were not resized upon setting `max_background_jobs` dynamically through the `SetDBOptions` interface.
 
 ### New Features
 * It is now possible to enable periodic compactions for the base DB when using BlobDB.

db/db_impl/db_impl.cc

@@ -1005,12 +1005,36 @@ Status DBImpl::SetDBOptions(
       }
     }
     if (s.ok()) {
-      if (new_options.max_background_compactions >
+      const BGJobLimits current_bg_job_limits =
-          mutable_db_options_.max_background_compactions) {
+          GetBGJobLimits(immutable_db_options_.max_background_flushes,
-        env_->IncBackgroundThreadsIfNeeded(
+                         mutable_db_options_.max_background_compactions,
-            new_options.max_background_compactions, Env::Priority::LOW);
+                         mutable_db_options_.max_background_jobs,
+                         /* parallelize_compactions */ true);
+      const BGJobLimits new_bg_job_limits = GetBGJobLimits(
+          immutable_db_options_.max_background_flushes,
+          new_options.max_background_compactions,
+          new_options.max_background_jobs, /* parallelize_compactions */ true);
+
+      const bool max_flushes_increased =
+          new_bg_job_limits.max_flushes > current_bg_job_limits.max_flushes;
+      const bool max_compactions_increased =
+          new_bg_job_limits.max_compactions >
+          current_bg_job_limits.max_compactions;
+
+      if (max_flushes_increased || max_compactions_increased) {
+        if (max_flushes_increased) {
+          env_->IncBackgroundThreadsIfNeeded(new_bg_job_limits.max_flushes,
+                                             Env::Priority::HIGH);
+        }
+
+        if (max_compactions_increased) {
+          env_->IncBackgroundThreadsIfNeeded(new_bg_job_limits.max_compactions,
+                                             Env::Priority::LOW);
+        }
+
         MaybeScheduleFlushOrCompaction();
       }
+
       if (new_options.stats_dump_period_sec !=
           mutable_db_options_.stats_dump_period_sec) {
         if (thread_dump_stats_) {

db/db_options_test.cc

@@ -423,16 +423,22 @@ TEST_F(DBOptionsTest, SetBackgroundJobs) {
             std::to_string(options.max_background_jobs)}}));
     }
 
-    ASSERT_EQ(options.max_background_jobs / 4,
+    const int expected_max_flushes = options.max_background_jobs / 4;
-              dbfull()->TEST_BGFlushesAllowed());
+
+    ASSERT_EQ(expected_max_flushes, dbfull()->TEST_BGFlushesAllowed());
     ASSERT_EQ(1, dbfull()->TEST_BGCompactionsAllowed());
 
     auto stop_token = dbfull()->TEST_write_controler().GetStopToken();
 
-    ASSERT_EQ(options.max_background_jobs / 4,
+    const int expected_max_compactions = 3 * expected_max_flushes;
-              dbfull()->TEST_BGFlushesAllowed());
+
-    ASSERT_EQ(3 * options.max_background_jobs / 4,
+    ASSERT_EQ(expected_max_flushes, dbfull()->TEST_BGFlushesAllowed());
-              dbfull()->TEST_BGCompactionsAllowed());
+    ASSERT_EQ(expected_max_compactions, dbfull()->TEST_BGCompactionsAllowed());
+
+    ASSERT_EQ(expected_max_flushes,
+              env_->GetBackgroundThreads(Env::Priority::HIGH));
+    ASSERT_EQ(expected_max_compactions,
+              env_->GetBackgroundThreads(Env::Priority::LOW));
   }
 }