Add Header cache to KeyValueStore

In the performance statistic recently, the header lookup becomes the main time
consuming for the read/write operations. Most of time it occur 50% to deal with
header lookup, decode/encode logics.

Now adding header cache using SharedLRU structure which will maintain the header
cache and caller will get the pointer to the real header. It also avoid too much
header copy operations overhead.

Signed-off-by: Haomai Wang <haomaiwang@gmail.com>

diff --git a/src/os/KeyValueStore.cc b/src/os/KeyValueStore.cc
index fba5290b4fc3b09ac4985191c511107a90958a5f..314e73d1c3f33c5eac55d371a0ce6979d3a5d6f2 100644 (file)
--- a/src/os/KeyValueStore.cc
+++ b/src/os/KeyValueStore.cc
@@ -69,36 +69,48 @@ const string KeyValueStore::COLLECTION_ATTR = "__COLL_ATTR__";
 
 // ============== StripObjectMap Implementation =================
 
-int StripObjectMap::save_strip_header(StripObjectHeader &strip_header,
+int StripObjectMap::save_strip_header(StripObjectHeaderRef strip_header,
                                       KeyValueDB::Transaction t)
 {
-  strip_header.header->data.clear();
-  ::encode(strip_header, strip_header.header->data);
+  strip_header->header->data.clear();
+  ::encode(*strip_header, strip_header->header->data);
 
-  set_header(strip_header.cid, strip_header.oid, *(strip_header.header), t);
+  set_header(strip_header->cid, strip_header->oid, *(strip_header->header), t);
   return 0;
 }
 
 int StripObjectMap::create_strip_header(const coll_t &cid,
                                         const ghobject_t &oid,
-                                        StripObjectHeader &strip_header,
+                                        StripObjectHeaderRef *strip_header,
                                         KeyValueDB::Transaction t)
 {
   Header header = generate_new_header(cid, oid, Header(), t);
   if (!header)
     return -EINVAL;
 
-  strip_header.oid = oid;
-  strip_header.cid = cid;
-  strip_header.header = header;
+  _StripObjectHeader *tmp = new _StripObjectHeader();
+  tmp->oid = oid;
+  tmp->cid = cid;
+  tmp->header = header;
 
+  {
+    Mutex::Locker l(lock);
+    *strip_header = caches.add(oid, tmp);
+  }
   return 0;
 }
 
 int StripObjectMap::lookup_strip_header(const coll_t &cid,
                                         const ghobject_t &oid,
-                                        StripObjectHeader &strip_header)
+                                        StripObjectHeaderRef *strip_header)
 {
+  if (cid != coll_t()) {
+    Mutex::Locker l(lock);
+    *strip_header = caches.lookup(oid);
+    if (*strip_header) {
+      return 0;
+    }
+  }
   Header header = lookup_header(cid, oid);
 
   if (!header) {
@@ -107,18 +119,24 @@ int StripObjectMap::lookup_strip_header(const coll_t &cid,
     return -ENOENT;
   }
 
+
+  _StripObjectHeader *tmp = new _StripObjectHeader();
   if (header->data.length()) {
     bufferlist::iterator bliter = header->data.begin();
-    ::decode(strip_header, bliter);
+    ::decode(*tmp, bliter);
   }
 
-  if (strip_header.strip_size == 0)
-    strip_header.strip_size = default_strip_size;
+  if (tmp->strip_size == 0)
+    tmp->strip_size = default_strip_size;
 
-  strip_header.oid = oid;
-  strip_header.cid = cid;
-  strip_header.header = header;
+  tmp->oid = oid;
+  tmp->cid = cid;
+  tmp->header = header;
 
+  {
+    Mutex::Locker l(lock);
+    *strip_header = caches.add(oid, tmp);
+  }
   dout(10) << "lookup_strip_header done " << " cid " << cid << " oid "
            << oid << dendl;
   return 0;
@@ -162,34 +180,28 @@ int StripObjectMap::file_to_extents(uint64_t offset, size_t len,
   return 0;
 }
 
-void StripObjectMap::clone_wrap(StripObjectHeader &old_header,
+void StripObjectMap::clone_wrap(StripObjectHeaderRef old_header,
                                 const coll_t &cid, const ghobject_t &oid,
                                 KeyValueDB::Transaction t,
-                                StripObjectHeader *origin_header,
-                                StripObjectHeader *target_header)
+                                StripObjectHeaderRef *target_header)
 {
   Header new_origin_header;
+  StripObjectHeaderRef tmp = StripObjectHeaderRef(new _StripObjectHeader());
 
-  if (target_header)
-    *target_header = old_header;
-  if (origin_header)
-    *origin_header = old_header;
-
-  clone(old_header.header, cid, oid, t, &new_origin_header,
-        &target_header->header);
+  clone(old_header->header, cid, oid, t, &new_origin_header,
+        &tmp->header);
 
-  if(origin_header)
-    origin_header->header = new_origin_header;
+  tmp->oid = oid;
+  tmp->cid = cid;
+  old_header->header = new_origin_header;
 
-  if (target_header) {
-    target_header->oid = oid;
-    target_header->cid = cid;
-  }
+  if (target_header)
+    *target_header = tmp;
 }
 
 void StripObjectMap::rename_wrap(const coll_t &cid, const ghobject_t &oid,
                                  KeyValueDB::Transaction t,
-                                 StripObjectHeader *header)
+                                 StripObjectHeaderRef header)
 {
   assert(header);
   assert(header->header);
@@ -199,19 +211,19 @@ void StripObjectMap::rename_wrap(const coll_t &cid, const ghobject_t &oid,
   header->cid = cid;
 }
 
-int StripObjectMap::get_values_with_header(const StripObjectHeader &header,
+int StripObjectMap::get_values_with_header(const StripObjectHeaderRef header,
                                            const string &prefix,
                                            const set<string> &keys,
                                            map<string, bufferlist> *out)
 {
-  return scan(header.header, prefix, keys, 0, out);
+  return scan(header->header, prefix, keys, 0, out);
 }
 
-int StripObjectMap::get_keys_with_header(const StripObjectHeader &header,
+int StripObjectMap::get_keys_with_header(const StripObjectHeaderRef header,
                                          const string &prefix,
                                          set<string> *keys)
 {
-  ObjectMap::ObjectMapIterator iter = _get_iterator(header.header, prefix);
+  ObjectMap::ObjectMapIterator iter = _get_iterator(header->header, prefix);
   for (; iter->valid(); iter->next()) {
     if (iter->status())
       return iter->status();
@@ -220,10 +232,10 @@ int StripObjectMap::get_keys_with_header(const StripObjectHeader &header,
   return 0;
 }
 
-int StripObjectMap::get_with_header(const StripObjectHeader &header,
+int StripObjectMap::get_with_header(const StripObjectHeaderRef header,
                         const string &prefix, map<string, bufferlist> *out)
 {
-  ObjectMap::ObjectMapIterator iter = _get_iterator(header.header, prefix);
+  ObjectMap::ObjectMapIterator iter = _get_iterator(header->header, prefix);
   for (iter->seek_to_first(); iter->valid(); iter->next()) {
     if (iter->status())
       return iter->status();
@@ -237,18 +249,18 @@ int StripObjectMap::get_with_header(const StripObjectHeader &header,
 
 int KeyValueStore::BufferTransaction::lookup_cached_header(
     const coll_t &cid, const ghobject_t &oid,
-    StripObjectMap::StripObjectHeader **strip_header,
+    StripObjectMap::StripObjectHeaderRef *strip_header,
     bool create_if_missing)
 {
-  StripObjectMap::StripObjectHeader header;
+  StripObjectMap::StripObjectHeaderRef header;
   int r = 0;
 
   StripHeaderMap::iterator it = strip_headers.find(make_pair(cid, oid));
   if (it != strip_headers.end()) {
 
-    if (!it->second.deleted) {
+    if (!it->second->deleted) {
       if (strip_header)
-        *strip_header = &it->second;
+        *strip_header = it->second;
       return 0;
     } else if (!create_if_missing) {
       return -ENOENT;
@@ -257,11 +269,11 @@ int KeyValueStore::BufferTransaction::lookup_cached_header(
     // If (it->second.deleted && create_if_missing) go down
     r = -ENOENT;
   } else {
-    r = store->backend->lookup_strip_header(cid, oid, header);
+    r = store->backend->lookup_strip_header(cid, oid, &header);
   }
 
   if (r == -ENOENT && create_if_missing) {
-    r = store->backend->create_strip_header(cid, oid, header, t);
+    r = store->backend->create_strip_header(cid, oid, &header, t);
   }
 
   if (r < 0) {
@@ -272,21 +284,21 @@ int KeyValueStore::BufferTransaction::lookup_cached_header(
 
   strip_headers[make_pair(cid, oid)] = header;
   if (strip_header)
-    *strip_header = &strip_headers[make_pair(cid, oid)];
+    *strip_header = strip_headers[make_pair(cid, oid)];
   return r;
 }
 
 int KeyValueStore::BufferTransaction::get_buffer_keys(
-    StripObjectMap::StripObjectHeader &strip_header, const string &prefix,
+    StripObjectMap::StripObjectHeaderRef strip_header, const string &prefix,
     const set<string> &keys, map<string, bufferlist> *out)
 {
   set<string> need_lookup;
 
   for (set<string>::iterator it = keys.begin(); it != keys.end(); ++it) {
     map<pair<string, string>, bufferlist>::iterator i =
-        strip_header.buffers.find(make_pair(prefix, *it));
+        strip_header->buffers.find(make_pair(prefix, *it));
 
-    if (i != strip_header.buffers.end()) {
+    if (i != strip_header->buffers.end()) {
       (*out)[*it].swap(i->second);
     } else {
       need_lookup.insert(*it);
@@ -297,8 +309,8 @@ int KeyValueStore::BufferTransaction::get_buffer_keys(
     int r = store->backend->get_values_with_header(strip_header, prefix,
                                                    need_lookup, out);
     if (r < 0) {
-      dout(10) << __func__  << " " << strip_header.cid << "/"
-               << strip_header.oid << " " << " r = " << r << dendl;
+      dout(10) << __func__  << " " << strip_header->cid << "/"
+               << strip_header->oid << " " << " r = " << r << dendl;
       return r;
     }
   }
@@ -307,74 +319,78 @@ int KeyValueStore::BufferTransaction::get_buffer_keys(
 }
 
 void KeyValueStore::BufferTransaction::set_buffer_keys(
-     StripObjectMap::StripObjectHeader &strip_header,
+     StripObjectMap::StripObjectHeaderRef strip_header,
      const string &prefix, map<string, bufferlist> &values)
 {
-  store->backend->set_keys(strip_header.header, prefix, values, t);
+  store->backend->set_keys(strip_header->header, prefix, values, t);
 
   for (map<string, bufferlist>::iterator iter = values.begin();
        iter != values.end(); ++iter) {
-    strip_header.buffers[make_pair(prefix, iter->first)].swap(iter->second);
+    strip_header->buffers[make_pair(prefix, iter->first)].swap(iter->second);
   }
 }
 
 int KeyValueStore::BufferTransaction::remove_buffer_keys(
-     StripObjectMap::StripObjectHeader &strip_header, const string &prefix,
+     StripObjectMap::StripObjectHeaderRef strip_header, const string &prefix,
      const set<string> &keys)
 {
   for (set<string>::iterator iter = keys.begin(); iter != keys.end(); ++iter) {
-    strip_header.buffers[make_pair(prefix, *iter)] = bufferlist();
+    strip_header->buffers[make_pair(prefix, *iter)] = bufferlist();
   }
 
-  return store->backend->rm_keys(strip_header.header, prefix, keys, t);
+  return store->backend->rm_keys(strip_header->header, prefix, keys, t);
 }
 
 void KeyValueStore::BufferTransaction::clear_buffer_keys(
-     StripObjectMap::StripObjectHeader &strip_header, const string &prefix)
+     StripObjectMap::StripObjectHeaderRef strip_header, const string &prefix)
 {
-  for (map<pair<string, string>, bufferlist>::iterator iter = strip_header.buffers.begin();
-       iter != strip_header.buffers.end(); ++iter) {
+  for (map<pair<string, string>, bufferlist>::iterator iter = strip_header->buffers.begin();
+       iter != strip_header->buffers.end(); ++iter) {
     if (iter->first.first == prefix)
       iter->second = bufferlist();
   }
 }
 
 int KeyValueStore::BufferTransaction::clear_buffer(
-     StripObjectMap::StripObjectHeader &strip_header)
+     StripObjectMap::StripObjectHeaderRef strip_header)
 {
-  strip_header.deleted = true;
+  strip_header->deleted = true;
+
+  InvalidCacheContext *c = new InvalidCacheContext(store, strip_header->oid);
+  finishes.push_back(c);
 
-  return store->backend->clear(strip_header.header, t);
+  return store->backend->clear(strip_header->header, t);
 }
 
 void KeyValueStore::BufferTransaction::clone_buffer(
-    StripObjectMap::StripObjectHeader &old_header,
+    StripObjectMap::StripObjectHeaderRef old_header,
     const coll_t &cid, const ghobject_t &oid)
 {
   // Remove target ahead to avoid dead lock
   strip_headers.erase(make_pair(cid, oid));
 
-  StripObjectMap::StripObjectHeader new_origin_header, new_target_header;
+  StripObjectMap::StripObjectHeaderRef new_target_header;
 
-  store->backend->clone_wrap(old_header, cid, oid, t,
-                             &new_origin_header, &new_target_header);
+  store->backend->clone_wrap(old_header, cid, oid, t, &new_target_header);
 
   // FIXME: Lacking of lock for origin header(now become parent), it will
   // cause other operation can get the origin header while submitting
   // transactions
-  strip_headers[make_pair(cid, old_header.oid)] = new_origin_header;
   strip_headers[make_pair(cid, oid)] = new_target_header;
 }
 
 void KeyValueStore::BufferTransaction::rename_buffer(
-    StripObjectMap::StripObjectHeader &old_header,
+    StripObjectMap::StripObjectHeaderRef old_header,
     const coll_t &cid, const ghobject_t &oid)
 {
   // FIXME: Lacking of lock for origin header, it will cause other operation
   // can get the origin header while submitting transactions
-  store->backend->rename_wrap(cid, oid, t, &old_header);
+  store->backend->rename_wrap(cid, oid, t, old_header);
 
-  strip_headers.erase(make_pair(old_header.cid, old_header.oid));
+  InvalidCacheContext *c = new InvalidCacheContext(store, oid);
+  finishes.push_back(c);
+
+  strip_headers.erase(make_pair(old_header->cid, old_header->oid));
   strip_headers[make_pair(cid, oid)] = old_header;
 }
 
@@ -384,9 +400,9 @@ int KeyValueStore::BufferTransaction::submit_transaction()
 
   for (StripHeaderMap::iterator header_iter = strip_headers.begin();
        header_iter != strip_headers.end(); ++header_iter) {
-    StripObjectMap::StripObjectHeader header = header_iter->second;
+    StripObjectMap::StripObjectHeaderRef header = header_iter->second;
 
-    if (header.deleted)
+    if (header->deleted)
       continue;
 
     r = store->backend->save_strip_header(header, t);
@@ -398,6 +414,10 @@ int KeyValueStore::BufferTransaction::submit_transaction()
   }
 
 out:
+  for (list<Context*>::iterator it = finishes.begin();
+       it != finishes.end(); ++it) {
+    (*it)->complete(r);
+  }
 
   dout(5) << __func__ << " r = " << r << dendl;
   return store->backend->submit_transaction(t);
@@ -1467,9 +1487,9 @@ bool KeyValueStore::exists(coll_t cid, const ghobject_t& oid)
   dout(10) << __func__ << "collection: " << cid << " object: " << oid
            << dendl;
   int r;
-  StripObjectMap::StripObjectHeader header;
+  StripObjectMap::StripObjectHeaderRef header;
 
-  r = backend->lookup_strip_header(cid, oid, header);
+  r = backend->lookup_strip_header(cid, oid, &header);
   if (r < 0) {
     return false;
   }
@@ -1482,42 +1502,42 @@ int KeyValueStore::stat(coll_t cid, const ghobject_t& oid,
 {
   dout(10) << "stat " << cid << "/" << oid << dendl;
 
-  StripObjectMap::StripObjectHeader header;
+  StripObjectMap::StripObjectHeaderRef header;
 
-  int r = backend->lookup_strip_header(cid, oid, header);
+  int r = backend->lookup_strip_header(cid, oid, &header);
   if (r < 0) {
     dout(10) << "stat " << cid << "/" << oid << "=" << r << dendl;
     return -ENOENT;
   }
 
-  st->st_blocks = header.max_size / header.strip_size;
-  if (header.max_size % header.strip_size)
+  st->st_blocks = header->max_size / header->strip_size;
+  if (header->max_size % header->strip_size)
     st->st_blocks++;
   st->st_nlink = 1;
-  st->st_size = header.max_size;
-  st->st_blksize = header.strip_size;
+  st->st_size = header->max_size;
+  st->st_blksize = header->strip_size;
 
   return r;
 }
 
-int KeyValueStore::_generic_read(StripObjectMap::StripObjectHeader &header,
+int KeyValueStore::_generic_read(StripObjectMap::StripObjectHeaderRef header,
                                  uint64_t offset, size_t len, bufferlist& bl,
                                  bool allow_eio, BufferTransaction *bt)
 {
-  if (header.max_size < offset) {
-    dout(10) << __func__ << " " << header.cid << "/" << header.oid << ")"
+  if (header->max_size < offset) {
+    dout(10) << __func__ << " " << header->cid << "/" << header->oid << ")"
              << " offset exceed the length of bl"<< dendl;
     return 0;
   }
 
   if (len == 0)
-    len = header.max_size - offset;
+    len = header->max_size - offset;
 
-  if (offset + len > header.max_size)
-    len = header.max_size - offset;
+  if (offset + len > header->max_size)
+    len = header->max_size - offset;
 
   vector<StripObjectMap::StripExtent> extents;
-  StripObjectMap::file_to_extents(offset, len, header.strip_size,
+  StripObjectMap::file_to_extents(offset, len, header->strip_size,
                                   extents);
   map<string, bufferlist> out;
   set<string> keys;
@@ -1527,23 +1547,23 @@ int KeyValueStore::_generic_read(StripObjectMap::StripObjectHeader &header,
     bufferlist old;
     string key = strip_object_key(iter->no);
 
-    if (bt && header.buffers.count(make_pair(OBJECT_STRIP_PREFIX, key))) {
+    if (bt && header->buffers.count(make_pair(OBJECT_STRIP_PREFIX, key))) {
       // use strip_header buffer
-      assert(header.bits[iter->no]);
-      out[key] = header.buffers[make_pair(OBJECT_STRIP_PREFIX, key)];
-    } else if (header.bits[iter->no]) {
+      assert(header->bits[iter->no]);
+      out[key] = header->buffers[make_pair(OBJECT_STRIP_PREFIX, key)];
+    } else if (header->bits[iter->no]) {
       keys.insert(key);
     }
   }
 
   int r = backend->get_values_with_header(header, OBJECT_STRIP_PREFIX, keys, &out);
   if (r < 0) {
-    dout(10) << __func__ << " " << header.cid << "/" << header.oid << " "
+    dout(10) << __func__ << " " << header->cid << "/" << header->oid << " "
              << offset << "~" << len << " = " << r << dendl;
     return r;
   } else if (out.size() != keys.size()) {
     dout(0) << __func__ << " broken header or missing data in backend "
-            << header.cid << "/" << header.oid << " " << offset << "~"
+            << header->cid << "/" << header->oid << " " << offset << "~"
             << len << " = " << r << dendl;
     return -EBADF;
   }
@@ -1552,8 +1572,8 @@ int KeyValueStore::_generic_read(StripObjectMap::StripObjectHeader &header,
        iter != extents.end(); ++iter) {
     string key = strip_object_key(iter->no);
 
-    if (header.bits[iter->no]) {
-      if (iter->len == header.strip_size) {
+    if (header->bits[iter->no]) {
+      if (iter->len == header->strip_size) {
         bl.claim_append(out[key]);
       } else {
         out[key].copy(iter->offset, iter->len, bl);
@@ -1563,7 +1583,7 @@ int KeyValueStore::_generic_read(StripObjectMap::StripObjectHeader &header,
     }
   }
 
-  dout(10) << __func__ << " " << header.cid << "/" << header.oid << " "
+  dout(10) << __func__ << " " << header->cid << "/" << header->oid << " "
            << offset << "~" << bl.length() << "/" << len << " r = " << r
            << dendl;
 
@@ -1577,9 +1597,9 @@ int KeyValueStore::read(coll_t cid, const ghobject_t& oid, uint64_t offset,
   dout(15) << __func__ << " " << cid << "/" << oid << " " << offset << "~"
            << len << dendl;
 
-  StripObjectMap::StripObjectHeader header;
+  StripObjectMap::StripObjectHeaderRef header;
 
-  int r = backend->lookup_strip_header(cid, oid, header);
+  int r = backend->lookup_strip_header(cid, oid, &header);
 
   if (r < 0) {
     dout(10) << __func__ << " " << cid << "/" << oid << " " << offset << "~"
@@ -1596,9 +1616,9 @@ int KeyValueStore::fiemap(coll_t cid, const ghobject_t& oid,
   dout(10) << __func__ << " " << cid << " " << oid << " " << offset << "~"
            << len << dendl;
   int r;
-  StripObjectMap::StripObjectHeader header;
+  StripObjectMap::StripObjectHeaderRef header;
 
-  r = backend->lookup_strip_header(cid, oid, header);
+  r = backend->lookup_strip_header(cid, oid, &header);
   if (r < 0) {
     dout(10) << "fiemap " << cid << "/" << oid << " " << offset << "~" << len
              << " failed to get header: r = " << r << dendl;
@@ -1606,7 +1626,7 @@ int KeyValueStore::fiemap(coll_t cid, const ghobject_t& oid,
   }
 
   vector<StripObjectMap::StripExtent> extents;
-  StripObjectMap::file_to_extents(offset, len, header.strip_size,
+  StripObjectMap::file_to_extents(offset, len, header->strip_size,
                                   extents);
 
   map<uint64_t, uint64_t> m;
@@ -1625,7 +1645,7 @@ int KeyValueStore::_remove(coll_t cid, const ghobject_t& oid,
   dout(15) << __func__ << " " << cid << "/" << oid << dendl;
 
   int r;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   r = t.lookup_cached_header(cid, oid, &header, false);
   if (r < 0) {
@@ -1636,7 +1656,7 @@ int KeyValueStore::_remove(coll_t cid, const ghobject_t& oid,
 
   header->max_size = 0;
   header->bits.clear();
-  r = t.clear_buffer(*header);
+  r = t.clear_buffer(header);
 
   dout(10) << __func__ << " " << cid << "/" << oid << " = " << r << dendl;
   return r;
@@ -1649,7 +1669,7 @@ int KeyValueStore::_truncate(coll_t cid, const ghobject_t& oid, uint64_t size,
            << dendl;
 
   int r;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   r = t.lookup_cached_header(cid, oid, &header, false);
   if (r < 0) {
@@ -1675,7 +1695,7 @@ int KeyValueStore::_truncate(coll_t cid, const ghobject_t& oid, uint64_t size,
       string key = strip_object_key(iter->no);
 
       lookup_keys.insert(key);
-      r = t.get_buffer_keys(*header, OBJECT_STRIP_PREFIX,
+      r = t.get_buffer_keys(header, OBJECT_STRIP_PREFIX,
                             lookup_keys, &values);
       if (r < 0) {
         dout(10) << __func__ << " " << cid << "/" << oid << " "
@@ -1693,7 +1713,7 @@ int KeyValueStore::_truncate(coll_t cid, const ghobject_t& oid, uint64_t size,
       assert(value.length() == header->strip_size);
       value.swap(values[key]);
 
-      t.set_buffer_keys(*header, OBJECT_STRIP_PREFIX, values);
+      t.set_buffer_keys(header, OBJECT_STRIP_PREFIX, values);
       ++iter;
     }
 
@@ -1704,7 +1724,7 @@ int KeyValueStore::_truncate(coll_t cid, const ghobject_t& oid, uint64_t size,
         header->bits[iter->no] = 0;
       }
     }
-    r = t.remove_buffer_keys(*header, OBJECT_STRIP_PREFIX, keys);
+    r = t.remove_buffer_keys(header, OBJECT_STRIP_PREFIX, keys);
     if (r < 0) {
       dout(10) << __func__ << " " << cid << "/" << oid << " "
                << size << " = " << r << dendl;
@@ -1726,7 +1746,7 @@ int KeyValueStore::_touch(coll_t cid, const ghobject_t& oid,
   dout(15) << __func__ << " " << cid << "/" << oid << dendl;
 
   int r;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   r = t.lookup_cached_header(cid, oid, &header, true);
   if (r < 0) {
@@ -1740,7 +1760,7 @@ int KeyValueStore::_touch(coll_t cid, const ghobject_t& oid,
   return r;
 }
 
-int KeyValueStore::_generic_write(StripObjectMap::StripObjectHeader &header,
+int KeyValueStore::_generic_write(StripObjectMap::StripObjectHeaderRef header,
                                   uint64_t offset, size_t len,
                                   const bufferlist& bl, BufferTransaction &t,
                                   bool replica)
@@ -1748,34 +1768,34 @@ int KeyValueStore::_generic_write(StripObjectMap::StripObjectHeader &header,
   if (len > bl.length())
     len = bl.length();
 
-  if (len + offset > header.max_size) {
-    header.max_size = len + offset;
-    header.bits.resize(header.max_size/header.strip_size+1);
+  if (len + offset > header->max_size) {
+    header->max_size = len + offset;
+    header->bits.resize(header->max_size/header->strip_size+1);
   }
 
   vector<StripObjectMap::StripExtent> extents;
-  StripObjectMap::file_to_extents(offset, len, header.strip_size,
+  StripObjectMap::file_to_extents(offset, len, header->strip_size,
                                   extents);
 
   map<string, bufferlist> out;
   set<string> keys;
   for (vector<StripObjectMap::StripExtent>::iterator iter = extents.begin();
        iter != extents.end(); ++iter) {
-    if (header.bits[iter->no] && !(iter->offset == 0 &&
-                                   iter->len == header.strip_size))
+    if (header->bits[iter->no] && !(iter->offset == 0 &&
+                                   iter->len == header->strip_size))
       keys.insert(strip_object_key(iter->no));
   }
 
   int r = t.get_buffer_keys(header, OBJECT_STRIP_PREFIX, keys, &out);
   if (r < 0) {
-    dout(10) << __func__ << " failed to get value " << header.cid << "/"
-              << header.oid << " " << offset << "~" << len << " = " << r
+    dout(10) << __func__ << " failed to get value " << header->cid << "/"
+              << header->oid << " " << offset << "~" << len << " = " << r
               << dendl;
     return r;
   } else if (keys.size() != out.size()) {
     // Error on header.bits or the corresponding key/value pair is missing
     dout(0) << __func__ << " broken header or missing data in backend "
-            << header.cid << "/" << header.oid << " " << offset << "~"
+            << header->cid << "/" << header->oid << " " << offset << "~"
             << len << " = " << r << dendl;
     return -EBADF;
   }
@@ -1786,19 +1806,19 @@ int KeyValueStore::_generic_write(StripObjectMap::StripObjectHeader &header,
        iter != extents.end(); ++iter) {
     bufferlist value;
     string key = strip_object_key(iter->no);
-    if (header.bits[iter->no]) {
-      if (iter->offset == 0 && iter->len == header.strip_size) {
+    if (header->bits[iter->no]) {
+      if (iter->offset == 0 && iter->len == header->strip_size) {
         bl.copy(bl_offset, iter->len, value);
         bl_offset += iter->len;
       } else {
-        assert(out[key].length() == header.strip_size);
+        assert(out[key].length() == header->strip_size);
 
         out[key].copy(0, iter->offset, value);
         bl.copy(bl_offset, iter->len, value);
         bl_offset += iter->len;
 
-        if (value.length() != header.strip_size)
-          out[key].copy(value.length(), header.strip_size-value.length(),
+        if (value.length() != header->strip_size)
+          out[key].copy(value.length(), header->strip_size-value.length(),
                         value);
       }
     } else {
@@ -1807,18 +1827,18 @@ int KeyValueStore::_generic_write(StripObjectMap::StripObjectHeader &header,
       bl.copy(bl_offset, iter->len, value);
       bl_offset += iter->len;
 
-      if (value.length() < header.strip_size)
-        value.append_zero(header.strip_size-value.length());
+      if (value.length() < header->strip_size)
+        value.append_zero(header->strip_size-value.length());
 
-      header.bits[iter->no] = 1;
+      header->bits[iter->no] = 1;
     }
-    assert(value.length() == header.strip_size);
+    assert(value.length() == header->strip_size);
     values[key].swap(value);
   }
   assert(bl_offset == len);
 
   t.set_buffer_keys(header, OBJECT_STRIP_PREFIX, values);
-  dout(10) << __func__ << " " << header.cid << "/" << header.oid << " "
+  dout(10) << __func__ << " " << header->cid << "/" << header->oid << " "
            << offset << "~" << len << " = " << r << dendl;
 
   return r;
@@ -1832,7 +1852,7 @@ int KeyValueStore::_write(coll_t cid, const ghobject_t& oid,
            << len << dendl;
 
   int r;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   r = t.lookup_cached_header(cid, oid, &header, true);
   if (r < 0) {
@@ -1841,7 +1861,7 @@ int KeyValueStore::_write(coll_t cid, const ghobject_t& oid,
     return r;
   }
 
-  return _generic_write(*header, offset, len, bl, t, replica);
+  return _generic_write(header, offset, len, bl, t, replica);
 }
 
 int KeyValueStore::_zero(coll_t cid, const ghobject_t& oid, uint64_t offset,
@@ -1870,7 +1890,7 @@ int KeyValueStore::_clone(coll_t cid, const ghobject_t& oldoid,
     return 0;
 
   int r;
-  StripObjectMap::StripObjectHeader *old_header;
+  StripObjectMap::StripObjectHeaderRef old_header;
 
   r = t.lookup_cached_header(cid, oldoid, &old_header, false);
   if (r < 0) {
@@ -1879,7 +1899,7 @@ int KeyValueStore::_clone(coll_t cid, const ghobject_t& oldoid,
     return r;
   }
 
-  t.clone_buffer(*old_header, cid, newoid);
+  t.clone_buffer(old_header, cid, newoid);
 
   dout(10) << __func__ << " " << cid << "/" << oldoid << " -> " << cid << "/"
            << newoid << " = " << r << dendl;
@@ -1898,7 +1918,7 @@ int KeyValueStore::_clone_range(coll_t cid, const ghobject_t& oldoid,
   int r;
   bufferlist bl;
 
-  StripObjectMap::StripObjectHeader *old_header, *new_header;
+  StripObjectMap::StripObjectHeaderRef old_header, new_header;
 
   r = t.lookup_cached_header(cid, oldoid, &old_header, false);
   if (r < 0) {
@@ -1916,11 +1936,11 @@ int KeyValueStore::_clone_range(coll_t cid, const ghobject_t& oldoid,
     return r;
   }
 
-  r = _generic_read(*old_header, srcoff, len, bl, &t);
+  r = _generic_read(old_header, srcoff, len, bl, &t);
   if (r < 0)
     goto out;
 
-  r = _generic_write(*new_header, dstoff, len, bl, t);
+  r = _generic_write(new_header, dstoff, len, bl, t);
 
  out:
   dout(10) << __func__ << " " << cid << "/" << oldoid << " -> " << cid << "/"
@@ -1940,9 +1960,17 @@ int KeyValueStore::getattr(coll_t cid, const ghobject_t& oid, const char *name,
   int r;
   map<string, bufferlist> got;
   set<string> to_get;
+  StripObjectMap::StripObjectHeaderRef header;
 
   to_get.insert(string(name));
-  r = backend->get_values(cid, oid, OBJECT_XATTR, to_get, &got);
+
+  r = backend->lookup_strip_header(cid, oid, &header);
+  if (r < 0) {
+    dout(10) << __func__ << " lookup_strip_header failed: r =" << r << dendl;
+    return r;
+  }
+
+  r = backend->get_values_with_header(header, OBJECT_XATTR, to_get, &got);
   if (r < 0 && r != -ENOENT) {
     dout(10) << __func__ << " get_xattrs err r =" << r << dendl;
     goto out;
@@ -1998,7 +2026,7 @@ int KeyValueStore::_setattrs(coll_t cid, const ghobject_t& oid,
 
   int r;
 
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
   map<string, bufferlist> attrs;
 
   r = t.lookup_cached_header(cid, oid, &header, false);
@@ -2010,7 +2038,7 @@ int KeyValueStore::_setattrs(coll_t cid, const ghobject_t& oid,
     attrs[it->first].push_back(it->second);
   }
 
-  t.set_buffer_keys(*header, OBJECT_XATTR, attrs);
+  t.set_buffer_keys(header, OBJECT_XATTR, attrs);
 
 out:
   dout(10) << __func__ << " " << cid << "/" << oid << " = " << r << dendl;
@@ -2026,7 +2054,7 @@ int KeyValueStore::_rmattr(coll_t cid, const ghobject_t& oid, const char *name,
 
   int r;
   set<string> to_remove;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   r = t.lookup_cached_header(cid, oid, &header, false);
   if (r < 0) {
@@ -2036,7 +2064,7 @@ int KeyValueStore::_rmattr(coll_t cid, const ghobject_t& oid, const char *name,
   }
 
   to_remove.insert(string(name));
-  r = t.remove_buffer_keys(*header, OBJECT_XATTR, to_remove);
+  r = t.remove_buffer_keys(header, OBJECT_XATTR, to_remove);
 
   dout(10) << __func__ << " " << cid << "/" << oid << " '" << name << "' = "
            << r << dendl;
@@ -2051,7 +2079,7 @@ int KeyValueStore::_rmattrs(coll_t cid, const ghobject_t& oid,
   int r;
   set<string> attrs;
 
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   r = t.lookup_cached_header(cid, oid, &header, false);
   if (r < 0) {
@@ -2060,14 +2088,14 @@ int KeyValueStore::_rmattrs(coll_t cid, const ghobject_t& oid,
     return r;
   }
 
-  r = backend->get_keys_with_header(*header, OBJECT_XATTR, &attrs);
+  r = backend->get_keys_with_header(header, OBJECT_XATTR, &attrs);
   if (r < 0 && r != -ENOENT) {
     dout(10) << __func__ << " could not get attrs r = " << r << dendl;
     return r;
   }
 
-  r = t.remove_buffer_keys(*header, OBJECT_XATTR, attrs);
-  t.clear_buffer_keys(*header, OBJECT_XATTR);
+  r = t.remove_buffer_keys(header, OBJECT_XATTR, attrs);
+  t.clear_buffer_keys(header, OBJECT_XATTR);
 
   dout(10) << __func__ <<  " " << cid << "/" << oid << " = " << r << dendl;
   return r;
@@ -2110,10 +2138,18 @@ int KeyValueStore::collection_getattr(coll_t c, const char *name,
 
   set<string> keys;
   map<string, bufferlist> out;
+  StripObjectMap::StripObjectHeaderRef header;
+
   keys.insert(string(name));
 
-  int r = backend->get_values(get_coll_for_coll(), make_ghobject_for_coll(c),
-                              COLLECTION_ATTR, keys, &out);
+  int r = backend->lookup_strip_header(get_coll_for_coll(),
+                                       make_ghobject_for_coll(c), &header);
+  if (r < 0) {
+    dout(10) << __func__ << " lookup_strip_header failed: r =" << r << dendl;
+    return r;
+  }
+
+  r = backend->get_values_with_header(header, COLLECTION_ATTR, keys, &out);
   if (r < 0) {
     dout(10) << __func__ << " could not get key" << string(name) << dendl;
     r = -EINVAL;
@@ -2134,14 +2170,21 @@ int KeyValueStore::collection_getattrs(coll_t cid,
 
   map<string, bufferlist> out;
   set<string> keys;
+  StripObjectMap::StripObjectHeaderRef header;
 
   for (map<string, bufferptr>::iterator it = aset.begin();
        it != aset.end(); ++it) {
       keys.insert(it->first);
   }
 
-  int r = backend->get_values(get_coll_for_coll(), make_ghobject_for_coll(cid),
-                              COLLECTION_ATTR, keys, &out);
+  int r = backend->lookup_strip_header(get_coll_for_coll(),
+                                       make_ghobject_for_coll(cid), &header);
+  if (r < 0) {
+    dout(10) << __func__ << " lookup_strip_header failed: r =" << r << dendl;
+    return r;
+  }
+
+  r = backend->get_values_with_header(header, COLLECTION_ATTR, keys, &out);
   if (r < 0) {
     dout(10) << __func__ << " could not get keys" << dendl;
     r = -EINVAL;
@@ -2169,7 +2212,7 @@ int KeyValueStore::_collection_setattr(coll_t c, const char *name,
   int r;
   bufferlist bl;
   map<string, bufferlist> out;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   r = t.lookup_cached_header(get_coll_for_coll(),
                              make_ghobject_for_coll(c),
@@ -2182,7 +2225,7 @@ int KeyValueStore::_collection_setattr(coll_t c, const char *name,
   bl.append(reinterpret_cast<const char*>(value), size);
   out.insert(make_pair(string(name), bl));
 
-  t.set_buffer_keys(*header, COLLECTION_ATTR, out);
+  t.set_buffer_keys(header, COLLECTION_ATTR, out);
 
   dout(10) << __func__ << " " << c << " '"
            << name << "' len " << size << " = " << r << dendl;
@@ -2196,7 +2239,7 @@ int KeyValueStore::_collection_rmattr(coll_t c, const char *name,
 
   bufferlist bl;
   set<string> out;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   int r = t.lookup_cached_header(get_coll_for_coll(),
                                  make_ghobject_for_coll(c), &header, false);
@@ -2206,7 +2249,7 @@ int KeyValueStore::_collection_rmattr(coll_t c, const char *name,
   }
 
   out.insert(string(name));
-  r = t.remove_buffer_keys(*header, COLLECTION_ATTR, out);
+  r = t.remove_buffer_keys(header, COLLECTION_ATTR, out);
 
   dout(10) << __func__ << " " << c << " = " << r << dendl;
   return r;
@@ -2219,7 +2262,7 @@ int KeyValueStore::_collection_setattrs(coll_t cid,
   dout(15) << __func__ << " " << cid << dendl;
 
   map<string, bufferlist> attrs;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
   int r = t.lookup_cached_header(get_coll_for_coll(),
                                  make_ghobject_for_coll(cid),
                                  &header, false);
@@ -2233,7 +2276,7 @@ int KeyValueStore::_collection_setattrs(coll_t cid,
     attrs[it->first].push_back(it->second);
   }
 
-  t.set_buffer_keys(*header, COLLECTION_ATTR, attrs);
+  t.set_buffer_keys(header, COLLECTION_ATTR, attrs);
 
   dout(10) << __func__ << " " << cid << " = " << r << dendl;
   return r;
@@ -2247,7 +2290,7 @@ int KeyValueStore::_create_collection(coll_t c, BufferTransaction &t)
   dout(15) << __func__ << " " << c << dendl;
 
   int r;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
   bufferlist bl;
 
   r = t.lookup_cached_header(get_coll_for_coll(),
@@ -2272,7 +2315,7 @@ int KeyValueStore::_destroy_collection(coll_t c, BufferTransaction &t)
 
   int r;
   uint64_t modified_object = 0;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
   vector<ghobject_t> oids;
 
   r = t.lookup_cached_header(get_coll_for_coll(), make_ghobject_for_coll(c),
@@ -2289,7 +2332,7 @@ int KeyValueStore::_destroy_collection(coll_t c, BufferTransaction &t)
       continue;
 
     modified_object++;
-    if (!iter->second.deleted) {
+    if (!iter->second->deleted) {
       r = -ENOTEMPTY;
       goto out;
     }
@@ -2311,7 +2354,7 @@ int KeyValueStore::_destroy_collection(coll_t c, BufferTransaction &t)
     }
   }
 
-  r = t.clear_buffer(*header);
+  r = t.clear_buffer(header);
 
 out:
   dout(10) << __func__ << " " << c << " = " << r << dendl;
@@ -2327,7 +2370,7 @@ int KeyValueStore::_collection_add(coll_t c, coll_t oldcid,
            << o << dendl;
 
   bufferlist bl;
-  StripObjectMap::StripObjectHeader *header, *old_header;
+  StripObjectMap::StripObjectHeaderRef header, old_header;
 
   int r = t.lookup_cached_header(oldcid, o, &old_header, false);
   if (r < 0) {
@@ -2342,13 +2385,13 @@ int KeyValueStore::_collection_add(coll_t c, coll_t oldcid,
     goto out;
   }
 
-  r = _generic_read(*old_header, 0, old_header->max_size, bl, &t);
+  r = _generic_read(old_header, 0, old_header->max_size, bl, &t);
   if (r < 0) {
     r = -EINVAL;
     goto out;
   }
 
-  r = _generic_write(*header, 0, bl.length(), bl, t);
+  r = _generic_write(header, 0, bl.length(), bl, t);
   if (r < 0) {
     r = -EINVAL;
   }
@@ -2367,7 +2410,7 @@ int KeyValueStore::_collection_move_rename(coll_t oldcid,
   dout(15) << __func__ << " " << c << "/" << o << " from " << oldcid << "/"
            << oldoid << dendl;
   int r;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   r = t.lookup_cached_header(c, o, &header, false);
   if (r == 0) {
@@ -2383,7 +2426,7 @@ int KeyValueStore::_collection_move_rename(coll_t oldcid,
     return r;
   }
 
-  t.rename_buffer(*header, c, o);
+  t.rename_buffer(header, c, o);
 
   dout(10) << __func__ << " " << c << "/" << o << " from " << oldcid << "/"
            << oldoid << " = " << r << dendl;
@@ -2395,7 +2438,7 @@ int KeyValueStore::_collection_remove_recursive(const coll_t &cid,
 {
   dout(15) << __func__ << " " << cid << dendl;
 
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   int r = t.lookup_cached_header(get_coll_for_coll(),
                                  make_ghobject_for_coll(cid),
@@ -2420,7 +2463,7 @@ int KeyValueStore::_collection_remove_recursive(const coll_t &cid,
     }
   }
 
-  r = t.clear_buffer(*header);
+  r = t.clear_buffer(header);
 
   dout(10) << __func__ << " " << cid  << " r = " << r << dendl;
   return 0;
@@ -2432,7 +2475,7 @@ int KeyValueStore::_collection_rename(const coll_t &cid, const coll_t &ncid,
   dout(10) << __func__ << " origin cid " << cid << " new cid " << ncid
            << dendl;
 
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   int r = t.lookup_cached_header(get_coll_for_coll(),
                                  make_ghobject_for_coll(ncid),
@@ -2474,7 +2517,7 @@ int KeyValueStore::_collection_rename(const coll_t &cid, const coll_t &ncid,
     current = next;
   }
 
-  t.rename_buffer(*header, get_coll_for_coll(), make_ghobject_for_coll(ncid));
+  t.rename_buffer(header, get_coll_for_coll(), make_ghobject_for_coll(ncid));
 
   dout(10) << __func__ << " origin cid " << cid << " new cid " << ncid
            << dendl;
@@ -2502,9 +2545,9 @@ bool KeyValueStore::collection_exists(coll_t c)
 {
   dout(10) << __func__ << " " << dendl;
 
-  StripObjectMap::StripObjectHeader header;
+  StripObjectMap::StripObjectHeaderRef header;
   int r = backend->lookup_strip_header(get_coll_for_coll(),
-                                       make_ghobject_for_coll(c), header);
+                                       make_ghobject_for_coll(c), &header);
   if (r < 0) {
     return false;
   }
@@ -2594,15 +2637,14 @@ int KeyValueStore::omap_get(coll_t c, const ghobject_t &hoid,
 {
   dout(15) << __func__ << " " << c << "/" << hoid << dendl;
 
-  StripObjectMap::StripObjectHeader header;
+  StripObjectMap::StripObjectHeaderRef header;
 
-  int r = backend->lookup_strip_header(c, hoid, header);
+  int r = backend->lookup_strip_header(c, hoid, &header);
   if (r < 0) {
     dout(10) << __func__ << " lookup_strip_header failed: r =" << r << dendl;
     return r;
   }
 
-
   r = backend->get_with_header(header, OBJECT_OMAP, out);
   if (r < 0 && r != -ENOENT) {
     dout(10) << __func__ << " err r =" << r << dendl;
@@ -2634,9 +2676,16 @@ int KeyValueStore::omap_get_header(coll_t c, const ghobject_t &hoid,
 
   set<string> keys;
   map<string, bufferlist> got;
+  StripObjectMap::StripObjectHeaderRef header;
+
+  int r = backend->lookup_strip_header(c, hoid, &header);
+  if (r < 0) {
+    dout(10) << __func__ << " lookup_strip_header failed: r =" << r << dendl;
+    return r;
+  }
 
   keys.insert(OBJECT_OMAP_HEADER_KEY);
-  int r = backend->get_values(c, hoid, OBJECT_OMAP_HEADER, keys, &got);
+  r = backend->get_values_with_header(header, OBJECT_OMAP_HEADER, keys, &got);
   if (r < 0 && r != -ENOENT) {
     dout(10) << __func__ << " err r =" << r << dendl;
     return r;
@@ -2654,7 +2703,14 @@ int KeyValueStore::omap_get_keys(coll_t c, const ghobject_t &hoid, set<string> *
 {
   dout(15) << __func__ << " " << c << "/" << hoid << dendl;
 
-  int r = backend->get_keys(c, hoid, OBJECT_OMAP, keys);
+  StripObjectMap::StripObjectHeaderRef header;
+  int r = backend->lookup_strip_header(c, hoid, &header);
+  if (r < 0) {
+    dout(10) << __func__ << " lookup_strip_header failed: r =" << r << dendl;
+    return r;
+  }
+
+  r = backend->get_keys_with_header(header, OBJECT_OMAP, keys);
   if (r < 0 && r != -ENOENT) {
     return r;
   }
@@ -2667,7 +2723,14 @@ int KeyValueStore::omap_get_values(coll_t c, const ghobject_t &hoid,
 {
   dout(15) << __func__ << " " << c << "/" << hoid << dendl;
 
-  int r = backend->get_values(c, hoid, OBJECT_OMAP, keys, out);
+  StripObjectMap::StripObjectHeaderRef header;
+  int r = backend->lookup_strip_header(c, hoid, &header);
+  if (r < 0) {
+    dout(10) << __func__ << " lookup_strip_header failed: r =" << r << dendl;
+    return r;
+  }
+
+  r = backend->get_values_with_header(header, OBJECT_OMAP, keys, out);
   if (r < 0 && r != -ENOENT) {
     return r;
   }
@@ -2698,7 +2761,7 @@ int KeyValueStore::_omap_clear(coll_t cid, const ghobject_t &hoid,
 {
   dout(15) << __func__ << " " << cid << "/" << hoid << dendl;
 
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   int r = t.lookup_cached_header(cid, hoid, &header, false);
   if (r < 0) {
@@ -2708,13 +2771,13 @@ int KeyValueStore::_omap_clear(coll_t cid, const ghobject_t &hoid,
   }
 
   set<string> keys;
-  r = backend->get_keys_with_header(*header, OBJECT_OMAP, &keys);
+  r = backend->get_keys_with_header(header, OBJECT_OMAP, &keys);
   if (r < 0 && r != -ENOENT) {
     dout(10) << __func__ << " could not get omap_keys r = " << r << dendl;
     return r;
   }
 
-  r = t.remove_buffer_keys(*header, OBJECT_OMAP, keys);
+  r = t.remove_buffer_keys(header, OBJECT_OMAP, keys);
   if (r < 0) {
     dout(10) << __func__ << " could not remove keys r = " << r << dendl;
     return r;
@@ -2722,13 +2785,13 @@ int KeyValueStore::_omap_clear(coll_t cid, const ghobject_t &hoid,
 
   keys.clear();
   keys.insert(OBJECT_OMAP_HEADER_KEY);
-  r = t.remove_buffer_keys(*header, OBJECT_OMAP_HEADER, keys);
+  r = t.remove_buffer_keys(header, OBJECT_OMAP_HEADER, keys);
   if (r < 0) {
     dout(10) << __func__ << " could not remove keys r = " << r << dendl;
     return r;
   }
 
-  t.clear_buffer_keys(*header, OBJECT_OMAP_HEADER);
+  t.clear_buffer_keys(header, OBJECT_OMAP_HEADER);
 
   dout(10) << __func__ << " " << cid << "/" << hoid << " r = " << r << dendl;
   return 0;
@@ -2740,7 +2803,7 @@ int KeyValueStore::_omap_setkeys(coll_t cid, const ghobject_t &hoid,
 {
   dout(15) << __func__ << " " << cid << "/" << hoid << dendl;
 
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   int r = t.lookup_cached_header(cid, hoid, &header, false);
   if (r < 0) {
@@ -2749,7 +2812,7 @@ int KeyValueStore::_omap_setkeys(coll_t cid, const ghobject_t &hoid,
     return r;
   }
 
-  t.set_buffer_keys(*header, OBJECT_OMAP, aset);
+  t.set_buffer_keys(header, OBJECT_OMAP, aset);
 
   return 0;
 }
@@ -2760,7 +2823,7 @@ int KeyValueStore::_omap_rmkeys(coll_t cid, const ghobject_t &hoid,
 {
   dout(15) << __func__ << " " << cid << "/" << hoid << dendl;
 
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   int r = t.lookup_cached_header(cid, hoid, &header, false);
   if (r < 0) {
@@ -2769,7 +2832,7 @@ int KeyValueStore::_omap_rmkeys(coll_t cid, const ghobject_t &hoid,
     return r;
   }
 
-  r = t.remove_buffer_keys(*header, OBJECT_OMAP, keys);
+  r = t.remove_buffer_keys(header, OBJECT_OMAP, keys);
 
   dout(10) << __func__ << " " << cid << "/" << hoid << " r = " << r << dendl;
   return r;
@@ -2803,7 +2866,7 @@ int KeyValueStore::_omap_setheader(coll_t cid, const ghobject_t &hoid,
   dout(15) << __func__ << " " << cid << "/" << hoid << dendl;
 
   map<string, bufferlist> sets;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef header;
 
   int r = t.lookup_cached_header(cid, hoid, &header, false);
   if (r < 0) {
@@ -2813,7 +2876,7 @@ int KeyValueStore::_omap_setheader(coll_t cid, const ghobject_t &hoid,
   }
 
   sets[OBJECT_OMAP_HEADER_KEY] = bl;
-  t.set_buffer_keys(*header, OBJECT_OMAP_HEADER, sets);
+  t.set_buffer_keys(header, OBJECT_OMAP_HEADER, sets);
   return 0;
 }
 
@@ -2823,7 +2886,7 @@ int KeyValueStore::_split_collection(coll_t cid, uint32_t bits, uint32_t rem,
   {
     dout(15) << __func__ << " " << cid << " bits: " << bits << dendl;
 
-    StripObjectMap::StripObjectHeader *header;
+    StripObjectMap::StripObjectHeaderRef header;
 
     int r = t.lookup_cached_header(get_coll_for_coll(),
                                    make_ghobject_for_coll(cid),
@@ -2919,7 +2982,7 @@ int KeyValueStore::_set_alloc_hint(coll_t cid, const ghobject_t& oid,
            << expected_write_size << dendl;
 
   int r = 0;
-  StripObjectMap::StripObjectHeader *header;
+  StripObjectMap::StripObjectHeaderRef *header;
 
   r = t.lookup_cached_header(cid, oid, &header, false);
   if (r < 0) {

diff --git a/src/os/KeyValueStore.h b/src/os/KeyValueStore.h
index 59e01a9c4a1a32920247d48b2ab77fb08fb030c1..2d0b39b4524c1e8c358527e076d837266ecc6beb 100644 (file)
--- a/src/os/KeyValueStore.h
+++ b/src/os/KeyValueStore.h
@@ -37,6 +37,7 @@ using namespace std;
 #include "common/Mutex.h"
 #include "GenericObjectMap.h"
 #include "KeyValueDB.h"
+#include "common/shared_cache.hpp"
 
 #include "include/uuid.h"
 
@@ -93,43 +94,53 @@ class StripObjectMap: public GenericObjectMap {
       DECODE_FINISH(bl);
     }
   };
+  typedef ceph::shared_ptr<StripObjectHeader> StripObjectHeaderRef;
 
   static int file_to_extents(uint64_t offset, size_t len, uint64_t strip_size,
                              vector<StripExtent> &extents);
   int lookup_strip_header(const coll_t & cid, const ghobject_t &oid,
-                          StripObjectHeader &header);
-  int save_strip_header(StripObjectHeader &header, KeyValueDB::Transaction t);
+                          StripObjectHeaderRef *header);
+  int save_strip_header(StripObjectHeaderRef header, KeyValueDB::Transaction t);
   int create_strip_header(const coll_t &cid, const ghobject_t &oid,
-                          StripObjectHeader &strip_header,
+                          StripObjectHeaderRef *strip_header,
                           KeyValueDB::Transaction t);
-  void clone_wrap(StripObjectHeader &old_header,
+  void clone_wrap(StripObjectHeaderRef old_header,
                   const coll_t &cid, const ghobject_t &oid,
                   KeyValueDB::Transaction t,
-                  StripObjectHeader *origin_header,
-                  StripObjectHeader *target_header);
+                  StripObjectHeaderRef *target_header);
   void rename_wrap(const coll_t &cid, const ghobject_t &oid,
                    KeyValueDB::Transaction t,
-                   StripObjectHeader *header);
+                   StripObjectHeaderRef header);
   // Already hold header to avoid lock header seq again
   int get_with_header(
-    const StripObjectHeader &header,
+    const StripObjectHeaderRef header,
     const string &prefix,
     map<string, bufferlist> *out
     );
 
   int get_values_with_header(
-    const StripObjectHeader &header,
+    const StripObjectHeaderRef header,
     const string &prefix,
     const set<string> &keys,
     map<string, bufferlist> *out
     );
   int get_keys_with_header(
-    const StripObjectHeader &header,
+    const StripObjectHeaderRef header,
     const string &prefix,
     set<string> *keys
     );
 
-  StripObjectMap(KeyValueDB *db): GenericObjectMap(db) {}
+  Mutex lock;
+  void invalidate_cache(const ghobject_t &oid) {
+    Mutex::Locker l(lock);
+    caches.clear(oid);
+  }
+
+  SharedLRU<ghobject_t, _StripObjectHeader> caches;
+  StripObjectMap(KeyValueDB *db): GenericObjectMap(db), lock("StripObjectMap::lock"),
+                                  caches(4096){}
+
+  static const uint64_t default_strip_size = 1024;
 };
 
 
@@ -221,31 +232,32 @@ class KeyValueStore : public ObjectStore,
   // 4. Clone or rename
   struct BufferTransaction {
     typedef pair<coll_t, ghobject_t> uniq_id;
-    typedef map<uniq_id, StripObjectMap::StripObjectHeader> StripHeaderMap;
+    typedef map<uniq_id, StripObjectMap::StripObjectHeaderRef> StripHeaderMap;
 
     //Dirty records
     StripHeaderMap strip_headers;
+    list<Context*> finishes;
 
     KeyValueStore *store;
 
     KeyValueDB::Transaction t;
 
     int lookup_cached_header(const coll_t &cid, const ghobject_t &oid,
-                             StripObjectMap::StripObjectHeader **strip_header,
+                             StripObjectMap::StripObjectHeaderRef *strip_header,
                              bool create_if_missing);
-    int get_buffer_keys(StripObjectMap::StripObjectHeader &strip_header,
+    int get_buffer_keys(StripObjectMap::StripObjectHeaderRef strip_header,
                         const string &prefix, const set<string> &keys,
                         map<string, bufferlist> *out);
-    void set_buffer_keys(StripObjectMap::StripObjectHeader &strip_header,
+    void set_buffer_keys(StripObjectMap::StripObjectHeaderRef strip_header,
                          const string &prefix, map<string, bufferlist> &bl);
-    int remove_buffer_keys(StripObjectMap::StripObjectHeader &strip_header,
+    int remove_buffer_keys(StripObjectMap::StripObjectHeaderRef strip_header,
                            const string &prefix, const set<string> &keys);
-    void clear_buffer_keys(StripObjectMap::StripObjectHeader &strip_header,
+    void clear_buffer_keys(StripObjectMap::StripObjectHeaderRef strip_header,
                            const string &prefix);
-    int clear_buffer(StripObjectMap::StripObjectHeader &strip_header);
-    void clone_buffer(StripObjectMap::StripObjectHeader &old_header,
+    int clear_buffer(StripObjectMap::StripObjectHeaderRef strip_header);
+    void clone_buffer(StripObjectMap::StripObjectHeaderRef old_header,
                       const coll_t &cid, const ghobject_t &oid);
-    void rename_buffer(StripObjectMap::StripObjectHeader &old_header,
+    void rename_buffer(StripObjectMap::StripObjectHeaderRef old_header,
                        const coll_t &cid, const ghobject_t &oid);
     int submit_transaction();
 
@@ -254,6 +266,17 @@ class KeyValueStore : public ObjectStore,
     }
   };
 
+  struct InvalidCacheContext : public Context {
+    KeyValueStore *store;
+    const ghobject_t object;
+    InvalidCacheContext(KeyValueStore *s,
+                        const ghobject_t &oid) : store(s), object(oid) {}
+    void finish(int r) {
+      if (r >= 0)
+        store->backend->invalidate_cache(object);
+    }
+  };
+
   // -- op workqueue --
   struct Op {
     utime_t start;
@@ -435,10 +458,10 @@ class KeyValueStore : public ObjectStore,
   // ------------------
   // objects
 
-  int _generic_read(StripObjectMap::StripObjectHeader &header,
+  int _generic_read(StripObjectMap::StripObjectHeaderRef header,
                     uint64_t offset, size_t len, bufferlist& bl,
                     bool allow_eio = false, BufferTransaction *bt = 0);
-  int _generic_write(StripObjectMap::StripObjectHeader &header,
+  int _generic_write(StripObjectMap::StripObjectHeaderRef header,
                      uint64_t offset, size_t len, const bufferlist& bl,
                      BufferTransaction &t, bool replica = false);
 
@@ -583,6 +606,6 @@ class KeyValueStore : public ObjectStore,
   static const uint32_t COLLECTION_VERSION = 1;
 };
 
-WRITE_CLASS_ENCODER(StripObjectMap::StripObjectHeader)
+WRITE_CLASS_ENCODER(StripObjectMap::_StripObjectHeader)
 
 #endif