Merge branch 'wip-4207'

Feature: #4207: osd/librados: add ops to list snaps for an object

Signed-off-by: David Zafman <david.zafman@inktank.com>
Reported-by: Sam Just <sam.just@inktank.com>

12 files changed, 620 insertions, 24 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index 4a1d7633a38..af4a259efbc 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -1178,6 +1178,7 @@ rados_includedir = $(includedir)/rados
 rados_include_DATA = \
 	$(srcdir)/include/rados/librados.h \
 	$(srcdir)/include/rados/rados_types.h \
+	$(srcdir)/include/rados/rados_types.hpp \
 	$(srcdir)/include/rados/librados.hpp \
 	$(srcdir)/include/buffer.h \
 	$(srcdir)/include/page.h \
@@ -1642,6 +1643,7 @@ noinst_HEADERS = \
         include/xlist.h\
 	include/rados/librados.h\
 	include/rados/rados_types.h\
+	include/rados/rados_types.hpp\
 	include/rados/librados.hpp\
 	include/rados/librgw.h\
 	include/rados/page.h\
diff --git a/src/include/rados.h b/src/include/rados.h
index 093a04baf86..f4f120a8f15 100644
--- a/src/include/rados.h
+++ b/src/include/rados.h
@@ -179,6 +179,8 @@ enum {
 
 	CEPH_OSD_OP_LIST_WATCHERS = CEPH_OSD_OP_MODE_RD | CEPH_OSD_OP_TYPE_DATA | 9,
 
+	CEPH_OSD_OP_LIST_SNAPS = CEPH_OSD_OP_MODE_RD | CEPH_OSD_OP_TYPE_DATA | 10,
+
 	/* write */
 	CEPH_OSD_OP_WRITE     = CEPH_OSD_OP_MODE_WR | CEPH_OSD_OP_TYPE_DATA | 1,
 	CEPH_OSD_OP_WRITEFULL = CEPH_OSD_OP_MODE_WR | CEPH_OSD_OP_TYPE_DATA | 2,
diff --git a/src/include/rados/librados.hpp b/src/include/rados/librados.hpp
index ccb19438bcc..1463a34996d 100644
--- a/src/include/rados/librados.hpp
+++ b/src/include/rados/librados.hpp
@@ -12,6 +12,7 @@
 #include "buffer.h"
 
 #include "librados.h"
+#include "include/rados/rados_types.hpp"
 
 namespace librados
 {
@@ -26,7 +27,6 @@ namespace librados
   class RadosClient;
 
   typedef void *list_ctx_t;
-  typedef uint64_t snap_t;
   typedef uint64_t auid_t;
   typedef void *config_t;
 
@@ -52,11 +52,6 @@ namespace librados
   typedef void *completion_t;
   typedef void (*callback_t)(completion_t cb, void *arg);
 
-  struct SnapContext {
-    snap_t seq;
-    std::vector<snap_t> snaps;
-  };
-
   class ObjectIterator : public std::iterator <std::forward_iterator_tag, std::string> {
   public:
     static const ObjectIterator __EndObjectIterator;
@@ -333,6 +328,7 @@ namespace librados
      * @param prval [out] place error code in prval upon completion
      */
     void list_watchers(std::list<obj_watch_t> *out_watchers, int *prval);
+    void list_snaps(snap_set_t *out_snaps, int *prval);
 
   };
 
@@ -506,6 +502,7 @@ namespace librados
     int unwatch(const std::string& o, uint64_t handle);
     int notify(const std::string& o, uint64_t ver, bufferlist& bl);
     int list_watchers(const std::string& o, std::list<obj_watch_t> *out_watchers);
+    int list_snaps(const std::string& o, snap_set_t *out_snaps);
     void set_notify_timeout(uint32_t timeout);
 
     // assert version for next sync operations
diff --git a/src/include/rados/rados_types.hpp b/src/include/rados/rados_types.hpp
new file mode 100644
index 00000000000..eb28d4f8b6e
--- /dev/null
+++ b/src/include/rados/rados_types.hpp
@@ -0,0 +1,25 @@
+#ifndef CEPH_RADOS_TYPES_HPP
+#define CEPH_RADOS_TYPES_HPP
+
+#include <utility>
+#include <vector>
+#include "include/inttypes.h"
+
+namespace librados {
+
+typedef uint64_t snap_t;
+
+struct clone_info_t {
+  static const snap_t HEAD = ((snap_t)-1);
+  snap_t cloneid;
+  std::vector<snap_t> snaps;          // ascending
+  std::vector< std::pair<uint64_t,uint64_t> > overlap;
+  uint64_t size;
+};
+
+struct snap_set_t {
+  std::vector<clone_info_t> clones;   // ascending
+};
+
+}
+#endif
diff --git a/src/librados/librados.cc b/src/librados/librados.cc
index c03a20f8d12..59ff0d1c3e8 100644
--- a/src/librados/librados.cc
+++ b/src/librados/librados.cc
@@ -219,6 +219,14 @@ void librados::ObjectReadOperation::list_watchers(
   o->list_watchers(out_watchers, prval);
 }
 
+void librados::ObjectReadOperation::list_snaps(
+  snap_set_t *out_snaps,
+  int *prval)
+{
+  ::ObjectOperation *o = (::ObjectOperation *)impl;
+  o->list_snaps(out_snaps, prval);
+}
+
 int librados::IoCtx::omap_get_vals(const std::string& oid,
                                    const std::string& start_after,
                                    const std::string& filter_prefix,
@@ -1040,6 +1048,20 @@ int librados::IoCtx::list_watchers(const std::string& oid,
   return r;
 }
 
+int librados::IoCtx::list_snaps(const std::string& oid,
+                                   snap_set_t *out_snaps)
+{
+  ObjectReadOperation op;
+  int r;
+  op.list_snaps(out_snaps, &r);
+  bufferlist bl;
+  int ret = operate(oid, &op, &bl);
+  if (ret < 0)
+    return ret;
+
+  return r;
+}
+
 void librados::IoCtx::set_notify_timeout(uint32_t timeout)
 {
   io_ctx_impl->set_notify_timeout(timeout);
diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index b1f60d4f0dc..ea71f2b81e8 100644
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -2252,6 +2252,94 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
         break;
       }
 
+    case CEPH_OSD_OP_LIST_SNAPS:
+      {
+        obj_list_snap_response_t resp;
+
+        if (!ssc) {
+            ssc = ctx->obc->ssc = get_snapset_context(soid.oid,
+                soid.get_key(), soid.hash, false);
+        }
+
+        assert(ssc);
+
+        vector<snapid_t>::reverse_iterator snap_iter =
+            ssc->snapset.snaps.rbegin();
+
+        int clonecount = ssc->snapset.clones.size();
+        if (ssc->snapset.head_exists)
+          clonecount++;
+        resp.clones.reserve(clonecount);
+        for (vector<snapid_t>::const_iterator clone_iter = ssc->snapset.clones.begin();
+               clone_iter != ssc->snapset.clones.end(); ++clone_iter) {
+          clone_info ci;
+
+          dout(20) << "List clones id=" << *clone_iter << dendl;
+
+          ci.cloneid = *clone_iter;
+
+          for (;snap_iter != ssc->snapset.snaps.rend()
+               && (*snap_iter <= ci.cloneid); snap_iter++) {
+
+            dout(20) << "List snaps id=" << *snap_iter << dendl;
+
+            assert(*snap_iter != CEPH_NOSNAP);
+            assert(*snap_iter != CEPH_SNAPDIR);
+
+            ci.snaps.push_back(*snap_iter);
+          }
+
+          map<snapid_t, interval_set<uint64_t> >::const_iterator coi;
+          coi = ssc->snapset.clone_overlap.find(ci.cloneid);
+          if (coi == ssc->snapset.clone_overlap.end()) {
+            osd->clog.error() << "osd." << osd->whoami << ": inconsistent clone_overlap found for oid "
+                    << soid << " clone " << *clone_iter;
+            result = EINVAL;
+            break;
+          }
+          const interval_set<uint64_t> &o = coi->second;
+          ci.overlap.reserve(o.num_intervals());
+          for (interval_set<uint64_t>::const_iterator r = o.begin();
+               r != o.end(); ++r) {
+            ci.overlap.push_back(pair<uint64_t,uint64_t>(r.get_start(), r.get_len()));
+          }
+
+          map<snapid_t, uint64_t>::const_iterator si;
+          si = ssc->snapset.clone_size.find(ci.cloneid);
+          if (si == ssc->snapset.clone_size.end()) {
+            osd->clog.error() << "osd." << osd->whoami << ": inconsistent clone_size found for oid "
+                    << soid << " clone " << *clone_iter;
+            result = EINVAL;
+            break;
+          }
+          ci.size = si->second;
+
+          resp.clones.push_back(ci);
+        }
+        if (ssc->snapset.head_exists) {
+          clone_info ci;
+
+          assert(obs.exists);
+
+          ci.cloneid = clone_info::HEAD;
+
+          //Put remaining snapshots into head clone
+          for (;snap_iter != ssc->snapset.snaps.rend(); snap_iter++)
+            ci.snaps.push_back(*snap_iter);
+
+          //Size for HEAD is oi.size
+          ci.size = oi.size;
+
+          resp.clones.push_back(ci);
+        }
+
+        resp.encode(osd_op.outdata);
+        result = 0;
+
+        ctx->delta_stats.num_rd++;
+        break;
+      }
+
     case CEPH_OSD_OP_ASSERT_SRC_VERSION:
       {
 	uint64_t ver = op.watch.ver;
diff --git a/src/osd/osd_types.cc b/src/osd/osd_types.cc
index 219c1bfdec8..9c890397e8d 100644
--- a/src/osd/osd_types.cc
+++ b/src/osd/osd_types.cc
@@ -20,6 +20,8 @@ extern "C" {
 #include "PG.h"
 #include "OSDMap.h"
 
+const snapid_t clone_info::HEAD((uint64_t)-1);
+
 // -- osd_reqid_t --
 void osd_reqid_t::encode(bufferlist &bl) const
 {
diff --git a/src/osd/osd_types.h b/src/osd/osd_types.h
index a0755825e1e..3f464852398 100644
--- a/src/osd/osd_types.h
+++ b/src/osd/osd_types.h
@@ -1671,7 +1671,7 @@ WRITE_CLASS_ENCODER(interval_set<uint64_t>)
 struct SnapSet {
   snapid_t seq;
   bool head_exists;
-  vector<snapid_t> snaps;    // ascending
+  vector<snapid_t> snaps;    // descending
   vector<snapid_t> clones;   // ascending
   map<snapid_t, interval_set<uint64_t> > clone_overlap;  // overlap w/ next newest
   map<snapid_t, uint64_t> clone_size;
@@ -2093,4 +2093,113 @@ struct obj_list_watch_response_t {
 
 WRITE_CLASS_ENCODER(obj_list_watch_response_t)
 
+struct clone_info {
+  static const snapid_t HEAD;
+
+  snapid_t cloneid;
+  vector<snapid_t> snaps;  // ascending
+  vector< pair<uint64_t,uint64_t> > overlap;
+  uint64_t size;
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(cloneid, bl);
+    ::encode(snaps, bl);
+    ::encode(overlap, bl);
+    ::encode(size, bl);
+    ENCODE_FINISH(bl);
+  }
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(cloneid, bl);
+    ::decode(snaps, bl);
+    ::decode(overlap, bl);
+    ::decode(size, bl);
+    DECODE_FINISH(bl);
+  }
+  void dump(Formatter *f) const {
+    if (cloneid == HEAD)
+      f->dump_string("cloneid", "HEAD");
+    else
+      f->dump_unsigned("cloneid", cloneid.val);
+    f->open_array_section("snapshots");
+    for (vector<snapid_t>::const_iterator p = snaps.begin(); p != snaps.end(); ++p) {
+      f->open_object_section("snap");
+      f->dump_unsigned("id", p->val);
+      f->close_section();
+    }
+    f->close_section();
+    f->open_array_section("overlaps");
+    for (vector< pair<uint64_t,uint64_t> >::const_iterator q = overlap.begin();
+          q != overlap.end(); ++q) {
+      f->open_object_section("overlap");
+      f->dump_unsigned("offset", q->first);
+      f->dump_unsigned("length", q->second);
+      f->close_section();
+    }
+    f->close_section();
+    f->dump_unsigned("size", size);
+  }
+  static void generate_test_instances(list<clone_info*>& o) {
+    o.push_back(new clone_info);
+    o.push_back(new clone_info);
+    o.back()->cloneid = 1;
+    o.back()->snaps.push_back(1);
+    o.back()->overlap.push_back(pair<uint64_t,uint64_t>(0,4096));
+    o.back()->overlap.push_back(pair<uint64_t,uint64_t>(8192,4096));
+    o.back()->size = 16384;
+    o.push_back(new clone_info);
+    o.back()->cloneid = HEAD;
+    o.back()->size = 32768;
+  }
+};
+
+WRITE_CLASS_ENCODER(clone_info)
+
+/**
+ * obj list snaps response format
+ *
+ */
+struct obj_list_snap_response_t {
+  vector<clone_info> clones;   // ascending
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(clones, bl);
+    ENCODE_FINISH(bl);
+  }
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(clones, bl);
+    DECODE_FINISH(bl);
+  }
+  void dump(Formatter *f) const {
+    f->open_array_section("clones");
+    for (vector<clone_info>::const_iterator p = clones.begin(); p != clones.end(); ++p) {
+      f->open_object_section("clone");
+      p->dump(f);
+      f->close_section();
+    }
+    f->close_section();
+  }
+  static void generate_test_instances(list<obj_list_snap_response_t*>& o) {
+    o.push_back(new obj_list_snap_response_t);
+    o.push_back(new obj_list_snap_response_t);
+    clone_info cl;
+    cl.cloneid = 1;
+    cl.snaps.push_back(1);
+    cl.overlap.push_back(pair<uint64_t,uint64_t>(0,4096));
+    cl.overlap.push_back(pair<uint64_t,uint64_t>(8192,4096));
+    cl.size = 16384;
+    o.back()->clones.push_back(cl);
+    cl.cloneid = clone_info::HEAD;
+    cl.snaps.clear();
+    cl.overlap.clear();
+    cl.size = 32768;
+    o.back()->clones.push_back(cl);
+  }
+};
+
+WRITE_CLASS_ENCODER(obj_list_snap_response_t)
+
 #endif
diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 6cc3cb88426..f9583400c6e 100644
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -25,6 +25,7 @@
 #include "common/admin_socket.h"
 #include "common/Timer.h"
 #include "include/rados/rados_types.h"
+#include "include/rados/rados_types.hpp"
 
 #include <list>
 #include <map>
@@ -344,6 +345,43 @@ struct ObjectOperation {
       }	
     }
   };
+  struct C_ObjectOperation_decodesnaps : public Context {
+    bufferlist bl;
+    librados::snap_set_t *psnaps;
+    int *prval;
+    C_ObjectOperation_decodesnaps(librados::snap_set_t *ps, int *pr)
+      : psnaps(ps), prval(pr) {}
+    void finish(int r) {
+      if (r >= 0) {
+	bufferlist::iterator p = bl.begin();
+	try {
+          obj_list_snap_response_t resp;
+	  ::decode(resp, p);
+	  if (psnaps) {
+
+            psnaps->clones.clear();
+            vector<clone_info>::iterator ci;
+            for (ci = resp.clones.begin(); ci != resp.clones.end(); ci++) {
+              librados::clone_info_t clone;
+
+              clone.cloneid = ci->cloneid;
+              clone.snaps.reserve(ci->snaps.size());
+              clone.snaps.insert(clone.snaps.end(), ci->snaps.begin(), ci->snaps.end());
+              clone.overlap = ci->overlap;
+              clone.size = ci->size;
+
+              psnaps->clones.push_back(clone);
+            }
+          }
+          *prval = 0;
+	}
+	catch (buffer::error& e) {
+	  if (prval)
+	    *prval = -EIO;
+	}
+      }
+    }
+  };
   void getxattrs(std::map<std::string,bufferlist> *pattrs, int *prval) {
     add_op(CEPH_OSD_OP_GETXATTRS);
     if (pattrs || prval) {
@@ -539,6 +577,18 @@ struct ObjectOperation {
     }
   }
 
+  void list_snaps(librados::snap_set_t *out, int *prval) {
+    (void)add_op(CEPH_OSD_OP_LIST_SNAPS);
+    if (prval || out) {
+      unsigned p = ops.size() - 1;
+      C_ObjectOperation_decodesnaps *h =
+	new C_ObjectOperation_decodesnaps(out, prval);
+      out_handler[p] = h;
+      out_bl[p] = &h->bl;
+      out_rval[p] = prval;
+    }
+  }
+
   void assert_version(uint64_t ver) {
     bufferlist bl;
     add_watch(CEPH_OSD_OP_ASSERT_VER, 0, ver, 0, bl);
diff --git a/src/rados.cc b/src/rados.cc
index bbd3badc082..e4679bea887 100644
--- a/src/rados.cc
+++ b/src/rados.cc
@@ -15,7 +15,7 @@
 #include "include/types.h"
 
 #include "include/rados/librados.hpp"
-#include "include/rados/rados_types.h"
+#include "include/rados/rados_types.hpp"
 #include "rados_sync.h"
 using namespace librados;
 
@@ -83,6 +83,7 @@ void usage(ostream& out)
 "   rmsnap <snap-name>               remove snap <snap-name>\n"
 "   rollback <obj-name> <snap-name>  roll back object to snap <snap-name>\n"
 "\n"
+"   listsnaps <obj-name>             list the snapshots of this object\n"
 "   bench <seconds> write|seq|rand [-t concurrent_operations] [--no-cleanup]\n"
 "                                    default is 16 concurrent IOs and 4 MB ops\n"
 "                                    default is to clean up after write benchmark\n"
@@ -2022,6 +2023,130 @@ static int rados_tool_common(const std::map < std::string, std::string > &opts,
     for (i = lw.begin(); i != lw.end(); i++) {
       cout << "watcher=client." << i->watcher_id << " cookie=" << i->cookie << std::endl;
     }
+  } else if (strcmp(nargs[0], "listsnaps") == 0) {
+    if (!pool_name || nargs.size() < 2)
+      usage_exit();
+
+    string oid(nargs[1]);
+    snap_set_t ls;
+
+    ret = io_ctx.list_snaps(oid, &ls);
+    if (ret < 0) {
+      cerr << "error listing snap shots " << pool_name << "/" << oid << ": " << strerror_r(-ret, buf, sizeof(buf)) << std::endl;
+      return 1;
+    }
+    else
+      ret = 0;
+
+    map<snap_t,string> snamemap;
+    if (formatter || pretty_format) {
+      vector<snap_t> snaps;
+      io_ctx.snap_list(&snaps);
+      for (vector<snap_t>::iterator i = snaps.begin();
+          i != snaps.end(); i++) {
+        string s;
+        if (io_ctx.snap_get_name(*i, &s) < 0)
+          continue;
+        snamemap.insert(pair<snap_t,string>(*i, s));
+      }
+    }
+
+    if (formatter) {
+      formatter->open_object_section("object");
+      formatter->dump_string("name", oid);
+      formatter->open_array_section("clones");
+    } else {
+      cout << oid << ":" << std::endl;
+      cout << "cloneid	snaps	size	overlap" << std::endl;
+    }
+
+    for (std::vector<clone_info_t>::iterator ci = ls.clones.begin();
+          ci != ls.clones.end(); ci++) {
+
+      if (formatter) formatter->open_object_section("clone");
+
+      if (ci->cloneid == clone_info_t::HEAD) {
+        if (formatter)
+          formatter->dump_string("id", "head");
+        else
+          cout << "head";
+      } else {
+        if (formatter)
+          formatter->dump_unsigned("id", ci->cloneid);
+        else
+          cout << ci->cloneid;
+      }
+
+      if (formatter)
+        formatter->open_array_section("snapshots");
+      else
+        cout << "\t";
+
+      if (!formatter && ci->snaps.empty()) {
+        cout << "-";
+      }
+      for (std::vector<snap_t>::const_iterator snapindex = ci->snaps.begin();
+          snapindex != ci->snaps.end(); ++snapindex) {
+
+        map<snap_t,string>::iterator si;
+
+        if (formatter || pretty_format) si = snamemap.find(*snapindex);
+
+        if (formatter) {
+          formatter->open_object_section("snapshot");
+          formatter->dump_unsigned("id", *snapindex);
+          if (si != snamemap.end())
+            formatter->dump_string("name", si->second);
+          formatter->close_section(); //snapshot
+        } else {
+          if (snapindex != ci->snaps.begin()) cout << ",";
+          if (!pretty_format || (si == snamemap.end()))
+            cout << *snapindex;
+          else
+            cout << si->second << "(" << *snapindex << ")";
+        }
+      }
+
+      if (formatter) {
+        formatter->close_section();	//Snapshots
+        formatter->dump_unsigned("size", ci->size);
+      } else {
+        cout << "\t" << ci->size;
+      }
+
+      if (ci->cloneid != clone_info_t::HEAD) {
+        if (formatter)
+          formatter->open_array_section("overlaps");
+        else
+          cout << "\t[";
+
+        for (std::vector< std::pair<uint64_t,uint64_t> >::iterator ovi = ci->overlap.begin();
+            ovi != ci->overlap.end(); ovi++) {
+          if (formatter) {
+            formatter->open_object_section("section");
+            formatter->dump_unsigned("start", ovi->first);
+            formatter->dump_unsigned("length", ovi->second);
+            formatter->close_section(); //section
+          } else {
+            if (ovi != ci->overlap.begin()) cout << ",";
+            cout << ovi->first << "~" << ovi->second;
+          }
+        }
+        if (formatter)
+          formatter->close_section(); //overlaps
+        else
+          cout << "]" << std::endl;
+      }
+      if (formatter) formatter->close_section(); //clone
+    }
+    if (formatter) {
+      formatter->close_section(); //clones
+      formatter->close_section(); //object
+      formatter->flush(cout);
+    } else {
+      cout << std::endl;
+    }
+
   } else {
     cerr << "unrecognized command " << nargs[0] << std::endl;
     usage_exit();
diff --git a/src/test/encoding/types.h b/src/test/encoding/types.h
index b69bb9fd07c..a554d7e114e 100644
--- a/src/test/encoding/types.h
+++ b/src/test/encoding/types.h
@@ -63,6 +63,8 @@ TYPE(ObjectRecoveryProgress)
 TYPE(ScrubMap::object)
 TYPE(ScrubMap)
 TYPE(osd_peer_stat_t)
+TYPE(clone_info)
+TYPE(obj_list_snap_response_t)
 
 #include "os/ObjectStore.h"
 TYPE(ObjectStore::Transaction)
diff --git a/src/test/librados/snapshots.cc b/src/test/librados/snapshots.cc
index 439fe3fd6a8..1aec45456ba 100644
--- a/src/test/librados/snapshots.cc
+++ b/src/test/librados/snapshots.cc
@@ -1,4 +1,4 @@
-#include "include/rados/librados.h"
+#include "include/rados/librados.hpp"
 #include "test/librados/test.h"
 
 #include <algorithm>
@@ -9,8 +9,10 @@
 using namespace librados;
 using std::string;
 
+const int bufsize = 128;
+
 TEST(LibRadosSnapshots, SnapList) {
-  char buf[128];
+  char buf[bufsize];
   rados_t cluster;
   rados_ioctx_t ioctx;
   std::string pool_name = get_temp_pool_name();
@@ -35,7 +37,7 @@ TEST(LibRadosSnapshots, SnapListPP) {
   ASSERT_EQ("", create_one_pool_pp(pool_name, cluster));
   IoCtx ioctx;
   cluster.ioctx_create(pool_name.c_str(), ioctx);
-  char buf[128];
+  char buf[bufsize];
   memset(buf, 0xcc, sizeof(buf));
   bufferlist bl1;
   bl1.append(buf, sizeof(buf));
@@ -52,7 +54,7 @@ TEST(LibRadosSnapshots, SnapListPP) {
 }
 
 TEST(LibRadosSnapshots, SnapRemove) {
-  char buf[128];
+  char buf[bufsize];
   rados_t cluster;
   rados_ioctx_t ioctx;
   std::string pool_name = get_temp_pool_name();
@@ -71,7 +73,7 @@ TEST(LibRadosSnapshots, SnapRemove) {
 }
 
 TEST(LibRadosSnapshots, SnapRemovePP) {
-  char buf[128];
+  char buf[bufsize];
   Rados cluster;
   IoCtx ioctx;
   std::string pool_name = get_temp_pool_name();
@@ -91,7 +93,7 @@ TEST(LibRadosSnapshots, SnapRemovePP) {
 }
 
 TEST(LibRadosSnapshots, Rollback) {
-  char buf[128];
+  char buf[bufsize];
   rados_t cluster;
   rados_ioctx_t ioctx;
   std::string pool_name = get_temp_pool_name();
@@ -112,7 +114,7 @@ TEST(LibRadosSnapshots, Rollback) {
 }
 
 TEST(LibRadosSnapshots, RollbackPP) {
-  char buf[128];
+  char buf[bufsize];
   Rados cluster;
   IoCtx ioctx;
   std::string pool_name = get_temp_pool_name();
@@ -137,7 +139,7 @@ TEST(LibRadosSnapshots, RollbackPP) {
 }
 
 TEST(LibRadosSnapshots, SnapGetName) {
-  char buf[128];
+  char buf[bufsize];
   rados_t cluster;
   rados_ioctx_t ioctx;
   std::string pool_name = get_temp_pool_name();
@@ -160,7 +162,7 @@ TEST(LibRadosSnapshots, SnapGetName) {
 }
 
 TEST(LibRadosSnapshots, SnapGetNamePP) {
-  char buf[128];
+  char buf[bufsize];
   Rados cluster;
   IoCtx ioctx;
   std::string pool_name = get_temp_pool_name();
@@ -197,7 +199,7 @@ TEST(LibRadosSnapshots, SelfManagedSnapTest) {
   ASSERT_EQ(0, rados_ioctx_selfmanaged_snap_set_write_ctx(ioctx, my_snaps[0],
 					&my_snaps[0], my_snaps.size()));
   ::std::reverse(my_snaps.begin(), my_snaps.end());
-  char buf[128];
+  char buf[bufsize];
   memset(buf, 0xcc, sizeof(buf));
   ASSERT_EQ((int)sizeof(buf), rados_write(ioctx, "foo", buf, sizeof(buf), 0));
 
@@ -237,7 +239,7 @@ TEST(LibRadosSnapshots, SelfManagedRollbackTest) {
   ASSERT_EQ(0, rados_ioctx_selfmanaged_snap_set_write_ctx(ioctx, my_snaps[0],
 					&my_snaps[0], my_snaps.size()));
   ::std::reverse(my_snaps.begin(), my_snaps.end());
-  char buf[128];
+  char buf[bufsize];
   memset(buf, 0xcc, sizeof(buf));
   ASSERT_EQ((int)sizeof(buf), rados_write(ioctx, "foo", buf, sizeof(buf), 0));
 
@@ -276,7 +278,7 @@ TEST(LibRadosSnapshots, SelfManagedSnapTestPP) {
   ::std::reverse(my_snaps.begin(), my_snaps.end());
   ASSERT_EQ(0, ioctx.selfmanaged_snap_set_write_ctx(my_snaps[0], my_snaps));
   ::std::reverse(my_snaps.begin(), my_snaps.end());
-  char buf[128];
+  char buf[bufsize];
   memset(buf, 0xcc, sizeof(buf));
   bufferlist bl1;
   bl1.append(buf, sizeof(buf));
@@ -319,11 +321,24 @@ TEST(LibRadosSnapshots, SelfManagedSnapRollbackPP) {
   ::std::reverse(my_snaps.begin(), my_snaps.end());
   ASSERT_EQ(0, ioctx.selfmanaged_snap_set_write_ctx(my_snaps[0], my_snaps));
   ::std::reverse(my_snaps.begin(), my_snaps.end());
-  char buf[128];
+  char buf[bufsize];
   memset(buf, 0xcc, sizeof(buf));
   bufferlist bl1;
   bl1.append(buf, sizeof(buf));
+  //Write 3 consecutive buffers
   ASSERT_EQ((int)sizeof(buf), ioctx.write("foo", bl1, sizeof(buf), 0));
+  ASSERT_EQ((int)sizeof(buf), ioctx.write("foo", bl1, sizeof(buf), bufsize));
+  ASSERT_EQ((int)sizeof(buf), ioctx.write("foo", bl1, sizeof(buf), bufsize*2));
+
+  snap_set_t ss;
+
+  snap_t head = clone_info_t::HEAD;
+  ASSERT_EQ(0, ioctx.list_snaps("foo", &ss));
+  ASSERT_EQ(1u, ss.clones.size());
+  ASSERT_EQ(head, ss.clones[0].cloneid);
+  ASSERT_EQ(1u, ss.clones[0].snaps.size());	//this could go away in the future
+  ASSERT_EQ(0u, ss.clones[0].overlap.size());
+  ASSERT_EQ(384u, ss.clones[0].size);
 
   my_snaps.push_back(-2);
   ASSERT_EQ(0, ioctx.selfmanaged_snap_create(&my_snaps.back()));
@@ -334,13 +349,170 @@ TEST(LibRadosSnapshots, SelfManagedSnapRollbackPP) {
   memset(buf2, 0xdd, sizeof(buf2));
   bufferlist bl2;
   bl2.append(buf2, sizeof(buf2));
-  ASSERT_EQ((int)sizeof(buf2), ioctx.write("foo", bl2, sizeof(buf2), 0));
+  //Change the middle buffer
+  ASSERT_EQ((int)sizeof(buf2), ioctx.write("foo", bl2, sizeof(buf2), bufsize));
+  //Add another after
+  ASSERT_EQ((int)sizeof(buf2), ioctx.write("foo", bl2, sizeof(buf2), bufsize*3));
+
+  ASSERT_EQ(0, ioctx.list_snaps("foo", &ss));
+  ASSERT_EQ(2u, ss.clones.size());
+  ASSERT_EQ(3u, ss.clones[0].cloneid);
+  ASSERT_EQ(2u, ss.clones[0].snaps.size());
+  ASSERT_EQ(2u, ss.clones[0].snaps[0]);	//this could go away in the future
+  ASSERT_EQ(3u, ss.clones[0].snaps[1]);
+  ASSERT_EQ(2u, ss.clones[0].overlap.size());
+  ASSERT_EQ(0u, ss.clones[0].overlap[0].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[0].second);
+  ASSERT_EQ(256u, ss.clones[0].overlap[1].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[1].second);
+  ASSERT_EQ(384u, ss.clones[0].size);
+  ASSERT_EQ(head, ss.clones[1].cloneid);
+  ASSERT_EQ(0u, ss.clones[1].snaps.size());
+  ASSERT_EQ(0u, ss.clones[1].overlap.size());
+  ASSERT_EQ(512u, ss.clones[1].size);
+
+  ioctx.selfmanaged_snap_rollback("foo", my_snaps[1]);
 
-  string foo_str("foo");
-  ioctx.selfmanaged_snap_rollback(foo_str, my_snaps[1]);
   bufferlist bl3;
   ASSERT_EQ((int)sizeof(buf), ioctx.read("foo", bl3, sizeof(buf), 0));
   ASSERT_EQ(0, memcmp(bl3.c_str(), buf, sizeof(buf)));
+  ASSERT_EQ((int)sizeof(buf), ioctx.read("foo", bl3, sizeof(buf), bufsize));
+  ASSERT_EQ(0, memcmp(bl3.c_str(), buf, sizeof(buf)));
+  ASSERT_EQ((int)sizeof(buf), ioctx.read("foo", bl3, sizeof(buf), bufsize*2));
+  ASSERT_EQ(0, memcmp(bl3.c_str(), buf, sizeof(buf)));
+  ASSERT_EQ((int)0, ioctx.read("foo", bl3, sizeof(buf), bufsize*3));
+
+  ASSERT_EQ(0, ioctx.selfmanaged_snap_remove(my_snaps.back()));
+  my_snaps.pop_back();
+  ASSERT_EQ(0, ioctx.selfmanaged_snap_remove(my_snaps.back()));
+  my_snaps.pop_back();
+  ioctx.close();
+  ASSERT_EQ(0, destroy_one_pool_pp(pool_name, cluster));
+}
+
+TEST(LibRadosSnapshots, SelfManagedSnapOverlapPP) {
+  std::vector<uint64_t> my_snaps;
+  Rados cluster;
+  IoCtx ioctx;
+  std::string pool_name = get_temp_pool_name();
+  ASSERT_EQ("", create_one_pool_pp(pool_name, cluster));
+  ASSERT_EQ(0, cluster.ioctx_create(pool_name.c_str(), ioctx));
+
+  my_snaps.push_back(-2);
+  ASSERT_EQ(0, ioctx.selfmanaged_snap_create(&my_snaps.back()));
+  ::std::reverse(my_snaps.begin(), my_snaps.end());
+  ASSERT_EQ(0, ioctx.selfmanaged_snap_set_write_ctx(my_snaps[0], my_snaps));
+  ::std::reverse(my_snaps.begin(), my_snaps.end());
+  char buf[bufsize];
+  memset(buf, 0xcc, sizeof(buf));
+  bufferlist bl1;
+  bl1.append(buf, sizeof(buf));
+  ASSERT_EQ((int)sizeof(buf), ioctx.write("foo", bl1, sizeof(buf), 0));
+  ASSERT_EQ((int)sizeof(buf), ioctx.write("foo", bl1, sizeof(buf), bufsize*2));
+  ASSERT_EQ((int)sizeof(buf), ioctx.write("foo", bl1, sizeof(buf), bufsize*4));
+  ASSERT_EQ((int)sizeof(buf), ioctx.write("foo", bl1, sizeof(buf), bufsize*6));
+  ASSERT_EQ((int)sizeof(buf), ioctx.write("foo", bl1, sizeof(buf), bufsize*8));
+
+  snap_set_t ss;
+  snap_t head = clone_info_t::HEAD;
+  ASSERT_EQ(0, ioctx.list_snaps("foo", &ss));
+  ASSERT_EQ(1u, ss.clones.size());
+  ASSERT_EQ(head, ss.clones[0].cloneid);
+  ASSERT_EQ(1u, ss.clones[0].snaps.size());
+  ASSERT_EQ(2u, ss.clones[0].snaps[0]);	//this could go away in the future
+  ASSERT_EQ(0u, ss.clones[0].overlap.size());
+  ASSERT_EQ(1152u, ss.clones[0].size);
+
+  my_snaps.push_back(-2);
+  ASSERT_EQ(0, ioctx.selfmanaged_snap_create(&my_snaps.back()));
+  ::std::reverse(my_snaps.begin(), my_snaps.end());
+  ASSERT_EQ(0, ioctx.selfmanaged_snap_set_write_ctx(my_snaps[0], my_snaps));
+  ::std::reverse(my_snaps.begin(), my_snaps.end());
+  char buf2[sizeof(buf)];
+  memset(buf2, 0xdd, sizeof(buf2));
+  bufferlist bl2;
+  bl2.append(buf2, sizeof(buf2));
+  ASSERT_EQ((int)sizeof(buf2), ioctx.write("foo", bl2, sizeof(buf2), bufsize*1));
+  ASSERT_EQ((int)sizeof(buf2), ioctx.write("foo", bl2, sizeof(buf2), bufsize*3));
+  ASSERT_EQ((int)sizeof(buf2), ioctx.write("foo", bl2, sizeof(buf2), bufsize*5));
+  ASSERT_EQ((int)sizeof(buf2), ioctx.write("foo", bl2, sizeof(buf2), bufsize*7));
+  ASSERT_EQ((int)sizeof(buf2), ioctx.write("foo", bl2, sizeof(buf2), bufsize*9));
+
+  ASSERT_EQ(0, ioctx.list_snaps("foo", &ss));
+  ASSERT_EQ(2u, ss.clones.size());
+  ASSERT_EQ(3u, ss.clones[0].cloneid);
+  ASSERT_EQ(2u, ss.clones[0].snaps.size());
+  ASSERT_EQ(2u, ss.clones[0].snaps[0]);	//this could go away in the future
+  ASSERT_EQ(3u, ss.clones[0].snaps[1]);
+  ASSERT_EQ(5u, ss.clones[0].overlap.size());
+  ASSERT_EQ(0u, ss.clones[0].overlap[0].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[0].second);
+  ASSERT_EQ(256u, ss.clones[0].overlap[1].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[1].second);
+  ASSERT_EQ(512u, ss.clones[0].overlap[2].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[2].second);
+  ASSERT_EQ(768u, ss.clones[0].overlap[3].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[3].second);
+  ASSERT_EQ(1024u, ss.clones[0].overlap[4].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[4].second);
+  ASSERT_EQ(1152u, ss.clones[0].size);
+  ASSERT_EQ(head, ss.clones[1].cloneid);
+  ASSERT_EQ(0u, ss.clones[1].snaps.size());
+  ASSERT_EQ(0u, ss.clones[1].overlap.size());
+  ASSERT_EQ(1280u, ss.clones[1].size);
+
+  my_snaps.push_back(-2);
+  ASSERT_EQ(0, ioctx.selfmanaged_snap_create(&my_snaps.back()));
+  ::std::reverse(my_snaps.begin(), my_snaps.end());
+  ASSERT_EQ(0, ioctx.selfmanaged_snap_set_write_ctx(my_snaps[0], my_snaps));
+  ::std::reverse(my_snaps.begin(), my_snaps.end());
+
+  char buf3[sizeof(buf)];
+  memset(buf3, 0xee, sizeof(buf3));
+  bufferlist bl4;
+  bl4.append(buf3, sizeof(buf3));
+  ASSERT_EQ((int)sizeof(buf3), ioctx.write("foo", bl2, sizeof(buf3), bufsize*1));
+  ASSERT_EQ((int)sizeof(buf3), ioctx.write("foo", bl2, sizeof(buf3), bufsize*4));
+  ASSERT_EQ((int)sizeof(buf3), ioctx.write("foo", bl2, sizeof(buf3), bufsize*5));
+  ASSERT_EQ((int)sizeof(buf3), ioctx.write("foo", bl2, sizeof(buf3), bufsize*8));
+
+  ASSERT_EQ(0, ioctx.list_snaps("foo", &ss));
+  ASSERT_EQ(3u, ss.clones.size());
+  ASSERT_EQ(3u, ss.clones[0].cloneid);
+  ASSERT_EQ(2u, ss.clones[0].snaps.size());
+  ASSERT_EQ(2u, ss.clones[0].snaps[0]);	//this could go away in the future
+  ASSERT_EQ(3u, ss.clones[0].snaps[1]);
+  ASSERT_EQ(5u, ss.clones[0].overlap.size());
+  ASSERT_EQ(0u, ss.clones[0].overlap[0].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[0].second);
+  ASSERT_EQ(256u, ss.clones[0].overlap[1].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[1].second);
+  ASSERT_EQ(512u, ss.clones[0].overlap[2].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[2].second);
+  ASSERT_EQ(768u, ss.clones[0].overlap[3].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[3].second);
+  ASSERT_EQ(1024u, ss.clones[0].overlap[4].first);
+  ASSERT_EQ(128u, ss.clones[0].overlap[4].second);
+  ASSERT_EQ(1152u, ss.clones[0].size);
+
+  ASSERT_EQ(4u, ss.clones[1].cloneid);
+  ASSERT_EQ(1u, ss.clones[1].snaps.size());
+  ASSERT_EQ(4u, ss.clones[1].snaps[0]);
+  ASSERT_EQ(4u, ss.clones[1].overlap.size());
+  ASSERT_EQ(0u, ss.clones[1].overlap[0].first);
+  ASSERT_EQ(128u, ss.clones[1].overlap[0].second);
+  ASSERT_EQ(256u, ss.clones[1].overlap[1].first);
+  ASSERT_EQ(256u, ss.clones[1].overlap[1].second);
+  ASSERT_EQ(768u, ss.clones[1].overlap[2].first);
+  ASSERT_EQ(256u, ss.clones[1].overlap[2].second);
+  ASSERT_EQ(1152u, ss.clones[1].overlap[3].first);
+  ASSERT_EQ(128u, ss.clones[1].overlap[3].second);
+  ASSERT_EQ(1280u, ss.clones[1].size);
+
+  ASSERT_EQ(head, ss.clones[2].cloneid);
+  ASSERT_EQ(0u, ss.clones[2].snaps.size());
+  ASSERT_EQ(0u, ss.clones[2].overlap.size());
+  ASSERT_EQ(1280u, ss.clones[2].size);
 
   ASSERT_EQ(0, ioctx.selfmanaged_snap_remove(my_snaps.back()));
   my_snaps.pop_back();