msg: add alternate statuses for ms_dispatch2 handling

Many dispatchers return false to allow other dispatchers also common messages
like MOSDMap or MFSMap. They implicitly depend on some dispatcher which is
always at the "tail" of the dispatcher queue to return "true" indicating the
msg was processed to avoid messages like:

    2025-02-18T05:31:17.738+0000 7f5206546640  0 ms_deliver_dispatch: unhandled message 0x5632d05f0700 fsmap(e 9) from mon.0 v2:172.21.3.230:40412/0

but this cannot always happen when some libraries like the RadosClient used standalone.

So, add a variant for encapsulating other indications for how the message was
processed by dispatch2.  For example, a message may be "acknowledged" but
explicitly allow other dispatchers to try processing the message.

Note: we're using a variant to avoid updating all of the ms_dispatch code to
use the sentinel classes.

Signed-off-by: Patrick Donnelly <pdonnell@ibm.com>
(cherry picked from commit c9d0913f53b0c4d632746075c05c2f2187d77c02)

Conflicts:
	src/nvmeof/NVMeofGwMonitorClient.cc: not present in squid
	src/nvmeof/NVMeofGwMonitorClient.h: not present in squid

diff --git a/src/client/Client.cc b/src/client/Client.cc
index 9aca9e9d4b29996fec924186cfabc27a0123acf7..f62136f8d3bc37e959d403d58a080bd82e184cfa 100644 (file)
--- a/src/client/Client.cc
+++ b/src/client/Client.cc
@@ -3065,7 +3065,7 @@ void Client::handle_osd_map(const MConstRef<MOSDMap>& m)
 // incoming messages
 
 
-bool Client::ms_dispatch2(const MessageRef &m)
+Dispatcher::dispatch_result_t Client::ms_dispatch2(const MessageRef &m)
 {
   RWRef_t iref_reader(initialize_state, CLIENT_INITIALIZED);
   if (!iref_reader.is_state_satisfied()) {

diff --git a/src/client/Client.h b/src/client/Client.h
index 19cc7a81d6585b92a3376d6a76f193ee6701ea7a..76b9a1d8ea4ec959b1822efc397d4b5753d365b2 100644 (file)
--- a/src/client/Client.h
+++ b/src/client/Client.h
@@ -1160,7 +1160,7 @@ protected:
 
   void dump_status(Formatter *f);  // debug
 
-  bool ms_dispatch2(const MessageRef& m) override;
+  Dispatcher::dispatch_result_t ms_dispatch2(const MessageRef& m) override;
 
   void ms_handle_connect(Connection *con) override;
   bool ms_handle_reset(Connection *con) override;

diff --git a/src/mds/Beacon.cc b/src/mds/Beacon.cc
index 4673a010b43df8d62ccf618b541f30a1e019d2a9..01d016130960be610aaa17c74c4644beabb6d117 100644 (file)
--- a/src/mds/Beacon.cc
+++ b/src/mds/Beacon.cc
@@ -103,7 +103,7 @@ void Beacon::init(const MDSMap &mdsmap)
   });
 }
 
