osd: add flag to prevent truncate_seq copy in copy-from operation

It doesn't make sense to copy truncate_seq from the base object for the
CephFS use-case (when using the copy_file_range syscall).  Doing so results
in issues when performing an object copy after a truncate operation.

This patch adds a flag to the copy-from op that clients can set so that the
truncate_seq and truncate_size to set in the destination object are sent in
the op payload.

Clients can check if 'copy-from' without truncate_seq copy is supported by
checking the SERVER_OCTOPUS feature bit.

Fixes: https://tracker.ceph.com/issues/37378
Signed-off-by: Luis Henriques <lhenriques@suse.com>

diff --git a/src/include/rados.h b/src/include/rados.h
index 44cbacffcad87553add75de1f94626056a0e6a8d..5fc54cfbda8981dfb1d83411f773d1de91e20de2 100644 (file)
--- a/src/include/rados.h
+++ b/src/include/rados.h
@@ -509,6 +509,7 @@ enum {
        CEPH_OSD_COPY_FROM_FLAG_MAP_SNAP_CLONE = 8, /* map snap direct to
                                                     * cloneid */
        CEPH_OSD_COPY_FROM_FLAG_RWORDERED = 16, /* order with write */
+       CEPH_OSD_COPY_FROM_FLAG_TRUNCATE_SEQ = 32, /* send truncate_{seq,size} */
 };
 
 enum {

diff --git a/src/osd/PrimaryLogPG.cc b/src/osd/PrimaryLogPG.cc
index 1667781b890333b9681106ed1331b7ffd080876d..c8274b3c041cdda6a9ded21ca0c12b43a63ecd3b 100644 (file)
--- a/src/osd/PrimaryLogPG.cc
+++ b/src/osd/PrimaryLogPG.cc
@@ -280,6 +280,9 @@ public:
   PrimaryLogPG::CopyResults *results = nullptr;
   PrimaryLogPG::OpContext *ctx;
   OSDOp &osd_op;
+  uint32_t truncate_seq;
+  uint64_t truncate_size;
+  bool have_truncate = false;
 
   CopyFromCallback(PrimaryLogPG::OpContext *ctx, OSDOp &osd_op)
     : ctx(ctx), osd_op(osd_op) {
@@ -290,6 +293,13 @@ public:
     results = results_.get<1>();
     int r = results_.get<0>();
 
+    // Only use truncate_{seq,size} from the original object if the client
+    // did not sent us these parameters
+    if (!have_truncate) {
+      truncate_seq = results->truncate_seq;
+      truncate_size = results->truncate_size;
+    }
+
     // for finish_copyfrom
     ctx->user_at_version = results->user_version;
 
@@ -313,6 +323,11 @@ public:
   uint64_t get_data_size() {
     return results->object_size;
   }
+  void set_truncate(uint32_t seq, uint64_t size) {
+    truncate_seq = seq;
+    truncate_size = size;
+    have_truncate = true;
+  }
 };
 
 struct CopyFromFinisher : public PrimaryLogPG::OpFinisher {
@@ -7668,11 +7683,21 @@ int PrimaryLogPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
       {
        object_t src_name;
        object_locator_t src_oloc;
+       uint32_t truncate_seq = 0;
+       uint64_t truncate_size = 0;
+       bool have_truncate = false;
        snapid_t src_snapid = (uint64_t)op.copy_from.snapid;
        version_t src_version = op.copy_from.src_version;
        try {
          decode(src_name, bp);
          decode(src_oloc, bp);
+         // check if client sent us truncate_seq and truncate_size
+         if ((op.copy_from.flags & CEPH_OSD_COPY_FROM_FLAG_TRUNCATE_SEQ) &&
+             !bp.end()) {
+           decode(truncate_seq, bp);
+           decode(truncate_size, bp);
+           have_truncate = true;
+         }
        }
        catch (buffer::error& e) {
          result = -EINVAL;
@@ -7713,6 +7738,8 @@ int PrimaryLogPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
            break;
          }
          CopyFromCallback *cb = new CopyFromCallback(ctx, osd_op);
+         if (have_truncate)
+           cb->set_truncate(truncate_seq, truncate_size);
           ctx->op_finishers[ctx->current_osd_subop_num].reset(
             new CopyFromFinisher(cb));
          start_copy(cb, ctx->obc, src, src_oloc, src_version,
@@ -9544,8 +9571,8 @@ void PrimaryLogPG::finish_copyfrom(CopyFromCallback *cb)
     obs.oi.clear_omap_digest();
   }
 
-  obs.oi.truncate_seq = cb->results->truncate_seq;
-  obs.oi.truncate_size = cb->results->truncate_size;
+  obs.oi.truncate_seq = cb->truncate_seq;
+  obs.oi.truncate_size = cb->truncate_size;
 
   obs.oi.mtime = ceph::real_clock::to_timespec(cb->results->mtime);
   ctx->mtime = utime_t();