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

7 files changed, 62 insertions, 0 deletions

diff --git a/src/include/ceph_strings.cc b/src/include/ceph_strings.cc
index f14f29ce0e9..e86aae4fd50 100644
--- 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 27291a7440e..178c171c445 100644
--- 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 bc0bcc95ceb..5a750cbc0d1 100644
--- 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 12372d960b1..852228ed383 100644
--- 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 1c1b457002c..3451d520ff2 100644
--- 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 91f62551729..154ee410fde 100644
--- 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 6cb7cf5452a..9fe6427d3f8 100644
--- 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);