-bool Beacon::ms_dispatch2(const ref_t<Message>& m)
+Dispatcher::dispatch_result_t Beacon::ms_dispatch2(const ref_t<Message>& m)
 {
   dout(25) << __func__ << ": processing " << m << dendl;
   if (m->get_type() == MSG_MDS_BEACON) {

diff --git a/src/mds/Beacon.h b/src/mds/Beacon.h
index fb7d1cffa2941fddabeb0ed161e5b74ceea279cb..3f0c336f2bc50c7b38cdab405e11c1d6ddab740f 100644 (file)
--- a/src/mds/Beacon.h
+++ b/src/mds/Beacon.h
@@ -53,7 +53,7 @@ public:
   void init(const MDSMap &mdsmap);
   void shutdown();
 
-  bool ms_dispatch2(const ref_t<Message> &m) override;
+  Dispatcher::dispatch_result_t ms_dispatch2(const ref_t<Message> &m) override;
   void ms_handle_connect(Connection *c) override {}
   bool ms_handle_reset(Connection *c) override {return false;}
   void ms_handle_remote_reset(Connection *c) override {}

diff --git a/src/mds/MDSDaemon.cc b/src/mds/MDSDaemon.cc
index 6e54b4b9a3ad42bfe1b68070c32a8cf1f91164c7..60fb3a874229789ef405ff55de4a1b0b85c59005 100644 (file)
--- a/src/mds/MDSDaemon.cc
+++ b/src/mds/MDSDaemon.cc
@@ -984,7 +984,7 @@ void MDSDaemon::respawn()
 
 
 
-bool MDSDaemon::ms_dispatch2(const ref_t<Message> &m)
+Dispatcher::dispatch_result_t MDSDaemon::ms_dispatch2(const ref_t<Message> &m)
 {
   dout(25) << __func__ << ": processing " << m << dendl;
   std::lock_guard l(mds_lock);

diff --git a/src/mds/MDSDaemon.h b/src/mds/MDSDaemon.h
index c1999a32029c832ecf93232f3dfb0b6f922507c6..e7cb3151bfa163f8bc0059225edd6acbdb6e1646 100644 (file)
--- a/src/mds/MDSDaemon.h
+++ b/src/mds/MDSDaemon.h
@@ -145,7 +145,7 @@ class MDSDaemon : public Dispatcher {
   class MDSSocketHook *asok_hook = nullptr;
 
  private:
-  bool ms_dispatch2(const ref_t<Message> &m) override;
+  Dispatcher::dispatch_result_t ms_dispatch2(const ref_t<Message> &m) override;
   bool ms_handle_fast_authentication(Connection *con) override;
   void ms_handle_accept(Connection *con) override;
   void ms_handle_connect(Connection *con) override;

diff --git a/src/mds/MetricAggregator.cc b/src/mds/MetricAggregator.cc
index 6cdc3ab193e988cbd86e7b01390caffcc1ba6648..d854d37dde681fddf4c6117aa1313b854697d790 100644 (file)
--- a/src/mds/MetricAggregator.cc
+++ b/src/mds/MetricAggregator.cc
@@ -126,7 +126,7 @@ void MetricAggregator::shutdown() {
   }
 }
 
-bool MetricAggregator::ms_dispatch2(const ref_t<Message> &m) {
+Dispatcher::dispatch_result_t 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) {

diff --git a/src/mds/MetricAggregator.h b/src/mds/MetricAggregator.h
index 72c37217e624d6288db3f8c044fa07d398216365..677cf39829c57f5466b63910ef78da8b1c0d6705 100644 (file)
--- a/src/mds/MetricAggregator.h
+++ b/src/mds/MetricAggregator.h
@@ -34,7 +34,7 @@ public:
 
   void notify_mdsmap(const MDSMap &mdsmap);
 
-  bool ms_dispatch2(const ref_t<Message> &m) override;
+  Dispatcher::dispatch_result_t ms_dispatch2(const ref_t<Message> &m) override;
 
   void ms_handle_connect(Connection *c) override {
   }

diff --git a/src/mds/MetricsHandler.cc b/src/mds/MetricsHandler.cc
index 833d7ef19984c0bd97e95fda3f38694a262e7ed2..6b7484ea9e95c622fa1f85c778c912c3f5f6a45e 100644 (file)
--- a/src/mds/MetricsHandler.cc
+++ b/src/mds/MetricsHandler.cc
@@ -20,7 +20,7 @@ MetricsHandler::MetricsHandler(CephContext *cct, MDSRank *mds)
     mds(mds) {
 }
 
-bool MetricsHandler::ms_dispatch2(const ref_t<Message> &m) {
+Dispatcher::dispatch_result_t MetricsHandler::ms_dispatch2(const ref_t<Message> &m) {
   if (m->get_type() == CEPH_MSG_CLIENT_METRICS &&
       m->get_connection()->get_peer_type() == CEPH_ENTITY_TYPE_CLIENT) {
     handle_client_metrics(ref_cast<MClientMetrics>(m));

diff --git a/src/mds/MetricsHandler.h b/src/mds/MetricsHandler.h
index 25ee208aa956226467da27e4294667589a7cfb01..9f6fc11b55f7e85ce17977c02893a9b10763a29e 100644 (file)
--- a/src/mds/MetricsHandler.h
+++ b/src/mds/MetricsHandler.h
@@ -25,7 +25,7 @@ class MetricsHandler : public Dispatcher {
 public:
   MetricsHandler(CephContext *cct, MDSRank *mds);
 
-  bool ms_dispatch2(const ref_t<Message> &m) override;
+  Dispatcher::dispatch_result_t ms_dispatch2(const ref_t<Message> &m) override;
 
   void ms_handle_connect(Connection *c) override {
   }

diff --git a/src/mgr/DaemonServer.cc b/src/mgr/DaemonServer.cc
index 60c0637d551a051ea679b12e780b53a14eb7c72e..34e18b0bf74deee16340a99a4a33a4e393015f41 100644 (file)
--- a/src/mgr/DaemonServer.cc
+++ b/src/mgr/DaemonServer.cc
@@ -339,7 +339,7 @@ bool DaemonServer::ms_handle_refused(Connection *con)
   return false;
 }
 
-bool DaemonServer::ms_dispatch2(const ref_t<Message>& m)
+Dispatcher::dispatch_result_t DaemonServer::ms_dispatch2(const ref_t<Message>& m)
 {
   // Note that we do *not* take ::lock here, in order to avoid
   // serializing all message handling.  It's up to each handler

diff --git a/src/mgr/DaemonServer.h b/src/mgr/DaemonServer.h
index c1c92883f3d5bc6de878dec50e689d96e790f808..719b44634b652709b88ff3cad8273ca52cabab2d 100644 (file)
--- a/src/mgr/DaemonServer.h
+++ b/src/mgr/DaemonServer.h
@@ -273,7 +273,7 @@ public:
               LogChannelRef auditcl);
   ~DaemonServer() override;
 
-  bool ms_dispatch2(const ceph::ref_t<Message>& m) override;
+  Dispatcher::dispatch_result_t ms_dispatch2(const ceph::ref_t<Message>& m) override;
   bool ms_handle_fast_authentication(Connection *con) override;
   void ms_handle_accept(Connection *con) override;
   bool ms_handle_reset(Connection *con) override;

diff --git a/src/mgr/MgrClient.cc b/src/mgr/MgrClient.cc
index 6250ea3b9f18ea0d0ddcae2fc4d0c3514e18e323..291ca5c5531cc1a209b3524d426abcd3419dd4e3 100644 (file)
--- a/src/mgr/MgrClient.cc
+++ b/src/mgr/MgrClient.cc
@@ -97,7 +97,7 @@ void MgrClient::shutdown()
   }
 }
 
-bool MgrClient::ms_dispatch2(const ref_t<Message>& m)
+Dispatcher::dispatch_result_t MgrClient::ms_dispatch2(const ref_t<Message>& m)
 {
   std::lock_guard l(lock);
 

diff --git a/src/mgr/MgrClient.h b/src/mgr/MgrClient.h
index 1f9bb397fbec16b2efa0b93de54a54be0daea2d5..8f7feaac7dd57af4e3681994bc0117e51560dfd1 100644 (file)
--- a/src/mgr/MgrClient.h
+++ b/src/mgr/MgrClient.h
@@ -120,7 +120,7 @@ public:
 
   void set_mgr_optional(bool optional_) {mgr_optional = optional_;}
 
-  bool ms_dispatch2(const ceph::ref_t<Message>& m) override;
+  Dispatcher::dispatch_result_t ms_dispatch2(const ceph::ref_t<Message>& m) override;
   bool ms_handle_reset(Connection *con) override;
   void ms_handle_remote_reset(Connection *con) override {}
   bool ms_handle_refused(Connection *con) override;

diff --git a/src/mgr/MgrStandby.cc b/src/mgr/MgrStandby.cc
index 3fc14d9eb29d76fa362fd7b36b6edf18e6acee4c..cb2bd10f92dd02f1028d01490539f0f3ddcb9ee9 100644 (file)
--- a/src/mgr/MgrStandby.cc
+++ b/src/mgr/MgrStandby.cc
@@ -402,7 +402,7 @@ void MgrStandby::handle_mgr_map(ref_t<MMgrMap> mmap)
   }
 }
 
-bool MgrStandby::ms_dispatch2(const ref_t<Message>& m)
+Dispatcher::dispatch_result_t MgrStandby::ms_dispatch2(const ref_t<Message>& m)
 {
   std::lock_guard l(lock);
   dout(10) << state_str() << " " << *m << dendl;

diff --git a/src/mgr/MgrStandby.h b/src/mgr/MgrStandby.h
index 35ce43c0c808582057906565183882c2dbe1c9e0..ab5a8d05cbf02f9461f5c8deff1ba3c51263ac55 100644 (file)
--- a/src/mgr/MgrStandby.h
+++ b/src/mgr/MgrStandby.h
@@ -71,7 +71,7 @@ public:
   MgrStandby(int argc, const char **argv);
   ~MgrStandby() override;
 
-  bool ms_dispatch2(const ceph::ref_t<Message>& m) override;
+  Dispatcher::dispatch_result_t ms_dispatch2(const ceph::ref_t<Message>& m) override;
   bool ms_handle_reset(Connection *con) override { return false; }
   void ms_handle_remote_reset(Connection *con) override {}
   bool ms_handle_refused(Connection *con) override;

diff --git a/src/msg/Dispatcher.h b/src/msg/Dispatcher.h
index 8b3c80542ac6c2a29581df6d2eade25289d2c050..312336a79e629347a755742b078210cb3affd640 100644 (file)
--- a/src/msg/Dispatcher.h
+++ b/src/msg/Dispatcher.h
@@ -22,6 +22,8 @@
 #include "include/common_fwd.h"
 #include "msg/MessageRef.h"
 
+#include <variant>
+
 class Messenger;
 class Connection;
 class CryptoKey;
@@ -124,7 +126,24 @@ public:
   }
 
   /* ms_dispatch2 because otherwise the child must define both */
-  virtual bool ms_dispatch2(const MessageRef &m) {
+  struct HANDLED {};
+  struct UNHANDLED {};
+  struct ACKNOWLEDGED {};
+  typedef std::variant<bool, HANDLED, UNHANDLED, ACKNOWLEDGED> dispatch_result_t;
+
+  static inline dispatch_result_t fold_dispatch_result(dispatch_result_t r) {
+    if (std::holds_alternative<bool>(r)) {
+      if (std::get<bool>(r)) {
+        return HANDLED();
+      } else {
+        return UNHANDLED();
+      }
+    } else {
+      return r;
+    }
+  }
+
+  virtual dispatch_result_t ms_dispatch2(const MessageRef &m) {
     /* allow old style dispatch handling that expects a Message * with a floating ref */
     MessageRef mr(m);
     if (ms_dispatch(mr.get())) {

diff --git a/src/msg/Messenger.h b/src/msg/Messenger.h
index 078418fe715fce45674f2fb157f02cd8db9da2de..53ffab4e67ef650797647320b4d0ae5651865c69 100644 (file)
--- a/src/msg/Messenger.h
+++ b/src/msg/Messenger.h
@@ -736,11 +736,17 @@ public:
    */
   void ms_deliver_dispatch(const ceph::ref_t<Message> &m) {
     m->set_dispatch_stamp(ceph_clock_now());
+    bool acked = false;
     for ([[maybe_unused]] const auto& [priority, dispatcher] : dispatchers) {
-      if (dispatcher->ms_dispatch2(m)) {
+      auto r = Dispatcher::fold_dispatch_result(dispatcher->ms_dispatch2(m));
+      if (std::holds_alternative<Dispatcher::HANDLED>(r)) {
         return;
+      } else if (std::holds_alternative<Dispatcher::ACKNOWLEDGED>(r)) {
+        acked = true;
       }
     }
+    if (acked)
+      return;
     lsubdout(cct, ms, 0) << "ms_deliver_dispatch: unhandled message " << m << " " << *m << " from "
                         << m->get_source_inst() << dendl;
     ceph_assert(!cct->_conf->ms_die_on_unhandled_msg);

diff --git a/src/tools/cephfs_mirror/ClusterWatcher.cc b/src/tools/cephfs_mirror/ClusterWatcher.cc
index 8d53a1c725412e618ce81d8646222a3d260750c3..133133a9d1190866a4da2e6fd0c54356d5dfe5da 100644 (file)
--- a/src/tools/cephfs_mirror/ClusterWatcher.cc
+++ b/src/tools/cephfs_mirror/ClusterWatcher.cc
@@ -31,7 +31,7 @@ ClusterWatcher::ClusterWatcher(CephContext *cct, MonClient *monc, ServiceDaemon
 ClusterWatcher::~ClusterWatcher() {
 }
 
-bool ClusterWatcher::ms_dispatch2(const ref_t<Message> &m) {
+Dispatcher::dispatch_result_t ClusterWatcher::ms_dispatch2(const ref_t<Message> &m) {
   if (m->get_type() == CEPH_MSG_FS_MAP) {
     if (m->get_connection()->get_peer_type() == CEPH_ENTITY_TYPE_MON) {
       handle_fsmap(ref_cast<MFSMap>(m));

diff --git a/src/tools/cephfs_mirror/ClusterWatcher.h b/src/tools/cephfs_mirror/ClusterWatcher.h
index 312aeccd63c4ef5c01232682e43858f1f72cb929..5b9c3e85dc137c5ddd7145898d31e5f1e9d98075 100644 (file)
--- a/src/tools/cephfs_mirror/ClusterWatcher.h
+++ b/src/tools/cephfs_mirror/ClusterWatcher.h
@@ -38,7 +38,7 @@ public:
                  Listener &listener);
   ~ClusterWatcher();
 
-  bool ms_dispatch2(const ref_t<Message> &m) override;
+  Dispatcher::dispatch_result_t ms_dispatch2(const ref_t<Message> &m) override;
 
   void ms_handle_connect(Connection *c) override {
   }