rgwlc: prevent lc for one bucket from exceeding time budget

Fixes: https://tracker.ceph.com/issues/57951
Signed-off-by: Matt Benjamin <mbenjamin@redhat.com>
(cherry picked from commit 617ffccbca0169ac0f1cd713962d44e8cc74a8af)

diff --git a/src/rgw/rgw_lc.cc b/src/rgw/rgw_lc.cc
index 6dad16efc8086e4508f682ab895aac58e04895c7..1a23988f8fc7f014bc7a92f887f0862d58443298 100644 (file)
--- a/src/rgw/rgw_lc.cc
+++ b/src/rgw/rgw_lc.cc
@@ -941,6 +941,7 @@ int RGWLC::handle_multipart_expiration(rgw::sal::Bucket* target,
     }
     params.prefix = prefix_iter->first;
     do {
+      auto offset = 0;
       results.objs.clear();
       ret = target->list(this, params, 1000, results, null_yield);
       if (ret < 0) {
@@ -950,7 +951,7 @@ int RGWLC::handle_multipart_expiration(rgw::sal::Bucket* target,
           return ret;
       }
 
-      for (auto obj_iter = results.objs.begin(); obj_iter != results.objs.end(); ++obj_iter) {
+      for (auto obj_iter = results.objs.begin(); obj_iter != results.objs.end(); ++obj_iter, ++offset) {
        std::tuple<lc_op, rgw_bucket_dir_entry> t1 =
          {prefix_iter->second, *obj_iter};
        worker->workpool->enqueue(WorkItem{t1});
@@ -959,6 +960,15 @@ int RGWLC::handle_multipart_expiration(rgw::sal::Bucket* target,
        }
       } /* for objs */
 
+      if ((offset % 100) == 0) {
+       if (worker_should_stop(stop_at, once)) {
+         ldpp_dout(this, 5) << __func__ << " interval budget EXPIRED worker "
+                            << worker->ix
+                            << dendl;
+         return 0;
+       }
+      }
+
       std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
     } while(results.is_truncated);
   } /* for prefix_map */
@@ -1822,10 +1832,18 @@ int RGWLC::bucket_lc_process(string& shard_id, LCWorker* worker,
     LCOpRule orule(oenv);
     orule.build(); // why can't ctor do it?
     rgw_bucket_dir_entry* o{nullptr};
-    for (; ol.get_obj(this, &o /* , fetch_barrier */); ol.next()) {
+    for (auto offset = 0; ol.get_obj(this, &o /* , fetch_barrier */); ++offset, ol.next()) {
       orule.update();
       std::tuple<LCOpRule, rgw_bucket_dir_entry> t1 = {orule, *o};
       worker->workpool->enqueue(WorkItem{t1});
+      if ((offset % 100) == 0) {
+       if (worker_should_stop(stop_at, once)) {
+         ldpp_dout(this, 5) << __func__ << " interval budget EXPIRED worker "
+                            << worker->ix
+                            << dendl;
+         return 0;
+       }
+      }
     }
     worker->workpool->drain();
   }