objecter, librados: add COPY_FROM operation

This operation will copy an entire object (data, attrs, omap)
atomically.  If the src_version does not match the source object, or
the source object is updated while the copy is in progress, we will
fail with a suitable error code.  By atomic we mean that it will either
successfully copy the entire object in its entirety or it will fail (and
require no cleanup).

Add to C++ librados API only for now.

Signed-off-by: Sage Weil <sage@inktank.com>
Conflicts:

	src/include/ceph_strings.cc
	src/include/rados.h
	src/osd/osd_types.cc

diff --git a/src/include/ceph_strings.cc b/src/include/ceph_strings.cc
index f14f29ce0e90743e5e2762dbb73686e2d9dc4dac..e86aae4fd502f7f2d1cd409710e8499cba97690d 100644 (file)
--- a/src/include/ceph_strings.cc
+++ b/src/include/ceph_strings.cc
@@ -49,6 +49,7 @@ const char *ceph_osd_op_name(int op)
        case CEPH_OSD_OP_WATCH: return "watch";
 
        case CEPH_OSD_OP_COPY_GET: return "copy-get";
+       case CEPH_OSD_OP_COPY_FROM: return "copy-from";
 
        case CEPH_OSD_OP_CLONERANGE: return "clonerange";
        case CEPH_OSD_OP_ASSERT_SRC_VERSION: return "assert-src-version";

diff --git a/src/include/rados.h b/src/include/rados.h
index 27291a7440e84bd51d2170c4ca78c8dc58f875a1..178c171c445ccdf3c5735b0b655dc51a01037c3b 100644 (file)
--- a/src/include/rados.h
+++ b/src/include/rados.h
@@ -217,6 +217,7 @@ enum {
        CEPH_OSD_OP_OMAPRMKEYS    = CEPH_OSD_OP_MODE_WR | CEPH_OSD_OP_TYPE_DATA | 24,
        CEPH_OSD_OP_OMAP_CMP      = CEPH_OSD_OP_MODE_RD | CEPH_OSD_OP_TYPE_DATA | 25,
 
+       CEPH_OSD_OP_COPY_FROM = CEPH_OSD_OP_MODE_WR | CEPH_OSD_OP_TYPE_DATA | 26,
        CEPH_OSD_OP_COPY_GET = CEPH_OSD_OP_MODE_RD | CEPH_OSD_OP_TYPE_DATA | 27,
 
        /** multi **/
@@ -410,6 +411,10 @@ struct ceph_osd_op {
                struct {
                        __le64 max;     /* max data in reply */
                } __attribute__ ((packed)) copy_get;
+               struct {
+                       __le64 snapid;
+                       __le64 src_version;
+               } __attribute__ ((packed)) copy_from;
        };
        __le32 payload_len;
 } __attribute__ ((packed));

diff --git a/src/include/rados/librados.hpp b/src/include/rados/librados.hpp
index bc0bcc95ceb54ffde9e55c95add8904f6d24fb51..5a750cbc0d13591d9be4ea5dd93dc971a0067b68 100644 (file)
--- a/src/include/rados/librados.hpp
+++ b/src/include/rados/librados.hpp
@@ -265,6 +265,19 @@ namespace librados
      */
     void omap_rm_keys(const std::set<std::string> &to_rm);
 
+    /**
+     * Copy an object
+     *
+     * Copies an object from another location.  The operation is atomic in that
+     * the copy either succeeds in its entirety or fails (e.g., because the
+     * source object was modified while the copy was in progress).
+     *
+     * @param src source object name
+     * @param src_ioctx ioctx for the source object
+     * @param version current version of the source object
+     */
+    void copy_from(const std::string& src, const IoCtx& src_ioctx, uint64_t src_version);
+
     friend class IoCtx;
   };
 
@@ -674,6 +687,7 @@ namespace librados
     IoCtx(IoCtxImpl *io_ctx_impl_);
 
     friend class Rados; // Only Rados can use our private constructor to create IoCtxes.
+    friend class ObjectWriteOperation;  // copy_from needs to see our IoCtxImpl
 
     IoCtxImpl *io_ctx_impl;
   };

