Merge pull request ceph#62887 from clwluvw/copy-obj-remote

rgw: report copy obj progress from the frontend CR

Reviewed-by: Adam C. Emerson <[email protected]>

Author: adamemerson

src/rgw/driver/rados/rgw_rados.cc

@@ -4989,19 +4989,31 @@ int RGWRados::copy_obj(RGWObjectCtx& src_obj_ctx,
   ldpp_dout(dpp, 5) << "Copy object " << src_obj.bucket << ":" << src_obj.get_oid() << " => " << dest_obj.bucket << ":" << dest_obj.get_oid() << dendl;
 
   if (remote_src || !source_zone.empty()) {
-    // null_yield resolves a crash when calling progress_cb(), because the beast
-    // frontend tried to use this same yield context to write the progress
-    // response to the frontend socket. call fetch_remote_obj() synchronously so
-    // that only one thread tries to suspend that coroutine
-    const req_context rctx{dpp, null_yield, nullptr};
-    const rgw_owner remote_user_owner(remote_user);
-    return fetch_remote_obj(dest_obj_ctx, &remote_user_owner, &remote_user, info, source_zone,
-               dest_obj, src_obj, dest_bucket_info, &src_bucket_info,
-               dest_placement, src_mtime, mtime, mod_ptr,
-               unmod_ptr, high_precision_time,
-               if_match, if_nomatch, attrs_mod, copy_if_newer, attrs, category,
-               olh_epoch, delete_at, ptag, petag, progress_cb, progress_data, rctx,
-               nullptr /* filter */, stat_follow_olh, stat_dest_obj, std::nullopt);
+    RGWCopyObj *op = static_cast<RGWCopyObj *>(progress_data);
+    auto& progress_tracker = op->get_progress_tracker();
+
+    int ret = 0;
+    boost::asio::spawn(driver->get_io_context(),
+      [&](boost::asio::yield_context yield) {
+        const req_context rctx{dpp, yield, nullptr};
+        const rgw_owner remote_user_owner(remote_user);
+        ret = fetch_remote_obj(dest_obj_ctx, &remote_user_owner, &remote_user, info, source_zone,
+                   dest_obj, src_obj, dest_bucket_info, &src_bucket_info,
+                   dest_placement, src_mtime, mtime, mod_ptr,
+                   unmod_ptr, high_precision_time,
+                   if_match, if_nomatch, attrs_mod, copy_if_newer, attrs, category,
+                   olh_epoch, delete_at, ptag, petag, progress_cb, progress_data, rctx,
+                   nullptr /* filter */, stat_follow_olh, stat_dest_obj, std::nullopt);
+
+        progress_tracker.done = true;
+        progress_tracker.cv.notify_one();
+      }, [] (std::exception_ptr eptr) {
+        if (eptr) std::rethrow_exception(eptr);
+      });
+
+    op->progress_cb_handler();
+
+    return ret;
   }
 
   map<string, bufferlist> src_attrs;

src/rgw/rgw_op.cc

@@ -5951,11 +5951,31 @@ void RGWCopyObj::progress_cb(off_t ofs)
     return;
   }
 
-  send_partial_response(ofs);
+  std::lock_guard<std::mutex> l(progress_tracker->mtx);
+  progress_tracker->ofs_queue.push(ofs);
+  progress_tracker->cv.notify_one();
 
   last_ofs = ofs;
 }
 
+void RGWCopyObj::progress_cb_handler()
+{
+  std::unique_lock<std::mutex> l(progress_tracker->mtx);
+  while(!progress_tracker->done || !progress_tracker->ofs_queue.empty()) {
+    progress_tracker->cv.wait(l, [&]() {
+      return progress_tracker->done || !progress_tracker->ofs_queue.empty();
+    });
+
+    while (!progress_tracker->ofs_queue.empty()) {
+      auto ofs = progress_tracker->ofs_queue.front();
+      progress_tracker->ofs_queue.pop();
+      l.unlock();
+      send_partial_response(ofs);
+      l.lock();
+    }
+  }
+}
+
 void RGWCopyObj::pre_exec()
 {
   rgw_bucket_object_pre_exec(s);

src/rgw/rgw_op.h

@@ -1611,6 +1611,17 @@ class RGWCopyObj : public RGWOp {
   RGWObjectRetention *obj_retention;
   RGWObjectLegalHold *obj_legal_hold;
 
+  // remote copy progress helper
+  struct ProgressTracker {
+    std::mutex mtx;
+    std::condition_variable cv;
+    std::queue<off_t> ofs_queue;
+    std::atomic<bool> done{false};
+
+    ProgressTracker() {}
+  };
+  std::optional<ProgressTracker> progress_tracker;
+
   int init_common();
 
 public:
@@ -1651,6 +1662,13 @@ class RGWCopyObj : public RGWOp {
   void pre_exec() override;
   void execute(optional_yield y) override;
   void progress_cb(off_t ofs);
+  void progress_cb_handler();
+  ProgressTracker& get_progress_tracker() {
+    if (!progress_tracker) {
+      progress_tracker.emplace();
+    }
+    return *progress_tracker;
+  }
 
   virtual int check_storage_class(const rgw_placement_rule& src_placement) {
     return 0;

src/test/rgw/rgw_multi/tests.py

@@ -5243,8 +5243,6 @@ def test_copy_obj_between_zonegroups(zonegroup):
     source_bucket = source_zone.create_bucket(gen_bucket_name())
 
     objname = 'dummy'
-    k = new_key(source_zone, source_bucket.name, objname)
-    k.set_contents_from_string('foo')
 
     for zg in realm.current_period.zonegroups:
         if zg.name == zonegroup.name:
@@ -5254,16 +5252,24 @@ def test_copy_obj_between_zonegroups(zonegroup):
         dest_bucket = dest_zone.create_bucket(gen_bucket_name())
         realm_meta_checkpoint(realm)
 
-        # copy object
-        dest_zone.s3_client.copy_object(
-            Bucket=dest_bucket.name,
-            CopySource=f'{source_bucket.name}/{objname}',
-            Key=objname
-        )
-
-        # check that object exists in destination bucket
-        k = get_key(dest_zone, dest_bucket, objname)
-        assert_equal(k.get_contents_as_string().decode('utf-8'), 'foo')
+        # try object sizes of 4K and 8MiB
+        # 4K to test no progress case
+        # 8MiB to test progress case
+        obj_sizes = [4096, 8 * 1024 * 1024]
+        for size in obj_sizes:
+            k = new_key(source_zone, source_bucket.name, objname)
+            k.set_contents_from_string('x' * size)
+
+            # copy object
+            dest_zone.s3_client.copy_object(
+                Bucket=dest_bucket.name,
+                CopySource=f'{source_bucket.name}/{objname}',
+                Key=objname
+            )
+
+            # check that object exists in destination bucket
+            k = get_key(dest_zone, dest_bucket, objname)
+            assert_equal(k.get_contents_as_string().decode('utf-8'), 'x' * size)
 
 @allow_bucket_replication
 def test_bucket_replication_alt_user_delete_forbidden():