From 6d5ccef84b4021f2efc3bedf6d347c45d29a0920 Mon Sep 17 00:00:00 2001
From: Pritha Srivastava <prsrivas@redhat.com>
Date: Thu, 13 Jun 2019 10:31:19 +0530
Subject: [PATCH] rgw: Integrate rgw gc with queue ops.

Signed-off-by: Pritha Srivastava <prsrivas@redhat.com>
---
 src/common/options.cc  |  6 +--
 src/rgw/CMakeLists.txt |  3 +-
 src/rgw/rgw_gc.cc      | 86 +++++++++++++++++++++++++++++++++++-------
 src/rgw/rgw_gc.h       |  3 +-
 src/rgw/rgw_rados.cc   |  4 +-
 5 files changed, 83 insertions(+), 19 deletions(-)

diff --git a/src/common/options.cc b/src/common/options.cc
index 70a01b7c03343..d2068d5a490f5 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -6419,7 +6419,7 @@ std::vector<Option> get_rgw_options() {
     .add_see_also({"rgw_gc_obj_min_wait", "rgw_gc_processor_max_time", "rgw_gc_processor_period", "rgw_gc_max_concurrent_io"}),
 
     Option("rgw_gc_obj_min_wait", Option::TYPE_INT, Option::LEVEL_ADVANCED)
-    .set_default(2_hr)
+    .set_default(300)
     .set_description("Garbage collection object expiration time")
     .set_long_description(
        "The length of time (in seconds) that the RGW collector will wait before purging "
@@ -6429,7 +6429,7 @@ std::vector<Option> get_rgw_options() {
     .add_see_also({"rgw_gc_max_objs", "rgw_gc_processor_max_time", "rgw_gc_processor_period", "rgw_gc_max_concurrent_io"}),
 
     Option("rgw_gc_processor_max_time", Option::TYPE_INT, Option::LEVEL_ADVANCED)
-    .set_default(1_hr)
+    .set_default(600)
     .set_description("Length of time GC processor can lease shard")
     .set_long_description(
         "Garbage collection thread in RGW process holds a lease on its data shards. These "
@@ -6441,7 +6441,7 @@ std::vector<Option> get_rgw_options() {
     .add_see_also({"rgw_gc_max_objs", "rgw_gc_obj_min_wait", "rgw_gc_processor_period", "rgw_gc_max_concurrent_io"}),
 
     Option("rgw_gc_processor_period", Option::TYPE_INT, Option::LEVEL_ADVANCED)
-    .set_default(1_hr)
+    .set_default(600)
     .set_description("Garbage collector cycle run time")
     .set_long_description(
         "The amount of time between the start of consecutive runs of the garbage collector "
diff --git a/src/rgw/CMakeLists.txt b/src/rgw/CMakeLists.txt
index 0db24bdee9608..730d004cf2954 100644
--- a/src/rgw/CMakeLists.txt
+++ b/src/rgw/CMakeLists.txt
@@ -214,7 +214,7 @@ target_link_libraries(rgw_a
   PRIVATE
   librados cls_otp_client cls_lock_client cls_rgw_client cls_refcount_client
   cls_log_client cls_timeindex_client cls_version_client
-  cls_user_client ceph-common common_utf8 global
+  cls_user_client cls_queue_client ceph-common common_utf8 global
   ${CURL_LIBRARIES}
   ${EXPAT_LIBRARIES}
   ${OPENLDAP_LIBRARIES} ${CRYPTO_LIBS}
@@ -371,6 +371,7 @@ target_link_libraries(rgw_admin_user PRIVATE
   cls_timeindex_client
   cls_version_client
   cls_user_client
+  cls_queue_client
   global
   ${CURL_LIBRARIES}
   ${EXPAT_LIBRARIES}
diff --git a/src/rgw/rgw_gc.cc b/src/rgw/rgw_gc.cc
index ccd88bac8261d..aca2f29143a87 100644
--- a/src/rgw/rgw_gc.cc
+++ b/src/rgw/rgw_gc.cc
@@ -7,6 +7,7 @@
 #include "include/scope_guard.h"
 #include "include/rados/librados.hpp"
 #include "cls/rgw/cls_rgw_client.h"
+#include "cls/queue/cls_queue_client.h"
 #include "cls/refcount/cls_refcount_client.h"
 #include "rgw_perf_counters.h"
 #include "cls/lock/cls_lock_client.h"
@@ -38,6 +39,21 @@ void RGWGC::initialize(CephContext *_cct, RGWRados *_store) {
     snprintf(buf, 32, ".%d", i);
     obj_names[i].append(buf);
   }
+
+  // Create GC queues
+  for (int i = 0; i < max_objs; i++) {
+    ldpp_dout(this, 20) << "RGWGC::initialize initing gc queue with name = " << obj_names[i] << dendl;
+    librados::ObjectWriteOperation op;
+    op.assert_exists();
+    uint64_t queue_size = 1048576, num_urgent_data_entries = 50;
+    cls_rgw_gc_init_queue(op, obj_names[i], queue_size, num_urgent_data_entries);
+    int ret = store->gc_operate(obj_names[i], &op);
+    if (ret == -ENOENT) {
+      ldpp_dout(this, 20) << "RGWGC::initialize creating gc queue with name = " << obj_names[i] << dendl;
+      cls_rgw_gc_create_queue(op, obj_names[i], queue_size, num_urgent_data_entries);
+      store->gc_operate(obj_names[i], &op);
+    }
+  }
 }
 
 void RGWGC::finalize()
@@ -56,7 +72,8 @@ void RGWGC::add_chain(ObjectWriteOperation& op, cls_rgw_obj_chain& chain, const
   info.chain = chain;
   info.tag = tag;
 
-  cls_rgw_gc_set_entry(op, cct->_conf->rgw_gc_obj_min_wait, info);
+  //cls_rgw_gc_set_entry(op, cct->_conf->rgw_gc_obj_min_wait, info);
+  cls_rgw_gc_enqueue(op, cct->_conf->rgw_gc_obj_min_wait, info);
 }
 
 int RGWGC::send_chain(cls_rgw_obj_chain& chain, const string& tag, bool sync)
@@ -66,16 +83,22 @@ int RGWGC::send_chain(cls_rgw_obj_chain& chain, const string& tag, bool sync)
 
   int i = tag_index(tag);
 
+  ldpp_dout(this, 20) << "RGWGC::send_chain - on object name: " << obj_names[i] << "tag is: " << tag << dendl;
   if (sync)
     return store->gc_operate(obj_names[i], &op);
 
   return store->gc_aio_operate(obj_names[i], &op);
 }
 
-int RGWGC::defer_chain(const string& tag, bool sync)
+int RGWGC::defer_chain(const string& tag, cls_rgw_obj_chain& chain, bool sync)
 {
+  cls_rgw_gc_obj_info info;
+  info.chain = chain;
+  info.tag = tag;
+
   ObjectWriteOperation op;
-  cls_rgw_gc_defer_entry(op, cct->_conf->rgw_gc_obj_min_wait, tag);
+  //cls_rgw_gc_defer_entry(op, cct->_conf->rgw_gc_obj_min_wait, tag);
+  cls_rgw_gc_defer_entry_queue(op, cct->_conf->rgw_gc_obj_min_wait, info);
 
   int i = tag_index(tag);
 
@@ -92,6 +115,13 @@ int RGWGC::remove(int index, const std::vector<string>& tags, AioCompletion **pc
   return store->gc_aio_operate(obj_names[index], &op, pc);
 }
 
+int RGWGC::remove(int index, string& marker, int num_entries, librados::AioCompletion **pc)
+{
+  ObjectWriteOperation op;
+  cls_rgw_gc_remove_entries_queue(op, marker, num_entries);
+  return store->gc_aio_operate(obj_names[index], &op, pc);
+}
+
 int RGWGC::list(int *index, string& marker, uint32_t max, bool expired_only, std::list<cls_rgw_gc_obj_info>& result, bool *truncated)
 {
   result.clear();
@@ -99,7 +129,8 @@ int RGWGC::list(int *index, string& marker, uint32_t max, bool expired_only, std
 
   for (; *index < max_objs && result.size() < max; (*index)++, marker.clear()) {
     std::list<cls_rgw_gc_obj_info> entries;
-    int ret = cls_rgw_gc_list(store->gc_pool_ctx, obj_names[*index], marker, max - result.size(), expired_only, entries, truncated, next_marker);
+    //int ret = cls_rgw_gc_list(store->gc_pool_ctx, obj_names[*index], marker, max - result.size(), expired_only, entries, truncated, next_marker);
+    int ret = cls_rgw_gc_list_queue(store->gc_pool_ctx, obj_names[*index], marker, max - result.size(), expired_only, entries, truncated, next_marker);
     if (ret == -ENOENT)
       continue;
     if (ret < 0)
@@ -203,7 +234,7 @@ public:
     if (ret == -ENOENT) {
       ret = 0;
     }
-
+#if 0
     if (io.type == IO::IndexIO) {
       if (ret < 0) {
         ldpp_dout(dpp, 0) << "WARNING: gc cleanup of tags on gc shard index=" <<
@@ -211,14 +242,14 @@ public:
       }
       goto done;
     }
-
+#endif
     if (ret < 0) {
       ldpp_dout(dpp, 0) << "WARNING: gc could not remove oid=" << io.oid <<
 	", ret=" << ret << dendl;
       goto done;
     }
 
-    schedule_tag_removal(io.index, io.tag);
+    //schedule_tag_removal(io.index, io.tag);
 
   done:
     ios.pop_front();
@@ -267,7 +298,7 @@ public:
 
   void drain() {
     drain_ios();
-    flush_remove_tags();
+    //flush_remove_tags();
     /* the tags draining might have generated more ios, drain those too */
     drain_ios();
   }
@@ -311,6 +342,19 @@ public:
       ++index;
     }
   }
+
+  int remove_queue_entries(int index, string& marker, int num_entries) {
+    IO index_io;
+    index_io.type = IO::IndexIO;
+    index_io.index = index;
+    int ret = gc->remove(index, marker, num_entries, &index_io.c);
+    if (ret < 0) {
+      ldpp_dout(dpp, 0) << "WARNING: failed to remove queue entries on index=" <<
+	    index << " ret=" << ret << dendl;
+      return ret;
+    }
+    return 0;
+  }
 }; // class RGWGCIOManger
 
 int RGWGC::process(int index, int max_secs, bool expired_only,
@@ -351,10 +395,12 @@ int RGWGC::process(int index, int max_secs, bool expired_only,
     int max = 100;
     std::list<cls_rgw_gc_obj_info> entries;
 
-    ret = cls_rgw_gc_list(store->gc_pool_ctx, obj_names[index], marker, max,
-			  expired_only, entries, &truncated, next_marker);
+    //ret = cls_rgw_gc_list(store->gc_pool_ctx, obj_names[index], marker, max,
+			//  expired_only, entries, &truncated, next_marker);
+
+    ret = cls_rgw_gc_list_queue(store->gc_pool_ctx, obj_names[index], marker, max, expired_only, entries, &truncated, next_marker);
     ldpp_dout(this, 20) <<
-      "RGWGC::process cls_rgw_gc_list returned with returned:" << ret <<
+      "RGWGC::process cls_rgw_gc_list_queue returned with returned:" << ret <<
       ", entries.size=" << entries.size() << ", truncated=" << truncated <<
       ", next_marker='" << next_marker << "'" << dendl;
 
@@ -365,7 +411,7 @@ int RGWGC::process(int index, int max_secs, bool expired_only,
     if (ret < 0)
       goto done;
 
-    marker = next_marker;
+    //marker = next_marker;
 
     string last_pool;
     std::list<cls_rgw_gc_obj_info>::iterator iter;
@@ -383,11 +429,14 @@ int RGWGC::process(int index, int max_secs, bool expired_only,
       if (now >= end) {
         goto done;
       }
-
+#if 0
       if (chain.objs.empty()) {
         io_manager.schedule_tag_removal(index, info.tag);
       } else {
         io_manager.add_tag_io_size(index, info.tag, chain.objs.size());
+      }
+#endif
+  if (! chain.objs.empty()) {
 	for (liter = chain.objs.begin(); liter != chain.objs.end(); ++liter) {
 	  cls_rgw_obj& obj = *liter;
 
@@ -426,6 +475,17 @@ int RGWGC::process(int index, int max_secs, bool expired_only,
 	} // chains loop
       } // else -- chains not empty
     } // entries loop
+    if (entries.size() > 0) {
+      //Remove the entries from the queue
+      ldpp_dout(this, 5) << "RGWGC::process removing entries, marker: " << marker << dendl;
+      ret = io_manager.remove_queue_entries(index, marker, entries.size());
+      if (ret < 0) {
+        ldpp_dout(this, 0) <<
+          "WARNING: failed to remove queue entries" << dendl;
+        goto done;
+      }
+      marker = next_marker;
+    }
   } while (truncated);
 
 done:
diff --git a/src/rgw/rgw_gc.h b/src/rgw/rgw_gc.h
index a58452b556906..3624b3d0a465c 100644
--- a/src/rgw/rgw_gc.h
+++ b/src/rgw/rgw_gc.h
@@ -50,8 +50,9 @@ public:
 
   void add_chain(librados::ObjectWriteOperation& op, cls_rgw_obj_chain& chain, const string& tag);
   int send_chain(cls_rgw_obj_chain& chain, const string& tag, bool sync);
-  int defer_chain(const string& tag, bool sync);
+  int defer_chain(const string& tag, cls_rgw_obj_chain& info, bool sync);
   int remove(int index, const std::vector<string>& tags, librados::AioCompletion **pc);
+  int remove(int index, string& marker, int num_entries, librados::AioCompletion **pc);
 
   void initialize(CephContext *_cct, RGWRados *_store);
   void finalize();
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 55fb8869cdaf7..7830fd2222e09 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -4618,7 +4618,9 @@ int RGWRados::defer_gc(void *ctx, const RGWBucketInfo& bucket_info, const rgw_ob
 
   ldout(cct, 0) << "defer chain tag=" << tag << dendl;
 
-  return gc->defer_chain(tag, false);
+  cls_rgw_obj_chain chain;
+  update_gc_chain(state->obj, state->manifest, &chain);
+  return gc->defer_chain(tag, chain, false);
 }
 
 void RGWRados::remove_rgw_head_obj(ObjectWriteOperation& op)
-- 
2.39.5