diff --git a/src/librados/librados.cc b/src/librados/librados.cc
index 12372d960b1b02d2ab28dd658995807fd1cc0d76..852228ed3834f718feb81dd27c52a28e3036c162 100644 (file)
--- a/src/librados/librados.cc
+++ b/src/librados/librados.cc
@@ -382,6 +382,14 @@ void librados::ObjectWriteOperation::omap_rm_keys(
   o->omap_rm_keys(to_rm);
 }
 
+void librados::ObjectWriteOperation::copy_from(const std::string& src,
+                                              const IoCtx& src_ioctx,
+                                              uint64_t src_version)
+{
+  ::ObjectOperation *o = (::ObjectOperation *)impl;
+  o->copy_from(object_t(src), src_ioctx.io_ctx_impl->snap_seq, src_ioctx.io_ctx_impl->oloc, src_version);
+}
+
 void librados::ObjectWriteOperation::tmap_put(const bufferlist &bl)
 {
   ::ObjectOperation *o = (::ObjectOperation *)impl;

diff --git a/src/osd/osd_types.cc b/src/osd/osd_types.cc
index 1c1b457002c07661fa4f2b5d3c889ca6e840aa2c..3451d520ff2805b2f1726bfd0b40c84aef5ac931 100644 (file)
--- a/src/osd/osd_types.cc
+++ b/src/osd/osd_types.cc
@@ -3484,6 +3484,8 @@ ostream& operator<<(ostream& out, const OSDOp& op)
       break;
     case CEPH_OSD_OP_COPY_GET:
       out << " max " << op.op.copy_get.max;
+    case CEPH_OSD_OP_COPY_FROM:
+      out << " ver " << op.op.copy_from.src_version;
       break;
     default:
       out << " " << op.op.extent.offset << "~" << op.op.extent.length;

diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 91f625517291ce68237cb60e90c2cd91268604f0..154ee410fdee62cfe6abb0ded0732fa869edf049 100644 (file)
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -753,6 +753,14 @@ struct ObjectOperation {
     OSDOp& osd_op = add_op(CEPH_OSD_OP_ROLLBACK);
     osd_op.op.snap.snapid = snapid;
   }
+
+  void copy_from(object_t src, snapid_t snapid, object_locator_t src_oloc, version_t src_version) {
+    OSDOp& osd_op = add_op(CEPH_OSD_OP_COPY_FROM);
+    osd_op.op.copy_from.snapid = snapid;
+    osd_op.op.copy_from.src_version = src_version;
+    ::encode(src, osd_op.indata);
+    ::encode(src_oloc, osd_op.indata);
+  }
 };
 
 

diff --git a/src/test/librados/misc.cc b/src/test/librados/misc.cc
index 6cb7cf5452ac4c4166f1a0119e6bd84c5b61c150..9fe6427d3f8302dadfc66cad392884876bcd33a1 100644 (file)
--- a/src/test/librados/misc.cc
+++ b/src/test/librados/misc.cc
@@ -564,6 +564,30 @@ TEST(LibRadosMisc, BigAttrPP) {
   ASSERT_EQ(0, destroy_one_pool_pp(pool_name, cluster));
 }
 
+TEST(LibRadosMisc, CopyPP) {
+  Rados cluster;
+  std::string pool_name = get_temp_pool_name();
+  ASSERT_EQ("", create_one_pool_pp(pool_name, cluster));
+  IoCtx ioctx;
+  ASSERT_EQ(0, cluster.ioctx_create(pool_name.c_str(), ioctx));
+
+  char buf[64];
+  memset(buf, 0xcc, sizeof(buf));
+  bufferlist bl;
+  bl.append(buf, sizeof(buf));
+
+  ASSERT_EQ(0, ioctx.write_full("foo", bl));
+
+
+  ObjectWriteOperation op;
+  op.copyfrom("foo", ioctx, ioctx.get_last_version());
+
+  ASSERT_EQ(0, ioctx.operate("bar", &op));
+
+  ioctx.close();
+  ASSERT_EQ(0, destroy_one_pool_pp(pool_name, cluster));
+}
+
 int main(int argc, char **argv)
 {
   ::testing::InitGoogleTest(&argc, argv);