mgr: Add one finisher thread per module

Adds a finisher thread for each module. Each module command is
executed via corresponding finisher thread. There is generic
finisher thread via which all other commands like notify, config
is run.

Fixes: https://tracker.ceph.com/issues/51177
Signed-off-by: Kotresh HR <khiremat@redhat.com>
(cherry picked from commit 46de6431603a56fa9ca2a7c7c9c795f126d16452)

diff --git a/src/mgr/ActivePyModule.h b/src/mgr/ActivePyModule.h
index 1cbf6d18ac2f4e1d9feec69102e7031ba61fb820..cd0ddc953cadb201ec050dc7c0c0cc4c643e5a68 100644 (file)
--- a/src/mgr/ActivePyModule.h
+++ b/src/mgr/ActivePyModule.h
@@ -21,6 +21,7 @@
 #include "common/cmdparse.h"
 #include "common/LogEntry.h"
 #include "common/Thread.h"
+#include "common/Finisher.h"
 #include "mon/health_check.h"
 #include "mgr/Gil.h"
 
@@ -45,12 +46,19 @@ private:
 
   std::string m_command_perms;
   const MgrSession* m_session = nullptr;
+  std::string fin_thread_name;
+public:
+  Finisher finisher; // per active module finisher to execute commands
 
 public:
   ActivePyModule(const PyModuleRef &py_module_,
       LogChannelRef clog_)
-    : PyModuleRunner(py_module_, clog_)
-  {}
+    : PyModuleRunner(py_module_, clog_),
+      fin_thread_name(std::string("m-fin-" + py_module->get_name()).substr(0,15)),
+      finisher(g_ceph_context, thread_name, fin_thread_name)
+
+  {
+  }
 
   int load(ActivePyModules *py_modules);
   void notify(const std::string &notify_type, const std::string &notify_id);
@@ -94,6 +102,11 @@ public:
     return uri;
   }
 
+  std::string get_fin_thread_name() const
+  {
+    return fin_thread_name;
+  }
+
   bool is_authorized(const std::map<std::string, std::string>& arguments) const;
 
 };

diff --git a/src/mgr/ActivePyModules.cc b/src/mgr/ActivePyModules.cc
index 6a8525b37b28ef4743b9d3cf78c4fd0b09028ed5..514f8d8ecb67e0ab73699567b233d09ccfcb3319 100644 (file)
--- a/src/mgr/ActivePyModules.cc
+++ b/src/mgr/ActivePyModules.cc
@@ -536,6 +536,9 @@ void ActivePyModules::start_one(PyModuleRef py_module)
 
       dout(4) << "Starting thread for " << name << dendl;
       active_module->thread.create(active_module->get_thread_name());
+      dout(4) << "Starting active module " << name <<" finisher thread "
+        << active_module->get_fin_thread_name() << dendl;
+      active_module->finisher.start();
     }
   }));
 }
