rgw: async refcount operate in copy_obj

Signed-off-by: Mingyuan Liang <liangmingyuan@baidu.com>

diff --git a/doc/radosgw/config-ref.rst b/doc/radosgw/config-ref.rst
index 2f8478b9ca147b047a8785e137be4e0b2b7aaeb3..3811499fbe66e43c024b157c0c2863b552f60a4a 100644 (file)
--- a/doc/radosgw/config-ref.rst
+++ b/doc/radosgw/config-ref.rst
@@ -43,6 +43,7 @@ instances or all radosgw-admin options can be put into the ``[global]`` or the
 .. confval:: rgw_curl_wait_timeout_ms
 .. confval:: rgw_copy_obj_progress
 .. confval:: rgw_copy_obj_progress_every_bytes
+.. confval:: rgw_max_copy_obj_concurrent_io
 .. confval:: rgw_admin_entry
 .. confval:: rgw_content_length_compat
 .. confval:: rgw_bucket_quota_ttl

diff --git a/src/common/options/rgw.yaml.in b/src/common/options/rgw.yaml.in
index e758da2e0033fff41d103bc9ced6f805b37aafad..4c0111e1de8d4f6922936f1cdecd6d4a83d6e0c1 100644 (file)
--- a/src/common/options/rgw.yaml.in
+++ b/src/common/options/rgw.yaml.in
@@ -1913,6 +1913,14 @@ options:
   services:
   - rgw
   with_legacy: true
+- name: rgw_max_copy_obj_concurrent_io
+  type: int
+  level: advanced
+  desc: Number of refcount operations to process concurrently when executing copy_obj
+  default: 10
+  services:
+  - rgw
+  with_legacy: true
 - name: rgw_sync_obj_etag_verify
   type: bool
   level: advanced

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index d862e83a681b1ac0466d4a94aa03e1294c66ab45..56db767a7cab8e589f5c17b8f187872394c5b2cb 100644 (file)
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -4453,12 +4453,6 @@ int RGWRados::copy_obj(RGWObjectCtx& obj_ctx,
     ++miter;
   }
 
-  rgw_rados_ref ref;
-  ret = get_raw_obj_ref(dpp, miter.get_location().get_raw_obj(store), &ref);
-  if (ret < 0) {
-    return ret;
-  }
-
   bufferlist first_chunk;
 
   const bool copy_itself = (dest_obj->get_obj() == src_obj->get_obj());
@@ -4478,7 +4472,10 @@ int RGWRados::copy_obj(RGWObjectCtx& obj_ctx,
     append_rand_alpha(cct, tag, tag, 32);
   }
 
+  std::unique_ptr<rgw::Aio> aio;
+  rgw::AioResultList all_results;
   if (!copy_itself) {
+    aio = rgw::make_throttle(cct->_conf->rgw_max_copy_obj_concurrent_io, y);
     attrs.erase(RGW_ATTR_TAIL_TAG);
     manifest = *amanifest;
     const rgw_bucket_placement& tail_placement = manifest.get_tail_placement();
@@ -4490,17 +4487,31 @@ int RGWRados::copy_obj(RGWObjectCtx& obj_ctx,
       ObjectWriteOperation op;
       ref_tag = tag + '\0';
       cls_refcount_get(op, ref_tag, true);
-      const rgw_raw_obj& loc = miter.get_location().get_raw_obj(store);
 
-      auto& ioctx = ref.pool.ioctx();
-      ioctx.locator_set_key(loc.loc);
+      auto obj = svc.rados->obj(miter.get_location().get_raw_obj(store));
+      ret = obj.open(dpp);
+      if (ret < 0) {
+        ldpp_dout(dpp, 0) << "failed to open rados context for " << obj << dendl;
+        goto done_ret;
+      }
 
-      ret = rgw_rados_operate(dpp, ioctx, loc.oid, &op, null_yield);
+      static constexpr uint64_t cost = 1; // 1 throttle unit per request
+      static constexpr uint64_t id = 0; // ids unused
+      rgw::AioResultList completed = aio->get(obj, rgw::Aio::librados_op(std::move(op), y), cost, id);
+      ret = rgw::check_for_errors(completed);
+      all_results.splice(all_results.end(), completed);
       if (ret < 0) {
+        ldpp_dout(dpp, 0) << "ERROR: failed to copy obj=" << obj << ", the error code = " << ret << dendl;
         goto done_ret;
       }
+    }
 
-      ref_objs.push_back(loc);
+    rgw::AioResultList completed = aio->drain();
+    ret = rgw::check_for_errors(completed);
+    all_results.splice(all_results.end(), completed);
+    if (ret < 0) {
+      ldpp_dout(dpp, 0) << "ERROR: failed to drain ios, the error code = " << ret <<dendl;
+      goto done_ret;
     }
 
     pmanifest = &manifest;
@@ -4541,21 +4552,34 @@ int RGWRados::copy_obj(RGWObjectCtx& obj_ctx,
 
 done_ret:
   if (!copy_itself) {
-    vector<rgw_raw_obj>::iterator riter;
+
+    /* wait all pending op done */
+    rgw::AioResultList completed = aio->drain();
+    all_results.splice(all_results.end(), completed);
 
     /* rollback reference */
     string ref_tag = tag + '\0';
-    for (riter = ref_objs.begin(); riter != ref_objs.end(); ++riter) {
+    int ret2 = 0;
+    for (auto& r : all_results) {
+      if (r.result < 0) {
+        continue; // skip errors
+      }
       ObjectWriteOperation op;
       cls_refcount_put(op, ref_tag, true);
 
-      ref.pool.ioctx().locator_set_key(riter->loc);
-
-      int r = rgw_rados_operate(dpp, ref.pool.ioctx(), riter->oid, &op, null_yield);
-      if (r < 0) {
-        ldpp_dout(dpp, 0) << "ERROR: cleanup after error failed to drop reference on obj=" << *riter << dendl;
+      static constexpr uint64_t cost = 1; // 1 throttle unit per request
+      static constexpr uint64_t id = 0; // ids unused
+      rgw::AioResultList completed = aio->get(r.obj, rgw::Aio::librados_op(std::move(op), y), cost, id);
+      ret2 = rgw::check_for_errors(completed);
+      if (ret2 < 0) {
+        ldpp_dout(dpp, 0) << "ERROR: cleanup after error failed to drop reference on obj=" << r.obj << dendl;
       }
     }
+    completed = aio->drain();
+    ret2 = rgw::check_for_errors(completed);
+    if (ret2 < 0) {
+      ldpp_dout(dpp, 0) << "ERROR: failed to drain rollback ios, the error code = " << ret2 <<dendl;
+    }
   }
   return ret;
 }