From fbc83116cbe2636683534b310a09be7087544216 Mon Sep 17 00:00:00 2001
From: myoungwon oh <ohmyoungwon@gmail.com>
Date: Wed, 24 Feb 2021 23:16:39 +0900
Subject: [PATCH] osd: recover unreadable snapshots when handling manifest
 object

The manifest object needs adjacent clones to increment/decrement
refcount when modifying the object. So, recovering the clones is needed
if the adjacent clones are unreadable.

Signed-off-by: Myoungwon Oh <myoungwon.oh@samsumg.com>
(cherry picked from commit b242a2c061182b724fa8016d1175f2b23e570d4b)
---
 src/osd/PrimaryLogPG.cc | 35 +++++++++++++++++++++++++++++++++++
 src/osd/PrimaryLogPG.h  |  1 +
 2 files changed, 36 insertions(+)

diff --git a/src/osd/PrimaryLogPG.cc b/src/osd/PrimaryLogPG.cc
index 4a17591cf194..24b53d696745 100644
--- a/src/osd/PrimaryLogPG.cc
+++ b/src/osd/PrimaryLogPG.cc
@@ -2287,6 +2287,9 @@ void PrimaryLogPG::do_op(OpRequestRef& op)
   }
 
   if (obc.get() && obc->obs.exists && obc->obs.oi.has_manifest()) {
+    if (recover_adjacent_clones(obc, op)) {
+      return;
+    }
     if (maybe_handle_manifest(op,
 			       write_ordered,
 			       obc))
@@ -3378,6 +3381,38 @@ int PrimaryLogPG::get_manifest_ref_count(ObjectContextRef obc, std::string& fp_o
   return cnt;
 }
 
+bool PrimaryLogPG::recover_adjacent_clones(ObjectContextRef obc, OpRequestRef op)
+{
+  if (!obc->obs.oi.manifest.is_chunked() || !obc->ssc || !obc->ssc->snapset.clones.size()) {
+    return false;
+  }
+
+  const SnapSet& snapset = obc->ssc->snapset;
+  auto s = std::find(snapset.clones.begin(), snapset.clones.end(), obc->obs.oi.soid.snap);
+  auto is_unreadable_snap = [this, obc, &snapset, op](auto iter) -> bool {
+    hobject_t cid = obc->obs.oi.soid;
+    cid.snap = (iter == snapset.clones.end()) ? snapid_t(CEPH_NOSNAP) : *iter;
+    if (is_unreadable_object(cid)) {
+      dout(10) << __func__ << ": clone " << cid
+	       << " is unreadable, waiting" << dendl;
+      wait_for_unreadable_object(cid, op);
+      return true;
+    }
+    return false;
+  };
+  if (s != snapset.clones.begin()) {
+    if (is_unreadable_snap(s - 1)) {
+      return true;
+    }
+  }
+  if (s != snapset.clones.end()) {
+    if (is_unreadable_snap(s + 1)) {
+      return true;
+    }
+  }
+  return false;
+}
+
 ObjectContextRef PrimaryLogPG::get_prev_clone_obc(ObjectContextRef obc)
 {
   auto s = std::find(obc->ssc->snapset.clones.begin(), obc->ssc->snapset.clones.end(),
diff --git a/src/osd/PrimaryLogPG.h b/src/osd/PrimaryLogPG.h
index 134b3ba1a937..1a8f8560870b 100644
--- a/src/osd/PrimaryLogPG.h
+++ b/src/osd/PrimaryLogPG.h
@@ -1467,6 +1467,7 @@ protected:
   void dec_refcount(const hobject_t& soid, const object_ref_delta_t& refs);
   void dec_refcount_by_dirty(OpContext* ctx);
   ObjectContextRef get_prev_clone_obc(ObjectContextRef obc);
+  bool recover_adjacent_clones(ObjectContextRef obc, OpRequestRef op);
   void get_adjacent_clones(ObjectContextRef src_obc, 
 			   ObjectContextRef& _l, ObjectContextRef& _g);
   bool inc_refcount_by_set(OpContext* ctx, object_manifest_t& tgt,
-- 
2.47.3