osd: introduce TMAP->OMAP conversion operation

Signed-off-by: Yan, Zheng <zheng.z.yan@intel.com>

diff --git a/src/include/rados.h b/src/include/rados.h
index f5fca376a42624738555d1c6530b95ea18049090..1994cd385ee3edc13357a1252ff5f6ac96654a1f 100644 (file)
--- a/src/include/rados.h
+++ b/src/include/rados.h
@@ -231,6 +231,9 @@ enum {
        CEPH_OSD_OP_CACHE_EVICT = CEPH_OSD_OP_MODE_WR | CEPH_OSD_OP_TYPE_DATA | 32,
        CEPH_OSD_OP_CACHE_TRY_FLUSH = CEPH_OSD_OP_MODE_WR | CEPH_OSD_OP_TYPE_DATA | 33,
 
+       /* convert tmap to omap */
+       CEPH_OSD_OP_TMAP2OMAP = CEPH_OSD_OP_MODE_RMW | CEPH_OSD_OP_TYPE_DATA | 34,
+
        /** multi **/
        CEPH_OSD_OP_CLONERANGE = CEPH_OSD_OP_MODE_WR | CEPH_OSD_OP_TYPE_MULTI | 1,
        CEPH_OSD_OP_ASSERT_SRC_VERSION = CEPH_OSD_OP_MODE_RD | CEPH_OSD_OP_TYPE_MULTI | 2,
@@ -386,6 +389,10 @@ enum {
        CEPH_OSD_COPY_FROM_FLAG_IGNORE_CACHE = 4, /* ignore osd cache logic */
 };
 
+enum {
+       CEPH_OSD_TMAP2OMAP_NULLOK = 1,
+};
+
 /*
  * an individual object operation.  each may be accompanied by some data
  * payload
@@ -442,6 +449,9 @@ struct ceph_osd_op {
                struct {
                        struct ceph_timespec stamp;
                } __attribute__ ((packed)) hit_set_get;
+               struct {
+                       __u8 flags;
+               } __attribute__ ((packed)) tmap2omap;
        };
        __le32 payload_len;
 } __attribute__ ((packed));

diff --git a/src/include/rados/librados.hpp b/src/include/rados/librados.hpp
index 106c6f87ec5a093257e198936f9f3980f1ff0bae..5e9eb312fd6aa67849a71e925659f23289b22c85 100644 (file)
--- a/src/include/rados/librados.hpp
+++ b/src/include/rados/librados.hpp
@@ -550,6 +550,7 @@ namespace librados
      */
     int tmap_put(const std::string& oid, bufferlist& bl);
     int tmap_get(const std::string& oid, bufferlist& bl);
+    int tmap_to_omap(const std::string& oid, bool nullok=false);
 
     int omap_get_vals(const std::string& oid,
                       const std::string& start_after,

diff --git a/src/librados/IoCtxImpl.cc b/src/librados/IoCtxImpl.cc
index f16c1caa8cf4988eb62f934990c8987f477bff69..929e105209f7e55a60c1568ccf06759ec359e16b 100644 (file)
--- a/src/librados/IoCtxImpl.cc
+++ b/src/librados/IoCtxImpl.cc
@@ -880,6 +880,13 @@ int librados::IoCtxImpl::tmap_get(const object_t& oid, bufferlist& bl)
   return operate_read(oid, &rd, NULL);
 }
 
+int librados::IoCtxImpl::tmap_to_omap(const object_t& oid, bool nullok)
+{
+  ::ObjectOperation wr;
+  prepare_assert_ops(&wr);
+  wr.tmap_to_omap(nullok);
+  return operate(oid, &wr, NULL);
+}
 
 int librados::IoCtxImpl::exec(const object_t& oid,
                              const char *cls, const char *method,

diff --git a/src/librados/IoCtxImpl.h b/src/librados/IoCtxImpl.h
index aa94aa83bba1981dca77e06590dea83b37ec9d3d..8693d6cba444e20c642368b77ea4b28f51b1c246 100644 (file)
--- a/src/librados/IoCtxImpl.h
+++ b/src/librados/IoCtxImpl.h
@@ -131,6 +131,7 @@ struct librados::IoCtxImpl {
   int tmap_update(const object_t& oid, bufferlist& cmdbl);
   int tmap_put(const object_t& oid, bufferlist& bl);
   int tmap_get(const object_t& oid, bufferlist& bl);
+  int tmap_to_omap(const object_t& oid, bool nullok=false);
 
   int exec(const object_t& oid, const char *cls, const char *method, bufferlist& inbl, bufferlist& outbl);
 

diff --git a/src/librados/librados.cc b/src/librados/librados.cc
index ad93689fa308b8c7b607c8cc2470138a26c543f1..ec3422c3f036fd1b5c7f77e6b57b1f190016aa29 100644 (file)
--- a/src/librados/librados.cc
+++ b/src/librados/librados.cc
@@ -856,6 +856,12 @@ int librados::IoCtx::tmap_get(const std::string& oid, bufferlist& bl)
   return io_ctx_impl->tmap_get(obj, bl);
 }
 
+int librados::IoCtx::tmap_to_omap(const std::string& oid, bool nullok)
+{
+  object_t obj(oid);
+  return io_ctx_impl->tmap_to_omap(obj, nullok);
+}
+
 int librados::IoCtx::omap_get_vals(const std::string& oid,
                                    const std::string& start_after,
                                    uint64_t max_return,
@@ -2623,6 +2629,13 @@ extern "C" int rados_tmap_get(rados_ioctx_t io, const char *o, char *buf, size_t
   return bl.length();
 }
 
+extern "C" int rados_tmap_to_omap(rados_ioctx_t io, const char *o, bool nullok)
+{
+  librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+  object_t oid(o);
+  return ctx->tmap_to_omap(oid, nullok);
+}
+
 extern "C" int rados_exec(rados_ioctx_t io, const char *o, const char *cls, const char *method,
                          const char *inbuf, size_t in_len, char *buf, size_t out_len)
 {

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 3c1715212ae97655e20cdef0028fad622d2a309a..025918643429144c1c4a1e63e698ea41bccdf9b9 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -2348,6 +2348,32 @@ int ReplicatedPG::do_xattr_cmp_str(int op, string& v1s, bufferlist& xattr)
 // ========================================================================
 // low level osd ops
 
+int ReplicatedPG::do_tmap2omap(OpContext *ctx, unsigned flags)
+{
+  dout(20) << " convert tmap to omap for " << ctx->new_obs.oi.soid << dendl;
+  bufferlist header, vals;
+  int r = _get_tmap(ctx, &header, &vals);
+  if (r < 0) {
+    if (r == -ENODATA && (flags & CEPH_OSD_TMAP2OMAP_NULLOK))
+      r = 0;
+    return r;
+  }
+
+  vector<OSDOp> ops(3);
+
+  ops[0].op.op = CEPH_OSD_OP_TRUNCATE;
+  ops[0].op.extent.offset = 0;
+  ops[0].op.extent.length = 0;
+
+  ops[1].op.op = CEPH_OSD_OP_OMAPSETHEADER;
+  ops[1].indata.claim(header);
+
+  ops[2].op.op = CEPH_OSD_OP_OMAPSETVALS;
+  ops[2].indata.claim(vals);
+
+  return do_osd_ops(ctx, ops);
+}
+
 int ReplicatedPG::do_tmapup_slow(OpContext *ctx, bufferlist::iterator& bp, OSDOp& osd_op,
                                    bufferlist& bl)
 {
@@ -3694,6 +3720,11 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
       result = do_tmapup(ctx, bp, osd_op);
       break;
 
+    case CEPH_OSD_OP_TMAP2OMAP:
+      ++ctx->num_write;
+      result = do_tmap2omap(ctx, op.tmap2omap.flags);
+      break;
+
       // OMAP Read ops
     case CEPH_OSD_OP_OMAPGETKEYS:
       ++ctx->num_read;
@@ -4014,10 +4045,12 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
   return result;
 }
 
-int ReplicatedPG::_get_tmap(OpContext *ctx,
-                           map<string, bufferlist> *out,
-                           bufferlist *header)
+int ReplicatedPG::_get_tmap(OpContext *ctx, bufferlist *header, bufferlist *vals)
 {
+  if (ctx->new_obs.oi.size == 0) {
+    dout(20) << "unable to get tmap for zero sized " << ctx->new_obs.oi.soid << dendl;
+    return -ENODATA;
+  }
   vector<OSDOp> nops(1);
   OSDOp &newop = nops[0];
   newop.op.op = CEPH_OSD_OP_TMAPGET;
@@ -4025,7 +4058,7 @@ int ReplicatedPG::_get_tmap(OpContext *ctx,
   try {
     bufferlist::iterator i = newop.outdata.begin();
     ::decode(*header, i);
-    ::decode(*out, i);
+    (*vals).substr_of(newop.outdata, i.get_off(), i.get_remaining());
   } catch (...) {
     dout(20) << "unsuccessful at decoding tmap for " << ctx->new_obs.oi.soid
             << dendl;

diff --git a/src/osd/ReplicatedPG.h b/src/osd/ReplicatedPG.h
index d42031b3af00ab8bdd7ecd837c2e5d043bb8cb1b..7679983542af72f9ec854f93fb48f5e1db304631 100644 (file)
--- a/src/osd/ReplicatedPG.h
+++ b/src/osd/ReplicatedPG.h
@@ -1016,6 +1016,8 @@ public:
   void snap_trimmer();
   int do_osd_ops(OpContext *ctx, vector<OSDOp>& ops);
 
+  int _get_tmap(OpContext *ctx, bufferlist *header, bufferlist *vals);
+  int do_tmap2omap(OpContext *ctx, unsigned flags);
   int do_tmapup(OpContext *ctx, bufferlist::iterator& bp, OSDOp& osd_op);
   int do_tmapup_slow(OpContext *ctx, bufferlist::iterator& bp, OSDOp& osd_op, bufferlist& bl);
 
@@ -1095,8 +1097,6 @@ private:
     boost::statechart::result react(const SnapTrim&);
   };
 
-  int _get_tmap(OpContext *ctx, map<string, bufferlist> *out,
-               bufferlist *header);
   int _delete_head(OpContext *ctx, bool no_whiteout);
   int _rollback_to(OpContext *ctx, ceph_osd_op& op);
 public:

diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 93058dc5024438ba030509f6593bc574a3b9d5ac..4b689424a3fa563b70704c6e855d9f7890ba8abd 100644 (file)
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -483,6 +483,12 @@ struct ObjectOperation {
   void tmap_get() {
     add_op(CEPH_OSD_OP_TMAPGET);
   }
+  void tmap_to_omap(bool nullok=false) {
+     OSDOp& osd_op = add_op(CEPH_OSD_OP_TMAP2OMAP);
+     osd_op.op.op = CEPH_OSD_OP_TMAP2OMAP;
+     if (nullok)
+       osd_op.op.tmap2omap.flags = CEPH_OSD_TMAP2OMAP_NULLOK;
+  }
 
   // objectmap
   void omap_get_keys(const string &start_after,

diff --git a/src/test/librados/misc.cc b/src/test/librados/misc.cc
index 9fe42dbb2d508cb65de5f99bce63ed82d08b65a8..bbe67ea0b5aa542ff732ece6431596592984a6f1 100644 (file)
--- a/src/test/librados/misc.cc
+++ b/src/test/librados/misc.cc
@@ -265,6 +265,71 @@ TEST(LibRadosMisc, TmapUpdateMisorderedPutPP) {
   ASSERT_EQ(0, destroy_one_pool_pp(pool_name, cluster));
 }
 
+TEST(LibRadosMisc, Tmap2OmapPP) {
+  Rados cluster;
+  std::string pool_name = get_temp_pool_name();
+  ASSERT_EQ("", create_one_pool_pp(pool_name, cluster));
+  IoCtx ioctx;
+  cluster.ioctx_create(pool_name.c_str(), ioctx);
+
+  // create tmap
+  bufferlist hdr;
+  hdr.append("header");
+  map<string, bufferlist> omap;
+  omap["1"].append("a");
+  omap["2"].append("b");
+  omap["3"].append("c");
+  {
+    bufferlist bl;
+    ::encode(hdr, bl);
+    ::encode(omap, bl);
+    ASSERT_EQ(0, ioctx.tmap_put("foo", bl));
+  }
+
+  // convert tmap to omap
+  ASSERT_EQ(0, ioctx.tmap_to_omap("foo", false));
+
+  // if tmap was truncated ?
+  {
+    uint64_t size;
+    time_t mtime;
+    ASSERT_EQ(0, ioctx.stat("foo", &size, &mtime));
+    ASSERT_EQ(0U, size);
+  }
+
+  // if 'nullok' works
+  ASSERT_EQ(0, ioctx.tmap_to_omap("foo", true));
+  ASSERT_LE(ioctx.tmap_to_omap("foo", false), 0);
+
+  {
+    // read omap
+    bufferlist got;
+    map<string, bufferlist> m;
+    ObjectReadOperation o;
+    o.omap_get_header(&got, NULL);
+    o.omap_get_vals("", 1024, &m, NULL);
+    ASSERT_EQ(0, ioctx.operate("foo", &o, NULL));
+
+    // compare header
+    ASSERT_TRUE(hdr.contents_equal(got));
+
+    // compare values
+    ASSERT_EQ(omap.size(), m.size());
+    bool same = true;
+    for (map<string, bufferlist>::iterator p = omap.begin(); p != omap.end(); ++p) {
+      map<string, bufferlist>::iterator q = m.find(p->first);
+      if (q == m.end() || !p->second.contents_equal(q->second)) {
+       same = false;
+       break;
+      }
+    }
+    ASSERT_TRUE(same);
+  }
+
+  ioctx.close();
+  ASSERT_EQ(0, destroy_one_pool_pp(pool_name, cluster));
+}
+
 TEST(LibRadosMisc, Exec) {
   char buf[128];
   rados_t cluster;