Merge pull request ceph#55340 from ronen-fr/wip-rf-reserver2

osd/scrub: use an AsyncReserver to handle scrub reservations on the replica side

Reviewed-by: Samuel Just <[email protected]>-

Author: ronen-fr

src/common/AsyncReserver.h

@@ -16,6 +16,9 @@
 #define ASYNC_RESERVER_H
 
 #include "common/Formatter.h"
+#include "common/ceph_context.h"
+#include "common/ceph_mutex.h"
+#include "include/Context.h"
 
 #define rdout(x) lgeneric_subdout(cct,reserver,x)
 
@@ -110,8 +113,10 @@ class AsyncReserver {
       if (it->second.empty()) {
 	queues.erase(it);
       }
-      f->queue(p.grant);
-      p.grant = nullptr;
+      if (p.grant) {
+	f->queue(p.grant);
+	p.grant = nullptr;
+      }
       in_progress[p.item] = p;
       if (p.preempt) {
 	preempt_by_prio.insert(std::make_pair(p.prio, p.item));
@@ -271,8 +276,7 @@ class AsyncReserver {
    * active reservations.
    */
   bool request_reservation_or_fail(
-      T item,		     ///< [in] reservation key
-      Context *on_reserved   ///< [in] callback to be called on reservation
+      T item		     ///< [in] reservation key
   )
   {
     std::lock_guard l(lock);
@@ -285,7 +289,7 @@ class AsyncReserver {
     }
 
     const unsigned prio = UINT_MAX;
-    Reservation r(item, prio, on_reserved, nullptr);
+    Reservation r(item, prio, nullptr, nullptr);
     queues[prio].push_back(r);
     queue_pointers.insert(std::make_pair(
 	item, std::make_pair(prio, --(queues[prio]).end())));

src/common/options/osd.yaml.in

@@ -535,6 +535,16 @@ options:
   see_also:
   - osd_scrub_slow_reservation_response
   with_legacy: false
+- name: osd_scrub_disable_reservation_queuing
+  type: bool
+  level: advanced
+  desc: Disable queuing of scrub reservations
+  long_desc: When set - scrub replica reservations are responded to immediately, with
+    either success or failure (the pre-Squid version behaviour). This configuration
+    option is introduced to support mixed-version clusters and debugging, and will
+    be removed in the next release.
+  default: false
+  with_legacy: false
 # where rados plugins are stored
 - name: osd_class_dir
   type: str

src/messages/MOSDScrubReserve.h

@@ -37,7 +37,7 @@ class MOSDScrubReserve : public MOSDFastDispatchOp {
   reservation_nonce_t reservation_nonce{0};
   /// 'false' if the (legacy) primary is expecting an immediate
   /// granted / denied response
-  bool wait_for_resources{false};
+  bool wait_for_resources{true};
 
   epoch_t get_map_epoch() const override {
     return map_epoch;
@@ -63,23 +63,23 @@ class MOSDScrubReserve : public MOSDFastDispatchOp {
   }
 
   void print(std::ostream& out) const {
-    out << "MOSDScrubReserve(" << pgid << " ";
+    out << "MOSDScrubReserve(" << pgid << ",";
     switch (type) {
     case REQUEST:
-      out << (wait_for_resources ? "QREQUEST " : "REQUEST ");
+      out << (wait_for_resources ? "QREQUEST" : "REQUEST");
       break;
     case GRANT:
-      out << "GRANT ";
+      out << "GRANT";
       break;
     case REJECT:
-      out << "REJECT ";
+      out << "REJECT";
       break;
     case RELEASE:
-      out << "RELEASE ";
+      out << "RELEASE";
       break;
     }
-    out << "e" << map_epoch << " from: " << from
-	<< " reservation_nonce: " << reservation_nonce << ")";
+    out << ",e:" << map_epoch << ",from:" << from
+	<< ",reservation_nonce:" << reservation_nonce << ")";
     return;
   }
 

src/osd/OSD.cc

@@ -273,6 +273,8 @@ OSDService::OSDService(OSD *osd, ceph::async::io_context_pool& poolctx) :
 		  cct->_conf->osd_min_recovery_priority),
   snap_reserver(cct, &reserver_finisher,
 		cct->_conf->osd_max_trimming_pgs),
+  scrub_reserver(cct, &reserver_finisher,
+		cct->_conf->osd_max_scrubs),
   recovery_ops_active(0),
   recovery_ops_reserved(0),
   recovery_paused(false),
@@ -2891,7 +2893,7 @@ will start to track new ops received afterwards.";
     f->close_section();
   } else if (prefix == "dump_scrub_reservations") {
     f->open_object_section("scrub_reservations");
-    service.get_scrub_services().resource_bookkeeper().dump_scrub_reservations(f);
+    service.get_scrub_services().dump_scrub_reservations(f);
     f->close_section();
   } else if (prefix == "get_latest_osdmap") {
     get_latest_osdmap();
@@ -9889,6 +9891,9 @@ void OSD::handle_conf_change(const ConfigProxy& conf,
   if (changed.count("osd_max_trimming_pgs")) {
     service.snap_reserver.set_max(cct->_conf->osd_max_trimming_pgs);
   }
+  if (changed.count("osd_max_scrubs")) {
+    service.scrub_reserver.set_max(cct->_conf->osd_max_scrubs);
+  }
   if (changed.count("osd_op_complaint_time") ||
       changed.count("osd_op_log_threshold")) {
     op_tracker.set_complaint_and_threshold(cct->_conf->osd_op_complaint_time,

src/osd/OSD.h

@@ -251,6 +251,14 @@ class OSDService : public Scrub::ScrubSchedListener {
    */
   std::optional<PGLockWrapper> get_locked_pg(spg_t pgid) final;
 
+  /**
+   * the entity that counts the number of active replica scrub
+   * operations, and grant scrub reservation requests asynchronously.
+   */
+  AsyncReserver<spg_t, Finisher>& get_scrub_reserver() {
+    return scrub_reserver;
+  }
+
  private:
   // -- agent shared state --
   ceph::mutex agent_lock = ceph::make_mutex("OSDService::agent_lock");
@@ -494,6 +502,8 @@ class OSDService : public Scrub::ScrubSchedListener {
   void send_pg_created();
 
   AsyncReserver<spg_t, Finisher> snap_reserver;
+  /// keeping track of replicas being reserved for scrubbing
+  AsyncReserver<spg_t, Finisher> scrub_reserver;
   void queue_recovery_context(PG *pg,
                               GenContext<ThreadPool::TPHandle&> *c,
                               uint64_t cost,

src/osd/scrubber/osd_scrub.cc

@@ -57,6 +57,14 @@ void OsdScrub::dump_scrubs(ceph::Formatter* f) const
   m_queue.dump_scrubs(f);
 }
 
+void OsdScrub::dump_scrub_reservations(ceph::Formatter* f) const
+{
+  m_resource_bookkeeper.dump_scrub_reservations(f);
+  f->open_array_section("remote_scrub_reservations");
+  m_osd_svc.get_scrub_reserver().dump(f);
+  f->close_section();
+}
+
 void OsdScrub::log_fwd(std::string_view text)
 {
   dout(20) << text << dendl;
@@ -468,21 +476,6 @@ void OsdScrub::dec_scrubs_local()
   m_resource_bookkeeper.dec_scrubs_local();
 }
 
-bool OsdScrub::inc_scrubs_remote(pg_t pgid)
-{
-  return m_resource_bookkeeper.inc_scrubs_remote(pgid);
-}
-
-void OsdScrub::enqueue_remote_reservation(pg_t pgid)
-{
-  m_resource_bookkeeper.enqueue_remote_reservation(pgid);
-}
-
-void OsdScrub::dec_scrubs_remote(pg_t pgid)
-{
-  m_resource_bookkeeper.dec_scrubs_remote(pgid);
-}
-
 void OsdScrub::mark_pg_scrub_blocked(spg_t blocked_pg)
 {
   m_queue.mark_pg_scrub_blocked(blocked_pg);

src/osd/scrubber/osd_scrub.h

@@ -51,6 +51,8 @@ class OsdScrub {
 
   void dump_scrubs(ceph::Formatter* f) const;  ///< fwd to the queue
 
+  void dump_scrub_reservations(ceph::Formatter* f) const;
+
   /**
    * on_config_change() (the refactored "OSD::sched_all_scrubs()")
    *
@@ -69,9 +71,6 @@ class OsdScrub {
   std::unique_ptr<Scrub::LocalResourceWrapper> inc_scrubs_local(
       bool is_high_priority);
   void dec_scrubs_local();
-  bool inc_scrubs_remote(pg_t pgid);
-  void enqueue_remote_reservation(pg_t pgid);
-  void dec_scrubs_remote(pg_t pgid);
 
   // counting the number of PGs stuck while scrubbing, waiting for objects
   void mark_pg_scrub_blocked(spg_t blocked_pg);

src/osd/scrubber/osd_scrub_sched.h

@@ -108,6 +108,7 @@ ScrubQueue interfaces (main functions):
 // clang-format on
 
 #include <optional>
+#include "common/AsyncReserver.h"
 #include "utime.h"
 #include "osd/scrubber/scrub_job.h"
 #include "osd/PG.h"
@@ -135,6 +136,12 @@ class ScrubSchedListener {
    */
   virtual std::optional<PGLockWrapper> get_locked_pg(spg_t pgid) = 0;
 
+  /**
+   * allow access to the scrub_reserver, the AsyncReserver that keeps track
+   * of 'remote replica reservations'.
+   */
+  virtual AsyncReserver<spg_t, Finisher>& get_scrub_reserver() = 0;
+
   virtual ~ScrubSchedListener() {}
 };
 

src/osd/scrubber/pg_scrubber.cc

@@ -397,6 +397,15 @@ void PgScrubber::send_scrub_is_finished(epoch_t epoch_queued)
   dout(10) << "scrubber event --<< " << __func__ << dendl;
 }
 
+void PgScrubber::send_granted_by_reserver(const AsyncScrubResData& req)
+{
+  dout(10) << "scrubber event -->> granted_by_reserver" << dendl;
+  if (check_interval(req.request_epoch)) {
+    m_fsm->process_event(Scrub::ReserverGranted{req});
+  }
+  dout(10) << "scrubber event --<< granted_by_reserver" << dendl;
+}
+
 // -----------------
 
 bool PgScrubber::is_reserving() const

src/osd/scrubber/pg_scrubber.h

@@ -227,6 +227,8 @@ class PgScrubber : public ScrubPgIF,
 
   void send_scrub_is_finished(epoch_t epoch_queued) final;
 
+  void send_granted_by_reserver(const AsyncScrubResData& req) final;
+
   /**
    *  we allow some number of preemptions of the scrub, which mean we do
    *  not block.  Then we start to block.  Once we start blocking, we do