crimson/os/seastore/EPM/BackgroundProcess: add promote and demote process

Signed-off-by: Zhang Song <zhangsong02@qianxin.com>
Signed-off-by: Xuehan Xu <xuxuehan@qianxin.com>

src/crimson/os/seastore/async_cleaner.h

@@ -453,6 +453,7 @@ struct BackgroundListener {
   virtual ~BackgroundListener() = default;
   virtual void maybe_wake_background() = 0;
   virtual void maybe_wake_blocked_io() = 0;
+  virtual void maybe_wake_promote() = 0;
   virtual state_t get_state() const = 0;
 
   bool is_ready() const {

src/crimson/os/seastore/extent_pinboard.cc

@@ -319,7 +319,8 @@ public:
       remove_extent(list.front(), extent_pin_state_t::Fresh);
     }
     if (should_run_promote()) {
-      // TODO: wake promote background process
+      assert(listener);
+      listener->maybe_wake_promote();
     }
   }
 

src/crimson/os/seastore/extent_placement_manager.cc

@@ -634,41 +634,49 @@ void ExtentPlacementManager::BackgroundProcess::start_background()
   ceph_assert(state == state_t::SCAN_SPACE);
   assert(!is_running());
   process_join = seastar::now();
+  promote_process_join = seastar::now();
   state = state_t::RUNNING;
   assert(is_running());
   process_join = run();
+  if (has_cold_tier()) {
+    promote_process_join = run_promote();
+  }
 }
 
 seastar::future<>
 ExtentPlacementManager::BackgroundProcess::stop_background()
 {
   LOG_PREFIX(BackgroundProcess::stop_background);
-  return seastar::futurize_invoke([this, FNAME] {
-    if (!is_running()) {
-      if (state != state_t::HALT) {
-        INFO("isn't RUNNING or HALT, STOP");
-        state = state_t::STOP;
-      } else {
-        INFO("isn't RUNNING, already HALT");
-      }
-      return seastar::now();
+  if (!is_running()) {
+    if (state != state_t::HALT) {
+      INFO("isn't RUNNING or HALT, STOP");
+      state = state_t::STOP;
+    } else {
+      INFO("isn't RUNNING, already HALT");
     }
-    INFO("is RUNNING, going to HALT...");
-    auto ret = std::move(*process_join);
-    process_join.reset();
-    state = state_t::HALT;
-    assert(!is_running());
-    do_wake_background();
-    return ret;
-  }).then([this, FNAME] {
-    INFO("done, {}, {}",
-         JournalTrimmerImpl::stat_printer_t{*trimmer, true},
-         AsyncCleaner::stat_printer_t{*main_cleaner, true});
-    if (has_cold_tier()) {
-      INFO("done, cold_cleaner: {}",
-           AsyncCleaner::stat_printer_t{*cold_cleaner, true});
-    }
-  });
+    co_return;
+  }
+  INFO("is RUNNING, going to HALT...");
+  std::vector<seastar::future<>> futs;
+  futs.emplace_back(std::move(*process_join));
+  process_join.reset();
+  if (promote_process_join) {
+    futs.emplace_back(std::move(*promote_process_join));
+    promote_process_join.reset();
+  }
+  state = state_t::HALT;
+  assert(!is_running());
+  do_wake_background();
+  do_wake_promote();
+  co_await seastar::when_all(futs.begin(), futs.end());
+  INFO("done, {}, {}",
+       JournalTrimmerImpl::stat_printer_t{*trimmer, true},
+       AsyncCleaner::stat_printer_t{*main_cleaner, true});
+  if (has_cold_tier()) {
+    INFO("done, cold_cleaner: {}",
+         AsyncCleaner::stat_printer_t{*cold_cleaner, true});
+  }
+  co_return;
 }
 
 seastar::future<>
@@ -1020,7 +1028,12 @@ ExtentPlacementManager::BackgroundProcess::do_background_cycle()
       proceed_clean_cold = true;
     }
 
