memstore: add Object interface to hide bufferlist

this prepares MemStore for a new object data implementation that
replaces bufferlist

Signed-off-by: Casey Bodley <cbodley@redhat.com>

diff --git a/src/os/MemStore.cc b/src/os/MemStore.cc
index 0683d6ab3daa9648bf401006bcd4365316efbbbe..788589d22cb1227d64b6364886d3e85f5b8204b3 100644 (file)
--- a/src/os/MemStore.cc
+++ b/src/os/MemStore.cc
@@ -293,7 +293,7 @@ int MemStore::stat(
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
-  st->st_size = o->data.length();
+  st->st_size = o->get_size();
   st->st_blksize = 4096;
   st->st_blocks = (st->st_size + st->st_blksize - 1) / st->st_blksize;
   st->st_nlink = 1;
@@ -319,16 +319,15 @@ int MemStore::read(
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
-  if (offset >= o->data.length())
+  if (offset >= o->get_size())
     return 0;
   size_t l = len;
   if (l == 0)  // note: len == 0 means read the entire object
-    l = o->data.length();
-  else if (offset + l > o->data.length())
-    l = o->data.length() - offset;
+    l = o->get_size();
+  else if (offset + l > o->get_size())
+    l = o->get_size() - offset;
   bl.clear();
-  bl.substr_of(o->data, offset, l);
-  return bl.length();
+  return o->read(offset, l, bl);
 }
 
 int MemStore::fiemap(coll_t cid, const ghobject_t& oid,
@@ -344,11 +343,11 @@ int MemStore::fiemap(coll_t cid, const ghobject_t& oid,
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
-  if (offset >= o->data.length())
+  if (offset >= o->get_size())
     return 0;
   size_t l = len;
-  if (offset + l > o->data.length())
-    l = o->data.length() - offset;
+  if (offset + l > o->get_size())
+    l = o->get_size() - offset;
   map<uint64_t, uint64_t> m;
   m[offset] = l;
   ::encode(m, bl);
@@ -957,7 +956,7 @@ int MemStore::_touch(coll_t cid, const ghobject_t& oid)
 
   ObjectRef o = c->get_object(oid);
   if (!o) {
-    o.reset(new Object);
+    o.reset(new BufferlistObject);
     c->object_map[oid] = o;
     c->object_hash[oid] = o;
   }
@@ -980,46 +979,18 @@ int MemStore::_write(coll_t cid, const ghobject_t& oid,
   ObjectRef o = c->get_object(oid);
   if (!o) {
     // write implicitly creates a missing object
-    o.reset(new Object);
+    o.reset(new BufferlistObject);
     c->object_map[oid] = o;
     c->object_hash[oid] = o;
   }
 
-  int old_size = o->data.length();
-  _write_into_bl(bl, offset, &o->data);
-  used_bytes += (o->data.length() - old_size);
+  const ssize_t old_size = o->get_size();
+  o->write(offset, bl);
+  used_bytes += (o->get_size() - old_size);
 
   return 0;
 }
 
-void MemStore::_write_into_bl(const bufferlist& src, unsigned offset,
-                             bufferlist *dst)
-{
-  unsigned len = src.length();
-
-  // before
-  bufferlist newdata;
-  if (dst->length() >= offset) {
-    newdata.substr_of(*dst, 0, offset);
-  } else {
-    newdata.substr_of(*dst, 0, dst->length());
-    bufferptr bp(offset - dst->length());
-    bp.zero();
-    newdata.append(bp);
-  }
-
-  newdata.append(src);
-
-  // after
-  if (dst->length() > offset + len) {
-    bufferlist tail;
-    tail.substr_of(*dst, offset + len, dst->length() - (offset + len));
-    newdata.append(tail);
-  }
-
-  dst->claim(newdata);
-}
-
 int MemStore::_zero(coll_t cid, const ghobject_t& oid,
                    uint64_t offset, size_t len)
 {
@@ -1043,20 +1014,10 @@ int MemStore::_truncate(coll_t cid, const ghobject_t& oid, uint64_t size)
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
-  if (o->data.length() > size) {
-    bufferlist bl;
-    bl.substr_of(o->data, 0, size);
-    used_bytes -= o->data.length() - size;
-    o->data.claim(bl);
-  } else if (o->data.length() == size) {
-    // do nothing
-  } else {
-    bufferptr bp(size - o->data.length());
-    bp.zero();
-    used_bytes += bp.length();
-    o->data.append(bp);
-  }
-  return 0;
+  const ssize_t old_size = o->get_size();
+  int r = o->truncate(size);
+  used_bytes += (o->get_size() - old_size);
+  return r;
 }
 
 int MemStore::_remove(coll_t cid, const ghobject_t& oid)
@@ -1073,7 +1034,7 @@ int MemStore::_remove(coll_t cid, const ghobject_t& oid)
   c->object_map.erase(oid);
   c->object_hash.erase(oid);
 
-  used_bytes -= o->data.length();
+  used_bytes -= o->get_size();
 
   return 0;
 }
@@ -1142,12 +1103,12 @@ int MemStore::_clone(coll_t cid, const ghobject_t& oldoid,
     return -ENOENT;
   ObjectRef no = c->get_object(newoid);
   if (!no) {
-    no.reset(new Object);
+    no.reset(new BufferlistObject);
     c->object_map[newoid] = no;
     c->object_hash[newoid] = no;
   }
-  used_bytes += oo->data.length() - no->data.length();
-  no->data = oo->data;
+  used_bytes += oo->get_size() - no->get_size();
+  no->clone(oo.get(), 0, oo->get_size(), 0);
   no->omap_header = oo->omap_header;
   no->omap = oo->omap;
   no->xattr = oo->xattr;
@@ -1172,20 +1133,18 @@ int MemStore::_clone_range(coll_t cid, const ghobject_t& oldoid,
     return -ENOENT;
   ObjectRef no = c->get_object(newoid);
   if (!no) {
-    no.reset(new Object);
+    no.reset(new BufferlistObject);
     c->object_map[newoid] = no;
     c->object_hash[newoid] = no;
   }
-  if (srcoff >= oo->data.length())
+  if (srcoff >= oo->get_size())
     return 0;
-  if (srcoff + len >= oo->data.length())
-    len = oo->data.length() - srcoff;
-  bufferlist bl;
-  bl.substr_of(oo->data, srcoff, len);
+  if (srcoff + len >= oo->get_size())
+    len = oo->get_size() - srcoff;
 
-  int old_size = no->data.length();
-  _write_into_bl(bl, dstoff, &no->data);
-  used_bytes += (no->data.length() - old_size);
+  const ssize_t old_size = no->get_size();
+  no->clone(oo.get(), srcoff, len, dstoff);
+  used_bytes += (no->get_size() - old_size);
 
   return len;
 }
@@ -1399,3 +1358,71 @@ int MemStore::_split_collection(coll_t cid, uint32_t bits, uint32_t match,
 
   return 0;
 }
+
+// BufferlistObject
+int MemStore::BufferlistObject::read(uint64_t offset, uint64_t len,
+                                     bufferlist &bl)
+{
+  bl.substr_of(data, offset, len);
+  return bl.length();
+}
+
+int MemStore::BufferlistObject::write(uint64_t offset, const bufferlist &src)
+{
+  unsigned len = src.length();
+
+  // before
+  bufferlist newdata;
+  if (get_size() >= offset) {
+    newdata.substr_of(data, 0, offset);
+  } else {
+    newdata.substr_of(data, 0, get_size());
+    bufferptr bp(offset - get_size());
+    bp.zero();
+    newdata.append(bp);
+  }
+
+  newdata.append(src);
+
+  // after
+  if (get_size() > offset + len) {
+    bufferlist tail;
+    tail.substr_of(data, offset + len, get_size() - (offset + len));
+    newdata.append(tail);
+  }
+
+  data.claim(newdata);
+  return 0;
+}
+
+int MemStore::BufferlistObject::clone(Object *src, uint64_t srcoff,
+                                      uint64_t len, uint64_t dstoff)
+{
+  auto srcbl = dynamic_cast<const BufferlistObject*>(src);
+  if (srcbl == nullptr)
+    return -ENOTSUP;
+
+  if (srcoff == dstoff && len == src->get_size()) {
+    data = srcbl->data;
+    return 0;
+  }
+  bufferlist bl;
+  bl.substr_of(srcbl->data, srcoff, len);
+  return write(dstoff, bl);
+}
+
+int MemStore::BufferlistObject::truncate(uint64_t size)
+{
+  if (get_size() > size) {
+    bufferlist bl;
+    bl.substr_of(data, 0, size);
+    data.claim(bl);
+  } else if (get_size() == size) {
+    // do nothing
+  } else {
+    bufferptr bp(size - get_size());
+    bp.zero();
+    data.append(bp);
+  }
+  return 0;
+}

diff --git a/src/os/MemStore.h b/src/os/MemStore.h
index a3bed1cbad1be0a0765c9aa045bc5def86092014..38159e9b7ce8203e524a5dc836ad327c28c8cdbd 100644 (file)
--- a/src/os/MemStore.h
+++ b/src/os/MemStore.h
@@ -29,7 +29,6 @@
 class MemStore : public ObjectStore {
 public:
   struct Object : public RefCountedObject {
-    bufferlist data;
     map<string,bufferptr> xattr;
     bufferlist omap_header;
     map<string,bufferlist> omap;
@@ -38,24 +37,29 @@ public:
     friend void intrusive_ptr_add_ref(Object *o) { o->get(); }
     friend void intrusive_ptr_release(Object *o) { o->put(); }
 
-    void encode(bufferlist& bl) const {
-      ENCODE_START(1, 1, bl);
-      ::encode(data, bl);
+    // interface for object data
+    virtual size_t get_size() const = 0;
+    virtual int read(uint64_t offset, uint64_t len, bufferlist &bl) = 0;
+    virtual int write(uint64_t offset, const bufferlist &bl) = 0;
+    virtual int clone(Object *src, uint64_t srcoff, uint64_t len,
+                      uint64_t dstoff) = 0;
+    virtual int truncate(uint64_t offset) = 0;
+    virtual void encode(bufferlist& bl) const = 0;
+    virtual void decode(bufferlist::iterator& p) = 0;
+
+    void encode_base(bufferlist& bl) const {
       ::encode(xattr, bl);
       ::encode(omap_header, bl);
       ::encode(omap, bl);
-      ENCODE_FINISH(bl);
     }
-    void decode(bufferlist::iterator& p) {
-      DECODE_START(1, p);
-      ::decode(data, p);
+    void decode_base(bufferlist::iterator& p) {
       ::decode(xattr, p);
       ::decode(omap_header, p);
       ::decode(omap, p);
-      DECODE_FINISH(p);
     }
+
     void dump(Formatter *f) const {
-      f->dump_int("data_len", data.length());
+      f->dump_int("data_len", get_size());
       f->dump_int("omap_header_len", omap_header.length());
 
       f->open_array_section("xattrs");
@@ -83,6 +87,31 @@ public:
   };
   typedef Object::Ref ObjectRef;
 
+  struct BufferlistObject : public Object {
+    bufferlist data;
+
+    size_t get_size() const override { return data.length(); }
+
+    int read(uint64_t offset, uint64_t len, bufferlist &bl) override;
+    int write(uint64_t offset, const bufferlist &bl) override;
+    int clone(Object *src, uint64_t srcoff, uint64_t len,
+              uint64_t dstoff) override;
+    int truncate(uint64_t offset) override;
+
+    void encode(bufferlist& bl) const override {
+      ENCODE_START(1, 1, bl);
+      ::encode(data, bl);
+      encode_base(bl);
+      ENCODE_FINISH(bl);
+    }
+    void decode(bufferlist::iterator& p) override {
+      DECODE_START(1, p);
+      ::decode(data, p);
+      decode_base(p);
+      DECODE_FINISH(p);
+    }
+  };
+
   struct Collection : public RefCountedObject {
     ceph::unordered_map<ghobject_t, ObjectRef> object_hash;  ///< for lookup
     map<ghobject_t, ObjectRef,ghobject_t::BitwiseComparator> object_map;        ///< for iteration
@@ -126,7 +155,7 @@ public:
       while (s--) {
        ghobject_t k;
        ::decode(k, p);
-       ObjectRef o(new Object);
+       ObjectRef o(new BufferlistObject);
        o->decode(p);
        object_map.insert(make_pair(k, o));
        object_hash.insert(make_pair(k, o));
@@ -139,7 +168,7 @@ public:
       for (map<ghobject_t, ObjectRef,ghobject_t::BitwiseComparator>::const_iterator p = object_map.begin();
           p != object_map.end();
           ++p) {
-        result += p->second->data.length();
+        result += p->second->get_size();
       }
 
       return result;