osd: revise missing map adjustment

Rewrite helpers in terms of how they are actually used.

diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index 8d27e2fa959d39b97d03e7886624ee6654dac3c1..f3be6a803c65590636f8a3d4865c82cb41a8d718 100644 (file)
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -283,8 +283,7 @@ void PG::merge_old_entry(ObjectStore::Transaction& t, Log::Entry& oe)
       } else {
        // old update, new update
        dout(20) << "merge_old_entry  had " << oe << " new " << ne << " : new item supercedes" << dendl;
-       missing.rm(oe.oid, oe.version);  // re-add older "new" entry to missing
-       missing.add_event(ne);
+       missing.revise_need(oe.oid, oe.version);
       }
     }
   } else {
@@ -348,7 +347,7 @@ void PG::merge_log(ObjectStore::Transaction& t, Info &oinfo, Log &olog, Missing
     for (; p != log.log.end(); p++) {
       Log::Entry &ne = *p;
       dout(10) << "merge_log merging " << ne << dendl;
-      missing.add_event(ne);
+      missing.add_next_event(ne);
       if (ne.is_delete())
        t.remove(info.pgid.to_coll(), pobject_t(info.pgid.pool(), 0, ne.oid));
     }
@@ -432,7 +431,7 @@ void PG::merge_log(ObjectStore::Transaction& t, Info &oinfo, Log &olog, Missing
        Log::Entry &ne = *p;
         dout(10) << "merge_log " << ne << dendl;
        log.index(ne);
-       missing.add_event(ne);
+       missing.add_next_event(ne);
        if (ne.is_delete())
          t.remove(info.pgid.to_coll(), pobject_t(info.pgid.pool(), 0, ne.oid));
       }
@@ -523,11 +522,11 @@ void PG::generate_backlog()
 
   /*
    * note that we don't create prior_version backlog entries for
-   * objects that no longer exist.  that's maybe a bit sloppy, but not
-   * a problem, since we mainly care about generating an accurate
-   * missing map, and an object that gets deleted will obviously not
-   * end up missing; in merge_log, we'll see the final remove entry,
-   * and missing.rm().
+   * objects that no longer exist (i.e., those for which there is a
+   * delete entry in the log).  that's maybe a bit sloppy, but not a
+   * problem, since we mainly care about generating an accurate
+   * missing map, and an object that was deleted should obviously not
+   * end up as missing.
    */
 
   int local = 0;

diff --git a/src/osd/PG.h b/src/osd/PG.h
index f68fe315ab89496b21c35dff2d57f7325b10438a..f3e089ea377e3b997c30b5f8bf0f04287e741ac1 100644 (file)
--- a/src/osd/PG.h
+++ b/src/osd/PG.h
@@ -431,29 +431,61 @@ public:
       return missing.count(oid) ? missing[oid].have : eversion_t();
     }
     
+    /*
+     * 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 add_next_event(Log::Entry& e) {
+      if (e.is_update()) {
+       if (e.prior_version == eversion_t()) {
+         // new object.
+         assert(missing.count(e.oid) == 0);
+         missing[e.oid].need = e.version;  // .have = nil
+       } else if (missing.count(e.oid)) {
+         // already missing (prior).
+         assert(missing[e.oid].need == e.prior_version);
+         rmissing.erase(e.prior_version);
+         missing[e.oid].need = e.version;  // .have unchanged.
+       } else {
+         // not missing, we must have prior_version (if any)
+         missing[e.oid] = item(e.version, e.prior_version);
+       }
+       rmissing[e.version] = e.oid;
+      } else
+       rm(e.oid, e.version);
+    }
+
     void add_event(Log::Entry& e) {
       if (e.is_update()) {
-       if (is_missing(e.oid, e.prior_version))
-         add(e.oid, e.version);
-       else
-         add(e.oid, e.version, e.prior_version);
+       if (missing.count(e.oid)) {
+         if (missing[e.oid].need >= e.version)
+           return;   // already missing same or newer.
+         // missing older, revise need
+         rmissing.erase(missing[e.oid].need);
+         missing[e.oid].need = e.version;
+       } else 
+         // not missing => have prior_version (if any)
+         missing[e.oid] = item(e.version, e.prior_version);
+       rmissing[e.version] = e.oid;
       } else
        rm(e.oid, e.version);
     }
 
-    void add(object_t oid, eversion_t need) {
-      eversion_t have;
-      add(oid, need, have);
+    void revise_need(object_t oid, eversion_t need) {
+      if (missing.count(oid)) {
+       rmissing.erase(missing[oid].need);
+       missing[oid].need = need;            // no not adjust .have
+      } else {
+       missing[oid] = item(need, eversion_t());
+      }
+      rmissing[need] = oid;
     }
+
     void add(object_t oid, eversion_t need, eversion_t have) {
-      if (missing.count(oid)) {
-        if (missing[oid].need > need) return;   // already missing newer.
-        rmissing.erase(missing[oid].need);
-       missing[oid].need = need;  // don't have .have
-      } else 
-       missing[oid] = item(need, have);
+      missing[oid] = item(need, have);
       rmissing[need] = oid;
     }
+    
     void rm(object_t oid, eversion_t when) {
       if (missing.count(oid) && missing[oid].need < when) {
         rmissing.erase(missing[oid].need);