rgw/rados: index operations use async_reads/writes()

replace the classes derived from CLSRGWConcurrentIO with classes
derived from Reader/Writer/RevertibleWriter and use the async algorithms

Signed-off-by: Casey Bodley <[email protected]>

Author: cbodley

src/cls/rgw/cls_rgw_client.cc

@@ -230,17 +230,22 @@ static bool issue_bucket_index_clean_op(librados::IoCtx& io_ctx,
   return manager->aio_operate(io_ctx, shard_id, oid, &op);
 }
 
+void cls_rgw_bucket_set_tag_timeout(librados::ObjectWriteOperation& op,
+                                    uint64_t timeout)
+{
+  const auto call = rgw_cls_tag_timeout_op{.tag_timeout = timeout};
+  bufferlist in;
+  encode(call, in);
+  op.exec(RGW_CLASS, RGW_BUCKET_SET_TAG_TIMEOUT, in);
+}
+
 static bool issue_bucket_set_tag_timeout_op(librados::IoCtx& io_ctx,
 					    const int shard_id,
 					    const string& oid,
 					    uint64_t timeout,
 					    BucketIndexAioManager *manager) {
-  bufferlist in;
-  rgw_cls_tag_timeout_op call;
-  call.tag_timeout = timeout;
-  encode(call, in);
   ObjectWriteOperation op;
-  op.exec(RGW_CLASS, RGW_BUCKET_SET_TAG_TIMEOUT, in);
+  cls_rgw_bucket_set_tag_timeout(op, timeout);
   return manager->aio_operate(io_ctx, shard_id, oid, &op);
 }
 
@@ -725,11 +730,16 @@ int CLSRGWIssueBILogTrim::issue_op(const int shard_id, const string& oid)
   return issue_bi_log_trim(io_ctx, oid, shard_id, start_marker_mgr, end_marker_mgr, &manager);
 }
 
+void cls_rgw_bucket_reshard_log_trim(librados::ObjectWriteOperation& op)
+{
+  bufferlist in;
+  op.exec(RGW_CLASS, RGW_RESHARD_LOG_TRIM, in);
+}
+
 static bool issue_reshard_log_trim(librados::IoCtx& io_ctx, const string& oid, int shard_id,
                                    BucketIndexAioManager *manager) {
-  bufferlist in;
   ObjectWriteOperation op;
-  op.exec(RGW_CLASS, RGW_RESHARD_LOG_TRIM, in);
+  cls_rgw_bucket_reshard_log_trim(op);
   return manager->aio_operate(io_ctx, shard_id, oid, &op);
 }
 
@@ -738,6 +748,20 @@ int CLSRGWIssueReshardLogTrim::issue_op(int shard_id, const string& oid)
   return issue_reshard_log_trim(io_ctx, oid, shard_id, &manager);
 }
 
+void cls_rgw_bucket_check_index(librados::ObjectReadOperation& op,
+                                bufferlist& out)
+{
+  bufferlist in;
+  op.exec(RGW_CLASS, RGW_BUCKET_CHECK_INDEX, in, &out, nullptr);
+}
+
+void cls_rgw_bucket_check_index_decode(const bufferlist& out,
+                                       rgw_cls_check_index_ret& result)
+{
+  auto p = out.cbegin();
+  decode(result, p);
+}
+
 static bool issue_bucket_check_index_op(IoCtx& io_ctx, const int shard_id, const string& oid, BucketIndexAioManager *manager,
     rgw_cls_check_index_ret *pdata) {
   bufferlist in;
@@ -752,11 +776,16 @@ int CLSRGWIssueBucketCheck::issue_op(int shard_id, const string& oid)
   return issue_bucket_check_index_op(io_ctx, shard_id, oid, &manager, &result[shard_id]);
 }
 
+void cls_rgw_bucket_rebuild_index(librados::ObjectWriteOperation& op)
+{
+  bufferlist in;
+  op.exec(RGW_CLASS, RGW_BUCKET_REBUILD_INDEX, in);
+}
+
 static bool issue_bucket_rebuild_index_op(IoCtx& io_ctx, const int shard_id, const string& oid,
     BucketIndexAioManager *manager) {
-  bufferlist in;
   librados::ObjectWriteOperation op;
-  op.exec(RGW_CLASS, RGW_BUCKET_REBUILD_INDEX, in);
+  cls_rgw_bucket_rebuild_index(op);
   return manager->aio_operate(io_ctx, shard_id, oid, &op);
 }
 
@@ -786,11 +815,16 @@ int CLSRGWIssueGetDirHeader::issue_op(const int shard_id, const string& oid)
 			      0, false, &manager, &result[shard_id]);
 }
 