-    if (!proceed_clean_main && !proceed_clean_cold) {
+    bool proceed_demote = false;
+    if (demote_should_run()) {
+      proceed_demote = true;
+    }
+
+    if (!proceed_clean_main && !proceed_clean_cold && !proceed_demote) {
       ceph_abort_msg("no background process will start");
     }
     return seastar::when_all(
@@ -1062,11 +1075,51 @@ ExtentPlacementManager::BackgroundProcess::do_background_cycle()
         ).finally([FNAME] {
           DEBUG("finished clean cold");
         });
+      },
+      [this, proceed_demote] {
+        if (!proceed_demote) {
+          return seastar::now();
+        }
+        return logical_bucket->demote();
       }
     ).discard_result();
   }
 }
 
+seastar::future<> ExtentPlacementManager::BackgroundProcess::run_promote()
+{
+  assert(pinboard);
+  assert(is_running());
+  return seastar::repeat([this] {
+    if (!is_running()) {
+      return seastar::make_ready_future<seastar::stop_iteration>(
+          seastar::stop_iteration::yes);
+    }
+
+    return seastar::futurize_invoke([this] {
+      if (pinboard->should_promote()) {
+        auto usage = cleaner_usage_t{pinboard->get_promotion_size(), 0};
+        auto res = try_reserve_cleaner(usage);
+        if (res.is_successful()) {
+          return pinboard->promote(
+          ).finally([this, usage, res] {
+            abort_cleaner_usage(usage, res);
+          });
+        } else {
+          // reserve usage failed, block
+          abort_cleaner_usage(usage, res);
+        }
+      } // shouldn't promote, block
+
+      ceph_assert(!blocking_promote);
+      blocking_promote = seastar::promise<>();
+      return blocking_promote->get_future();
+    }).then([] {
+      return seastar::stop_iteration::no;
+    });
+  });
+}
+
 void ExtentPlacementManager::BackgroundProcess::register_metrics(store_index_t store_index)
 {
   namespace sm = seastar::metrics;

src/crimson/os/seastore/extent_placement_manager.h

@@ -950,6 +950,15 @@ private:
 
     void maybe_wake_blocked_io() final;
 
+    void maybe_wake_promote() final {
+      if (!is_ready()) {
+        return;
+      }
+      if (pinboard && pinboard->should_promote()) {
+        do_wake_promote();
+      }
+    }
+
   private:
     // reserve helpers
     bool try_reserve_cold(std::size_t usage);
@@ -984,6 +993,13 @@ private:
       }
     }
 
+    void do_wake_promote() {
+      if (blocking_promote) {
+        blocking_promote->set_value();
+        blocking_promote = std::nullopt;
+      }
+    }
+
     // background_should_run() should be atomic with do_background_cycle()
     // to make sure the condition is consistent.
     bool background_should_run() {
@@ -991,7 +1007,8 @@ private:
       maybe_update_eviction_mode();
       return main_cleaner_should_run()
         || cold_cleaner_should_run()
-        || trimmer->should_trim();
+        || trimmer->should_trim()
+        || demote_should_run();
     }
 
     bool main_cleaner_should_fast_evict() const {
@@ -1013,6 +1030,13 @@ private:
         cold_cleaner->should_clean_space();
     }
 
+    bool demote_should_run() const {
+      return has_cold_tier() &&
+        logical_bucket->could_demote() &&
+        (eviction_state.is_fast_mode() ||
+         logical_bucket->should_demote());
+    }
+
     bool should_block_io() const {
       assert(is_ready());
       return trimmer->should_block_io_on_trim() ||
@@ -1142,6 +1166,7 @@ private:
     };
 
     seastar::future<> do_background_cycle();
+    seastar::future<> run_promote();
 
     void register_metrics(store_index_t store_index);
 
@@ -1177,6 +1202,8 @@ private:
     // giving the woken continuation a chance to retry the reservation
     // before the next background cycle.
     bool pending_user_io_wake = false;
+    std::optional<seastar::future<>> promote_process_join;
+    std::optional<seastar::promise<>> blocking_promote;
     bool is_running_until_halt = false;
     state_t state = state_t::STOP;
     eviction_state_t eviction_state;

src/crimson/os/seastore/logical_bucket.cc

@@ -50,6 +50,10 @@ public:
     } else {
       TRACE("create bucket: {}", laddr);
       index[laddr] = lru.emplace(lru.end(), laddr);
+      if (should_demote()) {
+	assert(listener);
+	listener->maybe_wake_background();
+      }
     }
   }