Merge pull request ceph#58148 from xxhdx1985126/wip-65696

crimson/osd/osd_operations: hang requests if the objects are unfound

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

Author: Matan-B

src/crimson/osd/osd_operations/background_recovery.cc

@@ -87,13 +87,6 @@ seastar::future<> BackgroundRecoveryT<T>::start()
           }, [](std::exception_ptr) {
             return seastar::make_ready_future<bool>(false);
           }, pg, epoch_started);
-        }).handle_exception_type([ref, this](const std::system_error& err) {
-	  LOG_PREFIX(BackgroundRecoveryT<T>::start);
-          if (err.code() == std::make_error_code(std::errc::interrupted)) {
-            DEBUGDPPI("recovery interruped: {}", *pg, err.what());
-            return seastar::now();
-          }
-          return seastar::make_exception_future<>(err);
         });
       });
   });

src/crimson/osd/osd_operations/client_request.cc

@@ -286,6 +286,7 @@ ClientRequest::recover_missing_snaps(
   ObjectContextRef head,
   std::set<snapid_t> &snaps)
 {
+  LOG_PREFIX(ClientRequest::process_op);
   co_await ihref.enter_stage<interruptor>(
     client_pp(*pg).recover_missing_snaps, *this);
   for (auto &snap : snaps) {
@@ -299,7 +300,12 @@ ClientRequest::recover_missing_snaps(
      * we skip the oid as there is no corresponding clone to recover.
      * See https://tracker.ceph.com/issues/63821 */
     if (oid) {
-      co_await do_recover_missing(pg, *oid, m->get_reqid());
+      auto unfound = co_await do_recover_missing(pg, *oid, m->get_reqid());
+      if (unfound) {
+        DEBUGDPP("{} unfound, hang it for now", *pg, m->get_hobj().get_head());
+        co_await interruptor::make_interruptible(
+          pg->get_recovery_backend()->add_unfound(m->get_hobj().get_head()));
+      }
     }
   }
 }
@@ -317,7 +323,14 @@ ClientRequest::process_op(
       "Skipping recover_missings on non primary pg for soid {}",
       *pg, m->get_hobj());
   } else {
-    co_await do_recover_missing(pg, m->get_hobj().get_head(), m->get_reqid());
+    auto unfound = co_await do_recover_missing(
+      pg, m->get_hobj().get_head(), m->get_reqid());
+    if (unfound) {
+      DEBUGDPP("{} unfound, hang it for now", *pg, m->get_hobj().get_head());
+      co_await interruptor::make_interruptible(
+        pg->get_recovery_backend()->add_unfound(m->get_hobj().get_head()));
+    }
+
     std::set<snapid_t> snaps = snaps_need_to_recover();
     if (!snaps.empty()) {
       // call with_obc() in order, but wait concurrently for loading.

src/crimson/osd/osd_operations/client_request_common.cc

@@ -13,7 +13,7 @@ namespace {
 
 namespace crimson::osd {
 
-typename InterruptibleOperation::template interruptible_future<>
+typename InterruptibleOperation::template interruptible_future<bool>
 CommonClientRequest::do_recover_missing(
   Ref<PG> pg,
   const hobject_t& soid,
@@ -45,22 +45,29 @@ CommonClientRequest::do_recover_missing(
   if (!needs_recovery_or_backfill) {
     logger().debug("{} reqid {} nothing to recover {}",
                    __func__, reqid, soid);
-    return seastar::now();
+    return seastar::make_ready_future<bool>(false);
   }
 
+  if (pg->get_peering_state().get_missing_loc().is_unfound(soid)) {
+    return seastar::make_ready_future<bool>(true);
+  }
   logger().debug("{} reqid {} need to wait for recovery, {} version {}",
                  __func__, reqid, soid, ver);
   if (pg->get_recovery_backend()->is_recovering(soid)) {
     logger().debug("{} reqid {} object {} version {}, already recovering",
                    __func__, reqid, soid, ver);
-    return pg->get_recovery_backend()->get_recovering(soid).wait_for_recovered();
+    return pg->get_recovery_backend()->get_recovering(
+      soid).wait_for_recovered(
+    ).then([] {
+      return seastar::make_ready_future<bool>(false);
+    });
   } else {
     logger().debug("{} reqid {} object {} version {}, starting recovery",
                    __func__, reqid, soid, ver);
     auto [op, fut] =
       pg->get_shard_services().start_operation<UrgentRecovery>(
         soid, ver, pg, pg->get_shard_services(), pg->get_osdmap_epoch());
-    return std::move(fut);
+    return fut.then([] { return seastar::make_ready_future<bool>(false); });
   }
 }
 

src/crimson/osd/osd_operations/client_request_common.h

@@ -11,7 +11,7 @@ namespace crimson::osd {
 
 struct CommonClientRequest {
 
-  static InterruptibleOperation::template interruptible_future<>
+  static InterruptibleOperation::template interruptible_future<bool>
   do_recover_missing(
     Ref<PG> pg,
     const hobject_t& soid,

src/crimson/osd/osd_operations/internal_client_request.cc

@@ -70,7 +70,12 @@ seastar::future<> InternalClientRequest::start()
             client_pp().recover_missing);
         }).then_interruptible([this] {
           return do_recover_missing(pg, get_target_oid(), osd_reqid_t());
-        }).then_interruptible([this] {
+        }).then_interruptible([this](bool unfound) {
+          if (unfound) {
+            throw std::system_error(
+              std::make_error_code(std::errc::operation_canceled),
+              fmt::format("{} is unfound, drop it!", get_target_oid()));
+          }
           return enter_stage<interruptor>(
             client_pp().get_obc);
         }).then_interruptible([this] () -> PG::load_obc_iertr::future<> {
@@ -128,6 +133,9 @@ seastar::future<> InternalClientRequest::start()
       }, pg, start_epoch);
     }).then([this] {
       track_event<CompletionEvent>();
+    }).handle_exception_type([](std::system_error &error) {
+      logger().debug("error {}, message: {}", error.code(), error.what());
+      return seastar::now();
     }).finally([this] {
       logger().debug("{}: exit", *this);
       handle.exit();

src/crimson/osd/pg_recovery.cc

@@ -433,6 +433,7 @@ void PGRecovery::on_global_recover (
   auto& recovery_waiter = pg->get_recovery_backend()->get_recovering(soid);
   recovery_waiter.set_recovered();
   pg->get_recovery_backend()->remove_recovering(soid);
+  pg->get_recovery_backend()->found_and_remove(soid);
 }
 
 void PGRecovery::on_failed_recover(

src/crimson/osd/recovery_backend.cc

@@ -43,7 +43,7 @@ void RecoveryBackend::clear_temp_obj(const hobject_t &oid)
 }
 
 void RecoveryBackend::clean_up(ceph::os::Transaction& t,
-			       std::string_view why)
+			       interrupt_cause_t why)
 {
   for_each_temp_obj([&](auto &soid) {
     t.remove(pg.get_collection_ref()->get_cid(),
@@ -65,6 +65,36 @@ void RecoveryBackend::clean_up(ceph::os::Transaction& t,
   recovering.clear();
 }
 
+void RecoveryBackend::WaitForObjectRecovery::interrupt(interrupt_cause_t why) {
+  switch(why) {
+  case interrupt_cause_t::INTERVAL_CHANGE:
+    if (readable) {
+      readable->set_exception(
+	crimson::common::actingset_changed(pg.is_primary()));
+      readable.reset();
+    }
+    if (recovered) {
+      recovered->set_exception(
+	crimson::common::actingset_changed(pg.is_primary()));
+      recovered.reset();
+    }
+    if (pulled) {
+      pulled->set_exception(
+	crimson::common::actingset_changed(pg.is_primary()));
+      pulled.reset();
+    }
+    for (auto& [pg_shard, pr] : pushes) {
+      pr.set_exception(
+	crimson::common::actingset_changed(pg.is_primary()));
+    }
+    pushes.clear();
+    break;
+  default:
+    ceph_abort("impossible");
+    break;
+  }
+}
+
 void RecoveryBackend::WaitForObjectRecovery::stop() {
   if (readable) {
     readable->set_exception(

src/crimson/osd/recovery_backend.h

@@ -45,10 +45,22 @@ class RecoveryBackend {
       backend{backend} {}
   virtual ~RecoveryBackend() {}
   std::pair<WaitForObjectRecovery&, bool> add_recovering(const hobject_t& soid) {
-    auto [it, added] = recovering.emplace(soid, new WaitForObjectRecovery{});
+    auto [it, added] = recovering.emplace(soid, new WaitForObjectRecovery(pg));
     assert(it->second);
     return {*(it->second), added};
   }
+  seastar::future<> add_unfound(const hobject_t &soid) {
+    auto [it, added] = unfound.emplace(soid, seastar::shared_promise());
+    return it->second.get_shared_future();
+  }
+  void found_and_remove(const hobject_t &soid) {
+    auto it = unfound.find(soid);
+    if (it != unfound.end()) {
+      auto &found_promise = it->second;
+      found_promise.set_value();
+      unfound.erase(it);
+    }
+  }
   WaitForObjectRecovery& get_recovering(const hobject_t& soid) {
     assert(is_recovering(soid));
     return *(recovering.at(soid));
@@ -82,14 +94,22 @@ class RecoveryBackend {
     std::int64_t min,
     std::int64_t max);
 
+  enum interrupt_cause_t : uint8_t {
+    INTERVAL_CHANGE,
+    MAX
+  };
   void on_peering_interval_change(ceph::os::Transaction& t) {
-    clean_up(t, "new peering interval");
+    clean_up(t, interrupt_cause_t::INTERVAL_CHANGE);
   }
 
   seastar::future<> stop() {
     for (auto& [soid, recovery_waiter] : recovering) {
       recovery_waiter->stop();
     }
+    for (auto& [soid, promise] : unfound) {
+      promise.set_exception(
+	crimson::common::system_shutdown_exception());
+    }
     return on_stop();
   }
 protected:
@@ -124,11 +144,14 @@ class RecoveryBackend {
     public boost::intrusive_ref_counter<
       WaitForObjectRecovery, boost::thread_unsafe_counter>,
     public crimson::BlockerT<WaitForObjectRecovery> {
+      crimson::osd::PG &pg;
     std::optional<seastar::shared_promise<>> readable, recovered, pulled;
     std::map<pg_shard_t, seastar::shared_promise<>> pushes;
   public:
     static constexpr const char* type_name = "WaitForObjectRecovery";
 
+    WaitForObjectRecovery(crimson::osd::PG &pg) : pg(pg) {}
+
     crimson::osd::ObjectContextRef obc;
     std::optional<pull_info_t> pull_info;
     std::map<pg_shard_t, push_info_t> pushing;
@@ -204,28 +227,7 @@ class RecoveryBackend {
 	pushes.erase(it);
       }
     }
-    void interrupt(std::string_view why) {
-      if (readable) {
-	readable->set_exception(std::system_error(
-	  std::make_error_code(std::errc::interrupted), why.data()));
-	readable.reset();
-      }
-      if (recovered) {
-	recovered->set_exception(std::system_error(
-	  std::make_error_code(std::errc::interrupted), why.data()));
-	recovered.reset();
-      }
-      if (pulled) {
-	pulled->set_exception(std::system_error(
-	  std::make_error_code(std::errc::interrupted), why.data()));
-	pulled.reset();
-      }
-      for (auto& [pg_shard, pr] : pushes) {
-	pr.set_exception(std::system_error(
-	  std::make_error_code(std::errc::interrupted), why.data()));
-      }
-      pushes.clear();
-    }
+    void interrupt(interrupt_cause_t why);
     void stop();
     void dump_detail(Formatter* f) const {
     }
@@ -235,6 +237,7 @@ class RecoveryBackend {
   using WaitForObjectRecoveryRef = boost::intrusive_ptr<WaitForObjectRecovery>;
 protected:
   std::map<hobject_t, WaitForObjectRecoveryRef> recovering;
+  std::map<hobject_t, seastar::shared_promise<>> unfound;
   hobject_t get_temp_recovery_object(
     const hobject_t& target,
     eversion_t version) const;
@@ -249,7 +252,7 @@ class RecoveryBackend {
     backend->clear_temp_objs();
   }
 
-  void clean_up(ceph::os::Transaction& t, std::string_view why);
+  void clean_up(ceph::os::Transaction& t, interrupt_cause_t why);
   virtual seastar::future<> on_stop() = 0;
 private:
   void handle_backfill_finish(