-static bool issue_resync_bi_log(librados::IoCtx& io_ctx, const int shard_id, const string& oid, BucketIndexAioManager *manager)
+void cls_rgw_bilog_start(ObjectWriteOperation& op)
 {
   bufferlist in;
-  librados::ObjectWriteOperation op;
   op.exec(RGW_CLASS, RGW_BI_LOG_RESYNC, in);
+}
+
+static bool issue_resync_bi_log(librados::IoCtx& io_ctx, const int shard_id, const string& oid, BucketIndexAioManager *manager)
+{
+  librados::ObjectWriteOperation op;
+  cls_rgw_bilog_start(op);
   return manager->aio_operate(io_ctx, shard_id, oid, &op);
 }
 
@@ -799,11 +833,16 @@ int CLSRGWIssueResyncBucketBILog::issue_op(const int shard_id, const string& oid
   return issue_resync_bi_log(io_ctx, shard_id, oid, &manager);
 }
 
-static bool issue_bi_log_stop(librados::IoCtx& io_ctx, const int shard_id, const string& oid, BucketIndexAioManager *manager)
+void cls_rgw_bilog_stop(ObjectWriteOperation& op)
 {
   bufferlist in;
-  librados::ObjectWriteOperation op;
   op.exec(RGW_CLASS, RGW_BI_LOG_STOP, in);
+}
+
+static bool issue_bi_log_stop(librados::IoCtx& io_ctx, const int shard_id, const string& oid, BucketIndexAioManager *manager)
+{
+  librados::ObjectWriteOperation op;
+  cls_rgw_bilog_stop(op);
   return manager->aio_operate(io_ctx, shard_id, oid, &op);
 }
 
@@ -1214,49 +1253,31 @@ void cls_rgw_reshard_remove(librados::ObjectWriteOperation& op, const cls_rgw_re
   op.exec(RGW_CLASS, RGW_RESHARD_REMOVE, in);
 }
 
-int cls_rgw_set_bucket_resharding(librados::IoCtx& io_ctx, const string& oid,
-				  const cls_rgw_bucket_instance_entry& entry)
-{
-  bufferlist in, out;
-  cls_rgw_set_bucket_resharding_op call;
-  call.entry = entry;
-  encode(call, in);
-  librados::ObjectWriteOperation op;
-  op.exec(RGW_CLASS, RGW_SET_BUCKET_RESHARDING, in);
-  return io_ctx.operate(oid, &op);
-}
-
-int cls_rgw_clear_bucket_resharding(librados::IoCtx& io_ctx, const string& oid)
+void cls_rgw_clear_bucket_resharding(librados::ObjectWriteOperation& op)
 {
-  bufferlist in, out;
+  bufferlist in;
   cls_rgw_clear_bucket_resharding_op call;
   encode(call, in);
-  librados::ObjectWriteOperation op;
   op.exec(RGW_CLASS, RGW_CLEAR_BUCKET_RESHARDING, in);
-  return io_ctx.operate(oid, &op);
 }
 
-int cls_rgw_get_bucket_resharding(librados::IoCtx& io_ctx, const string& oid,
-				  cls_rgw_bucket_instance_entry *entry)
+void cls_rgw_get_bucket_resharding(librados::ObjectReadOperation& op,
+                                   bufferlist& out)
 {
-  bufferlist in, out;
+  bufferlist in;
   cls_rgw_get_bucket_resharding_op call;
   encode(call, in);
-  int r= io_ctx.exec(oid, RGW_CLASS, RGW_GET_BUCKET_RESHARDING, in, out);
-  if (r < 0)
-    return r;
+  op.exec(RGW_CLASS, RGW_GET_BUCKET_RESHARDING, in, &out, nullptr);
+}
 
+void cls_rgw_get_bucket_resharding_decode(const bufferlist& out,
+                                          cls_rgw_bucket_instance_entry& entry)
+{
   cls_rgw_get_bucket_resharding_ret op_ret;
   auto iter = out.cbegin();
-  try {
-    decode(op_ret, iter);
-  } catch (ceph::buffer::error& err) {
-    return -EIO;
-  }
-
-  *entry = op_ret.new_instance;
+  decode(op_ret, iter);
 
-  return 0;
+  entry = std::move(op_ret.new_instance);
 }
 
 void cls_rgw_guard_bucket_resharding(librados::ObjectOperation& op, int ret_err)
@@ -1268,17 +1289,24 @@ void cls_rgw_guard_bucket_resharding(librados::ObjectOperation& op, int ret_err)
   op.exec(RGW_CLASS, RGW_GUARD_BUCKET_RESHARDING, in);
 }
 
