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 352bf0bd84d1c6137b0ed67f7232fb67a24c8ebd..9fa484029bbd3c59c3db7a7c2cc377c676ae333c 100644 (file)
--- 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 8a45177fd1d0464fac38fc12d13a653c0bc63049..4ddd8d03505712d13231588a1dd59e10a0ad767e 100644 (file)
--- 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());