From 09d611d25c660a700db012a913e38c7229f2110a Mon Sep 17 00:00:00 2001
From: Samuel Just <sam.just@inktank.com>
Date: Mon, 10 Feb 2014 14:07:07 -0800
Subject: [PATCH] PGLog::merge_old_entry: simplify the oe.version > ne.version
 case

If ne.version < oe.version, the correct answer is to rollback oe.version
if possible regardless of what the entries are.

Also, rearrange to deal with the fact that we cannot roll back a missing
object.

Signed-off-by: Samuel Just <sam.just@inktank.com>
---
 src/osd/PGLog.cc | 63 ++++++++++++++++++++++++++++++++++++------------
 1 file changed, 47 insertions(+), 16 deletions(-)

diff --git a/src/osd/PGLog.cc b/src/osd/PGLog.cc
index e845bd6a26e..a643a5a7fa7 100644
--- a/src/osd/PGLog.cc
+++ b/src/osd/PGLog.cc
@@ -269,7 +269,30 @@ bool PGLog::_merge_old_entry(
 	       << " new " << ne << " : same" << dendl;
       return true;
     }
-    if (oe.is_delete()) {
+
+    if (missing.is_missing(oe.soid)) {
+      if (!ne.is_delete()) {
+	dout(20) << __func__ << ": ne.version < oe.version && already missing, "
+		 << "revising missing need" << dendl;
+	if (rollbacker) {
+	  rollbacker->cant_rollback(oe);
+	}
+	missing.revise_need(ne.soid, ne.version);
+      } else {
+	dout(20) << __func__ << ": ne.version < oe.version && already missing, "
+		 << "ne is delete, clearing missing need" << dendl;
+	if (rollbacker) {
+	  rollbacker->remove(oe.soid);
+	  rollbacker->cant_rollback(oe);
+	}
+	missing.rm(oe.soid, oe.version);
+      }
+    } else if (oe.mod_desc.can_rollback() && oe.version > olog_can_rollback_to) {
+      dout(20) << __func__ << ": ne.version < oe.version && can rollback, "
+	       << "rolling back " << oe << dendl;
+      if (rollbacker)
+	rollbacker->rollback(oe);
+    } else if (oe.is_delete()) {
       if (ne.is_delete()) {
 	// old and new are delete
 	dout(20) << "merge_old_entry  had " << oe
@@ -291,22 +314,14 @@ bool PGLog::_merge_old_entry(
 	  rollbacker->remove(oe.soid);
 	  rollbacker->cant_rollback(oe);
 	}
-	if (missing.is_missing(oe.soid))
-	  missing.rm(oe.soid, oe.version);
+	assert(!missing.is_missing(oe.soid));
       } else {
 	// old update, new update
 	dout(20) << "merge_old_entry  had " << oe
 		 << " new " << ne << " : new item supercedes" << dendl;
-	if (oe.mod_desc.can_rollback() && oe.version > olog_can_rollback_to) {
-	  dout(20) << __func__ << ": ne.version < oe.version && can rollback, "
-		   << "rolling back " << oe << dendl;
-	  if (rollbacker)
-	    rollbacker->rollback(oe);
-	} else {
-	  missing.revise_need(ne.soid, ne.version);
-	  if (rollbacker)
-	    rollbacker->cant_rollback(oe);
-	}
+	missing.revise_need(ne.soid, ne.version);
+	if (rollbacker)
+	  rollbacker->cant_rollback(oe);
       }
     }
   } else if (oe.op == pg_log_entry_t::CLONE) {
@@ -338,7 +353,23 @@ bool PGLog::_merge_old_entry(
 	     << " oe.soid " << oe.soid
 	     << " must already have been merged" << dendl;
   } else {
-    if (oe.mod_desc.can_rollback() && oe.version > olog_can_rollback_to) {
+    if (missing.is_missing(oe.soid)) {
+      if (oe.prior_version > eversion_t()) {
+	dout(20) << __func__ << ": oe.prior_version > 0 && already missing, "
+		 << "revising missing need " << oe << dendl;
+	if (new_divergent_prior)
+	  *new_divergent_prior = make_pair(oe.prior_version, oe.soid);
+	missing.revise_need(oe.soid, oe.prior_version);
+      } else {
+	dout(20) << __func__ << ": oe.prior_version == 0 && already missing, "
+		 << "removing missing " << oe << dendl;
+	missing.rm(oe.soid, missing.missing[oe.soid].need);
+	if (rollbacker)
+	  rollbacker->remove(oe.soid);
+      }
+      if (rollbacker)
+	rollbacker->cant_rollback(oe);
+    } else if (oe.mod_desc.can_rollback() && oe.version > olog_can_rollback_to) {
       dout(20) << __func__ << ": rolling back " << oe << dendl;
       if (rollbacker)
 	rollbacker->rollback(oe);
@@ -356,8 +387,8 @@ bool PGLog::_merge_old_entry(
 	missing.revise_need(oe.soid, oe.prior_version);
 	if (rollbacker)
 	  rollbacker->cant_rollback(oe);
-      } else if (missing.is_missing(oe.soid)) {
-	missing.rm(oe.soid, missing.missing[oe.soid].need);
+      } else {
+	assert(!missing.is_missing(oe.soid));
       }
     }
   }