osdc/ObjectCacher: *do* pin object during write

This hopefully resolves #3431.

We originally did this in 46897fd4ffc2fe2fc2cc0c39b5d4c510df1e6c25, and
then reverted in caed0e917f8044d389d01de5ee5ecbf0d16ff349.

The current conundrum:
 - commit_set() will issue a write and queue a waiter on a tid
 - discard will discard all BufferHeads and unpin the object
 - trim will try to close and fail assert(ob->can_close())

But:
 - we can't wake the waiter on discard because we don't know what range(s)
   it is waiting for; discard needn't be the whole object.

So: pin the object so it doesn't get trimmed, and unpin when we write.

Adjust can_close() so that it is based on the lru pin status, and assert
that pinned implies the previous conditions are all true.

Signed-off-by: Sage Weil <sage@inktank.com>
Reviewed-by: Sam Lang <sam.lang@inktank.com>

2 files changed, 10 insertions, 6 deletions

diff --git a/src/osdc/ObjectCacher.cc b/src/osdc/ObjectCacher.cc
index c933bad164a..807a35e4fb6 100644
--- a/src/osdc/ObjectCacher.cc
+++ b/src/osdc/ObjectCacher.cc
@@ -734,7 +734,9 @@ void ObjectCacher::bh_write(BufferHead *bh)
 {
   assert(lock.is_locked());
   ldout(cct, 7) << "bh_write " << *bh << dendl;
-  
+
+  bh->ob->get();
+
   // finishers
   C_WriteCommit *oncommit = new C_WriteCommit(this, bh->ob->oloc.pool,
                                               bh->ob->get_soid(), bh->start(), bh->length());
@@ -842,6 +844,8 @@ void ObjectCacher::bh_write_commit(int64_t poolid, sobject_t oid, loff_t start,
 
     // is the entire object set now clean and fully committed?
     ObjectSet *oset = ob->oset;
+    ob->put();
+
     if (flush_set_callback &&
 	was_dirty_or_tx > 0 &&
 	oset->dirty_or_tx == 0) {        // nothing dirty/tx
diff --git a/src/osdc/ObjectCacher.h b/src/osdc/ObjectCacher.h
index b431463cdee..2d180cf817a 100644
--- a/src/osdc/ObjectCacher.h
+++ b/src/osdc/ObjectCacher.h
@@ -215,11 +215,11 @@ class ObjectCacher {
     void set_object_locator(object_locator_t& l) { oloc = l; }
 
     bool can_close() {
-      if (data.empty() &&
-	  waitfor_commit.empty() &&
-	  waitfor_rd.empty() && waitfor_wr.empty() &&
-	  dirty_or_tx == 0) {
-	assert(lru_is_expireable());
+      if (lru_is_expireable()) {
+	assert(data.empty());
+	assert(waitfor_commit.empty());
+	assert(waitfor_rd.empty());
+	assert(waitfor_wr.empty());
 	return true;
       }
       return false;