Merge pull request ceph#59189 from xxhdx1985126/wip-67508

crimson/osd/recovery_backend: restart object pulls that are blocked by down osds

Reviewed-by: Matan Breizman <[email protected]>

Author: Matan-B

src/crimson/common/interruptible_future.h

@@ -1330,6 +1330,27 @@ struct interruptor
       );
     }
   }
+
+  template <InvokeReturnsInterruptibleFuture AsyncAction>
+  [[gnu::always_inline]]
+  static auto repeat_eagain(AsyncAction&& action) {
+    return seastar::do_with(
+      std::forward<AsyncAction>(action),
+      [] (auto &f) {
+      return repeat([&f] {
+	return std::invoke(f
+	).si_then([] {
+	  return seastar::stop_iteration::yes;
+	}).handle_error_interruptible(
+	  [](const crimson::ct_error::eagain &e) {
+	    return seastar::stop_iteration::no;
+	  },
+	  crimson::ct_error::pass_further_all{}
+	);
+      });
+    });
+  }
+
   template <typename AsyncAction>
   requires (!InvokeReturnsInterruptibleFuture<AsyncAction>)
   [[gnu::always_inline]]

src/crimson/osd/pg.h

@@ -357,7 +357,20 @@ class PG : public boost::intrusive_ref_counter<
     shard_services.remove_want_pg_temp(orderer, pgid.pgid);
   }
   void check_recovery_sources(const OSDMapRef& newmap) final {
-    // Not needed yet
+    LOG_PREFIX(PG::check_recovery_sources);
+    recovery_backend->for_each_recovery_waiter(
+      [newmap, FNAME, this](auto &, auto &waiter) {
+        if (waiter->is_pulling() &&
+            newmap->is_down(waiter->pull_info->from.osd)) {
+          SUBDEBUGDPP(
+            osd,
+            " repeating pulling for {}, due to osd {} down",
+            *this,
+            waiter->pull_info->soid,
+            waiter->pull_info->from.osd);
+          waiter->repeat_pull();
+        }
+      });
   }
   void check_blocklisted_watchers() final;
   void clear_primary_state() final {

src/crimson/osd/recovery_backend.h

@@ -112,6 +112,13 @@ class RecoveryBackend {
     }
     return on_stop();
   }
+
+  template <typename Func>
+  void for_each_recovery_waiter(Func &&f) {
+    for (auto &[soid, recovery_waiter] : recovering) {
+      std::forward<Func>(f)(soid, recovery_waiter);
+    }
+  }
 protected:
   crimson::osd::PG& pg;
   crimson::osd::ShardServices& shard_services;
@@ -219,6 +226,13 @@ class RecoveryBackend {
 	pulled.reset();
       }
     }
+    void repeat_pull() {
+      ceph_assert(pulled);
+      pulled->set_exception(crimson::ct_error::eagain::exception_ptr());
+    }
+    bool is_pulling() const {
+      return (bool)pulled;
+    }
     void set_push_failed(pg_shard_t shard, std::exception_ptr e) {
       auto it = pushes.find(shard);
       if (it != pushes.end()) {

src/crimson/osd/replicated_recovery_backend.cc

@@ -113,28 +113,33 @@ ReplicatedRecoveryBackend::maybe_pull_missing_obj(
     // object is not missing, don't pull
     return seastar::make_ready_future<>();
   }
-  return pg.obc_loader.with_obc<RWState::RWREAD>(soid.get_head(),
-    [this, soid, need](auto head, auto) {
-    PullOp pull_op;
-    auto& recovery_waiter = get_recovering(soid);
-    recovery_waiter.pull_info =
-      std::make_optional<RecoveryBackend::pull_info_t>();
-    auto& pull_info = *recovery_waiter.pull_info;
-    prepare_pull(head, pull_op, pull_info, soid, need);
-    auto msg = crimson::make_message<MOSDPGPull>();
-    msg->from = pg.get_pg_whoami();
-    msg->set_priority(pg.get_recovery_op_priority());
-    msg->pgid = pg.get_pgid();
-    msg->map_epoch = pg.get_osdmap_epoch();
-    msg->min_epoch = pg.get_last_peering_reset();
-    msg->set_pulls({std::move(pull_op)});
-    return shard_services.send_to_osd(
-      pull_info.from.osd,
-      std::move(msg),
-      pg.get_osdmap_epoch());
-  }).si_then([this, soid] {
-    auto& recovery_waiter = get_recovering(soid);
-    return recovery_waiter.wait_for_pull();
+  return interruptor::repeat_eagain([this, soid, need] {
+    using prepare_pull_iertr =
+      crimson::osd::ObjectContextLoader::load_obc_iertr::extend<
+        crimson::ct_error::eagain>;
+    return pg.obc_loader.with_obc<RWState::RWREAD>(soid.get_head(),
+      [this, soid, need](auto head, auto) {
+      PullOp pull_op;
+      auto& recovery_waiter = get_recovering(soid);
+      recovery_waiter.pull_info =
+        std::make_optional<RecoveryBackend::pull_info_t>();
+      auto& pull_info = *recovery_waiter.pull_info;
+      prepare_pull(head, pull_op, pull_info, soid, need);
+      auto msg = crimson::make_message<MOSDPGPull>();
+      msg->from = pg.get_pg_whoami();
+      msg->set_priority(pg.get_recovery_op_priority());
+      msg->pgid = pg.get_pgid();
+      msg->map_epoch = pg.get_osdmap_epoch();
+      msg->min_epoch = pg.get_last_peering_reset();
+      msg->set_pulls({std::move(pull_op)});
+      return shard_services.send_to_osd(
+        pull_info.from.osd,
+        std::move(msg),
+        pg.get_osdmap_epoch());
+    }).si_then([this, soid]() -> prepare_pull_iertr::future<> {
+      auto& recovery_waiter = get_recovering(soid);
+      return recovery_waiter.wait_for_pull();
+    });
   }).handle_error_interruptible(
     crimson::ct_error::assert_all("unexpected error")
   );

src/test/crimson/test_interruptible_future.cc

@@ -282,6 +282,29 @@ TEST_F(seastar_test_suite_t, DISABLED_nested_interruptors)
   });
 }
 
+TEST_F(seastar_test_suite_t, interruptible_repeat_eagain)
+{
+  using interruptor =
+    interruptible::interruptor<TestInterruptCondition>;
+  run_async([] {
+    interruptor::with_interruption([] {
+      return seastar::do_with(
+	0,
+	[](auto &i) {
+	return interruptor::repeat_eagain([&i]() -> base_iertr::future<> {
+	  if (++i < 5) {
+	    return crimson::ct_error::eagain::make();
+	  }
+	  return base_iertr::now();
+	}).si_then([&i] {
+	  std::cout << i << std::endl;
+	  ceph_assert(i == 5);
+	});
+      });
+    }, [](std::exception_ptr) {}, false).unsafe_get();
+  });
+}
+
 #if 0
 // This seems to cause a hang in the gcc-9 linker on bionic
 TEST_F(seastar_test_suite_t, handle_error)