rgw/lc: advance head if the current entry doesn't exist

This is extension to https://github.com/ceph/ceph/pull/47595.

When skipping the non-exist LC entry, instead of get_next_entry(),
use advance_head() to fetch the next entry. In case the cycle
is finished for that shard, head should be reset to avoid lc process
being stuck in indefinite loop.

Fixes: https://tracker.ceph.com/issues/65160
Signed-off-by: Soumya Koduri <skoduri@redhat.com>

diff --git a/src/rgw/rgw_lc.cc b/src/rgw/rgw_lc.cc
index 7ae42cdfb7007d2a044c315eb96ea9254b5e73a5..280a12dcdab2a219b36acccc36e0a777dfc29474 100644 (file)
--- a/src/rgw/rgw_lc.cc
+++ b/src/rgw/rgw_lc.cc
@@ -2279,13 +2279,28 @@ int RGWLC::process(int index, int max_lock_secs, LCWorker* worker,
       /* fetches the entry pointed to by head.bucket */
       ret = sal_lc->get_entry(lc_shard, head->get_marker(), &entry);
       if (ret == -ENOENT) {
-        ret = sal_lc->get_next_entry(lc_shard, head->get_marker(), &entry);
-        if (ret < 0) {
-          ldpp_dout(this, 0) << "RGWLC::process() sal_lc->get_next_entry(lc_shard, "
-                             << "head.marker, entry) returned error ret==" << ret
-                             << dendl;
-          goto exit;
-        }
+        /* skip to next entry */
+       std::unique_ptr<rgw::sal::Lifecycle::LCEntry> tmp_entry = sal_lc->get_entry();
+       tmp_entry->set_bucket(head->get_marker());
+       if (advance_head(lc_shard, *head.get(), *tmp_entry.get(), now) < 0) {
+         goto exit;
+       }
+       /* done with this shard */
+       if (head->get_marker().empty()) {
+         ldpp_dout(this, 5) <<
+             "RGWLC::process() next_entry not found. cycle finished lc_shard="
+                              << lc_shard << " worker=" << worker->ix
+                              << dendl;
+          head->set_shard_rollover_date(ceph_clock_now());
+          ret = sal_lc->put_head(lc_shard, *head.get());
+          if (ret < 0) {
+           ldpp_dout(this, 0) << "RGWLC::process() failed to put head "
+                               << lc_shard
+                              << dendl;
+         }
+         goto exit;
+       }
+       continue;
       }
       if (ret < 0) {
        ldpp_dout(this, 0) << "RGWLC::process() sal_lc->get_entry(lc_shard, head.marker, entry) "