-static bool issue_set_bucket_resharding(librados::IoCtx& io_ctx,
-					const int shard_id, const string& oid,
-                                        const cls_rgw_bucket_instance_entry& entry,
-                                        BucketIndexAioManager *manager) {
+void cls_rgw_set_bucket_resharding(librados::ObjectWriteOperation& op,
+                                   cls_rgw_reshard_status status)
+{
   bufferlist in;
   cls_rgw_set_bucket_resharding_op call;
-  call.entry = entry;
+  call.entry.reshard_status = status;
   encode(call, in);
-  librados::ObjectWriteOperation op;
+
   op.assert_exists(); // the shard must exist; if not fail rather than recreate
   op.exec(RGW_CLASS, RGW_SET_BUCKET_RESHARDING, in);
+}
+
+static bool issue_set_bucket_resharding(librados::IoCtx& io_ctx,
+					const int shard_id, const string& oid,
+                                        const cls_rgw_bucket_instance_entry& entry,
+                                        BucketIndexAioManager *manager) {
+  librados::ObjectWriteOperation op;
+  cls_rgw_set_bucket_resharding(op, entry.reshard_status);
   return manager->aio_operate(io_ctx, shard_id, oid, &op);
 }
 

src/cls/rgw/cls_rgw_client.h

@@ -302,6 +302,8 @@ class CLSRGWConcurrentIO {
   int operator()();
 }; // class CLSRGWConcurrentIO
 
+void cls_rgw_bucket_set_tag_timeout(librados::ObjectWriteOperation& op,
+                                    uint64_t timeout);
 
 class CLSRGWIssueBucketIndexInit : public CLSRGWConcurrentIO {
 protected:
@@ -542,6 +544,12 @@ class CLSRGWIssueReshardLogTrim : public CLSRGWConcurrentIO {
       CLSRGWConcurrentIO(io_ctx, _bucket_objs, max_aio) {}
 };
 
+void cls_rgw_bucket_check_index(librados::ObjectReadOperation& op,
+                                bufferlist& out);
+// decode the response; may throw buffer::error
+void cls_rgw_bucket_check_index_decode(const bufferlist& out,
+                                       rgw_cls_check_index_ret& result);
+
 /**
  * Check the bucket index.
  *
@@ -563,6 +571,8 @@ class CLSRGWIssueBucketCheck : public CLSRGWConcurrentIO /*<std::map<std::string
   virtual ~CLSRGWIssueBucketCheck() override {}
 };
 
+void cls_rgw_bucket_rebuild_index(librados::ObjectWriteOperation& op);
+
 class CLSRGWIssueBucketRebuild : public CLSRGWConcurrentIO {
 protected:
   int issue_op(int shard_id, const std::string& oid) override;
@@ -594,6 +604,9 @@ class CLSRGWIssueSetBucketResharding : public CLSRGWConcurrentIO {
   virtual ~CLSRGWIssueSetBucketResharding() override {}
 };
 
+void cls_rgw_bilog_start(librados::ObjectWriteOperation& op);
+void cls_rgw_bilog_stop(librados::ObjectWriteOperation& op);
+
 class CLSRGWIssueResyncBucketBILog : public CLSRGWConcurrentIO {
 protected:
   int issue_op(int shard_id, const std::string& oid);
@@ -684,12 +697,15 @@ int cls_rgw_reshard_get(librados::IoCtx& io_ctx, const std::string& oid, cls_rgw
 // cls_rgw in the T+4 (X) release.
 void cls_rgw_guard_bucket_resharding(librados::ObjectOperation& op, int ret_err);
 
-// these overloads which call io_ctx.operate() should not be called in the rgw.
-// rgw_rados_operate() should be called after the overloads w/o calls to io_ctx.operate()
-#ifndef CLS_CLIENT_HIDE_IOCTX
-int cls_rgw_set_bucket_resharding(librados::IoCtx& io_ctx, const std::string& oid,
-                                  const cls_rgw_bucket_instance_entry& entry);
-int cls_rgw_clear_bucket_resharding(librados::IoCtx& io_ctx, const std::string& oid);
-int cls_rgw_get_bucket_resharding(librados::IoCtx& io_ctx, const std::string& oid,
-                                  cls_rgw_bucket_instance_entry *entry);
-#endif
+void cls_rgw_set_bucket_resharding(librados::ObjectWriteOperation& op,
+                                   cls_rgw_reshard_status status);
+void cls_rgw_clear_bucket_resharding(librados::ObjectWriteOperation& op);
+void cls_rgw_get_bucket_resharding(librados::ObjectReadOperation& op,
+                                   bufferlist& out);
+// decode the entry; may throw buffer::error
+void cls_rgw_get_bucket_resharding_decode(const bufferlist& out,
+                                          cls_rgw_bucket_instance_entry& entry);
+
+// Try to remove all reshard log entries from the bucket index. Return success
+// if any entries were removed, and -ENODATA once they're all gone.
+void cls_rgw_bucket_reshard_log_trim(librados::ObjectWriteOperation& op);

src/cls/rgw/cls_rgw_ops.h

@@ -7,9 +7,7 @@
 
 struct rgw_cls_tag_timeout_op
 {
-  uint64_t tag_timeout;
-
-  rgw_cls_tag_timeout_op() : tag_timeout(0) {}
+  uint64_t tag_timeout = 0;
 
   void encode(ceph::buffer::list &bl) const {
     ENCODE_START(1, 1, bl);