From: Igor Fedotov <ifedotov@mirantis.com>
Date: Fri, 23 Dec 2016 17:00:38 +0000 (+0300)
Subject: os/bluestore: replace Blob's ref_map with reference counting
X-Git-Tag: v12.0.0~30^2~7
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=51a002f456f14d4766e738d65fafe122945c153c;p=ceph.git

os/bluestore: replace Blob's ref_map with reference counting

Signed-off-by: Igor Fedotov <ifedotov@mirantis.com>
---

diff --git a/src/os/bluestore/BlueStore.cc b/src/os/bluestore/BlueStore.cc
index 2339948fc4e..bd7ece3b881 100644
--- a/src/os/bluestore/BlueStore.cc
+++ b/src/os/bluestore/BlueStore.cc
@@ -1425,7 +1425,7 @@ ostream& operator<<(ostream& out, const BlueStore::Blob& b)
   if (b.is_spanning()) {
     out << " spanning " << b.id;
   }
-  out << " " << b.get_blob() << " " << b.get_ref_map()
+  out << " " << b.get_blob() << " " << b.get_blob_use_tracker()
       << " " << *b.shared_blob
       << ")";
   return out;
@@ -1459,6 +1459,7 @@ void BlueStore::Blob::discard_unallocated(Collection *coll)
     if (blob.can_prune_tail()) {
       dirty_blob();
       blob.prune_tail();
+      used_in_blob.prune_tail(blob.get_ondisk_length());
       auto cct = coll->store->cct; //used by dout
       dout(20) << __func__ << " pruned tail, now " << blob << dendl;
     }
@@ -1466,26 +1467,84 @@ void BlueStore::Blob::discard_unallocated(Collection *coll)
 }
 
 void BlueStore::Blob::get_ref(
-  uint64_t offset,
-  uint64_t length)
+  Collection *coll,
+  uint32_t offset,
+  uint32_t length)
 {
-  ref_map.get(offset, length);
+  // Caller has to initialize Blob's logical length prior to increment 
+  // references.  Otherwise one is neither unable to determine required
+  // amount of counters in case of per-au tracking nor obtain min_release_size
+  // for single counter mode.
+  assert(get_blob().get_logical_length() != 0);
+  auto cct = coll->store->cct;
+  dout(20) << __func__ << " 0x" << std::hex << offset << "~" << length
+           << std::dec << " " << *this << dendl;
+
+  if (used_in_blob.is_empty()) {
+    uint32_t min_release_size =
+      blob.get_release_size(coll->store->min_alloc_size);
+    uint64_t l = blob.get_logical_length();
+    dout(20) << __func__ << " init 0x" << std::hex << l << ", " << min_release_size
+             << std::dec << dendl;
+    used_in_blob.init(l, min_release_size);
+  }
+  used_in_blob.get(
+    offset,
+    length);
 }
 
