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 13c1c200429859834419fa0295788717ceed047a..187fb68f846eba1fca71f1f17e4f33a5b9f38fd1 100644 (file)
--- a/src/mgr/ActivePyModule.h
+++ b/src/mgr/ActivePyModule.h
@@ -20,6 +20,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 8c070dd8cbf414148f2ec5264744db69a341c25d..d2dc638da8383cd8e495a4aeaebc1e0a96ca2b31 100644 (file)
--- a/src/mgr/ActivePyModules.cc
+++ b/src/mgr/ActivePyModules.cc
@@ -544,6 +544,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();
     }
   }));
 }
@@ -552,6 +555,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();
@@ -590,8 +600,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);
     }));
@@ -614,8 +625,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);
     }));
   }
@@ -1308,8 +1320,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 6015084e3cdcd057426853e63a0343aea185291a..d697cb061714580153015590afdbc482f0cb6282 100644 (file)
--- a/src/mgr/ActivePyModules.h
+++ b/src/mgr/ActivePyModules.h
@@ -189,6 +189,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 4afae0c20d8c35a0ae7c5e16332aaa97f5a0722a..bd873108263bc6f841647d23a6ab1b710a35534f 100644 (file)
--- a/src/mgr/DaemonServer.cc
+++ b/src/mgr/DaemonServer.cc
@@ -2452,9 +2452,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 4cd98e8827e487f7308146469020733d1e2d2640..c0e951ed7b67ab6053c235b472453634a5c18ae5 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)
 };