Merge pull request #660 from ceph/wip-fs-crc

sloppy / opportunistic CRC tracking in the filestore

Reviewed-by: Samuel Just <sam.just@inktank.com>

11 files changed, 561 insertions, 3 deletions

diff --git a/src/common/Makefile.am b/src/common/Makefile.am
index 3526118205f..deddc5d831c 100644
--- a/src/common/Makefile.am
+++ b/src/common/Makefile.am
@@ -4,6 +4,7 @@ libcommon_la_SOURCES = \
 	common/LogClient.cc \
 	common/LogEntry.cc \
 	common/PrebufferedStreambuf.cc \
+	common/SloppyCRCMap.cc \
 	common/BackTrace.cc \
 	common/perf_counters.cc \
 	common/Mutex.cc \
@@ -120,6 +121,7 @@ noinst_HEADERS += \
 	common/LogClient.h \
 	common/LogEntry.h \
 	common/Preforker.h \
+	common/SloppyCRCMap.h \
 	common/WorkQueue.h \
 	common/PrioritizedQueue.h \
 	common/ceph_argparse.h \
diff --git a/src/common/SloppyCRCMap.cc b/src/common/SloppyCRCMap.cc
new file mode 100644
index 00000000000..7924ae6e8a7
--- /dev/null
+++ b/src/common/SloppyCRCMap.cc
@@ -0,0 +1,180 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+
+#include "common/SloppyCRCMap.h"
+#include "common/Formatter.h"
+
+void SloppyCRCMap::write(uint64_t offset, uint64_t len, const bufferlist& bl,
+			 std::ostream *out)
+{
+  int64_t left = len;
+  uint64_t pos = offset;
+  unsigned o = offset % block_size;
+  if (o) {
+    crc_map.erase(offset - o);
+    if (out)
+      *out << "write invalidate " << (offset - o) << "\n";
+    pos += (block_size - o);
+    left -= (block_size - o);
+  }
+  while (left >= block_size) {
+    bufferlist t;
+    t.substr_of(bl, pos - offset, block_size);
+    crc_map[pos] = t.crc32c(crc_iv);
+    if (out)
+      *out << "write set " << pos << " " << crc_map[pos] << "\n";
+    pos += block_size;
+    left -= block_size;
+  }
+  if (left > 0) {
+    crc_map.erase(pos);
+    if (out)
+      *out << "write invalidate " << pos << "\n";
+  }
+}
+
+int SloppyCRCMap::read(uint64_t offset, uint64_t len, const bufferlist& bl,
+		       std::ostream *err)
+{
+  int errors = 0;
+  int64_t left = len;
+  uint64_t pos = offset;
+  unsigned o = offset % block_size;
+  if (o) {
+    pos += (block_size - o);
+    left -= (block_size - o);
+  }
+  while (left >= block_size) {
+    // FIXME: this could be more efficient if we avoid doing a find()
+    // on each iteration
+    std::map<uint64_t,uint32_t>::iterator p = crc_map.find(pos);
+    if (p != crc_map.end()) {
+      bufferlist t;
+      t.substr_of(bl, pos - offset, block_size);
+      uint32_t crc = t.crc32c(crc_iv);
+      if (p->second != crc) {
+	errors++;
+	if (err)
+	  *err << "offset " << pos << " len " << block_size
+	       << " has crc " << crc << " expected " << p->second << "\n";
+      }
+    }
+    pos += block_size;
+    left -= block_size;
+  }
+  return errors;  
+}
+
+void SloppyCRCMap::truncate(uint64_t offset)
+{
+  offset -= offset % block_size;
+  std::map<uint64_t,uint32_t>::iterator p = crc_map.lower_bound(offset);
+  while (p != crc_map.end())
+    crc_map.erase(p++);
+}
+
+void SloppyCRCMap::zero(uint64_t offset, uint64_t len)
+{
+  int64_t left = len;
+  uint64_t pos = offset;
+  unsigned o = offset % block_size;
+  if (o) {
+    crc_map.erase(offset - o);
+    pos += (block_size - o);
+    left -= (block_size - o);
+  }
+  while (left >= block_size) {
+    crc_map[pos] = zero_crc;
+    pos += block_size;
+    left -= block_size;
+  }
+  if (left > 0)
+    crc_map.erase(pos);
+}
+
+void SloppyCRCMap::clone_range(uint64_t offset, uint64_t len,
+			       uint64_t srcoff, const SloppyCRCMap& src,
+			       std::ostream *out)
+{
+  int64_t left = len;
+  uint64_t pos = offset;
+  uint64_t srcpos = srcoff;
+  unsigned o = offset % block_size;
+  if (o) {
+    crc_map.erase(offset - o);
+    pos += (block_size - o);
+    srcpos += (block_size - o);
+    left -= (block_size - o);
+    if (out)
+      *out << "clone_range invalidate " << (offset - o) << "\n";
+  }
+  while (left >= block_size) {
+    // FIXME: this could be more efficient.
+    if (block_size == src.block_size) {
+      map<uint64_t,uint32_t>::const_iterator p = src.crc_map.find(srcpos);
+      if (p != src.crc_map.end()) {
+	crc_map[pos] = p->second;
+	if (out)
+	  *out << "clone_range copy " << pos << " " << p->second << "\n";
+      } else {
+	crc_map.erase(pos);
+	if (out)
+	  *out << "clone_range invalidate " << pos << "\n";
+      }
+    } else {
+      crc_map.erase(pos);
+      if (out)
+	*out << "clone_range invalidate " << pos << "\n";
+    }
+    pos += block_size;
+    srcpos += block_size;
+    left -= block_size;
+  }
+  if (left > 0) {
+    crc_map.erase(pos);
+    if (out)
+      *out << "clone_range invalidate " << pos << "\n";
+  }
+}
+
+void SloppyCRCMap::encode(bufferlist& bl) const
+{
+  ENCODE_START(1, 1, bl);
+  ::encode(block_size, bl);
+  ::encode(crc_map, bl);
+  ENCODE_FINISH(bl);
+}
+
+void SloppyCRCMap::decode(bufferlist::iterator& bl)
+{
+  DECODE_START(1, bl);
+  uint32_t bs;
+  ::decode(bs, bl);
+  set_block_size(bs);
+  ::decode(crc_map, bl);
+  DECODE_FINISH(bl);
+}
+
+void SloppyCRCMap::dump(Formatter *f) const
+{
+  f->dump_unsigned("block_size", block_size);
+  f->open_array_section("crc_map");
+  for (map<uint64_t,uint32_t>::const_iterator p = crc_map.begin(); p != crc_map.end(); ++p) {
+    f->open_object_section("crc");
+    f->dump_unsigned("offset", p->first);
+    f->dump_unsigned("crc", p->second);
+    f->close_section();
+  }
+  f->close_section();
+}
+
+void SloppyCRCMap::generate_test_instances(list<SloppyCRCMap*>& ls)
+{
+  ls.push_back(new SloppyCRCMap);
+  ls.push_back(new SloppyCRCMap(2));
+  bufferlist bl;
+  bl.append("some data");
+  ls.back()->write(1, bl.length(), bl);
+  ls.back()->write(10, bl.length(), bl);
+  ls.back()->zero(4, 2);
+}
diff --git a/src/common/SloppyCRCMap.h b/src/common/SloppyCRCMap.h
new file mode 100644
index 00000000000..c07b4d9bb9d
--- /dev/null
+++ b/src/common/SloppyCRCMap.h
@@ -0,0 +1,78 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_COMMON_SLOPPYCRCMAP_H
+#define CEPH_COMMON_SLOPPYCRCMAP_H
+
+#include "include/types.h"
+#include "include/encoding.h"
+
+#include <map>
+#include <ostream>
+
+/**
+ * SloppyCRCMap
+ *
+ * Opportunistically track CRCs on any reads or writes that cover full
+ * blocks.  Verify read results when we have CRC data available for
+ * the given extent.
+ */
+class SloppyCRCMap {
+  static const int crc_iv = 0xffffffff;
+
+  std::map<uint64_t, uint32_t> crc_map;  // offset -> crc(-1)
+  uint32_t block_size;
+  uint32_t zero_crc;
+
+public:
+  SloppyCRCMap(uint32_t b=0) {
+    set_block_size(b);
+  }
+
+  void set_block_size(uint32_t b) {
+    block_size = b;
+    //zero_crc = ceph_crc32c(0xffffffff, NULL, block_size);
+    if (b) {
+      bufferlist bl;
+      bufferptr bp(block_size);
+      bp.zero();
+      bl.append(bp);
+      zero_crc = bl.crc32c(crc_iv);
+    } else {
+      zero_crc = crc_iv;
+    }
+  }
+
+  /// update based on a write
+  void write(uint64_t offset, uint64_t len, const bufferlist& bl,
+	     std::ostream *out = NULL);
+
+  /// update based on a truncate
+  void truncate(uint64_t offset);
+
+  /// update based on a zero/punch_hole
+  void zero(uint64_t offset, uint64_t len);
+
+  /// update based on a zero/punch_hole
+  void clone_range(uint64_t offset, uint64_t len, uint64_t srcoff, const SloppyCRCMap& src,
+		   std::ostream *out = NULL);
+
+  /**
+   * validate a read result
+   *
+   * @param offset offset
+   * @param length length
+   * @param bl data read
+   * @param err option ostream to describe errors in detail
+   * @returns error count, 0 for success
+   */
+  int read(uint64_t offset, uint64_t len, const bufferlist& bl, std::ostream *err);
+
+  void encode(bufferlist& bl) const;
+  void decode(bufferlist::iterator& bl);
+  void dump(Formatter *f) const;
+  static void generate_test_instances(list<SloppyCRCMap*>& ls);
+};
+WRITE_CLASS_ENCODER(SloppyCRCMap)
+
+#endif
diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index 1443fabdcd4..d7505306b6e 100644
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -557,6 +557,9 @@ OPTION(filestore_max_inline_xattr_size, OPT_U32, 512)
 // for more than filestore_max_inline_xattrs attrs
 OPTION(filestore_max_inline_xattrs, OPT_U32, 2)
 
