ReplicatedPG: register_object_context and register_snapset_context cleanup

Previously, get_object_context and get_snapset_context did not register
the resulting objects.  In some cases, these objects would not get
registered and multiple copies would end up created.  This caused a bug
in find_object_context where get_snapset_context could return an object
distinct from the one referenced by the object returned from
get_object_context.

Signed-off-by: Samuel Just <samuelj@hq.newdream.net>

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 204d52c5871e1b8efe1cbebb80e802deef4b3fbf..3bc87f8d3e9fb25f9edc161462306de623bc3705 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -516,7 +516,7 @@ void ReplicatedPG::do_op(MOSDOp *op)
   }
   
   // continuing on to write path, make sure object context is registered
-  register_object_context(obc);
+  assert(obc->registered);
 
   // issue replica writes
   tid_t rep_tid = osd->get_tid();
@@ -672,7 +672,7 @@ bool ReplicatedPG::snap_trimmer()
       ObjectContext *obc;
       int r = find_object_context(coid.oid, coi.oloc, sn, &obc, false);
       assert(r == 0);
-      register_object_context(obc);
+      assert(obc->registered);
 
       vector<OSDOp> ops;
       tid_t rep_tid = osd->get_tid();
@@ -761,7 +761,7 @@ bool ReplicatedPG::snap_trimmer()
 
       sobject_t snapoid(coid.oid, snapset.head_exists ? CEPH_NOSNAP:CEPH_SNAPDIR);
       ctx->snapset_obc = get_object_context(snapoid, coi.oloc, false);
-      register_object_context(ctx->snapset_obc);
+      assert(ctx->snapset_obc->registered);
       if (snapset.clones.empty() && !snapset.head_exists) {
        dout(10) << coid << " removing " << snapoid << dendl;
        ctx->log.push_back(Log::Entry(Log::Entry::DELETE, snapoid, ctx->at_version, 
@@ -1574,7 +1574,7 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,
               osd->client_messenger->send_message(notify_msg, session->con);
             }
           }
-         register_object_context(obc);
+         assert(obc->registered);
         } else {
          map<entity_name_t, watch_info_t>::iterator oi_iter = oi.watchers.find(entity);
          if (oi_iter != oi.watchers.end()) {
@@ -2183,7 +2183,7 @@ int ReplicatedPG::prepare_transaction(OpContext *ctx)
                                      osd_reqid_t(), ctx->mtime));
 
        ctx->snapset_obc->obs.exists = false;
-       register_object_context(ctx->snapset_obc);
+       assert(ctx->snapset_obc->registered);
       }
     }
   } else if (ctx->obs->ssc->snapset.clones.size()) {
@@ -2200,7 +2200,7 @@ int ReplicatedPG::prepare_transaction(OpContext *ctx)
     ctx->snapset_obc->obs.oi.version = ctx->at_version;
     ctx->snapset_obc->obs.oi.last_reqid = ctx->reqid;
     ctx->snapset_obc->obs.oi.mtime = ctx->mtime;
-    register_object_context(ctx->snapset_obc);
+    assert(ctx->snapset_obc->registered);
 
     bufferlist bv(sizeof(*poi));
     ::encode(ctx->snapset_obc->obs.oi, bv);
@@ -2666,6 +2666,7 @@ ReplicatedPG::ObjectContext *ReplicatedPG::get_object_context(const sobject_t& s
        ssc = get_snapset_context(soid.oid, true);
       obc = new ObjectContext(oi, true, ssc);
     }
+    register_object_context(obc);
 
     if (can_create)
       obc->obs.ssc = get_snapset_context(soid.oid, true);
@@ -2836,6 +2837,7 @@ ReplicatedPG::SnapSetContext *ReplicatedPG::get_snapset_context(const object_t&
        return NULL;
     }
     ssc = new SnapSetContext(oid);
+    register_snapset_context(ssc);
     if (r >= 0) {
       bufferlist::iterator bvp = bv.begin();
       ssc->snapset.decode(bvp);
@@ -3834,7 +3836,7 @@ void ReplicatedPG::sub_op_push(MOSDSubOp *op)
     if (is_primary()) {
       dout(10) << " setting up obc for " << soid << dendl;
       ObjectContext *obc = get_object_context(soid, op->oloc, true);
-      register_object_context(obc);
+      assert(obc->registered);
       obc->ondisk_write_lock();
       
       obc->obs.exists = true;