+// cut it out of extents
+struct vecbuilder {
+  PExtentVector v;
+  uint64_t invalid = 0;
+
+  void add_invalid(uint64_t length) {
+    invalid += length;
+  }
+  void flush() {
+    if (invalid) {
+      v.emplace_back(bluestore_pextent_t(bluestore_pextent_t::INVALID_OFFSET,
+	invalid));
+      invalid = 0;
+    }
+  }
+  void add(uint64_t offset, uint64_t length) {
+    if (offset == bluestore_pextent_t::INVALID_OFFSET) {
+      add_invalid(length);
+    }
+    else {
+      flush();
+      v.emplace_back(bluestore_pextent_t(offset, length));
+    }
+  }
+};
+
 bool BlueStore::Blob::put_ref(
   Collection *coll,
-  uint64_t offset,
-  uint64_t length,
+  uint32_t offset,
+  uint32_t length,
   PExtentVector *r)
 {
   PExtentVector logical;
-  ref_map.put(offset, length, &logical);
+
+  auto cct = coll->store->cct;
+  dout(20) << __func__ << " 0x" << std::hex << offset << "~" << length
+           << std::dec << " " << *this << dendl;
+  
+  bool empty = used_in_blob.put(
+    offset,
+    length,
+    &logical);
   r->clear();
+  // nothing to release
+  if (!empty && logical.empty()) {
+    return false;
+  }
 
   bluestore_blob_t& b = dirty_blob();
 
   // common case: all of it?
-  if (ref_map.empty()) {
+  if (empty) {
     uint64_t pos = 0;
     for (auto& e : b.extents) {
       if (e.is_valid()) {
@@ -1498,120 +1557,85 @@ bool BlueStore::Blob::put_ref(
     b.extents[0].length = pos;
     return true;
   }
-
-  // we cannot do partial deallocation on compressed blobs
-  if (b.has_flag(bluestore_blob_t::FLAG_COMPRESSED)) {
-    return false;
-  }
-
-  auto min_release_size = coll->store->min_alloc_size;
-  // we cannot release something smaller than our csum chunk size
-  if (b.has_csum() && b.get_csum_chunk_size() > min_release_size) {
-    min_release_size = b.get_csum_chunk_size();
-  }
-
-  // search from logical releases
-  for (auto le : logical) {
-    uint64_t r_off;
-    auto p = ref_map.ref_map.lower_bound(le.offset);
-    if (p != ref_map.ref_map.begin()) {
-      --p;
-      r_off = p->first + p->second.length;
-      ++p;
-    } else {
-      r_off = 0;
-    }
-    uint64_t end;
-    if (p == ref_map.ref_map.end()) {
-      end = b.get_ondisk_length();
+  // remove from pextents according to logical release list
+  vecbuilder vb;
+  auto loffs_it = logical.begin();
+  auto lend = logical.end();
+  uint32_t pext_loffs_start = 0; //starting loffset of the current pextent
+  uint32_t pext_loffs = 0; //current loffset
+  auto pext_it = b.extents.begin();
+  auto pext_end = b.extents.end();
+  while (pext_it != pext_end) {
+    if (loffs_it == lend || pext_loffs_start + pext_it->length <= loffs_it->offset) {
+      int delta0 = pext_loffs - pext_loffs_start;
+      assert(delta0 >= 0);
+      if ((uint32_t)delta0 < pext_it->length) {
+        vb.add(pext_it->offset + delta0, pext_it->length - delta0);
+      }
+      pext_loffs_start += pext_it->length;
+      pext_loffs = pext_loffs_start;
+      ++pext_it;
     } else {
-      end = p->first;
-    }
-    r_off = ROUND_UP_TO(r_off, min_release_size);
-    end -= end % min_release_size;
-    if (r_off >= end) {
-      continue;
-    }
-    uint64_t r_len = end - r_off;
-
-    // cut it out of extents
-    struct vecbuilder {
-      PExtentVector v;
-      uint64_t invalid = 0;
-
-      void add_invalid(uint64_t length) {
-	invalid += length;
-      }
-      void flush() {
-	if (invalid) {
-	  v.emplace_back(bluestore_pextent_t(bluestore_pextent_t::INVALID_OFFSET,
-					     invalid));
-	  invalid = 0;
-	}
-      }
-      void add(uint64_t offset, uint64_t length) {
-	if (offset == bluestore_pextent_t::INVALID_OFFSET) {
-	  add_invalid(length);
-	} else {
-	  flush();
-	  v.emplace_back(bluestore_pextent_t(offset, length));
-	}
-      }
-    } vb;
-
-    assert(r_len > 0);
-    auto q = b.extents.begin();
-    assert(q != b.extents.end());
-    while (r_off >= q->length) {
-      vb.add(q->offset, q->length);
-      r_off -= q->length;
-      ++q;
-      assert(q != b.extents.end());
-    }
-    while (r_len > 0) {
-      uint64_t l = MIN(r_len, q->length - r_off);
-      if (q->is_valid()) {
-	r->emplace_back(bluestore_pextent_t(q->offset + r_off, l));
-      }
-      if (r_off) {
-	vb.add(q->offset, r_off);
-      }
-      vb.add_invalid(l);
-      if (r_off + l < q->length) {
-	vb.add(q->offset + r_off + l, q->length - (r_off + l));
-      }
-      r_len -= l;
-      r_off = 0;
-      ++q;
-      assert(q != b.extents.end() || r_len == 0);
-    }
-    while (q != b.extents.end()) {
-      vb.add(q->offset, q->length);
-      ++q;
-    }
-    vb.flush();
-    b.extents.swap(vb.v);
-  }
+      //assert(pext_loffs == pext_loffs_start);
+      int delta0 = pext_loffs - pext_loffs_start;
+      assert(delta0 >= 0);
+
+      int delta = loffs_it->offset - pext_loffs;
+      assert(delta >= 0);
+      if (delta > 0 ) {
+        vb.add(pext_it->offset + delta0, delta);
+        pext_loffs += delta;
+      }
+  
+      PExtentVector::iterator last_r = r->end();
+      if (r->begin() != last_r) {
+        --last_r;
+      }
+      uint32_t to_release = loffs_it->length;
+      do {
+        uint32_t to_release_part =
+          MIN(pext_it->length - delta0 - delta, to_release);
+        auto o = pext_it->offset + delta0 + delta;
+        if (last_r != r->end() && last_r->offset + last_r->length == o) {
+          last_r->length += to_release_part;
+        } else {
+	  last_r = r->emplace(r->end(), o, to_release_part);
+        }
+        to_release -= to_release_part;
+        pext_loffs += to_release_part;
+        if (pext_loffs == pext_loffs_start + pext_it->length) {
+          pext_loffs_start += pext_it->length;
+          pext_loffs = pext_loffs_start;
+          pext_it++;
+          delta0 = delta = 0;
+        }            
+      } while (to_release > 0 && pext_it != pext_end);
+      vb.add_invalid(loffs_it->length - to_release);
+      ++loffs_it;
+    }
+  }
+  vb.flush();
+  b.extents.swap(vb.v);
   return false;
 }
 
-void BlueStore::Blob::pass_ref(Blob* other, uint64_t src_offset, uint64_t length, uint64_t dest_offset)
-{
-  ref_map.put(src_offset, length, nullptr);
-  other->ref_map.get(dest_offset, length);
-}
-
-void BlueStore::Blob::split(Collection *coll, size_t blob_offset, Blob *r)
+void BlueStore::Blob::split(Collection *coll, uint32_t blob_offset, Blob *r)
 {
   auto cct = coll->store->cct; //used by dout
   dout(10) << __func__ << " 0x" << std::hex << blob_offset << std::dec
 	   << " start " << *this << dendl;
   assert(blob.can_split());
+  assert(used_in_blob.can_split());
   bluestore_blob_t &lb = dirty_blob();
   bluestore_blob_t &rb = r->dirty_blob();
 
   unsigned i = 0;
   size_t left = blob_offset;
+
+  used_in_blob.split(
+    blob_offset,
+    &(r->used_in_blob));
+
   for (auto p = lb.extents.begin(); p != lb.extents.end(); ++p, ++i) {
     if (p->length <= left) {
       left -= p->length;
@@ -1778,7 +1802,6 @@ bool BlueStore::ExtentMap::update(KeyValueDB::Transaction t,
 
 void BlueStore::ExtentMap::reshard()
 {
-  auto min_alloc_size = onode->c->store->min_alloc_size;
   auto cct = onode->c->store->cct; //used by dout
 
   needs_reshard = false;
@@ -1894,8 +1917,7 @@ void BlueStore::ExtentMap::reshard()
 	  for (const auto& sh : shards) {
 	    if (bstart < sh.offset && bend > sh.offset) {
 	      uint32_t blob_offset = sh.offset - bstart;
-	      if (b->get_blob().can_split_at(blob_offset) &&
-		  blob_offset % min_alloc_size == 0) {
+	      if (b->can_split_at(blob_offset)) {
 		dout(20) << __func__ << "    splitting blob, bstart 0x"
 			 << std::hex << bstart << " blob_offset 0x"
 			 << blob_offset << std::dec << " " << *b << dendl;
@@ -1954,7 +1976,12 @@ bool BlueStore::ExtentMap::encode_some(
     denc_varint(0, bound); // logical_offset
     denc_varint(0, bound); // len
     denc_varint(0, bound); // blob_offset
-    p->blob->bound_encode(bound, struct_v, p->blob->shared_blob->sbid, false);
+
+    p->blob->bound_encode(
+      bound,
+      struct_v,
+      p->blob->shared_blob->sbid,
+      false);
   }
 
   {
@@ -2037,6 +2064,7 @@ void BlueStore::ExtentMap::decode_some(bufferlist& bl)
   uint64_t pos = 0;
   uint64_t prev_len = 0;
   unsigned n = 0;
+
   while (!p.end()) {
     Extent *le = new Extent();
     uint64_t blobid;
@@ -2073,7 +2101,10 @@ void BlueStore::ExtentMap::decode_some(bufferlist& bl)
 	le->assign_blob(b);
       }
       // we build ref_map dynamically for non-spanning blobs
-      le->blob->get_ref(le->blob_offset, le->length);
+      le->blob->get_ref(
+	onode->c,
+	le->blob_offset,
+	le->length);
     }
     pos += prev_len;
     ++n;
@@ -2377,7 +2408,14 @@ BlueStore::Extent *BlueStore::ExtentMap::set_lextent(
   extent_map_t *old_extents)
 {
   punch_hole(logical_offset, length, old_extents);
-  b->get_ref(blob_offset, length);
+
+  // We need to have completely initialized Blob to increment its ref counters.
+  // But that's not true for newly created blob and we defer the increment until
+  // blob is ready in _do_alloc_write. See Blob::get_ref BlueStore::_do_alloc_write 
+  // implementations for more details.
+  if (b->get_blob().get_logical_length() != 0) {
+    b->get_ref(onode->c, blob_offset, length);
+  }
   Extent *le = new Extent(logical_offset, blob_offset, length, b);
   extent_map.insert(*le);
   if (!needs_reshard && spans_shard(logical_offset, length)) {
@@ -2411,14 +2449,13 @@ BlueStore::BlobRef BlueStore::ExtentMap::split_blob(
       size_t left = pos - ep->logical_offset;
       Extent *ne = new Extent(pos, 0, ep->length - left, rb);
       extent_map.insert(*ne);
-      lb->pass_ref(rb.get(), ep->blob_offset + left, ne->length, ne->blob_offset);
       ep->length = left;
       dout(30) << __func__ << "  split " << *ep << dendl;
       dout(30) << __func__ << "     to " << *ne << dendl;
     } else {
       // switch blob
       assert(ep->blob_offset >= blob_offset);
-      lb->pass_ref(rb.get(), ep->blob_offset, ep->length, ep->blob_offset - blob_offset);
+
       ep->blob = rb;
       ep->blob_offset -= blob_offset;
       dout(30) << __func__ << "  adjusted " << *ep << dendl;
@@ -4615,7 +4652,7 @@ int BlueStore::fsck(bool deep)
       }
       // lextents
       uint64_t pos = 0;
-      map<BlobRef,bluestore_extent_ref_map_t> ref_map;
+      map<BlobRef, bluestore_blob_use_tracker_t> ref_map;
       for (auto& l : o->extent_map.extent_map) {
 	dout(20) << __func__ << "    " << l << dendl;
 	if (l.logical_offset < pos) {
@@ -4635,23 +4672,32 @@ int BlueStore::fsck(bool deep)
 	pos = l.logical_offset + l.length;
 	expected_statfs.stored += l.length;
 	assert(l.blob);
-	ref_map[l.blob].get(l.blob_offset, l.length);
+	const bluestore_blob_t& blob = l.blob->get_blob();
+
+        auto& ref = ref_map[l.blob];
+        if (ref.is_empty()) {
+          uint32_t min_release_size = blob.get_release_size(min_alloc_size);
+          uint32_t l = blob.get_logical_length();
+          ref.init(l, min_release_size);
+        }
+	ref.get(
+	  l.blob_offset, 
+	  l.length);
 	++num_extents;
       }
       for (auto &i : ref_map) {
 	++num_blobs;
-	if (i.first->get_ref_map() != i.second) {
+	const bluestore_blob_t& blob = i.first->get_blob();
+	bool equal = i.first->get_blob_use_tracker().equal(i.second);
+	if (!equal) {
 	  derr << __func__ << " " << oid << " blob " << *i.first
 	       << " doesn't match expected ref_map " << i.second << dendl;
 	  ++errors;
 	}
-	const bluestore_blob_t& blob = i.first->get_blob();
 	if (blob.is_compressed()) {
 	  expected_statfs.compressed += blob.compressed_length;
-	  for (auto& r : i.first->get_ref_map().ref_map) {
-	    expected_statfs.compressed_original +=
-	      r.second.refs * r.second.length;
-	  }
+	  expected_statfs.compressed_original += 
+	    i.first->get_referenced_bytes();
 	}
 	if (blob.is_shared()) {
 	  if (i.first->shared_blob->sbid > blobid_max) {
@@ -7825,7 +7871,6 @@ void BlueStore::_do_write_small(
 	       << b_off << "~" << b_len
 	       << " pad 0x" << head_pad << " + 0x" << tail_pad
 	       << std::dec << " of mutable " << *b << dendl;
-      assert(b->is_unreferenced(b_off, b_len));
       _buffer_cache_write(txc, b, b_off, padded,
 			  wctx->buffered ? 0 : Buffer::FLAG_NOCACHE);
 
@@ -7930,14 +7975,15 @@ void BlueStore::_do_write_small(
   b = c->new_blob();
   unsigned alloc_len = min_alloc_size;
   uint64_t b_off = P2PHASE(offset, alloc_len);
+  uint64_t b_off0 = b_off;
   _buffer_cache_write(txc, b, b_off, bl,
 		      wctx->buffered ? 0 : Buffer::FLAG_NOCACHE);
-  _pad_zeros(&bl, &b_off, block_size);
-  Extent *le = o->extent_map.set_lextent(offset, P2PHASE(offset, alloc_len),
+  _pad_zeros(&bl, &b_off0, block_size);
+  Extent *le = o->extent_map.set_lextent(offset, b_off,
 			 length, b, &wctx->old_extents);
   txc->statfs_delta.stored() += le->length;
   dout(20) << __func__ << "  lex " << *le << dendl;
-  wctx->write(b, alloc_len, b_off, bl, true);
+  wctx->write(b, alloc_len, b_off0, bl, b_off, length, true);
   logger->inc(l_bluestore_write_small_new);
   return;
 }
@@ -7962,7 +8008,7 @@ void BlueStore::_do_write_big(
     bufferlist t;
     blp.copy(l, t);
     _buffer_cache_write(txc, b, 0, t, wctx->buffered ? 0 : Buffer::FLAG_NOCACHE);
-    wctx->write(b, l, 0, t, false);
+    wctx->write(b, l, 0, t, 0, l, false);
     Extent *le = o->extent_map.set_lextent(offset, 0, l,
                                            b, &wctx->old_extents);
     txc->statfs_delta.stored() += l;
@@ -8151,6 +8197,14 @@ int BlueStore::_do_alloc_write(
         dblob.add_unused(b_end, wi.blob_length - b_end);
       }
     }
+    
+    // Here we reattempt get_ref call deferred at set_lextent for newly created
+    // blobs. This is required since blob has logical length established at this 
+    // moment only. And the latter is required to initialize blob's reference
+    // counting machinery.
+    assert(!b->is_referenced());
+    b->get_ref(coll.get(), wi.b_off0, wi.length0); 
+                                                   
 
     // queue io
     if (!g_conf->bluestore_debug_omit_block_device_write) {
@@ -8225,7 +8279,7 @@ void BlueStore::_wctx_finish(
       }
     }
     delete &lo;
-    if (b->is_spanning() && b->get_ref_map().empty()) {
+    if (b->is_spanning() && !b->is_referenced()) {
       dout(20) << __func__ << "  spanning_blob_map removing empty " << *b
 	       << dendl;
       o->extent_map.spanning_blob_map.erase(b->id);
@@ -9006,7 +9060,7 @@ int BlueStore::_do_clone_range(
 			    e.blob_offset + skip_front,
 			    e.length - skip_front - skip_back, cb);
     newo->extent_map.extent_map.insert(*ne);
-    ne->blob->get_ref(ne->blob_offset, ne->length);
+    ne->blob->get_ref(c.get(), ne->blob_offset, ne->length);
     // fixme: we may leave parts of new blob unreferenced that could
     // be freed (relative to the shared_blob).
     txc->statfs_delta.stored() += ne->length;
diff --git a/src/os/bluestore/BlueStore.h b/src/os/bluestore/BlueStore.h
index 8ca4ccf659c..47f551e62b7 100644
--- a/src/os/bluestore/BlueStore.h
+++ b/src/os/bluestore/BlueStore.h
@@ -423,7 +423,7 @@ public:
     mutable bufferlist blob_bl;     ///< cached encoded blob, blob is dirty if empty
 #endif
     /// refs from this shard.  ephemeral if id<0, persisted if spanning.
-    bluestore_extent_ref_map_t ref_map;
+    bluestore_blob_use_tracker_t used_in_blob;
 
   public:
 
@@ -432,9 +432,16 @@ public:
 
     friend ostream& operator<<(ostream& out, const Blob &b);
 
-    const bluestore_extent_ref_map_t& get_ref_map() const {
-      return ref_map;
+    const bluestore_blob_use_tracker_t& get_blob_use_tracker() const {
+      return used_in_blob;
     }
+    bool is_referenced() const {
+      return used_in_blob.is_not_empty();
+    }
+    uint32_t get_referenced_bytes() const {
+      return used_in_blob.get_referenced_bytes();
+    }
+
     bool is_spanning() const {
       return id >= 0;
     }
@@ -442,7 +449,14 @@ public:
     bool can_split() const {
       std::lock_guard<std::recursive_mutex> l(shared_blob->get_cache()->lock);
       // splitting a BufferSpace writing list is too hard; don't try.
-      return shared_blob->bc.writing.empty() && get_blob().can_split();
+      return shared_blob->bc.writing.empty() &&
+             used_in_blob.can_split() &&
+             get_blob().can_split();
+    }
+
+    bool can_split_at(uint32_t blob_offset) const {
+      return used_in_blob.can_split_at(blob_offset) &&
+             get_blob().can_split_at(blob_offset);
     }
 
     void dup(Blob& o) {
@@ -462,23 +476,18 @@ public:
 #endif
       return blob;
     }
-    bool is_unreferenced(uint64_t offset, uint64_t length) const {
-      return !ref_map.intersects(offset, length);
-    }
 
     /// discard buffers for unallocated regions
     void discard_unallocated(Collection *coll);
 
     /// get logical references
-    void get_ref(uint64_t offset, uint64_t length);
+    void get_ref(Collection *coll, uint32_t offset, uint32_t length);
     /// put logical references, and get back any released extents
-    bool put_ref(Collection *coll, uint64_t offset, uint64_t length,
+    bool put_ref(Collection *coll, uint32_t offset, uint32_t length,
 		 PExtentVector *r);
-    /// pass references for specific range to other blob
-    void pass_ref(Blob* other, uint64_t src_offset, uint64_t length, uint64_t dest_offset);
 
     /// split the blob
-    void split(Collection *coll, size_t blob_offset, Blob *o);
+    void split(Collection *coll, uint32_t blob_offset, Blob *o);
 
     void get() {
       ++nref;
@@ -497,58 +506,74 @@ public:
 	assert(blob_bl.length());
       }
     }
-    void bound_encode(size_t& p, bool include_ref_map) const {
+    void bound_encode(
+      size_t& p,
+      bool include_ref_map) const {
       _encode();
       p += blob_bl.length();
       if (include_ref_map) {
-        ref_map.bound_encode(p);
+	used_in_blob.bound_encode(p);
       }
     }
-    void encode(bufferlist::contiguous_appender& p, bool include_ref_map) const {
+    void encode(
+      bufferlist::contiguous_appender& p,
+      bool include_ref_map) const {
       _encode();
       p.append(blob_bl);
       if (include_ref_map) {
-        ref_map.encode(p);
+	used_in_blob.encode(p);
       }
     }
-    void decode(bufferptr::iterator& p, bool include_ref_map) {
+    void decode(
+      bufferptr::iterator& p,
+      bool include_ref_map) {
       const char *start = p.get_pos();
       denc(blob, p);
       const char *end = p.get_pos();
       blob_bl.clear();
       blob_bl.append(start, end - start);
       if (include_ref_map) {
-        ref_map.decode(p);
+	used_in_blob.decode(p);
       }
     }
 #else
-    void bound_encode(size_t& p, uint64_t struct_v, uint64_t sbid, bool include_ref_map) const {
+    void bound_encode(
+      size_t& p,
+      uint64_t struct_v,
+      uint64_t sbid,
+      bool include_ref_map) const {
       denc(blob, p, struct_v);
       if (blob.is_shared()) {
         denc(sbid, p);
       }
       if (include_ref_map) {
-        ref_map.bound_encode(p);
+	used_in_blob.bound_encode(p);
       }
     }
-    void encode(bufferlist::contiguous_appender& p, uint64_t struct_v, uint64_t sbid,
-		bool include_ref_map) const {
+    void encode(
+      bufferlist::contiguous_appender& p,
+      uint64_t struct_v,
+      uint64_t sbid,
+      bool include_ref_map) const {
       denc(blob, p, struct_v);
       if (blob.is_shared()) {
         denc(sbid, p);
       }
       if (include_ref_map) {
-        ref_map.encode(p);
+	used_in_blob.encode(p);
       }
     }
-    void decode(bufferptr::iterator& p, uint64_t struct_v, uint64_t* sbid,
-		bool include_ref_map) {
+    void decode(
+      bufferptr::iterator& p,
+      uint64_t struct_v,
+      uint64_t* sbid,
+      bool include_ref_map) {
       denc(blob, p, struct_v);
       if (blob.is_shared()) {
         denc(*sbid, p);
       }
       if (include_ref_map) {
-        ref_map.decode(p);
+	used_in_blob.decode(p);
       }
     }
 #endif
@@ -2065,15 +2090,37 @@ private:
       uint64_t blob_length;
       uint64_t b_off;
       bufferlist bl;
+      uint64_t b_off0; ///< original offset in a blob prior to padding
+      uint64_t length0; ///< original data length prior to padding
+
       bool mark_unused;
 
-      write_item(BlobRef b, uint64_t blob_len, uint64_t o, bufferlist& bl, bool _mark_unused)
-       : b(b), blob_length(blob_len), b_off(o), bl(bl), mark_unused(_mark_unused) {}
+      write_item(
+        BlobRef b,
+        uint64_t blob_len,
+        uint64_t o,
+        bufferlist& bl,
+        uint64_t o0,
+        uint64_t l0,
+        bool _mark_unused)
+       : b(b),
+         blob_length(blob_len),
+         b_off(o),
+         bl(bl),
+         b_off0(o0),
+         length0(l0),
+         mark_unused(_mark_unused) {}
     };
     vector<write_item> writes;                 ///< blobs we're writing
 
-    void write(BlobRef b, uint64_t blob_len, uint64_t o, bufferlist& bl, bool _mark_unused) {
-      writes.emplace_back(write_item(b, blob_len, o, bl, _mark_unused));
+    void write(
+      BlobRef b,
+      uint64_t blob_len,
+      uint64_t o,
+      bufferlist& bl,
+      uint64_t o0,
+      uint64_t len0, bool _mark_unused) {
+      writes.emplace_back(write_item(b, blob_len, o, bl, o0, len0, _mark_unused));
     }
   };
 
diff --git a/src/os/bluestore/bluestore_types.cc b/src/os/bluestore/bluestore_types.cc
index 68f9730413b..42a454fcbd1 100644
--- a/src/os/bluestore/bluestore_types.cc
+++ b/src/os/bluestore/bluestore_types.cc
@@ -331,6 +331,215 @@ ostream& operator<<(ostream& out, const bluestore_extent_ref_map_t& m)
   return out;
 }
 
+// bluestore_blob_use_tracker_t
+
+void bluestore_blob_use_tracker_t::allocate()
+{
+  assert(num_au != 0);
+  bytes_per_au = new uint32_t[num_au];
+  for (uint32_t i = 0; i < num_au; ++i) {
+    bytes_per_au[i] = 0;
+  }
+}
+
+void bluestore_blob_use_tracker_t::init(
+  uint32_t full_length, uint32_t _au_size) {
+  assert(!au_size || is_empty()); 
+  assert(_au_size > 0);
+  assert(full_length > 0);
+  clear();  
+  uint32_t _num_au = ROUND_UP_TO(full_length, _au_size) / _au_size;
+  au_size = _au_size;
+  if( _num_au > 1 ) {
+    num_au = _num_au;
+    allocate();
+  }
+}
+
+void bluestore_blob_use_tracker_t::get(
+  uint32_t offset, uint32_t length)
+{
+  assert(au_size);
+  if (!num_au) {
+    total_bytes += length;
+  }else {
+    auto end = offset + length;
+
+    while (offset < end) {
+      auto phase = offset % au_size;
+      bytes_per_au[offset / au_size] += 
+	MIN(au_size - phase, end - offset);
+      offset += (phase ? au_size - phase : au_size);
+    }
+  }
+}
+
+bool bluestore_blob_use_tracker_t::put(
+  uint32_t offset, uint32_t length,
+  PExtentVector *release_units)
+{
+  assert(au_size);
+  if (release_units) {
+    release_units->clear();
+  }
+  bool maybe_empty = true;
+  if (!num_au) {
+    assert(total_bytes >= length);
+    total_bytes -= length;
+  } else {
+    auto end = offset + length;
+    uint64_t next_offs = 0;
+    while (offset < end) {
+      auto phase = offset % au_size;
+      size_t pos = offset / au_size;
+      auto diff = MIN(au_size - phase, end - offset);
+      assert(diff <= bytes_per_au[pos]);
+      bytes_per_au[pos] -= diff;
+      offset += (phase ? au_size - phase : au_size);
+      if (bytes_per_au[pos] == 0) {
+	if (release_units) {
+          if (release_units->empty() || next_offs != pos * au_size) {
+  	    release_units->emplace_back(pos * au_size, au_size);
+          } else {
+            release_units->back().length += au_size;
+          }
+          next_offs += au_size;
+	}
+      } else {
+	maybe_empty = false; // micro optimization detecting we aren't empty 
+	                     // even in the affected extent
+      }
+    }
+  }
+  bool empty = maybe_empty ? !is_not_empty() : false;
+  if (empty && release_units) {
+    release_units->clear();
+  }
+  return empty;
+}
+
+bool bluestore_blob_use_tracker_t::can_split() const
+{
+  return num_au > 0;
+}
+
+bool bluestore_blob_use_tracker_t::can_split_at(uint32_t blob_offset) const
+{
+  assert(au_size);
+  return (blob_offset % au_size) == 0 &&
+         blob_offset < num_au * au_size;
+}
+
+void bluestore_blob_use_tracker_t::split(
+  uint32_t blob_offset,
+  bluestore_blob_use_tracker_t* r)
+{
+  assert(au_size);
+  assert(can_split());
+  assert(can_split_at(blob_offset));
+  assert(r->is_empty());
+  
+  uint32_t new_num_au = blob_offset / au_size;
+  r->init( (num_au - new_num_au) * au_size, au_size);
+
+  for (auto i = new_num_au; i < num_au; i++) {
+    r->get((i - new_num_au) * au_size, bytes_per_au[i]);
+    bytes_per_au[i] = 0;
+  }
+  if (new_num_au == 0) {
+    clear();
+  } else if (new_num_au == 1) {
+    uint32_t tmp = bytes_per_au[0];
+    uint32_t _au_size = au_size;
+    clear();
+    au_size = _au_size;
+    total_bytes = tmp;
+  } else {
+    num_au = new_num_au;
+  }
+}
+
+bool bluestore_blob_use_tracker_t::equal(
+  const bluestore_blob_use_tracker_t& other) const
+{
+  if (!num_au && !other.num_au) {
+    return total_bytes == other.total_bytes && au_size == other.au_size;
+  } else if (num_au && other.num_au) {
+    if (num_au != other.num_au || au_size != other.au_size) {
+      return false;
+    }
+    for (size_t i = 0; i < num_au; i++) {
+      if (bytes_per_au[i] != other.bytes_per_au[i]) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  uint32_t n = num_au ? num_au : other.num_au;
+  uint32_t referenced = 
+    num_au ? other.get_referenced_bytes() : get_referenced_bytes();
+   auto bytes_per_au_tmp = num_au ? bytes_per_au : other.bytes_per_au;
+  uint32_t my_referenced = 0;
+  for (size_t i = 0; i < n; i++) {
+    my_referenced += bytes_per_au_tmp[i];
+    if (my_referenced > referenced) {
+      return false;
+    }
+  }
+  return my_referenced == referenced;
+}
+
+void bluestore_blob_use_tracker_t::dump(Formatter *f) const
+{
+  f->dump_unsigned("num_au", num_au);
+  f->dump_unsigned("au_size", au_size);
+  if (!num_au) {
+    f->dump_unsigned("total_bytes", total_bytes);
+  } else {
+    f->open_array_section("bytes_per_au");
+    for (size_t i = 0; i < num_au; ++i) {
+      f->dump_unsigned("", bytes_per_au[i]);
+    }
+    f->close_section();
+  }
+}
+
+void bluestore_blob_use_tracker_t::generate_test_instances(
+  list<bluestore_blob_use_tracker_t*>& o)
+{
+  o.push_back(new bluestore_blob_use_tracker_t());
+  o.back()->init(16, 16);
+  o.back()->get(10, 10);
+  o.back()->get(10, 5);
+  o.push_back(new bluestore_blob_use_tracker_t());
+  o.back()->init(60, 16);
+  o.back()->get(18, 22);
+  o.back()->get(20, 20);
+  o.back()->get(15, 20);
+}
+
+ostream& operator<<(ostream& out, const bluestore_blob_use_tracker_t& m)
+{
+  out << "use_tracker(" << std::hex;
+  if (!m.num_au) {
+    out << "0x" << m.au_size 
+        << " :"
+        << "0x" << m.total_bytes;
+  } else {
+    out << "0x" << m.num_au 
+        << "*0x" << m.au_size 
+	<< " :";
+    for (size_t i = 0; i < m.num_au; ++i) {
+      if (i != 0)
+	out << ",";
+      out << "0x" << m.bytes_per_au[i];
+    }
+  }
+  out << std::dec << ")";
+  return out;
+}
+
 // bluestore_pextent_t
 
 void bluestore_pextent_t::dump(Formatter *f) const
diff --git a/src/os/bluestore/bluestore_types.h b/src/os/bluestore/bluestore_types.h
index 59f72ded023..80990f1ebfd 100644
--- a/src/os/bluestore/bluestore_types.h
+++ b/src/os/bluestore/bluestore_types.h
@@ -275,6 +275,159 @@ static inline bool operator!=(const bluestore_extent_ref_map_t& l,
   return !(l == r);
 }
 
+/// blob_use_tracker: a set of per-alloc unit ref counters to track blob usage
+struct bluestore_blob_use_tracker_t {
+  // N.B.: There is no need to minimize au_size/num_au
+  //   as much as possible (e.g. have just a single byte for au_size) since:
+  //   1) Struct isn't packed hence it's padded. And even if it's packed see 2)
+  //   2) Mem manager has its own granularity, most probably >= 8 bytes
+  //
+  uint32_t au_size; // Allocation (=tracking) unit size,
+                    // == 0 if uninitialized
+  uint32_t num_au;  // Amount of allocation units tracked
+                    // == 0 if single unit or the whole blob is tracked
+                       
+  union {
+    uint32_t* bytes_per_au;
+    uint32_t total_bytes;
+  };
+  
+  bluestore_blob_use_tracker_t()
+    : au_size(0), num_au(0), bytes_per_au(nullptr) {
+  }
+  ~bluestore_blob_use_tracker_t() {
+    clear();
+  }
+
+  void clear() {
+    if (num_au != 0) {
+      delete[] bytes_per_au;
+    }
+    bytes_per_au = 0;
+    au_size = 0;
+    num_au = 0;
+  }
+
+  uint32_t get_referenced_bytes() const {
+    uint32_t total = 0;
+    if (!num_au) {
+      total = total_bytes;
+    } else {
+      for (size_t i = 0; i < num_au; ++i) {
+	total += bytes_per_au[i];
+      }
+    }
+    return total;
+  }
+  bool is_not_empty() const {
+    if (!num_au) {
+      return total_bytes != 0;
+    } else {
+      for (size_t i = 0; i < num_au; ++i) {
+	if (bytes_per_au[i]) {
+	  return true;
+	}
+      }
+    }
+    return false;
+  }
+  bool is_empty() const {
+    return !is_not_empty();
+  }
+  void prune_tail(uint32_t new_len) {
+    if (num_au) {
+      new_len = ROUND_UP_TO(new_len, au_size);
+      uint32_t _num_au = new_len / au_size;
+      assert(_num_au <= num_au);
+      if (_num_au) {
+        num_au = _num_au; // bytes_per_au array is left unmodified
+      } else {
+        clear();
+      }
+    }
+  }
+  
+  void init(
+    uint32_t full_length,
+    uint32_t _au_size);
+
+  void get(
+    uint32_t offset,
+    uint32_t len);
+
+  /// put: return true if the blob has no references any more after the call,
+  /// no release_units is filled for the sake of performance.
+  /// return false if there are some references to the blob,
+  /// in this case release_units contains pextents
+  /// (identified by their offsets relative to the blob start)
+  //  that are not used any more and can be safely deallocated. 
+  bool put(
+    uint32_t offset,
+    uint32_t len,
+    PExtentVector *release);
+
+  bool can_split() const;
+  bool can_split_at(uint32_t blob_offset) const;
+  void split(
+    uint32_t blob_offset,
+    bluestore_blob_use_tracker_t* r);
+
+  bool equal(
+    const bluestore_blob_use_tracker_t& other) const;
+    
+  void bound_encode(size_t& p) const {
+    denc_varint(au_size, p);
+    if (au_size) {
+      denc_varint(num_au, p);
+      if (!num_au) {
+        denc_varint(total_bytes, p);
+      } else {
+        size_t elem_size = 0;
+        denc_varint((uint32_t)0, elem_size);
+        p += elem_size * num_au;
+      }
+    }
+  }
+  void encode(bufferlist::contiguous_appender& p) const {
+    denc_varint(au_size, p);
+    if (au_size) {
+      denc_varint(num_au, p);
+      if (!num_au) {
+        denc_varint(total_bytes, p);
+      } else {
+        size_t elem_size = 0;
+        denc_varint((uint32_t)0, elem_size);
+        for (size_t i = 0; i < num_au; ++i) {
+          denc_varint(bytes_per_au[i], p);
+        }
+      }
+    }
+  }
+  void decode(bufferptr::iterator& p) {
+    clear();
+    denc_varint(au_size, p);
+    if (au_size) {
+      denc_varint(num_au, p);
+      if (!num_au) {
+        denc_varint(total_bytes, p);
+      } else {
+        allocate();
+        for (size_t i = 0; i < num_au; ++i) {
+	  denc_varint(bytes_per_au[i], p);
+        }
+      }
+    }
+  }
+
+  void dump(Formatter *f) const;
+  static void generate_test_instances(list<bluestore_blob_use_tracker_t*>& o);
+private:
+  void allocate();
+  void fall_back_to_per_au(uint32_t _num_au, uint32_t _au_size);
+};
+WRITE_CLASS_DENC(bluestore_blob_use_tracker_t)
+ostream& operator<<(ostream& out, const bluestore_blob_use_tracker_t& rm);
+
 /// blob: a piece of data on disk
 struct bluestore_blob_t {
   enum {
@@ -633,6 +786,16 @@ struct bluestore_blob_t {
 			    get_csum_value_size());
     }
   }
+  uint32_t get_release_size(uint32_t min_alloc_size) const {
+    if (is_compressed()) {
+      return get_logical_length();
+    }
+    uint32_t res = get_csum_chunk_size();
+    if (!has_csum() || res < min_alloc_size) {
+      res = min_alloc_size;
+    }
+    return res;
+  }
 };
 WRITE_CLASS_DENC_FEATURED(bluestore_blob_t)
 
diff --git a/src/test/objectstore/store_test.cc b/src/test/objectstore/store_test.cc
index a51589b5922..a998e5d9a71 100644
--- a/src/test/objectstore/store_test.cc
+++ b/src/test/objectstore/store_test.cc
@@ -1150,6 +1150,87 @@ TEST_P(StoreTest, SimpleObjectTest) {
   }
 }
 
+
+
+
+TEST_P(StoreTest, UnalignedBlobReleaseTest) {
+  if (string(GetParam()) != "bluestore")
+    return;
+
+  ObjectStore::Sequencer osr("test");
+  coll_t cid;
+
+  g_conf->set_val("bluestore_min_alloc_size", "4096");
+  g_ceph_context->_conf->apply_changes(NULL);
+  int r = store->umount();
+  ASSERT_EQ(r, 0);
+  r = store->mount(); //to force min_alloc_size update
+  ASSERT_EQ(r, 0);
+
+  ghobject_t hoid(hobject_t(sobject_t("Object 1", CEPH_NOSNAP)));
+  {
+    ObjectStore::Transaction t;
+    t.create_collection(cid, 0);
+    cerr << "Creating collection " << cid << std::endl;
+    r = apply_transaction(store, &osr, std::move(t));
+    ASSERT_EQ(r, 0);
+  }
+  {
+    bool exists = store->exists(cid, hoid);
+    ASSERT_TRUE(!exists);
+
+    ObjectStore::Transaction t;
+    t.touch(cid, hoid);
+    cerr << "Creating object " << hoid << std::endl;
+    r = apply_transaction(store, &osr, std::move(t));
+    ASSERT_EQ(r, 0);
+
+    exists = store->exists(cid, hoid);
+    ASSERT_EQ(true, exists);
+  }
+  {
+    ObjectStore::Transaction t;
+    bufferlist bl, bl2, orig;
+    string s(0x600, 'a'), s2(0x6000,'b');
+    bl.append(s);
+    bl2.append(s2);
+    t.write(cid, hoid, 0x9000, 0x6000, bl2);
+    r = apply_transaction(store, &osr, std::move(t));
+    ASSERT_EQ(r, 0);
+
+  g_conf->set_val("bluestore_min_alloc_size", "16384");
+  g_ceph_context->_conf->apply_changes(NULL);
+  r = store->umount();
+  ASSERT_EQ(r, 0);
+  r = store->mount(); //to force min_alloc_size update
+  ASSERT_EQ(r, 0);
+
+    {
+      ObjectStore::Transaction t;
+      t.truncate(cid, hoid, 0xb200);
+      r = apply_transaction(store, &osr, std::move(t));
+      ASSERT_EQ(r, 0);
+    }
+    {
+      ObjectStore::Transaction t;
+      t.truncate(cid, hoid, 0);
+      r = apply_transaction(store, &osr, std::move(t));
+      ASSERT_EQ(r, 0);
+    }
+  }
+
+  {
+    ObjectStore::Transaction t;
+    t.remove(cid, hoid);
+    t.remove_collection(cid);
+    cerr << "Cleaning" << std::endl;
+    r = apply_transaction(store, &osr, std::move(t));
+    ASSERT_EQ(r, 0);
+  }
+  g_conf->set_val("bluestore_min_alloc_size", "0");
+  g_ceph_context->_conf->apply_changes(NULL);
+}
+
 TEST_P(StoreTest, BluestoreStatFSTest) {
   if(string(GetParam()) != "bluestore")
     return;
diff --git a/src/test/objectstore/test_bluestore_types.cc b/src/test/objectstore/test_bluestore_types.cc
index a6498fd0c48..87d16cd67c0 100644
--- a/src/test/objectstore/test_bluestore_types.cc
+++ b/src/test/objectstore/test_bluestore_types.cc
@@ -330,17 +330,20 @@ TEST(Blob, put_ref)
     b.dirty_blob().extents.push_back(
       bluestore_pextent_t(bluestore_pextent_t::INVALID_OFFSET, 0x8000));
     b.dirty_blob().extents.push_back(bluestore_pextent_t(0x4071f000, 0x5000));
-    b.get_ref(0, 0x1200);
-    b.get_ref(0xae00, 0x4200);
+    b.get_ref(&coll, 0, 0x1200);
+    b.get_ref(&coll, 0xae00, 0x4200);
+    ASSERT_EQ(0x5400u, b.get_referenced_bytes());
     cout << b << std::endl;
     PExtentVector r;
 
     b.put_ref(&coll, 0, 0x1200, &r);
+    ASSERT_EQ(0x4200u, b.get_referenced_bytes());
     cout << " r " << r << std::endl;
     cout << b << std::endl;
 
     r.clear();
     b.put_ref(&coll, 0xae00, 0x4200, &r);
+    ASSERT_EQ(0u, b.get_referenced_bytes());
     cout << " r " << r << std::endl;
     cout << b << std::endl;
   }
@@ -358,9 +361,11 @@ TEST(Blob, put_ref)
     bluestore_blob_t& b = B.dirty_blob();
     PExtentVector r;
     b.extents.push_back(bluestore_pextent_t(0, mas*2));
-    B.get_ref(0, mas*2);
+    B.get_ref(&coll, 0, mas*2);
+    ASSERT_EQ(mas * 2, B.get_referenced_bytes());
     ASSERT_TRUE(b.is_allocated(0, mas*2));
     B.put_ref(&coll, 0, mas*2, &r);
+    ASSERT_EQ(0u, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(1u, r.size());
     ASSERT_EQ(0u, r[0].offset);
@@ -378,12 +383,16 @@ TEST(Blob, put_ref)
     bluestore_blob_t& b = B.dirty_blob();
     PExtentVector r;
     b.extents.push_back(bluestore_pextent_t(123, mas*2));
-    B.get_ref(0, mas*2);
+    B.get_ref(&coll, 0, mas*2);
+    ASSERT_EQ(mas * 2, B.get_referenced_bytes());
     B.put_ref(&coll, 0, mas, &r);
+    ASSERT_EQ(mas, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*2));
     B.put_ref(&coll, mas, mas, &r);
+    ASSERT_EQ(0u, B.get_referenced_bytes());
+    ASSERT_EQ(0u, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(1u, r.size());
     ASSERT_EQ(123u, r[0].offset);
@@ -402,18 +411,22 @@ TEST(Blob, put_ref)
     b.extents.push_back(bluestore_pextent_t(2, mas));
     b.extents.push_back(bluestore_pextent_t(3, mas));
     b.extents.push_back(bluestore_pextent_t(4, mas));
-    B.get_ref(0, mas*4);
+    B.get_ref(&coll, 0, mas*4);
+    ASSERT_EQ(mas * 4, B.get_referenced_bytes());
     B.put_ref(&coll, mas, mas, &r);
+    ASSERT_EQ(mas * 3, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*4));
     ASSERT_TRUE(b.is_allocated(mas, mas));
     B.put_ref(&coll, mas*2, mas, &r);
+    ASSERT_EQ(mas * 2, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(mas*2, mas));
     ASSERT_TRUE(b.is_allocated(0, mas*4));
     B.put_ref(&coll, mas*3, mas, &r);
+    ASSERT_EQ(mas, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(2u, r.size());
     ASSERT_EQ(3u, r[0].offset);
@@ -439,16 +452,20 @@ TEST(Blob, put_ref)
     b.extents.push_back(bluestore_pextent_t(4, mas));
     b.extents.push_back(bluestore_pextent_t(5, mas));
     b.extents.push_back(bluestore_pextent_t(6, mas));
-    B.get_ref(0, mas*6);
+    B.get_ref(&coll, 0, mas*6);
+    ASSERT_EQ(mas * 6, B.get_referenced_bytes());
     B.put_ref(&coll, mas, mas, &r);
+    ASSERT_EQ(mas * 5, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*6));
     B.put_ref(&coll, mas*2, mas, &r);
+    ASSERT_EQ(mas * 4, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*6));
     B.put_ref(&coll, mas*3, mas, &r);
+    ASSERT_EQ(mas * 3, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(2u, r.size());
     ASSERT_EQ(3u, r[0].offset);
@@ -472,16 +489,20 @@ TEST(Blob, put_ref)
     bluestore_blob_t& b = B.dirty_blob();
     PExtentVector r;
     b.extents.push_back(bluestore_pextent_t(1, mas * 6));
-    B.get_ref(0, mas*6);
+    B.get_ref(&coll, 0, mas*6);
+    ASSERT_EQ(mas * 6, B.get_referenced_bytes());
     B.put_ref(&coll, mas, mas, &r);
+    ASSERT_EQ(mas * 5, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*6));
     B.put_ref(&coll, mas*2, mas, &r);
+    ASSERT_EQ(mas * 4, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*6));
     B.put_ref(&coll, mas*3, mas, &r);
+    ASSERT_EQ(mas * 3, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(1u, r.size());
     ASSERT_EQ(0x2001u, r[0].offset);
@@ -503,16 +524,20 @@ TEST(Blob, put_ref)
     b.extents.push_back(bluestore_pextent_t(1, mas * 4));
     b.extents.push_back(bluestore_pextent_t(2, mas * 4));
     b.extents.push_back(bluestore_pextent_t(3, mas * 4));
-    B.get_ref(0, mas*12);
+    B.get_ref(&coll, 0, mas*12);
+    ASSERT_EQ(mas * 12, B.get_referenced_bytes());
     B.put_ref(&coll, mas, mas, &r);
+    ASSERT_EQ(mas * 11, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*12));
     B.put_ref(&coll, mas*9, mas, &r);
+    ASSERT_EQ(mas * 10, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*12));
     B.put_ref(&coll, mas*2, mas*7, &r);
+    ASSERT_EQ(mas * 3, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(3u, r.size());
     ASSERT_EQ(0x2001u, r[0].offset);
@@ -538,16 +563,20 @@ TEST(Blob, put_ref)
     b.extents.push_back(bluestore_pextent_t(1, mas * 4));
     b.extents.push_back(bluestore_pextent_t(2, mas * 4));
     b.extents.push_back(bluestore_pextent_t(3, mas * 4));
-    B.get_ref(0, mas*12);
+    B.get_ref(&coll, 0, mas*12);
+    ASSERT_EQ(mas * 12, B.get_referenced_bytes());
     B.put_ref(&coll, mas, mas, &r);
+    ASSERT_EQ(mas * 11, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*12));
     B.put_ref(&coll, mas*9, mas, &r);
+    ASSERT_EQ(mas * 10, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*12));
     B.put_ref(&coll, mas*2, mas*7, &r);
+    ASSERT_EQ(mas * 3, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(3u, r.size());
     ASSERT_EQ(0x2001u, r[0].offset);
@@ -564,6 +593,7 @@ TEST(Blob, put_ref)
     ASSERT_FALSE(b.extents[1].is_valid());
     ASSERT_TRUE(b.extents[2].is_valid());
     B.put_ref(&coll, 0, mas, &r);
+    ASSERT_EQ(mas * 2, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(1u, r.size());
     ASSERT_EQ(0x1u, r[0].offset);
@@ -572,6 +602,7 @@ TEST(Blob, put_ref)
     ASSERT_FALSE(b.extents[0].is_valid());
     ASSERT_TRUE(b.extents[1].is_valid());
     B.put_ref(&coll, mas*10, mas*2, &r);
+    ASSERT_EQ(mas * 0, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(1u, r.size());
     ASSERT_EQ(0x2003u, r[0].offset);
@@ -588,16 +619,20 @@ TEST(Blob, put_ref)
     b.extents.push_back(bluestore_pextent_t(1, mas * 4));
     b.extents.push_back(bluestore_pextent_t(2, mas * 4));
     b.extents.push_back(bluestore_pextent_t(3, mas * 4));
-    B.get_ref(0, mas*12);
+    B.get_ref(&coll, 0, mas*12);
+    ASSERT_EQ(mas * 12, B.get_referenced_bytes());
     B.put_ref(&coll, mas, mas, &r);
+    ASSERT_EQ(mas * 11, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*12));
     B.put_ref(&coll, mas*9, mas, &r);
+    ASSERT_EQ(mas * 10, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*12));
     B.put_ref(&coll, mas*2, mas*7, &r);
+    ASSERT_EQ(mas * 3, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(3u, r.size());
     ASSERT_EQ(0x2001u, r[0].offset);
@@ -614,6 +649,7 @@ TEST(Blob, put_ref)
     ASSERT_FALSE(b.extents[1].is_valid());
     ASSERT_TRUE(b.extents[2].is_valid());
     B.put_ref(&coll, mas*10, mas*2, &r);
+    ASSERT_EQ(mas * 1, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(1u, r.size());
     ASSERT_EQ(0x2003u, r[0].offset);
@@ -622,6 +658,7 @@ TEST(Blob, put_ref)
     ASSERT_TRUE(b.extents[0].is_valid());
     ASSERT_FALSE(b.extents[1].is_valid());
     B.put_ref(&coll, 0, mas, &r);
+    ASSERT_EQ(mas * 0, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(1u, r.size());
     ASSERT_EQ(0x1u, r[0].offset);
@@ -636,20 +673,25 @@ TEST(Blob, put_ref)
     bluestore_blob_t& b = B.dirty_blob();
     PExtentVector r;
     b.extents.push_back(bluestore_pextent_t(1, mas * 8));
-    B.get_ref(0, mas*8);
+    B.get_ref(&coll, 0, mas*8);
+    ASSERT_EQ(mas * 8, B.get_referenced_bytes());
     B.put_ref(&coll, 0, mas, &r);
+    ASSERT_EQ(mas * 7, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*8));
     B.put_ref(&coll, mas*7, mas, &r);
+    ASSERT_EQ(mas * 6, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*8));
     B.put_ref(&coll, mas*2, mas, &r);
+    ASSERT_EQ(mas * 5, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, 8));
     B.put_ref(&coll, mas*3, mas*4, &r);
+    ASSERT_EQ(mas * 1, B.get_referenced_bytes());
     ASSERT_EQ(1u, r.size());
     ASSERT_EQ(0x2001u, r[0].offset);
     ASSERT_EQ(mas*6, r[0].length);
@@ -659,6 +701,7 @@ TEST(Blob, put_ref)
     ASSERT_TRUE(b.extents[0].is_valid());
     ASSERT_FALSE(b.extents[1].is_valid());
     B.put_ref(&coll, mas, mas, &r);
+    ASSERT_EQ(mas * 0, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(1u, r.size());
     ASSERT_EQ(0x1u, r[0].offset);
@@ -675,9 +718,11 @@ TEST(Blob, put_ref)
     PExtentVector r;
     b.extents.push_back(bluestore_pextent_t(0, mas*4));
     b.init_csum(Checksummer::CSUM_CRC32C, 14, mas * 4);
-    B.get_ref(0, mas*4);
+    B.get_ref(&coll, 0, mas*4);
+    ASSERT_EQ(mas * 4, B.get_referenced_bytes());
     ASSERT_TRUE(b.is_allocated(0, mas*4));
     B.put_ref(&coll, 0, mas*3, &r);
+    ASSERT_EQ(mas * 1, B.get_referenced_bytes());
     cout << "r " << r << " " << b << std::endl;
     ASSERT_EQ(0u, r.size());
     ASSERT_TRUE(b.is_allocated(0, mas*4));
@@ -689,21 +734,92 @@ TEST(Blob, put_ref)
     B.shared_blob = new BlueStore::SharedBlob(nullptr);
     B.shared_blob->get();  // hack to avoid dtor from running
     bluestore_blob_t& b = B.dirty_blob();
-    B.get_ref(0x0, 0x3800);
-    B.get_ref(0x17c00, 0x6400);
     b.extents.push_back(bluestore_pextent_t(0x40101000, 0x4000));
     b.extents.push_back(bluestore_pextent_t(bluestore_pextent_t::INVALID_OFFSET,
 					    0x13000));
     b.extents.push_back(bluestore_pextent_t(0x40118000, 0x7000));
+    B.get_ref(&coll, 0x0, 0x3800);
+    B.get_ref(&coll, 0x17c00, 0x6400);
+    ASSERT_EQ(0x3800u + 0x6400u, B.get_referenced_bytes());
     b.set_flag(bluestore_blob_t::FLAG_SHARED);
     b.init_csum(Checksummer::CSUM_CRC32C, 12, 0x1e000);
 
     cout << "before: " << B << std::endl;
     PExtentVector r;
     B.put_ref(&coll, 0x1800, 0x2000, &r);
+    ASSERT_EQ(0x3800u + 0x6400u - 0x2000u, B.get_referenced_bytes());
     cout << "after: " << B << std::endl;
     cout << "r " << r << std::endl;
   }
+  {
+    BlueStore::Blob B;
+    B.shared_blob = new BlueStore::SharedBlob(nullptr);
+    B.shared_blob->get();  // hack to avoid dtor from running
+    bluestore_blob_t& b = B.dirty_blob();
+    b.extents.push_back(bluestore_pextent_t(1, 0x5000));
+    b.extents.push_back(bluestore_pextent_t(2, 0x5000));
+    B.get_ref(&coll, 0x0, 0xa000);
+    ASSERT_EQ(0xa000u, B.get_referenced_bytes());
+    cout << "before: " << B << std::endl;
+    PExtentVector r;
+    B.put_ref(&coll, 0x8000, 0x2000, &r);
+    cout << "after: " << B << std::endl;
+    cout << "r " << r << std::endl;
+    ASSERT_EQ(0x8000u, B.get_referenced_bytes());
+    ASSERT_EQ(1u, r.size());
+    ASSERT_EQ(0x3002u, r[0].offset);
+    ASSERT_EQ(0x2000u, r[0].length);
+  }
+  {
+    BlueStore::Blob B;
+    B.shared_blob = new BlueStore::SharedBlob(nullptr);
+    B.shared_blob->get();  // hack to avoid dtor from running
+    bluestore_blob_t& b = B.dirty_blob();
+    b.extents.push_back(bluestore_pextent_t(1, 0x7000));
+    b.extents.push_back(bluestore_pextent_t(2, 0x7000));
+    B.get_ref(&coll, 0x0, 0xe000);
+    ASSERT_EQ(0xe000u, B.get_referenced_bytes());
+    cout << "before: " << B << std::endl;
+    PExtentVector r;
+    B.put_ref(&coll, 0, 0xb000, &r);
+    ASSERT_EQ(0x3000u, B.get_referenced_bytes());
+    cout << "after: " << B << std::endl;
+    cout << "r " << r << std::endl;
+    ASSERT_EQ(0x3000u, B.get_referenced_bytes());
+    ASSERT_EQ(2u, r.size());
+    ASSERT_EQ(1u, r[0].offset);
+    ASSERT_EQ(0x7000u, r[0].length);
+    ASSERT_EQ(2u, r[1].offset);
+    ASSERT_EQ(0x3000u, r[1].length); // we have 0x1000 bytes less due to 
+                                     // alignment caused by min_alloc_size = 0x2000
+  }
+  {
+    BlueStore store(g_ceph_context, "", 0x4000);
+    BlueStore::Cache *cache = BlueStore::Cache::create(
+      g_ceph_context, "lru", NULL);
+    BlueStore::Collection coll(&store, cache, coll_t());
+    BlueStore::Blob B;
+    B.shared_blob = new BlueStore::SharedBlob(nullptr);
+    B.shared_blob->get();  // hack to avoid dtor from running
+    bluestore_blob_t& b = B.dirty_blob();
+    b.extents.push_back(bluestore_pextent_t(1, 0x5000));
+    b.extents.push_back(bluestore_pextent_t(2, 0x7000));
+    B.get_ref(&coll, 0x0, 0xc000);
+    ASSERT_EQ(0xc000u, B.get_referenced_bytes());
+    cout << "before: " << B << std::endl;
+    PExtentVector r;
+    B.put_ref(&coll, 0x2000, 0xa000, &r);
+    cout << "after: " << B << std::endl;
+    cout << "r " << r << std::endl;
+    ASSERT_EQ(0x2000u, B.get_referenced_bytes());
+    ASSERT_EQ(2u, r.size());
+    ASSERT_EQ(0x4001u, r[0].offset);
+    ASSERT_EQ(0x1000u, r[0].length);
+    ASSERT_EQ(2u, r[1].offset);
+    ASSERT_EQ(0x7000u, r[1].length);
+    ASSERT_EQ(1u, b.extents[0].offset);
+    ASSERT_EQ(0x4000u, b.extents[0].length);
+  }
 }
 
 TEST(bluestore_blob_t, can_split)
@@ -780,17 +896,20 @@ TEST(Blob, split)
     R.shared_blob->get();  // hack to avoid dtor from running
     L.dirty_blob().extents.emplace_back(bluestore_pextent_t(0x2000, 0x2000));
     L.dirty_blob().init_csum(Checksummer::CSUM_CRC32C, 12, 0x2000);
+    L.get_ref(&coll, 0, 0x2000);
     L.split(&coll, 0x1000, &R);
     ASSERT_EQ(0x1000u, L.get_blob().get_logical_length());
     ASSERT_EQ(4u, L.get_blob().csum_data.length());
     ASSERT_EQ(1u, L.get_blob().extents.size());
     ASSERT_EQ(0x2000u, L.get_blob().extents.front().offset);
     ASSERT_EQ(0x1000u, L.get_blob().extents.front().length);
+    ASSERT_EQ(0x1000u, L.get_referenced_bytes());
     ASSERT_EQ(0x1000u, R.get_blob().get_logical_length());
     ASSERT_EQ(4u, R.get_blob().csum_data.length());
     ASSERT_EQ(1u, R.get_blob().extents.size());
     ASSERT_EQ(0x3000u, R.get_blob().extents.front().offset);
     ASSERT_EQ(0x1000u, R.get_blob().extents.front().length);
+    ASSERT_EQ(0x1000u, R.get_referenced_bytes());
   }
   {
     BlueStore::Blob L, R;
@@ -801,17 +920,21 @@ TEST(Blob, split)
     L.dirty_blob().extents.emplace_back(bluestore_pextent_t(0x2000, 0x1000));
     L.dirty_blob().extents.emplace_back(bluestore_pextent_t(0x12000, 0x1000));
     L.dirty_blob().init_csum(Checksummer::CSUM_CRC32C, 12, 0x2000);
+    L.get_ref(&coll, 0, 0x1000);
+    L.get_ref(&coll, 0x1000, 0x1000);
     L.split(&coll, 0x1000, &R);
     ASSERT_EQ(0x1000u, L.get_blob().get_logical_length());
     ASSERT_EQ(4u, L.get_blob().csum_data.length());
     ASSERT_EQ(1u, L.get_blob().extents.size());
     ASSERT_EQ(0x2000u, L.get_blob().extents.front().offset);
     ASSERT_EQ(0x1000u, L.get_blob().extents.front().length);
+    ASSERT_EQ(0x1000u, L.get_referenced_bytes());
     ASSERT_EQ(0x1000u, R.get_blob().get_logical_length());
     ASSERT_EQ(4u, R.get_blob().csum_data.length());
     ASSERT_EQ(1u, R.get_blob().extents.size());
     ASSERT_EQ(0x12000u, R.get_blob().extents.front().offset);
     ASSERT_EQ(0x1000u, R.get_blob().extents.front().length);
+    ASSERT_EQ(0x1000u, R.get_referenced_bytes());
   }
 }