Merge pull request ceph#61536 from Matan-B/wip-matanb-backfill-scan

crimson/osd/recovery_backend: scan_for_backfill to use obc_manager

Reviewed-by: Xuehan Xu <[email protected]>

Author: Matan-B

src/crimson/osd/recovery_backend.cc

@@ -222,61 +222,45 @@ RecoveryBackend::handle_backfill_remove(
 
 RecoveryBackend::interruptible_future<BackfillInterval>
 RecoveryBackend::scan_for_backfill(
-  const hobject_t& start,
+  const hobject_t start,
   [[maybe_unused]] const std::int64_t min,
   const std::int64_t max)
 {
   LOG_PREFIX(RecoveryBackend::scan_for_backfill);
   DEBUGDPP("starting from {}", pg, start);
   auto version_map = seastar::make_lw_shared<std::map<hobject_t, eversion_t>>();
-  return backend->list_objects(start, max).then_interruptible(
-    [FNAME, this, start, version_map] (auto&& ret) {
-    auto&& [objects, next] = std::move(ret);
-    return seastar::do_with(
-      std::move(objects),
-      [FNAME, this, version_map](auto &objects) {
-      return interruptor::parallel_for_each(objects,
-        [FNAME, this, version_map] (const hobject_t& object)
-	-> interruptible_future<> {
-	crimson::osd::ObjectContextRef obc;
-	if (pg.is_primary()) {
-	  obc = pg.obc_registry.maybe_get_cached_obc(object);
-	}
-	if (obc) {
-	  if (obc->obs.exists) {
-	    DEBUGDPP("found (primary): {}  {}",
-		     pg, object, obc->obs.oi.version);
-	    version_map->emplace(object, obc->obs.oi.version);
-	  } else {
-	    // if the object does not exist here, it must have been removed
-	    // between the collection_list_partial and here.  This can happen
-	    // for the first item in the range, which is usually last_backfill.
-	  }
-	  return seastar::now();
-	} else {
-	  return backend->load_metadata(object).safe_then_interruptible(
-	    [FNAME, this, version_map, object] (auto md) {
-	    if (md->os.exists) {
-	      DEBUGDPP("found: {}  {}", pg,
-		       object, md->os.oi.version);
-	      version_map->emplace(object, md->os.oi.version);
-	    }
-	    return seastar::now();
-	  }, PGBackend::load_metadata_ertr::assert_all{});
-	}
-      });
-    }).then_interruptible([FNAME, version_map, start=std::move(start),
-			  next=std::move(next), this] {
-      BackfillInterval bi;
-      bi.begin = std::move(start);
-      bi.end = std::move(next);
-      bi.objects = std::move(*version_map);
-      DEBUGDPP("{} BackfillInterval filled, leaving, {}",
-	       "scan_for_backfill",
-	       pg, bi);
-      return seastar::make_ready_future<BackfillInterval>(std::move(bi));
-    });
-  });
+  auto&& [objects, next] = co_await backend->list_objects(start, max);
+  co_await interruptor::parallel_for_each(objects, seastar::coroutine::lambda([FNAME, this, version_map]
+    (const hobject_t& object) -> interruptible_future<> {
+    DEBUGDPP("querying obj:{}", pg, object);
+    auto obc_manager = pg.obc_loader.get_obc_manager(object);
+    co_await pg.obc_loader.load_and_lock(
+      obc_manager, RWState::RWREAD
+    ).handle_error_interruptible(
+      crimson::ct_error::assert_all("unexpected error")
+    );
+
+    if (obc_manager.get_obc()->obs.exists) {
+      auto version = obc_manager.get_obc()->obs.oi.version;
+      version_map->emplace(object, version);
+      DEBUGDPP("found: {}  {}", pg,
+               object, version);
+      co_return;
+    } else {
+      // if the object does not exist here, it must have been removed
+      // between the collection_list_partial and here.  This can happen
+      // for the first item in the range, which is usually last_backfill.
+      co_return;
+    }
+  }));
+  BackfillInterval bi;
+  bi.begin = std::move(start);
+  bi.end = std::move(next);
+  bi.objects = std::move(*version_map);
+  DEBUGDPP("{} BackfillInterval filled, leaving, {}",
+           "scan_for_backfill",
+           pg, bi);
+  co_return std::move(bi);
 }
 
 RecoveryBackend::interruptible_future<>

src/crimson/osd/recovery_backend.h

@@ -90,7 +90,7 @@ class RecoveryBackend {
     eversion_t need) = 0;
 
   interruptible_future<BackfillInterval> scan_for_backfill(
-    const hobject_t& from,
+    const hobject_t from,
     std::int64_t min,
     std::int64_t max);
 

src/crimson/osd/replicated_recovery_backend.cc

@@ -773,10 +773,6 @@ ReplicatedRecoveryBackend::handle_pull(Ref<MOSDPGPull> m)
 {
   LOG_PREFIX(ReplicatedRecoveryBackend::handle_pull);
   DEBUGDPP("{}", pg, *m);
-  if (pg.can_discard_replica_op(*m)) {
-    DEBUGDPP("discarding {}", pg, *m);
-    return seastar::now();
-  }
   return seastar::do_with(m->take_pulls(), [FNAME, this, from=m->from](auto& pulls) {
     return interruptor::parallel_for_each(
       pulls,
@@ -841,39 +837,42 @@ ReplicatedRecoveryBackend::_handle_pull_response(
     pull_info.recovery_info.version = push_op.version;
 
   if (pull_info.recovery_progress.first) {
-    auto fut = pg.obc_loader.with_obc<RWState::RWNONE>(
-      pull_info.recovery_info.soid,
-      [FNAME, this, &pull_info, &recovery_waiter, &push_op](auto, auto obc) {
-        pull_info.obc = obc;
-        recovery_waiter.obc = obc;
-        obc->obs.oi.decode_no_oid(push_op.attrset.at(OI_ATTR),
-                                  push_op.soid);
-        auto ss_attr_iter = push_op.attrset.find(SS_ATTR);
-        if (ss_attr_iter != push_op.attrset.end()) {
-          if (!obc->ssc) {
-            obc->ssc = new crimson::osd::SnapSetContext(
-              push_op.soid.get_snapdir());
-          }
-          try {
-            obc->ssc->snapset = SnapSet(ss_attr_iter->second);
-            obc->ssc->exists = true;
-          } catch (const buffer::error&) {
-            WARNDPP("unable to decode SnapSet", pg);
-            throw crimson::osd::invalid_argument();
-          }
-          assert(!pull_info.obc->ssc->exists ||
-                 obc->ssc->snapset.seq == pull_info.obc->ssc->snapset.seq);
-        }
-        pull_info.recovery_info.oi = obc->obs.oi;
-        if (pull_info.recovery_info.soid.snap &&
-            pull_info.recovery_info.soid.snap < CEPH_NOSNAP) {
-            recalc_subsets(pull_info.recovery_info,
-                           pull_info.obc->ssc);
-        }
-        return crimson::osd::PG::load_obc_ertr::now();
-      }, false).handle_error_interruptible(crimson::ct_error::assert_all{});
-    co_await std::move(fut);
-  };
+    auto obc_manager = pg.obc_loader.get_obc_manager(pull_info.recovery_info.soid);
+    co_await pg.obc_loader.load_and_lock(
+      obc_manager, RWState::RWNONE
+    ).handle_error_interruptible(
+      crimson::ct_error::assert_all("unexpected error")
+    );
+
+    auto obc = obc_manager.get_obc();
+    pull_info.obc = obc;
+    recovery_waiter.obc = obc;
+    // TODO: move to ObjectContextLoader once constructing obc from attrset is supported
+    obc->obs.oi.decode_no_oid(push_op.attrset.at(OI_ATTR),
+                              push_op.soid);
+    auto ss_attr_iter = push_op.attrset.find(SS_ATTR);
+    if (ss_attr_iter != push_op.attrset.end()) {
+      if (!obc->ssc) {
+        obc->ssc = new crimson::osd::SnapSetContext(
+          push_op.soid.get_snapdir());
+      }
+      try {
+        obc->ssc->snapset = SnapSet(ss_attr_iter->second);
+        obc->ssc->exists = true;
+      } catch (const buffer::error&) {
+        WARNDPP("unable to decode SnapSet", pg);
+        throw crimson::osd::invalid_argument();
+      }
+      assert(!pull_info.obc->ssc->exists ||
+             obc->ssc->snapset.seq == pull_info.obc->ssc->snapset.seq);
+    }
+    pull_info.recovery_info.oi = obc->obs.oi;
+    if (pull_info.recovery_info.soid.snap &&
+        pull_info.recovery_info.soid.snap < CEPH_NOSNAP) {
+        recalc_subsets(pull_info.recovery_info,
+                       pull_info.obc->ssc);
+    }
+  }
 
   const bool first = pull_info.recovery_progress.first;
   pull_info.recovery_progress = push_op.after_progress;
@@ -943,10 +942,6 @@ ReplicatedRecoveryBackend::handle_pull_response(
   Ref<MOSDPGPush> m)
 {
   LOG_PREFIX(ReplicatedRecoveryBackend::handle_pull_response);
-  if (pg.can_discard_replica_op(*m)) {
-    DEBUGDPP("discarding {}", pg, *m);
-    co_return;
-  }
   PushOp& push_op = m->pushes[0]; //TODO: only one push per message for now.
   if (push_op.version == eversion_t()) {
     // replica doesn't have it!

src/test/crimson/test_crimson_coroutine.cc

@@ -5,6 +5,7 @@
 #include <numeric>
 
 #include "seastar/core/sleep.hh"
+#include "seastar/core/loop.hh"
 
 #include "crimson/common/coroutine.h"
 #include "crimson/common/errorator.h"
@@ -103,6 +104,28 @@ TEST_F(coroutine_test_t, test_coroutine)
   });
 }
 
+TEST_F(coroutine_test_t, test_coroutine_loops)
+{
+  run_scl([]() -> seastar::future<> {
+    int CHECK = 0;
+    std::vector<int> v = {1,2,3};
+    co_await seastar::parallel_for_each(v,
+      [&CHECK] (auto i) -> seastar::future<> {
+      CHECK++;
+      co_return;
+    });
+    EXPECT_EQ(CHECK, v.size());
+
+    co_await seastar::do_until(
+      [&CHECK] { return CHECK == 0; },
+      [&CHECK] () -> seastar::future<> {
+      CHECK--;
+      co_return;
+    });
+    EXPECT_EQ(CHECK, 0);
+  });
+}
+
 TEST_F(coroutine_test_t, test_ertr_coroutine_basic)
 {
   run_ertr_scl([]() -> ertr::future<> {