Merge PR ceph#57469 into main

* refs/pull/57469/head:
	mds: set dispatcher order
	mds: use regular dispatch for processing beacons
	msg: add priority to dispatcher invocation order
	mds: note when dispatcher is called

Reviewed-by: Leonid Usov <[email protected]>

Author: batrick

src/mds/Beacon.cc

@@ -102,19 +102,9 @@ void Beacon::init(const MDSMap &mdsmap)
   });
 }
 
-bool Beacon::ms_can_fast_dispatch2(const cref_t<Message>& m) const
-{
-  return m->get_type() == MSG_MDS_BEACON;
-}
-
-void Beacon::ms_fast_dispatch2(const ref_t<Message>& m)
-{
-  bool handled = ms_dispatch2(m);
-  ceph_assert(handled);
-}
-
 bool Beacon::ms_dispatch2(const ref_t<Message>& m)
 {
+  dout(25) << __func__ << ": processing " << m << dendl;
   if (m->get_type() == MSG_MDS_BEACON) {
     if (m->get_connection()->get_peer_type() == CEPH_ENTITY_TYPE_MON) {
       handle_mds_beacon(ref_cast<MMDSBeacon>(m));

src/mds/Beacon.h

@@ -53,9 +53,6 @@ class Beacon : public Dispatcher
   void init(const MDSMap &mdsmap);
   void shutdown();
 
-  bool ms_can_fast_dispatch_any() const override { return true; }
-  bool ms_can_fast_dispatch2(const cref_t<Message>& m) const override;
-  void ms_fast_dispatch2(const ref_t<Message>& m) override;
   bool ms_dispatch2(const ref_t<Message> &m) override;
   void ms_handle_connect(Connection *c) override {}
   bool ms_handle_reset(Connection *c) override {return false;}

src/mds/MDSDaemon.cc

@@ -573,8 +573,10 @@ int MDSDaemon::init()
   dout(10) << sizeof(Capability) << "\tCapability" << dendl;
   dout(10) << sizeof(xlist<void*>::item) << "\txlist<>::item" << dendl;
 
-  messenger->add_dispatcher_tail(&beacon);
-  messenger->add_dispatcher_tail(this);
+  // Ensure beacons are processed ahead of most other dispatchers.
+  messenger->add_dispatcher_head(&beacon, Dispatcher::PRIORITY_HIGH);
+  // order last as MDSDaemon::ms_dispatch2 first acquires the mds_lock
+  messenger->add_dispatcher_head(this, Dispatcher::PRIORITY_LOW);
 
   // init monc
   monc->set_messenger(messenger);
@@ -979,6 +981,7 @@ void MDSDaemon::respawn()
 
 bool MDSDaemon::ms_dispatch2(const ref_t<Message> &m)
 {
+  dout(25) << __func__ << ": processing " << m << dendl;
   std::lock_guard l(mds_lock);
   if (stopping) {
     return false;

src/mds/MDSRank.cc

@@ -611,7 +611,7 @@ MDSRank::~MDSRank()
 void MDSRankDispatcher::init()
 {
   objecter->init();
-  messenger->add_dispatcher_head(objecter);
+  messenger->add_dispatcher_tail(objecter); // the default priority
 
   objecter->start();
 
@@ -2137,7 +2137,7 @@ void MDSRank::active_start()
 
   dout(10) << __func__ << ": initializing metrics handler" << dendl;
   metrics_handler.init();
-  messenger->add_dispatcher_tail(&metrics_handler);
+  messenger->add_dispatcher_tail(&metrics_handler, Dispatcher::PRIORITY_HIGH);
 
   // metric aggregation is solely done by rank 0
   if (is_rank0()) {

src/mds/MetricAggregator.cc

@@ -127,6 +127,7 @@ void MetricAggregator::shutdown() {
 }
 
 bool MetricAggregator::ms_dispatch2(const ref_t<Message> &m) {
+  dout(25) << " processing " << m << dendl;
   if (m->get_type() == MSG_MDS_METRICS &&
       m->get_connection()->get_peer_type() == CEPH_ENTITY_TYPE_MDS) {
     const Message *msg = m.get();

src/mon/MonClient.cc

@@ -131,7 +131,7 @@ int MonClient::get_monmap_and_config()
   messenger = Messenger::create_client_messenger(
     cct, "temp_mon_client");
   ceph_assert(messenger);
-  messenger->add_dispatcher_head(this);
+  messenger->add_dispatcher_head(this, Dispatcher::PRIORITY_HIGH);
   messenger->start();
   auto shutdown_msgr = make_scope_guard([this] {
     messenger->shutdown();
@@ -265,7 +265,7 @@ int MonClient::ping_monitor(const string &mon_id, string *result_reply)
 						result_reply);
 
   Messenger *smsgr = Messenger::create_client_messenger(cct, "temp_ping_client");
-  smsgr->add_dispatcher_head(pinger);
+  smsgr->add_dispatcher_head(pinger, Dispatcher::PRIORITY_HIGH);
   smsgr->set_auth_client(pinger);
   smsgr->start();
 
@@ -295,6 +295,7 @@ int MonClient::ping_monitor(const string &mon_id, string *result_reply)
 
 bool MonClient::ms_dispatch(Message *m)
 {
+  ldout(cct, 25) << __func__ << " processing " << m << dendl;
   // we only care about these message types
   switch (m->get_type()) {
   case CEPH_MSG_MON_MAP:
@@ -511,7 +512,7 @@ int MonClient::init()
   initialized = true;
 
   messenger->set_auth_client(this);
-  messenger->add_dispatcher_head(this);
+  messenger->add_dispatcher_head(this, Dispatcher::PRIORITY_HIGH);
 
   timer.init();
   schedule_tick();

src/msg/Dispatcher.h

@@ -29,6 +29,12 @@ class KeyStore;
 
 class Dispatcher {
 public:
+  /* Ordering of dispatch for a list of Dispatchers. */
+  using priority_t = uint32_t;
+  static constexpr priority_t PRIORITY_HIGH = std::numeric_limits<priority_t>::max() / 4;
+  static constexpr priority_t PRIORITY_DEFAULT = std::numeric_limits<priority_t>::max() / 2;
+  static constexpr priority_t PRIORITY_LOW = (std::numeric_limits<priority_t>::max() / 4) * 3;
+
   explicit Dispatcher(CephContext *cct_)
     : cct(cct_)
   {

src/msg/Messenger.h

@@ -17,9 +17,9 @@
 #ifndef CEPH_MESSENGER_H
 #define CEPH_MESSENGER_H
 
-#include <deque>
 #include <map>
 #include <optional>
+#include <vector>
 
 #include <errno.h>
 #include <sstream>
@@ -92,8 +92,18 @@ struct Interceptor {
 
 class Messenger {
 private:
-  std::deque<Dispatcher*> dispatchers;
-  std::deque<Dispatcher*> fast_dispatchers;
+  struct PriorityDispatcher {
+    using priority_t = Dispatcher::priority_t;
+    priority_t priority;
+    Dispatcher* dispatcher;
+
+    bool operator<(const PriorityDispatcher& other) const {
+      return priority < other.priority;
+    }
+  };
+  std::vector<PriorityDispatcher> dispatchers;
+  std::vector<PriorityDispatcher> fast_dispatchers;
+
   ZTracer::Endpoint trace_endpoint;
 
 protected:
@@ -389,11 +399,14 @@ class Messenger {
    *
    * @param d The Dispatcher to insert into the list.
    */
-  void add_dispatcher_head(Dispatcher *d) {
+  void add_dispatcher_head(Dispatcher *d, PriorityDispatcher::priority_t priority=Dispatcher::PRIORITY_DEFAULT) {
     bool first = dispatchers.empty();
-    dispatchers.push_front(d);
-    if (d->ms_can_fast_dispatch_any())
-      fast_dispatchers.push_front(d);
+    dispatchers.insert(dispatchers.begin(), PriorityDispatcher{priority, d});
+    std::stable_sort(dispatchers.begin(), dispatchers.end());
+    if (d->ms_can_fast_dispatch_any()) {
+      fast_dispatchers.insert(fast_dispatchers.begin(), PriorityDispatcher{priority, d});
+      std::stable_sort(fast_dispatchers.begin(), fast_dispatchers.end());
+    }
     if (first)
       ready();
   }
@@ -404,11 +417,14 @@ class Messenger {
    *
    * @param d The Dispatcher to insert into the list.
    */
-  void add_dispatcher_tail(Dispatcher *d) {
+  void add_dispatcher_tail(Dispatcher *d, PriorityDispatcher::priority_t priority=Dispatcher::PRIORITY_DEFAULT) {
     bool first = dispatchers.empty();
-    dispatchers.push_back(d);
-    if (d->ms_can_fast_dispatch_any())
-      fast_dispatchers.push_back(d);
+    dispatchers.push_back(PriorityDispatcher{priority, d});
+    std::stable_sort(dispatchers.begin(), dispatchers.end());
+    if (d->ms_can_fast_dispatch_any()) {
+      fast_dispatchers.push_back(PriorityDispatcher{priority, d});
+      std::stable_sort(fast_dispatchers.begin(), fast_dispatchers.end());
+    }
     if (first)
       ready();
   }
@@ -667,9 +683,10 @@ class Messenger {
    * @param m The Message we are testing.
    */
   bool ms_can_fast_dispatch(const ceph::cref_t<Message>& m) {
-    for (const auto &dispatcher : fast_dispatchers) {
-      if (dispatcher->ms_can_fast_dispatch2(m))
-	return true;
+    for ([[maybe_unused]] const auto& [priority, dispatcher] : fast_dispatchers) {
+      if (dispatcher->ms_can_fast_dispatch2(m)) {
+        return true;
+      }
     }
     return false;
   }
@@ -682,10 +699,10 @@ class Messenger {
    */
   void ms_fast_dispatch(const ceph::ref_t<Message> &m) {
     m->set_dispatch_stamp(ceph_clock_now());
-    for (const auto &dispatcher : fast_dispatchers) {
+    for ([[maybe_unused]] const auto& [priority, dispatcher] : fast_dispatchers) {
       if (dispatcher->ms_can_fast_dispatch2(m)) {
-	dispatcher->ms_fast_dispatch2(m);
-	return;
+        dispatcher->ms_fast_dispatch2(m);
+        return;
       }
     }
     ceph_abort();
@@ -697,7 +714,7 @@ class Messenger {
    *
    */
   void ms_fast_preprocess(const ceph::ref_t<Message> &m) {
-    for (const auto &dispatcher : fast_dispatchers) {
+    for ([[maybe_unused]] const auto& [priority, dispatcher] : fast_dispatchers) {
       dispatcher->ms_fast_preprocess2(m);
     }
   }
@@ -710,9 +727,10 @@ class Messenger {
    */
   void ms_deliver_dispatch(const ceph::ref_t<Message> &m) {
     m->set_dispatch_stamp(ceph_clock_now());
-    for (const auto &dispatcher : dispatchers) {
-      if (dispatcher->ms_dispatch2(m))
-	return;
+    for ([[maybe_unused]] const auto& [priority, dispatcher] : dispatchers) {
+      if (dispatcher->ms_dispatch2(m)) {
+        return;
+      }
     }
     lsubdout(cct, ms, 0) << "ms_deliver_dispatch: unhandled message " << m << " " << *m << " from "
 			 << m->get_source_inst() << dendl;
@@ -729,7 +747,7 @@ class Messenger {
    * @param con Pointer to the new Connection.
    */
   void ms_deliver_handle_connect(Connection *con) {
-    for (const auto& dispatcher : dispatchers) {
+    for ([[maybe_unused]] const auto& [priority, dispatcher] : dispatchers) {
       dispatcher->ms_handle_connect(con);
     }
   }
@@ -742,7 +760,7 @@ class Messenger {
    * @param con Pointer to the new Connection.
    */
   void ms_deliver_handle_fast_connect(Connection *con) {
-    for (const auto& dispatcher : fast_dispatchers) {
+    for ([[maybe_unused]] const auto& [priority, dispatcher] : fast_dispatchers) {
       dispatcher->ms_handle_fast_connect(con);
     }
   }
@@ -754,7 +772,7 @@ class Messenger {
    * @param con Pointer to the new Connection.
    */
   void ms_deliver_handle_accept(Connection *con) {
-    for (const auto& dispatcher : dispatchers) {
+    for ([[maybe_unused]] const auto& [priority, dispatcher] : dispatchers) {
       dispatcher->ms_handle_accept(con);
     }
   }
@@ -766,7 +784,7 @@ class Messenger {
    * @param con Pointer to the new Connection.
    */
   void ms_deliver_handle_fast_accept(Connection *con) {
-    for (const auto& dispatcher : fast_dispatchers) {
+    for ([[maybe_unused]] const auto& [priority, dispatcher] : fast_dispatchers) {
       dispatcher->ms_handle_fast_accept(con);
     }
   }
@@ -779,9 +797,10 @@ class Messenger {
    * @param con Pointer to the broken Connection.
    */
   void ms_deliver_handle_reset(Connection *con) {
-    for (const auto& dispatcher : dispatchers) {
-      if (dispatcher->ms_handle_reset(con))
-	return;
+    for ([[maybe_unused]] const auto& [priority, dispatcher] : dispatchers) {
+      if (dispatcher->ms_handle_reset(con)) {
+        return;
+      }
     }
   }
   /**
@@ -792,7 +811,7 @@ class Messenger {
    * @param con Pointer to the broken Connection.
    */
   void ms_deliver_handle_remote_reset(Connection *con) {
-    for (const auto& dispatcher : dispatchers) {
+    for ([[maybe_unused]] const auto& [priority, dispatcher] : dispatchers) {
       dispatcher->ms_handle_remote_reset(con);
     }
   }
@@ -806,9 +825,10 @@ class Messenger {
    * @param con Pointer to the broken Connection.
    */
   void ms_deliver_handle_refused(Connection *con) {
-    for (const auto& dispatcher : dispatchers) {
-      if (dispatcher->ms_handle_refused(con))
+    for ([[maybe_unused]] const auto& [priority, dispatcher] : dispatchers) {
+      if (dispatcher->ms_handle_refused(con)) {
         return;
+      }
     }
   }