osd: use (robust) helper for setting exists or clearing whiteout

The current blanket check in prepare_transaction() will trigger only when
there is a net obs.exists change from the commited obs to new_obs.
However, this misses the case where the first osd_op is a delete and then a
subsequent osd_op recreates the object.  Changing the whiteout check to
look only at new_obs does not work because it fails to understand when
_delete_oid sets the whiteout and will simply clear it again.

In order to support sequences of delete + create in general, we need to do
the whiteout flag clearing when the actual create happens (to match the
fact that we set it when we process the delete osd_op).  Use a helper to
do this and consolidate most other obs.exists = true code to use it.

Backport: giant, firefly
Signed-off-by: Sage Weil <sage@redhat.com>

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 057564c37a1d6b2f6a5ed734b486702a290ce02d..b3b3f1ba49d1823d8c6c49f19783f568662fa39d 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -3505,6 +3505,22 @@ static string list_entries(const T& m) {
   return s;
 }
 
+bool ReplicatedPG::maybe_create_new_object(OpContext *ctx)
+{
+  ObjectState& obs = ctx->new_obs;
+  if (!obs.exists) {
+    ctx->delta_stats.num_objects++;
+    obs.exists = true;
+    obs.oi.new_object();
+    return true;
+  } else if (obs.oi.is_whiteout()) {
+    dout(10) << __func__ << " clearing whiteout on " << obs.oi.soid << dendl;
+    ctx->new_obs.oi.clear_flag(object_info_t::FLAG_WHITEOUT);
+    --ctx->delta_stats.num_whiteouts;
+  }
+  return false;
+}
+
 int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
 {
   int result = 0;
@@ -4275,13 +4291,10 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
           result = -EOPNOTSUPP;
           break;
         }
-        if (!obs.exists) {
+       if (maybe_create_new_object(ctx)) {
           ctx->mod_desc.create();
           t->touch(soid);
-          ctx->delta_stats.num_objects++;
-          obs.exists = true;
-         obs.oi.new_object();
-        }
+       }
         t->set_alloc_hint(soid, op.alloc_hint.expected_object_size,
                           op.alloc_hint.expected_write_size);
         ctx->delta_stats.num_wr++;
@@ -4365,11 +4378,7 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
        }
        write_update_size_and_usage(ctx->delta_stats, oi, ctx->modified_ranges,
                                    op.extent.offset, op.extent.length, true);
-       if (!obs.exists) {
-         ctx->delta_stats.num_objects++;
-         obs.exists = true;
-         obs.oi.set_omap_digest(-1);
-       }
+       maybe_create_new_object(ctx);
        if (op.extent.offset == 0 && op.extent.length == oi.size)
          obs.oi.set_data_digest(osd_op.indata.crc32c(-1));
        else
@@ -4426,11 +4435,7 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
          }
          t->write(soid, op.extent.offset, op.extent.length, osd_op.indata, op.flags);
        }
-       if (!obs.exists) {
-         ctx->delta_stats.num_objects++;
-         obs.exists = true;
-         obs.oi.set_omap_digest(-1);  // no omap yet
-       }
+       maybe_create_new_object(ctx);
        obs.oi.set_data_digest(osd_op.indata.crc32c(-1));
 
        interval_set<uint64_t> ch;
@@ -4500,14 +4505,10 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
            // category is no longer implemented.
          }
           if (result >= 0) {
-            if (!obs.exists)
+           if (maybe_create_new_object(ctx)) {
               ctx->mod_desc.create();
+           }
             t->touch(soid);
-            if (!obs.exists) {
-              ctx->delta_stats.num_objects++;
-              obs.exists = true;
-             obs.oi.new_object();
-            }
           }
        }
       }