+OPTION(filestore_sloppy_crc, OPT_BOOL, false)         // track sloppy crcs
+OPTION(filestore_sloppy_crc_block_size, OPT_INT, 65536)
+
 OPTION(filestore_max_sync_interval, OPT_DOUBLE, 5)    // seconds
 OPTION(filestore_min_sync_interval, OPT_DOUBLE, .01)  // seconds
 OPTION(filestore_btrfs_snap, OPT_BOOL, true)
diff --git a/src/os/FileStore.cc b/src/os/FileStore.cc
index 343fb25c0e4..a470e63dc1c 100644
--- a/src/os/FileStore.cc
+++ b/src/os/FileStore.cc
@@ -167,12 +167,17 @@ int FileStore::lfn_find(coll_t cid, const ghobject_t& oid, IndexedPath *path)
 int FileStore::lfn_truncate(coll_t cid, const ghobject_t& oid, off_t length)
 {
   IndexedPath path;
-  int r = lfn_find(cid, oid, &path);
+  FDRef fd;
+  int r = lfn_open(cid, oid, false, &fd, &path);
   if (r < 0)
     return r;
-  r = ::truncate(path->path(), length);
+  r = ::ftruncate(**fd, length);
   if (r < 0)
     r = -errno;
+  if (r >= 0 && m_filestore_sloppy_crc) {
+    int rc = backend->_crc_update_truncate(**fd, length);
+    assert(rc >= 0);
+  }
   assert(!m_filestore_fail_eio || r != -EIO);
   return r;
 }
