PG: fix add_next_event and merge add_event

Previously, we would assume that we had an object at the prior_version
in the log event if we encounter it but don't see the object in missing.
Now, if prior_version < log_tail, we assume that we do not have the
object.

Signed-off-by: Samuel Just <samuel.just@dreamhost.com>

diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index f01c114fecd91687c761516b251abb31cae34179..7213e9a376a0345602b53c72b3177c0b6a9aa54c 100644 (file)
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -427,7 +427,7 @@ void PG::merge_log(ObjectStore::Transaction& t,
     for (; p != log.log.end(); p++) {
       Log::Entry &ne = *p;
       dout(20) << "merge_log merging " << ne << dendl;
-      missing.add_next_event(ne);
+      missing.add_next_event(ne, oinfo);
       if (ne.is_delete())
        t.remove(coll, ne.soid);
     }
@@ -513,7 +513,7 @@ void PG::merge_log(ObjectStore::Transaction& t,
        Log::Entry &ne = *p;
         dout(20) << "merge_log " << ne << dendl;
        log.index(ne);
-       missing.add_next_event(ne);
+       missing.add_next_event(ne, info);
        if (ne.is_delete())
          t.remove(coll, ne.soid);
       }
@@ -1635,7 +1635,7 @@ void PG::activate(ObjectStore::Transaction& t, list<Context*>& tfin,
              p != m->log.log.end();
              p++) 
           if (p->version > plu)
-            pm.add_event(*p);
+            pm.add_next_event(*p, pi);
       }
       
       if (m) {
@@ -3727,7 +3727,7 @@ eversion_t PG::Missing::have_old(const sobject_t& oid) const
  * this needs to be called in log order as we extend the log.  it
  * assumes missing is accurate up through the previous log entry.
  */
-void PG::Missing::add_next_event(Log::Entry& e)
+void PG::Missing::add_next_event(Log::Entry& e, const Info &info)
 {
   if (e.is_update()) {
     if (e.prior_version == eversion_t()) {
@@ -3741,7 +3741,10 @@ void PG::Missing::add_next_event(Log::Entry& e)
       //assert(missing[e.soid].need == e.prior_version);
       rmissing.erase(missing[e.soid].need.version);
       missing[e.soid].need = e.version;  // leave .have unchanged.
-    } else if (e.is_backlog()) {
+    } else if (e.is_backlog() ||
+              e.prior_version <= info.log_tail ||
+              e.prior_version > info.last_update) {
+      // May not have prior version
       missing[e.soid].need = e.version;
     } else {
       // not missing, we must have prior_version (if any)
@@ -3752,25 +3755,6 @@ void PG::Missing::add_next_event(Log::Entry& e)
     rm(e.soid, e.version);
 }
 
-void PG::Missing::add_event(Log::Entry& e)
-{
-  if (e.is_update()) {
-    if (missing.count(e.soid)) {
-      if (missing[e.soid].need >= e.version)
-       return;   // already missing same or newer.
-      // missing older, revise need
-      rmissing.erase(missing[e.soid].need.version);
-      missing[e.soid].need = e.version;
-    } else if (e.is_backlog()) {
-      missing[e.soid].need = e.version;
-    } else
-      // not missing => have prior_version (if any)
-      missing[e.soid] = item(e.version, e.prior_version);
-    rmissing[e.version.version] = e.soid;
-  } else
-    rm(e.soid, e.version);
-}
-
 void PG::Missing::revise_need(sobject_t oid, eversion_t need)
 {
   if (missing.count(oid)) {

diff --git a/src/osd/PG.h b/src/osd/PG.h
index e3b31e66ba8fd7d6e8f71937acde82059557d2d2..50969a6f946c3c87d592fcdccb710d79ddf012fa 100644 (file)
--- a/src/osd/PG.h
+++ b/src/osd/PG.h
@@ -661,8 +661,7 @@ public:
     bool is_missing(const sobject_t& oid) const;
     bool is_missing(const sobject_t& oid, eversion_t v) const;
     eversion_t have_old(const sobject_t& oid) const;
-    void add_next_event(Log::Entry& e);
-    void add_event(Log::Entry& e);
+    void add_next_event(Log::Entry& e, const Info &info);
     void revise_need(sobject_t oid, eversion_t need);
     void add(const sobject_t& oid, eversion_t need, eversion_t have);
     void rm(const sobject_t& oid, eversion_t v);