Feature/342 object search

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Change Log
 
+## [0.0.40](https://github.com/ceph-dovecot/dovecot-ceph-plugin/tree/0.0.39) (2022-09-22)
+- #342 multithreading object search for doveadm force-resync (feature toggle) 
+       new config params: 
+       # search method default = 0 | 1 multithreading
+       rbox_object_search_method=1 
+       # number of threads to use in case of search_method=1
+       rbox_object_search_threads=4
+
 ## [0.0.39](https://github.com/ceph-dovecot/dovecot-ceph-plugin/tree/0.0.39) (2022-08-25)
 - #339 fail with assert if rados_config cannot be found due to network/connection issue 
        retry ceph read operations / read / xattr with timeout 

configure.ac

@@ -9,7 +9,7 @@
 AC_PREREQ([2.59])
 
 
-AC_INIT([dovecot-ceph-plugin], [0.0.39], [https://github.com/ceph-dovecot/dovecot-ceph-plugin/issues/new], ,[https://github.com/ceph-dovecot/dovecot-ceph-plugin])
+AC_INIT([dovecot-ceph-plugin], [0.0.40], [https://github.com/ceph-dovecot/dovecot-ceph-plugin/issues/new], ,[https://github.com/ceph-dovecot/dovecot-ceph-plugin])
 
 
 AC_CONFIG_AUX_DIR([.])

rpm/dovecot-ceph-plugin.spec

@@ -15,7 +15,7 @@ Name:		dovecot-ceph-plugin
 Summary:	Dovecot Ceph RADOS plugins
 
 
-Version:	0.0.39
+Version:	0.0.40
 
 Release:	0%{?dist}
 URL:		https://github.com/ceph-dovecot/dovecot-ceph-plugin

src/librmb/rados-ceph-config.h

@@ -58,7 +58,7 @@ class RadosCephConfig {
   void set_update_attributes(const std::string &update_attributes_) {
     config.set_update_attributes(update_attributes_);
   }
-
+  
   void update_mail_attribute(const char *value) { config.update_mail_attribute(value); }
   void update_updateable_attribute(const char *value) { config.update_updateable_attribute(value); }
 

src/librmb/rados-cluster-impl.cpp

@@ -17,7 +17,7 @@
 
 #include "rados-dictionary-impl.h"
 #include "rados-storage-impl.h"
-
+#include "rados-util.h"
 using std::list;
 using std::pair;
 using std::string;
@@ -72,6 +72,57 @@ int RadosClusterImpl::init(const std::string &clustername, const std::string &ra
   return ret;
 }
 
+
+std::vector<std::string> RadosClusterImpl::list_pgs_for_pool(std::string &pool_name) {
+    std::cout << " ola "  << RadosClusterImpl::cluster << std::endl;
+
+    if(is_connected()){
+      std::cout << " is connected YES" << std::endl;
+    }else{
+      std::cout << " is connected NO" << std::endl;
+      connect();
+    }
+
+    const string pool = "mail_storage";
+    const string cmd =
+    "{"
+    "\"prefix\": \"pg ls-by-pool\", "
+    "\"poolstr\": \"" + pool + "\""
+    "}";      
+
+    std::cout << "cmd: " << cmd << std::endl;
+
+    librados::bufferlist inbl;
+    librados::bufferlist outbl;
+    int res = RadosClusterImpl::cluster->mon_command(cmd, inbl, &outbl, nullptr);
+    std::cout << "inbl command " << inbl  <<std::endl;
+    std::cout << "outbl command " << outbl.c_str()  <<std::endl;
+
+    std::vector<std::string> list = RadosUtils::extractPgs(std::string(outbl.c_str()));
+
+    for (auto const &token: list) {
+          std::cout << token << std::endl;        
+    }
+    return list;
+}
+
+std::map<std::string, std::vector<std::string>> RadosClusterImpl::list_pgs_osd_for_pool(std::string &pool_name) {    
+
+    if(!is_connected()){      
+      connect();
+    }
+
+    const string cmd =
+    "{"
+    "\"prefix\": \"pg ls-by-pool\", "
+    "\"poolstr\": \"" + pool_name + "\""
+    "}";      
+
+    librados::bufferlist inbl;
+    librados::bufferlist outbl;
+    RadosClusterImpl::cluster->mon_command(cmd, inbl, &outbl, nullptr);
+    return RadosUtils::extractPgAndPrimaryOsd(std::string(outbl.c_str()));
+}
 int RadosClusterImpl::initialize() {
   int ret = 0;
 

src/librmb/rados-cluster-impl.h

@@ -40,6 +40,9 @@ class RadosClusterImpl : public RadosCluster {
   librados::Rados &get_cluster() { return *cluster; }
   void set_config_option(const char *option, const char *value);
 
+  std::vector<std::string> list_pgs_for_pool(std::string &pool_name) override;
+  std::map<std::string, std::vector<std::string>> list_pgs_osd_for_pool(std::string &pool_name) override;
+
  private:
   int initialize();
 

src/librmb/rados-cluster.h

@@ -76,6 +76,13 @@ class RadosCluster {
    * @return true if connected
    */
   virtual bool is_connected() = 0;
+
+  /*! get placement groups for mailbox storage pool 
+  */
+  virtual std::vector<std::string> list_pgs_for_pool(std::string &pool_name) = 0;
+  virtual std::map<std::string, std::vector<std::string>> list_pgs_osd_for_pool(std::string &pool_name) = 0;
+
+
 };
 
 }  // namespace librmb

src/librmb/rados-dovecot-ceph-cfg-impl.h

@@ -58,6 +58,8 @@ class RadosDovecotCephCfgImpl : public RadosDovecotCephCfg {
       rados_cfg.set_cfg_object_name(dovecot_cfg.get_rbox_cfg_object_name());
     }
   }
+  int get_object_search_method()  override { return std::stoi(dovecot_cfg.get_object_search_method()); }
+  int get_object_search_threads() override { return std::stoi(dovecot_cfg.get_object_search_threads()); }
 
   void set_rbox_cfg_object_name(const std::string &value) override { dovecot_cfg.set_rbox_cfg_object_name(value); }
 

src/librmb/rados-dovecot-ceph-cfg.h

@@ -44,6 +44,9 @@ class RadosDovecotCephCfg {
   virtual int get_chunk_size() = 0;
   virtual int get_write_method() = 0;
 
+  virtual int get_object_search_method()  = 0;
+  virtual int get_object_search_threads() = 0;
+
   virtual const std::string &get_pool_name_metadata_key() = 0;
   virtual const std::string &get_update_attributes_key() = 0;
   virtual const std::string &get_mail_attributes_key() = 0;

src/librmb/rados-dovecot-config.cpp

@@ -29,7 +29,9 @@ RadosConfig::RadosConfig()
       rbox_ceph_aio_wait_for_safe_and_cb("rbox_ceph_aio_wait_for_safe_and_cb"),
       rbox_ceph_write_chunks("rbox_ceph_write_chunks"),
       rbox_chunk_size("rbox_chunk_size"),
-      rbox_write_method("rbox_write_method") {
+      rbox_write_method("rbox_write_method"),
+      rbox_object_search_method("rbox_object_search_method"),
+      rbox_object_search_threads("rbox_object_search_threads") {
 
   config[pool_name] = "mail_storage";
 
@@ -43,6 +45,9 @@ RadosConfig::RadosConfig()
   config[rbox_ceph_write_chunks] = "false";
   config[rbox_chunk_size] = "10240";
   config[rbox_write_method] = "0";
+  config[rbox_object_search_method] = "0";
+  config[rbox_object_search_threads] = "4";
+
   is_valid = false;
 }
 
@@ -79,8 +84,10 @@ std::string RadosConfig::to_string() {
   ss << "  " << rbox_ceph_aio_wait_for_safe_and_cb << "=" << config[rbox_ceph_aio_wait_for_safe_and_cb] << std::endl;
   ss << "  " << rbox_ceph_write_chunks << "=" << config[rbox_ceph_write_chunks] << std::endl;
   ss << "  " << rbox_write_method << "=" << config[rbox_write_method] << std::endl;
-  ss << "  " << rbox_chunk_size << "=" << config[rbox_chunk_size]
-     << std::endl;
+  ss << "  " << rbox_chunk_size << "=" << config[rbox_chunk_size] << std::endl;
+  ss << "  " << rbox_object_search_method << "=" << config[rbox_object_search_method] << std::endl;
+  ss << "  " << rbox_object_search_threads << "=" << config[rbox_object_search_threads] << std::endl;
+
   return ss.str();
 }
 

src/librmb/rados-dovecot-config.h

@@ -49,6 +49,10 @@ class RadosConfig {
 
   const std::string &get_rbox_cluster_name() { return config[rbox_cluster_name]; }
   const std::string &get_rados_username() { return config[rados_username]; }
+
+  const std::string &get_object_search_method()  { return config[rbox_object_search_method]; }
+  const std::string &get_object_search_threads() { return config[rbox_object_search_threads]; }
+
   void update_metadata(const std::string &key, const char *value_);
   bool is_ceph_posix_bugfix_enabled() {
     return config[bugfix_cephfs_posix_hardlinks].compare("true") == 0 ? true : false;
@@ -87,6 +91,8 @@ class RadosConfig {
   std::string rbox_ceph_write_chunks;
   std::string rbox_chunk_size;
   std::string rbox_write_method;
+  std::string rbox_object_search_method;
+  std::string rbox_object_search_threads;
   bool is_valid;
 };
 

src/librmb/rados-storage-impl.cpp

@@ -16,8 +16,13 @@
 #include <set>
 #include <string>
 #include <utility>
+#include <thread>
+#include <mutex>
+
+#include "rados-util.h"
 
 #include <rados/librados.hpp>
+
 #include "encoding.h"
 #include "limits.h"
 
@@ -173,7 +178,6 @@ librados::NObjectIterator RadosStorageImpl::find_mails(const RadosMetadata *attr
   }
 
   if (attr != nullptr) {
-   // int hashpos = get_io_ctx().get_object_hash_position("t1_u");
     std::string filter_name = PLAIN_FILTER_NAME;
     ceph::bufferlist filter_bl;
 
@@ -186,7 +190,57 @@ librados::NObjectIterator RadosStorageImpl::find_mails(const RadosMetadata *attr
     return get_io_ctx().nobjects_begin();
   }
 }
-
+/**
+ * POC Implementation: 
+ * 
+ * see in prod how it behaves. 
+ * 
+ **/
+std::set<std::string> RadosStorageImpl::find_mails_async(const RadosMetadata *attr, 
+                                                         std::string &pool_name,
+                                                         int num_threads){
+
+    std::set<std::string> oid_list;
+    std::mutex oid_list_mutex;
+
+    // Define a Lambda Expression
+    auto f = [](const std::vector<std::string> &list, std::mutex &oid_mutex, std::set<std::string> &oids, librados::IoCtx *io_ctx) {
+
+        std::lock_guard<std::mutex> guard(oid_mutex);
+        for (auto const &pg: list) {
+          uint64_t ppool;
+          uint32_t pseed;
+          int r = sscanf(pg.c_str(), "%llu.%x", (long long unsigned *)&ppool, &pseed);
+
+          librados::NObjectIterator iter= io_ctx->nobjects_begin(pseed);
+
+          while (iter != librados::NObjectIterator::__EndObjectIterator) {
+            std::string oid = iter->get_oid();
+            oids.insert(oid);  
+            iter++;
+          }          
+        }           
+    };
+
+    //std::string pool_mame = "mail_storage";
+    std::map<std::string, std::vector<std::string>> osd_pg_map = cluster->list_pgs_osd_for_pool(pool_name);
+    std::vector<std::thread> threads;
+
+    for (const auto& x : osd_pg_map)
+    {
+      if(threads.size() == num_threads){
+        for (auto const &thread: threads) {
+          thread.join();
+        }   
+        threads.clear();      
+      }
+      threads.push_back(std::thread(f, std::ref(x.second),std::ref(oid_list_mutex),std::ref(oid_list), &get_io_ctx()));
+    }
+    for (auto const &thread: threads) {
+        thread.join();
+    }   
+    return oid_list;
+}
 librados::IoCtx &RadosStorageImpl::get_io_ctx() { return io_ctx; }
 
 int RadosStorageImpl::open_connection(const std::string &poolname, const std::string &clustername,

src/librmb/rados-storage-impl.h

@@ -18,6 +18,8 @@
 #include <string>
 #include <cstdint>
 #include <list>
+#include <set>
+
 #include <algorithm>
 #include <rados/librados.hpp>
 
@@ -50,6 +52,9 @@ class RadosStorageImpl : public RadosStorage {
   int aio_operate(librados::IoCtx *io_ctx_, const std::string &oid, librados::AioCompletion *c,
                   librados::ObjectWriteOperation *op) override;
   librados::NObjectIterator find_mails(const RadosMetadata *attr) override;
+
+  std::set<std::string> find_mails_async(const RadosMetadata *attr, std::string &pool_name, int num_threads) override;
+
   int open_connection(const std::string &poolname) override;
   int open_connection(const std::string &poolname, const std::string &clustername,
                       const std::string &rados_username) override;
@@ -72,6 +77,7 @@ class RadosStorageImpl : public RadosStorage {
 
   void free_rados_mail(librmb::RadosMail *mail) override;
 
+
  private:
   int create_connection(const std::string &poolname);
 

src/librmb/rados-storage.h

@@ -104,6 +104,11 @@ class RadosStorage {
    *
    * @return object iterator or librados::NObjectIterator::__EndObjectIterator */
   virtual librados::NObjectIterator find_mails(const RadosMetadata *attr) = 0;
+
+
+  virtual std::set<std::string> find_mails_async(const RadosMetadata *attr, std::string &pool_name, int num_threads) = 0;
+
+
   /*! open the rados connections with default cluster and username
    * @param[in] poolname the poolname to connect to, in case this one does not exists, it will be created.
    * */
@@ -199,6 +204,7 @@ class RadosStorage {
    * */
   virtual void free_rados_mail(librmb::RadosMail *mail) = 0;
 
+
 };
 
 }  // namespace librmb