@@ -544,6 +547,13 @@ void ActivePyModules::shutdown()
 {
   std::lock_guard locker(lock);
 
+  // Stop per active module finisher thread
+  for (auto& [name, module] : modules) {
+      dout(4) << "Stopping active module " << name << " finisher thread" << dendl;
+      module->finisher.wait_for_empty();
+      module->finisher.stop();
+  }
+
   // Signal modules to drop out of serve() and/or tear down resources
   for (auto& [name, module] : modules) {
     lock.unlock();
@@ -582,8 +592,9 @@ void ActivePyModules::notify_all(const std::string &notify_type,
     // Send all python calls down a Finisher to avoid blocking
     // C++ code, and avoid any potential lock cycles.
     dout(15) << "queuing notify (" << notify_type << ") to " << name << dendl;
+    Finisher& mod_finisher = py_module_registry.get_active_module_finisher(name);
     // workaround for https://bugs.llvm.org/show_bug.cgi?id=35984
-    finisher.queue(new LambdaContext([module=module, notify_type, notify_id]
+    mod_finisher.queue(new LambdaContext([module=module, notify_type, notify_id]
       (int r){
         module->notify(notify_type, notify_id);
     }));
@@ -606,8 +617,9 @@ void ActivePyModules::notify_all(const LogEntry &log_entry)
     // log_entry: we take a copy because caller's instance is
     // probably ephemeral.
     dout(15) << "queuing notify (clog) to " << name << dendl;
+    Finisher& mod_finisher = py_module_registry.get_active_module_finisher(name);
     // workaround for https://bugs.llvm.org/show_bug.cgi?id=35984
-    finisher.queue(new LambdaContext([module=module, log_entry](int r){
+    mod_finisher.queue(new LambdaContext([module=module, log_entry](int r){
       module->notify_clog(log_entry);
     }));
   }
@@ -1288,8 +1300,9 @@ void ActivePyModules::config_notify()
     // Send all python calls down a Finisher to avoid blocking
     // C++ code, and avoid any potential lock cycles.
     dout(15) << "notify (config) " << name << dendl;
+    Finisher& mod_finisher = py_module_registry.get_active_module_finisher(name);
     // workaround for https://bugs.llvm.org/show_bug.cgi?id=35984
-    finisher.queue(new LambdaContext([module=module](int r){
+    mod_finisher.queue(new LambdaContext([module=module](int r){
       module->config_notify();
     }));
   }

diff --git a/src/mgr/ActivePyModules.h b/src/mgr/ActivePyModules.h
index 2a22b98bf38f4f3d0a417e0988b1f1e6b065f465..61837d157f29a826457d5a7fceb1d2062aef89c8 100644 (file)
--- a/src/mgr/ActivePyModules.h
+++ b/src/mgr/ActivePyModules.h
@@ -188,6 +188,10 @@ public:
                   const std::string &notify_id);
   void notify_all(const LogEntry &log_entry);
 
+  auto& get_module_finisher(const std::string &name) {
+    return modules.at(name)->finisher;
+  }
+
   bool is_pending(std::string_view name) const {
     return pending_modules.count(name) > 0;
   }

diff --git a/src/mgr/DaemonServer.cc b/src/mgr/DaemonServer.cc
index 15cd9f0e7e02e0b3e46555b2a57b5ece2a80c118..800178fa5a5665da53919a3276e6e1df5261b2c9 100644 (file)
--- a/src/mgr/DaemonServer.cc
+++ b/src/mgr/DaemonServer.cc
@@ -2451,9 +2451,21 @@ bool DaemonServer::_handle_command(
     return true;
   }
 
+  // Validate that the module is active
+  auto& mod_name = py_command.module_name;
+  if (!py_modules.is_module_active(mod_name)) {
+    ss << "Module '" << mod_name << "' is not enabled/loaded (required by "
+          "command '" << prefix << "'): use `ceph mgr module enable "
+          << mod_name << "` to enable it";
+    dout(4) << ss.str() << dendl;
+    cmdctx->reply(-EOPNOTSUPP, ss);
+    return true;
+  }
+
   dout(10) << "passing through command '" << prefix << "' size " << cmdctx->cmdmap.size() << dendl;
-  finisher.queue(new LambdaContext([this, cmdctx, session, py_command, prefix]
-                                   (int r_) mutable {
+  Finisher& mod_finisher = py_modules.get_active_module_finisher(mod_name);
+  mod_finisher.queue(new LambdaContext([this, cmdctx, session, py_command, prefix]
+                                       (int r_) mutable {
     std::stringstream ss;
 
     dout(10) << "dispatching command '" << prefix << "' size " << cmdctx->cmdmap.size() << dendl;

diff --git a/src/mgr/PyModuleRegistry.h b/src/mgr/PyModuleRegistry.h
index 6c72af8932a127fa8b7d9c94406714743e8fe842..9bc923f8d6cdeba375d0d9d2110fa8eaad561902 100644 (file)
--- a/src/mgr/PyModuleRegistry.h
+++ b/src/mgr/PyModuleRegistry.h
@@ -227,5 +227,15 @@ public:
     return v;
   }
 
+  bool is_module_active(const std::string &name) {
+    ceph_assert(active_modules);
+    return active_modules->module_exists(name);
+  }
+
+  auto& get_active_module_finisher(const std::string &name) {
+    ceph_assert(active_modules);
+    return active_modules->get_module_finisher(name);
+  }
+
   // <<< (end of ActivePyModules cheeky call-throughs)
 };