From: Samuel Just <sam.just@inktank.com>
Date: Mon, 17 Mar 2014 21:28:16 +0000 (-0700)
Subject: ReplicatedPG: if !obc->obs->exists, lock snapdir obc if exists
X-Git-Tag: v0.78~2^2
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=refs%2Fpull%2F1505%2Fhead;p=ceph.git

ReplicatedPG: if !obc->obs->exists, lock snapdir obc if exists

Otherwise, backfill might slip in during a copyfrom.

Fixes: #7728
Signed-off-by: Samuel Just <sam.just@inktank.com>
---

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 352bf0bd84d..9fa484029bb 100644
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -1421,6 +1421,10 @@ void ReplicatedPG::do_op(OpRequestRef op)
 				 this);
   ctx->op_t = pgbackend->get_transaction();
   ctx->obc = obc;
+
+  if (!obc->obs.exists)
+    ctx->snapset_obc = get_object_context(obc->obs.oi.soid.get_snapdir(), false);
+
   if (m->get_flags() & CEPH_OSD_FLAG_SKIPRWLOCKS) {
     dout(20) << __func__ << ": skipping rw locks" << dendl;
   } else if (m->get_flags() & CEPH_OSD_FLAG_FLUSH) {
@@ -4987,15 +4991,8 @@ void ReplicatedPG::finish_ctx(OpContext *ctx, int log_op_type)
 
     if (ctx->new_obs.exists) {
       if (!ctx->obs->exists) {
-	// if we logically recreated the head, remove old _snapdir object
-	hobject_t snapoid(soid.oid, soid.get_key(), CEPH_SNAPDIR, soid.hash,
-			  info.pgid.pool(), soid.get_namespace());
-
-	ctx->snapset_obc = get_object_context(snapoid, false);
 	if (ctx->snapset_obc && ctx->snapset_obc->obs.exists) {
-	  bool got = ctx->snapset_obc->get_write(ctx->op);
-	  assert(got);
-	  ctx->release_snapset_obc = true;
+	  hobject_t snapoid = soid.get_snapdir();
 	  ctx->log.push_back(pg_log_entry_t(pg_log_entry_t::DELETE, snapoid,
 	      ctx->at_version,
 	      ctx->snapset_obc->obs.oi.version,
diff --git a/src/osd/ReplicatedPG.h b/src/osd/ReplicatedPG.h
index 8a45177fd1d..4ddd8d03505 100644
--- a/src/osd/ReplicatedPG.h
+++ b/src/osd/ReplicatedPG.h
@@ -643,11 +643,30 @@ protected:
    */
   bool get_rw_locks(OpContext *ctx) {
     if (ctx->op->may_write() || ctx->op->may_cache()) {
-      if (ctx->obc->get_write(ctx->op)) {
-	ctx->lock_to_release = OpContext::W_LOCK;
+      /* If snapset_obc, !obc->obs->exists and we need to
+       * get a write lock on the snapdir as well as the
+       * head.  Fortunately, we are guarranteed to get a
+       * write lock on the head if !obc->obs->exists
+       */
+      if (ctx->snapset_obc) {
+	assert(!ctx->obc->obs.exists);
+	if (ctx->snapset_obc->get_write(ctx->op)) {
+	  ctx->release_snapset_obc = true;
+	  ctx->lock_to_release = OpContext::W_LOCK;
+	} else {
+	  return false;
+	}
+	// we are creating it and have the only ref
+	bool got = ctx->obc->get_write(ctx->op);
+	assert(got);
 	return true;
       } else {
-	return false;
+	if (ctx->obc->get_write(ctx->op)) {
+	  ctx->lock_to_release = OpContext::W_LOCK;
+	  return true;
+	} else {
+	  return false;
+	}
       }
     } else {
       assert(ctx->op->may_read());