@@ -415,7 +420,9 @@ FileStore::FileStore(const std::string &base, const std::string &jdev, const cha
   m_filestore_queue_committing_max_ops(g_conf->filestore_queue_committing_max_ops),
   m_filestore_queue_committing_max_bytes(g_conf->filestore_queue_committing_max_bytes),
   m_filestore_do_dump(false),
-  m_filestore_dump_fmt(true)
+  m_filestore_dump_fmt(true),
+  m_filestore_sloppy_crc(g_conf->filestore_sloppy_crc),
+  m_filestore_sloppy_crc_block_size(g_conf->filestore_sloppy_crc_block_size)
 {
   m_filestore_kill_at.set(g_conf->filestore_kill_at);
 
@@ -2555,6 +2562,17 @@ int FileStore::read(
   }
   bptr.set_length(got);   // properly size the buffer
   bl.push_back(bptr);   // put it in the target bufferlist
+
+  if (m_filestore_sloppy_crc && (!replaying || backend->can_checkpoint())) {
+    ostringstream ss;
+    int errors = backend->_crc_verify_read(**fd, offset, got, bl, &ss);
+    if (errors > 0) {
+      dout(0) << "FileStore::read " << cid << "/" << oid << " " << offset << "~"
+	      << got << " ... BAD CRC:\n" << ss.str() << dendl;
+      assert(0 == "bad crc on read");
+    }
+  }
+
   lfn_close(fd);
 
   dout(10) << "FileStore::read " << cid << "/" << oid << " " << offset << "~"
@@ -2716,6 +2734,11 @@ int FileStore::_write(coll_t cid, const ghobject_t& oid,
   if (r == 0)
     r = bl.length();
 
+  if (r >= 0 && m_filestore_sloppy_crc) {
+    int rc = backend->_crc_update_write(**fd, offset, len, bl);
+    assert(rc >= 0);
+  }
+
   // flush?
   if (!replaying &&
       g_conf->filestore_wbthrottle_enable)
@@ -2747,6 +2770,11 @@ int FileStore::_zero(coll_t cid, const ghobject_t& oid, uint64_t offset, size_t
     ret = -errno;
   lfn_close(fd);
 
+  if (ret >= 0 && m_filestore_sloppy_crc) {
+    int rc = backend->_crc_update_zero(**fd, offset, len);
+    assert(rc >= 0);
+  }
+
   if (ret == 0)
     goto out;  // yay!
   if (ret != -EOPNOTSUPP)
@@ -2900,6 +2928,10 @@ int FileStore::_do_copy_range(int from, int to, uint64_t srcoff, uint64_t len, u
       break;
     pos += r;
   }
+  if (r >= 0 && m_filestore_sloppy_crc) {
+    int rc = backend->_crc_update_clone_range(from, to, srcoff, len, dstoff);
+    assert(rc >= 0);
+  }
   dout(20) << "_do_copy_range " << srcoff << "~" << len << " to " << dstoff << " = " << r << dendl;
   return r;
 }
@@ -4548,6 +4580,8 @@ const char** FileStore::get_tracked_conf_keys() const
     "filestore_kill_at",
     "filestore_fail_eio",
     "filestore_replica_fadvise",
+    "filestore_sloppy_crc",
+    "filestore_sloppy_crc_block_size",
     NULL
   };
   return KEYS;
@@ -4564,6 +4598,8 @@ void FileStore::handle_conf_change(const struct md_config_t *conf,
       changed.count("filestore_queue_committing_max_bytes") ||
       changed.count("filestore_kill_at") ||
       changed.count("filestore_fail_eio") ||
+      changed.count("filestore_sloppy_crc") ||
+      changed.count("filestore_sloppy_crc_block_size") ||
       changed.count("filestore_replica_fadvise")) {
     Mutex::Locker l(lock);
     m_filestore_min_sync_interval = conf->filestore_min_sync_interval;
@@ -4575,6 +4611,8 @@ void FileStore::handle_conf_change(const struct md_config_t *conf,
     m_filestore_kill_at.set(conf->filestore_kill_at);
     m_filestore_fail_eio = conf->filestore_fail_eio;
     m_filestore_replica_fadvise = conf->filestore_replica_fadvise;
+    m_filestore_sloppy_crc = conf->filestore_sloppy_crc;
+    m_filestore_sloppy_crc_block_size = conf->filestore_sloppy_crc_block_size;
   }
   if (changed.count("filestore_commit_timeout")) {
     Mutex::Locker l(sync_entry_timeo_lock);
diff --git a/src/os/FileStore.h b/src/os/FileStore.h
index b9017985a34..fdab0ece34f 100644
--- a/src/os/FileStore.h
+++ b/src/os/FileStore.h
@@ -591,6 +591,8 @@ private:
   std::ofstream m_filestore_dump;
   JSONFormatter m_filestore_dump_fmt;
   atomic_t m_filestore_kill_at;
+  bool m_filestore_sloppy_crc;
+  int m_filestore_sloppy_crc_block_size;
   FSSuperblock superblock;
 
   /**
@@ -643,6 +645,9 @@ protected:
   int _copy_range(int from, int to, uint64_t srcoff, uint64_t len, uint64_t dstoff) {
     return filestore->_do_copy_range(from, to, srcoff, len, dstoff);
   }
+  int get_crc_block_size() {
+    return filestore->m_filestore_sloppy_crc_block_size;
+  }
 public:
   FileStoreBackend(FileStore *fs) : filestore(fs) {}
   virtual ~FileStoreBackend() {};
@@ -658,6 +663,15 @@ public:
   virtual bool has_fiemap() = 0;
   virtual int do_fiemap(int fd, off_t start, size_t len, struct fiemap **pfiemap) = 0;
   virtual int clone_range(int from, int to, uint64_t srcoff, uint64_t len, uint64_t dstoff) = 0;
+
+  // hooks for (sloppy) crc tracking
+  virtual int _crc_update_write(int fd, loff_t off, size_t len, const bufferlist& bl) = 0;
+  virtual int _crc_update_truncate(int fd, loff_t off) = 0;
+  virtual int _crc_update_zero(int fd, loff_t off, size_t len) = 0;
+  virtual int _crc_update_clone_range(int srcfd, int destfd,
+				      loff_t srcoff, size_t len, loff_t dstoff) = 0;
+  virtual int _crc_verify_read(int fd, loff_t off, size_t len, const bufferlist& bl,
+			       ostream *out) = 0;
 };
 
 #endif
diff --git a/src/os/GenericFileStoreBackend.cc b/src/os/GenericFileStoreBackend.cc
index 461158fdfab..dad1a9c220c 100644
--- a/src/os/GenericFileStoreBackend.cc
+++ b/src/os/GenericFileStoreBackend.cc
@@ -40,6 +40,12 @@
 #include "common/config.h"
 #include "common/sync_filesystem.h"
 
+#include "common/SloppyCRCMap.h"
+#include "os/chain_xattr.h"
+
+#define SLOPPY_CRC_XATTR "user.cephos.scrc"
+
+
 #define dout_subsys ceph_subsys_filestore
 #undef dout_prefix
 #define dout_prefix *_dout << "genericfilestorebackend(" << get_basedir_path() << ") "
@@ -251,3 +257,104 @@ done_err:
   free(fiemap);
   return ret;
 }
+
+
+int GenericFileStoreBackend::_crc_load_or_init(int fd, SloppyCRCMap *cm)
+{
+  char buf[100];
+  bufferptr bp;
+  int l = chain_fgetxattr(fd, SLOPPY_CRC_XATTR, buf, sizeof(buf));
+  if (l == -ENODATA) {
+    return 0;
+  }
+  if (l >= 0) {
+    bp = buffer::create(l);
+    memcpy(bp.c_str(), buf, l);
+  } else if (l == -ERANGE) {
+    l = chain_fgetxattr(fd, SLOPPY_CRC_XATTR, 0, 0);
+    if (l > 0) {
+      bp = buffer::create(l);
+      l = chain_fgetxattr(fd, SLOPPY_CRC_XATTR, bp.c_str(), l);
+    }
+  }
+  bufferlist bl;
+  bl.append(bp);
+  bufferlist::iterator p = bl.begin();
+  try {
+    ::decode(*cm, p);
+  }
+  catch (buffer::error &e) {
+    return -EIO;
+  }
+  return 0;
+}
+
+int GenericFileStoreBackend::_crc_save(int fd, SloppyCRCMap *cm)
+{
+  bufferlist bl;
+  ::encode(*cm, bl);
+  return chain_fsetxattr(fd, SLOPPY_CRC_XATTR, bl.c_str(), bl.length());
+}
+
+int GenericFileStoreBackend::_crc_update_write(int fd, loff_t off, size_t len, const bufferlist& bl)
+{
+  SloppyCRCMap scm(get_crc_block_size());
+  int r = _crc_load_or_init(fd, &scm);
+  if (r < 0)
+    return r;
+  ostringstream ss;
+  scm.write(off, len, bl, &ss);
+  dout(30) << __func__ << "\n" << ss.str() << dendl;
+  r = _crc_save(fd, &scm);
+  return r;
+}
+
+int GenericFileStoreBackend::_crc_update_truncate(int fd, loff_t off)
+{
+  SloppyCRCMap scm(get_crc_block_size());
+  int r = _crc_load_or_init(fd, &scm);
+  if (r < 0)
+    return r;
+  scm.truncate(off);
+  r = _crc_save(fd, &scm);
+  return r;
+}
+
+int GenericFileStoreBackend::_crc_update_zero(int fd, loff_t off, size_t len)
+{
+  SloppyCRCMap scm(get_crc_block_size());
+  int r = _crc_load_or_init(fd, &scm);
+  if (r < 0)
+    return r;
+  scm.zero(off, len);
+  r = _crc_save(fd, &scm);
+  return r;
+}
+
+int GenericFileStoreBackend::_crc_update_clone_range(int srcfd, int destfd,
+						     loff_t srcoff, size_t len, loff_t dstoff)
+{
+  SloppyCRCMap scm_src(get_crc_block_size());
+  SloppyCRCMap scm_dst(get_crc_block_size());
+  int r = _crc_load_or_init(srcfd, &scm_src);
+  if (r < 0)
+    return r;
+  r = _crc_load_or_init(destfd, &scm_dst);
+  if (r < 0)
+    return r;
+  ostringstream ss;
+  scm_dst.clone_range(srcoff, len, dstoff, scm_src, &ss);
+  dout(30) << __func__ << "\n" << ss.str() << dendl;
+  r = _crc_save(destfd, &scm_dst);
+  return r;
+}
+
+int GenericFileStoreBackend::_crc_verify_read(int fd, loff_t off, size_t len, const bufferlist& bl,
+					      ostream *out)
+{
+  SloppyCRCMap scm(get_crc_block_size());
+  int r = _crc_load_or_init(fd, &scm);
+  if (r < 0)
+    return r;
+  return scm.read(off, len, bl, out);
+}
diff --git a/src/os/GenericFileStoreBackend.h b/src/os/GenericFileStoreBackend.h
index 95aca971708..5a09c2497a8 100644
--- a/src/os/GenericFileStoreBackend.h
+++ b/src/os/GenericFileStoreBackend.h
@@ -17,6 +17,8 @@
 
 #include "FileStore.h"
 
+class SloppyCRCMap;
+
 class GenericFileStoreBackend : public FileStoreBackend {
 private:
   bool ioctl_fiemap;
@@ -25,6 +27,7 @@ private:
 public:
   GenericFileStoreBackend(FileStore *fs);
   virtual ~GenericFileStoreBackend() {};
+
   virtual int detect_features();
   virtual int create_current();
   virtual bool can_checkpoint() { return false; };
@@ -39,5 +42,17 @@ public:
   virtual int clone_range(int from, int to, uint64_t srcoff, uint64_t len, uint64_t dstoff) {
     return _copy_range(from, to, srcoff, len, dstoff);
   }
+
+private:
+  int _crc_load_or_init(int fd, SloppyCRCMap *cm);
+  int _crc_save(int fd, SloppyCRCMap *cm);
+public:
+  virtual int _crc_update_write(int fd, loff_t off, size_t len, const bufferlist& bl);
+  virtual int _crc_update_truncate(int fd, loff_t off);
+  virtual int _crc_update_zero(int fd, loff_t off, size_t len);
+  virtual int _crc_update_clone_range(int srcfd, int destfd,
+				      loff_t srcoff, size_t len, loff_t dstoff);
+  virtual int _crc_verify_read(int fd, loff_t off, size_t len, const bufferlist& bl,
+			       ostream *out);
 };
 #endif
diff --git a/src/test/Makefile.am b/src/test/Makefile.am
index 32fee301e4c..debe18ff907 100644
--- a/src/test/Makefile.am
+++ b/src/test/Makefile.am
@@ -263,6 +263,11 @@ unittest_sharedptr_registry_CXXFLAGS = $(UNITTEST_CXXFLAGS)
 unittest_sharedptr_registry_LDADD = $(UNITTEST_LDADD) $(CEPH_GLOBAL)
 check_PROGRAMS += unittest_sharedptr_registry
 
+unittest_sloppy_crc_map_SOURCES = test/common/test_sloppy_crc_map.cc
+unittest_sloppy_crc_map_CXXFLAGS = $(UNITTEST_CXXFLAGS)
+unittest_sloppy_crc_map_LDADD = $(UNITTEST_LDADD) $(CEPH_GLOBAL)
+check_PROGRAMS += unittest_sloppy_crc_map
+
 unittest_util_SOURCES = test/common/test_util.cc
 unittest_util_CXXFLAGS = $(UNITTEST_CXXFLAGS)
 unittest_util_LDADD = $(LIBCOMMON) -lm $(UNITTEST_LDADD) $(CRYPTO_LIBS) $(EXTRALIBS)
diff --git a/src/test/common/test_sloppy_crc_map.cc b/src/test/common/test_sloppy_crc_map.cc
new file mode 100644
index 00000000000..2650f4f960d
--- /dev/null
+++ b/src/test/common/test_sloppy_crc_map.cc
@@ -0,0 +1,113 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "common/SloppyCRCMap.h"
+#include "common/Formatter.h"
+#include <gtest/gtest.h>
+
+void dump(const SloppyCRCMap& scm)
+{
+  Formatter *f = new_formatter("json-pretty");
+  f->open_object_section("map");
+  scm.dump(f);
+  f->close_section();
+  f->flush(cout);
+  delete f;
+}
+
+TEST(SloppyCRCMap, basic) {
+  SloppyCRCMap scm(4);
+
+  bufferlist a, b;
+  a.append("The quick brown fox jumped over a fence whose color I forget.");
+  b.append("asdf");
+
+  scm.write(0, a.length(), a);
+  if (0)
+    dump(scm);
+  ASSERT_EQ(0, scm.read(0, a.length(), a, &cout));
+
+  scm.write(12, b.length(), b);
+  if (0)
+    dump(scm);
+
+  ASSERT_EQ(0, scm.read(12, b.length(), b, &cout));
+  ASSERT_EQ(1, scm.read(0, a.length(), a, &cout));
+}
+
+TEST(SloppyCRCMap, truncate) {
+  SloppyCRCMap scm(4);
+
+  bufferlist a, b;
+  a.append("asdf");
+  b.append("qwer");
+
+  scm.write(0, a.length(), a);
+  scm.write(4, a.length(), a);
+  ASSERT_EQ(0, scm.read(4, 4, a, &cout));
+  ASSERT_EQ(1, scm.read(4, 4, b, &cout));
+  scm.truncate(4);
+  ASSERT_EQ(0, scm.read(4, 4, b, &cout));
+}
+
+TEST(SloppyCRCMap, zero) {
+  SloppyCRCMap scm(4);
+
+  bufferlist a, b;
+  a.append("asdf");
+  b.append("qwer");
+
+  scm.write(0, a.length(), a);
+  scm.write(4, a.length(), a);
+  ASSERT_EQ(0, scm.read(4, 4, a, &cout));
+  ASSERT_EQ(1, scm.read(4, 4, b, &cout));
+  scm.zero(4, 4);
+  ASSERT_EQ(1, scm.read(4, 4, a, &cout));
+  ASSERT_EQ(1, scm.read(4, 4, b, &cout));
+
+  bufferptr bp(4);
+  bp.zero();
+  bufferlist c;
+  c.append(bp);
+  ASSERT_EQ(0, scm.read(0, 4, a, &cout));
+  ASSERT_EQ(0, scm.read(4, 4, c, &cout));
+  scm.zero(0, 15);
+  ASSERT_EQ(1, scm.read(0, 4, a, &cout));
+  ASSERT_EQ(0, scm.read(0, 4, c, &cout));
+}
+
+TEST(SloppyCRCMap, clone_range) {
+  SloppyCRCMap src(4);
+  SloppyCRCMap dst(4);
+
+  bufferlist a, b;
+  a.append("asdfghjkl");
+  b.append("qwertyui");
+
+  src.write(0, a.length(), a);
+  src.write(8, a.length(), a);
+  src.write(16, a.length(), a);
+
+  dst.write(0, b.length(), b);
+  dst.clone_range(0, 8, 0, src);
+  ASSERT_EQ(2, dst.read(0, 8, b, &cout));
+  ASSERT_EQ(0, dst.read(8, 8, b, &cout));
+
+  dst.write(16, b.length(), b);
+  ASSERT_EQ(2, dst.read(16, 8, a, &cout));
+  dst.clone_range(16, 8, 16, src);
+  ASSERT_EQ(0, dst.read(16, 8, a, &cout));
+
+  dst.write(16, b.length(), b);
+  ASSERT_EQ(1, dst.read(16, 4, a, &cout));
+  dst.clone_range(16, 8, 2, src);
+  ASSERT_EQ(0, dst.read(16, 4, a, &cout));
+
+  dst.write(0, b.length(), b);
+  dst.write(8, b.length(), b);
+  ASSERT_EQ(2, dst.read(0, 8, a, &cout));
+  ASSERT_EQ(2, dst.read(8, 8, a, &cout));
+  dst.clone_range(2, 8, 0, src);
+  ASSERT_EQ(0, dst.read(0, 8, a, &cout));
+  ASSERT_EQ(0, dst.read(8, 4, a, &cout));
+}
diff --git a/src/test/encoding/types.h b/src/test/encoding/types.h
index 7f2b4d9db5d..6dd180bc198 100644
--- a/src/test/encoding/types.h
+++ b/src/test/encoding/types.h
@@ -16,6 +16,9 @@ TYPE(LogEntryKey)
 TYPE(LogEntry)
 TYPE(LogSummary)
 
+#include "common/SloppyCRCMap.h"
+TYPE(SloppyCRCMap)
+
 #include "msg/msg_types.h"
 TYPE(entity_name_t)
 TYPE(entity_addr_t)