@@ -4584,11 +4585,8 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
       ++ctx->num_read;
       ++ctx->num_write;
       {
-       if (!obs.exists) {
+       if (maybe_create_new_object(ctx)) {
          t->touch(obs.oi.soid);
-         ctx->delta_stats.num_objects++;
-         obs.exists = true;
-         obs.oi.new_object();
        }
        if (op.clonerange.src_offset + op.clonerange.length > src_obc->obs.oi.size) {
          dout(10) << " clonerange source " << osd_op.soid << " "
@@ -4699,12 +4697,9 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
          result = -ENAMETOOLONG;
          break;
        }
-       if (!obs.exists) {
+       if (maybe_create_new_object(ctx)) {
          ctx->mod_desc.create();
          t->touch(soid);
-         ctx->delta_stats.num_objects++;
-         obs.exists = true;
-         obs.oi.new_object();
        }
        string aname;
        bp.copy(op.xattr.name_len, aname);
@@ -5073,13 +5068,11 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
       ctx->mod_desc.mark_unrollbackable();
       ++ctx->num_write;
       {
-       if (!obs.exists) {
-         ctx->delta_stats.num_objects++;
-         obs.exists = true;
-         obs.oi.set_data_digest(-1);
+       if (maybe_create_new_object(ctx)) {
+         t->touch(soid);
+       } else {
+         obs.oi.clear_omap_digest();
        }
-       obs.oi.clear_omap_digest();
-       t->touch(soid);
        map<string, bufferlist> to_set;
        try {
          ::decode(to_set, bp);
@@ -5111,13 +5104,11 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
       ctx->mod_desc.mark_unrollbackable();
       ++ctx->num_write;
       {
-       if (!obs.exists) {
-         ctx->delta_stats.num_objects++;
-         obs.exists = true;
-         obs.oi.set_data_digest(-1);
+       if (maybe_create_new_object(ctx)) {
+         t->touch(soid);
+       } else {
+         obs.oi.clear_omap_digest();
        }
-       obs.oi.clear_omap_digest();
-       t->touch(soid);
        t->omap_setheader(soid, osd_op.indata);
        ctx->delta_stats.num_wr++;
       }
@@ -5502,10 +5493,7 @@ int ReplicatedPG::_rollback_to(OpContext *ctx, ceph_osd_op& op)
       }
 
       // Adjust the cached objectcontext
-      if (!obs.exists) {
-       obs.exists = true; //we're about to recreate it
-       ctx->delta_stats.num_objects++;
-      }
+      maybe_create_new_object(ctx);
       ctx->delta_stats.num_bytes -= obs.oi.size;
       ctx->delta_stats.num_bytes += rollback_to->obs.oi.size;
       obs.oi.size = rollback_to->obs.oi.size;
@@ -5859,16 +5847,6 @@ int ReplicatedPG::prepare_transaction(OpContext *ctx)
     return result;
   }
 
-  // cache: clear whiteout?
-  if (pool.info.cache_mode != pg_pool_t::CACHEMODE_NONE) {
-    if (ctx->user_modify &&
-       ctx->obc->obs.oi.is_whiteout()) {
-      dout(10) << __func__ << " clearing whiteout on " << soid << dendl;
-      ctx->new_obs.oi.clear_flag(object_info_t::FLAG_WHITEOUT);
-      --ctx->delta_stats.num_whiteouts;
-    }
-  }
-
   // clone, if necessary
   if (soid.snap == CEPH_NOSNAP)
     make_writeable(ctx);

diff --git a/src/osd/ReplicatedPG.h b/src/osd/ReplicatedPG.h
index f128c71e8d442f7f70038736211d16ead890711c..61477b5acbdd4699f9efa03ade1b2b6aaeb73413 100644 (file)
--- a/src/osd/ReplicatedPG.h
+++ b/src/osd/ReplicatedPG.h
@@ -1499,6 +1499,9 @@ private:
 
   int _verify_no_head_clones(const hobject_t& soid,
                             const SnapSet& ss);
+  // return true if we're creating a local object, false for a
+  // whiteout or no change.
+  bool maybe_create_new_object(OpContext *ctx);
   int _delete_oid(OpContext *ctx, bool no_whiteout);
   int _rollback_to(OpContext *ctx, ceph_osd_op& op);
 public: