// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
	// vim: ts=8 sw=2 smarttab
	
	#include <limits.h>
	
	#include "msg/Messenger.h"
	#include "ObjectCacher.h"
	#include "WritebackHandler.h"
	#include "common/errno.h"
	#include "common/perf_counters.h"
	
	#include "include/ceph_assert.h"
	
	#define MAX_FLUSH_UNDER_LOCK 20  ///< max bh's we start writeback on
	#define BUFFER_MEMORY_WEIGHT CEPH_PAGE_SHIFT  // memory usage of BufferHead, count in (1<<n)
	
	using std::chrono::seconds;
					 /// while holding the lock
	
	/*** ObjectCacher::BufferHead ***/
	
	
	/*** ObjectCacher::Object ***/
	
	#define dout_subsys ceph_subsys_objectcacher
	#undef dout_prefix
	#define dout_prefix *_dout << "objectcacher.object(" << oid << ") "
	
	
	
	class ObjectCacher::C_ReadFinish : public Context {
	  ObjectCacher *oc;
	  int64_t poolid;
	  sobject_t oid;
	  loff_t start;
	  uint64_t length;
	  xlist<C_ReadFinish*>::item set_item;
	  bool trust_enoent;
	  ceph_tid_t tid;
	  ZTracer::Trace trace;
	
	public:
	  bufferlist bl;
	  C_ReadFinish(ObjectCacher *c, Object *ob, ceph_tid_t t, loff_t s,
		       uint64_t l, const ZTracer::Trace &trace) :
	    oc(c), poolid(ob->oloc.pool), oid(ob->get_soid()), start(s), length(l),
	    set_item(this), trust_enoent(true),
	    tid(t), trace(trace) {
	    ob->reads.push_back(&set_item);
	  }
	
	  void finish(int r) override {
	    oc->bh_read_finish(poolid, oid, tid, start, length, bl, r, trust_enoent);
	    trace.event("finish");
	
	    // object destructor clears the list
	    if (set_item.is_on_list())
	      set_item.remove_myself();
	  }
	
	  void distrust_enoent() {
	    trust_enoent = false;
	  }
	};
	
	class ObjectCacher::C_RetryRead : public Context {
	  ObjectCacher *oc;
	  OSDRead *rd;
	  ObjectSet *oset;
	  Context *onfinish;
	  ZTracer::Trace trace;
	public:
	  C_RetryRead(ObjectCacher *_oc, OSDRead *r, ObjectSet *os, Context *c,
		      const ZTracer::Trace &trace)
	    : oc(_oc), rd(r), oset(os), onfinish(c), trace(trace) {
	  }
	  void finish(int r) override {
	    if (r >= 0) {
	      r = oc->_readx(rd, oset, onfinish, false, &trace);
	    }
	
	    if (r == 0) {
	      // read is still in-progress
	      return;
	    }
	
	    trace.event("finish");
	    if (onfinish) {
	      onfinish->complete(r);
	    }
	  }
	};
	
	ObjectCacher::BufferHead *ObjectCacher::Object::split(BufferHead *left,
							      loff_t off)
	{
	  ceph_assert(ceph_mutex_is_locked(oc->lock));
	  ldout(oc->cct, 20) << "split " << *left << " at " << off << dendl;
	
	  // split off right
	  ObjectCacher::BufferHead *right = new BufferHead(this);
	
	  //inherit and if later access, this auto clean.
	  right->set_dontneed(left->get_dontneed());
	  right->set_nocache(left->get_nocache());
	
	  right->last_write_tid = left->last_write_tid;
	  right->last_read_tid = left->last_read_tid;
	  right->set_state(left->get_state());
	  right->snapc = left->snapc;
	  right->set_journal_tid(left->journal_tid);
	
	  loff_t newleftlen = off - left->start();
	  right->set_start(off);
	  right->set_length(left->length() - newleftlen);
	
	  // shorten left
	  oc->bh_stat_sub(left);
	  left->set_length(newleftlen);
	  oc->bh_stat_add(left);
	
	  // add right
	  oc->bh_add(this, right);
	
	  // split buffers too
	  bufferlist bl;
	  bl.claim(left->bl);
	  if (bl.length()) {
	    ceph_assert(bl.length() == (left->length() + right->length()));
	    right->bl.substr_of(bl, left->length(), right->length());
	    left->bl.substr_of(bl, 0, left->length());
	  }
	
	  // move read waiters
	  if (!left->waitfor_read.empty()) {
	    map<loff_t, list<Context*> >::iterator start_remove
	      = left->waitfor_read.begin();
	    while (start_remove != left->waitfor_read.end() &&
		   start_remove->first < right->start())
	      ++start_remove;
	    for (map<loff_t, list<Context*> >::iterator p = start_remove;
		 p != left->waitfor_read.end(); ++p) {
	      ldout(oc->cct, 20) << "split  moving waiters at byte " << p->first
				 << " to right bh" << dendl;
	      right->waitfor_read[p->first].swap( p->second );
	      ceph_assert(p->second.empty());
	    }
	    left->waitfor_read.erase(start_remove, left->waitfor_read.end());
	  }
	
	  ldout(oc->cct, 20) << "split    left is " << *left << dendl;
	  ldout(oc->cct, 20) << "split   right is " << *right << dendl;
	  return right;
	}
	
	
	void ObjectCacher::Object::merge_left(BufferHead *left, BufferHead *right)
	{
	  ceph_assert(ceph_mutex_is_locked(oc->lock));
	
	  ldout(oc->cct, 10) << "merge_left " << *left << " + " << *right << dendl;
	  if (left->get_journal_tid() == 0) {
	    left->set_journal_tid(right->get_journal_tid());
	  }
	  right->set_journal_tid(0);
	
	  oc->bh_remove(this, right);
	  oc->bh_stat_sub(left);
	  left->set_length(left->length() + right->length());
	  oc->bh_stat_add(left);
	
	  // data
	  left->bl.claim_append(right->bl);
	
	  // version
	  // note: this is sorta busted, but should only be used for dirty buffers
	  left->last_write_tid =  std::max( left->last_write_tid, right->last_write_tid );
	  left->last_write = std::max( left->last_write, right->last_write );
	
	  left->set_dontneed(right->get_dontneed() ? left->get_dontneed() : false);
	  left->set_nocache(right->get_nocache() ? left->get_nocache() : false);
	
	  // waiters
	  for (map<loff_t, list<Context*> >::iterator p = right->waitfor_read.begin();
	       p != right->waitfor_read.end();
	       ++p)
	    left->waitfor_read[p->first].splice(left->waitfor_read[p->first].begin(),
						p->second );
	
	  // hose right
	  delete right;
	
	  ldout(oc->cct, 10) << "merge_left result " << *left << dendl;
	}
	
	bool ObjectCacher::Object::can_merge_bh(BufferHead *left, BufferHead *right)
	{
	  if (left->end() != right->start() ||
	      left->get_state() != right->get_state() ||
	      !left->can_merge_journal(right))
	    return false;
	  if (left->is_tx() && left->last_write_tid != right->last_write_tid)
	    return false;
	  return true;
	}
	
	void ObjectCacher::Object::try_merge_bh(BufferHead *bh)
	{
	  ceph_assert(ceph_mutex_is_locked(oc->lock));
	  ldout(oc->cct, 10) << "try_merge_bh " << *bh << dendl;
	
	  // do not merge rx buffers; last_read_tid may not match
	  if (bh->is_rx())
	    return;
	
	  // to the left?
	  map<loff_t,BufferHead*>::iterator p = data.find(bh->start());
	  ceph_assert(p->second == bh);
	  if (p != data.begin()) {
	    --p;
	    if (can_merge_bh(p->second, bh)) {
	      merge_left(p->second, bh);
	      bh = p->second;
	    } else {
	      ++p;
	    }
	  }
	  // to the right?
	  ceph_assert(p->second == bh);
	  ++p;
	  if (p != data.end() && can_merge_bh(bh, p->second))
	    merge_left(bh, p->second);
	
	  maybe_rebuild_buffer(bh);
	}
	
	void ObjectCacher::Object::maybe_rebuild_buffer(BufferHead *bh)
	{
	  auto& bl = bh->bl;
	  if (bl.get_num_buffers() <= 1)
	    return;
	
	  auto wasted = bl.get_wasted_space();
	  if (wasted * 2 > bl.length() &&
	      wasted > (1U << BUFFER_MEMORY_WEIGHT))
	    bl.rebuild();
	}
	
	/*
	 * count bytes we have cached in given range
	 */
	bool ObjectCacher::Object::is_cached(loff_t cur, loff_t left) const
	{
	  ceph_assert(ceph_mutex_is_locked(oc->lock));
	  map<loff_t, BufferHead*>::const_iterator p = data_lower_bound(cur);
	  while (left > 0) {
	    if (p == data.end())
	      return false;
	
	    if (p->first <= cur) {
	      // have part of it
	      loff_t lenfromcur = std::min(p->second->end() - cur, left);
	      cur += lenfromcur;
	      left -= lenfromcur;
	      ++p;
	      continue;
	    } else if (p->first > cur) {
	      // gap
	      return false;
	    } else
	      ceph_abort();
	  }
	
	  return true;
	}
	
	/*
	 * all cached data in this range[off, off+len]
	 */
	bool ObjectCacher::Object::include_all_cached_data(loff_t off, loff_t len)
	{
	  ceph_assert(ceph_mutex_is_locked(oc->lock));
	  if (data.empty())
	      return true;
	  map<loff_t, BufferHead*>::iterator first = data.begin();
	  map<loff_t, BufferHead*>::reverse_iterator last = data.rbegin();
	  if (first->second->start() >= off && last->second->end() <= (off + len))
	    return true;
	  else
	    return false;
	}
	
	/*
	 * map a range of bytes into buffer_heads.
	 * - create missing buffer_heads as necessary.
	 */
	int ObjectCacher::Object::map_read(ObjectExtent &ex,
	                                   map<loff_t, BufferHead*>& hits,
	                                   map<loff_t, BufferHead*>& missing,
	                                   map<loff_t, BufferHead*>& rx,
					   map<loff_t, BufferHead*>& errors)
	{
	  ceph_assert(ceph_mutex_is_locked(oc->lock));
	  ldout(oc->cct, 10) << "map_read " << ex.oid << " "
	                     << ex.offset << "~" << ex.length << dendl;
	
	  loff_t cur = ex.offset;
	  loff_t left = ex.length;
	
	  map<loff_t, BufferHead*>::const_iterator p = data_lower_bound(ex.offset);
	  while (left > 0) {
	    // at end?
	    if (p == data.end()) {
	      // rest is a miss.
	      BufferHead *n = new BufferHead(this);
	      n->set_start(cur);
	      n->set_length(left);
	      oc->bh_add(this, n);
	      if (complete) {
	        oc->mark_zero(n);
	        hits[cur] = n;
	        ldout(oc->cct, 20) << "map_read miss+complete+zero " << left << " left, " << *n << dendl;
	      } else {
	        missing[cur] = n;
	        ldout(oc->cct, 20) << "map_read miss " << left << " left, " << *n << dendl;
	      }
	      cur += left;
	      ceph_assert(cur == (loff_t)ex.offset + (loff_t)ex.length);
	      break;  // no more.
	    }
	
	    if (p->first <= cur) {
	      // have it (or part of it)
	      BufferHead *e = p->second;
	
	      if (e->is_clean() ||
	          e->is_dirty() ||
	          e->is_tx() ||
	          e->is_zero()) {
	        hits[cur] = e;     // readable!
	        ldout(oc->cct, 20) << "map_read hit " << *e << dendl;
	      } else if (e->is_rx()) {
	        rx[cur] = e;       // missing, not readable.
	        ldout(oc->cct, 20) << "map_read rx " << *e << dendl;
	      } else if (e->is_error()) {
	        errors[cur] = e;
	        ldout(oc->cct, 20) << "map_read error " << *e << dendl;
	      } else {
	        ceph_abort();
	      }
	
	      loff_t lenfromcur = std::min(e->end() - cur, left);
	      cur += lenfromcur;
	      left -= lenfromcur;
	      ++p;
	      continue;  // more?
	
	    } else if (p->first > cur) {
	      // gap.. miss
	      loff_t next = p->first;
	      BufferHead *n = new BufferHead(this);
	      loff_t len = std::min(next - cur, left);
	      n->set_start(cur);
	      n->set_length(len);
	      oc->bh_add(this,n);
	      if (complete) {
	        oc->mark_zero(n);
	        hits[cur] = n;
	        ldout(oc->cct, 20) << "map_read gap+complete+zero " << *n << dendl;
	      } else {
	        missing[cur] = n;
	        ldout(oc->cct, 20) << "map_read gap " << *n << dendl;
	      }
	      cur += std::min(left, n->length());
	      left -= std::min(left, n->length());
	      continue;    // more?
	    } else {
	      ceph_abort();
	    }
	  }
	  return 0;
	}
	
	void ObjectCacher::Object::audit_buffers()
	{
	  loff_t offset = 0;
	  for (map<loff_t, BufferHead*>::const_iterator it = data.begin();
	       it != data.end(); ++it) {
	    if (it->first != it->second->start()) {
	      lderr(oc->cct) << "AUDIT FAILURE: map position " << it->first
			     << " does not match bh start position: "
			     << *it->second << dendl;
	      ceph_assert(it->first == it->second->start());
	    }
	    if (it->first < offset) {
	      lderr(oc->cct) << "AUDIT FAILURE: " << it->first << " " << *it->second
			     << " overlaps with previous bh " << *((--it)->second)
			     << dendl;
	      ceph_assert(it->first >= offset);
	    }
	    BufferHead *bh = it->second;
	    map<loff_t, list<Context*> >::const_iterator w_it;
	    for (w_it = bh->waitfor_read.begin();
		 w_it != bh->waitfor_read.end(); ++w_it) {
	      if (w_it->first < bh->start() ||
		    w_it->first >= bh->start() + bh->length()) {
		lderr(oc->cct) << "AUDIT FAILURE: waiter at " << w_it->first
			       << " is not within bh " << *bh << dendl;
		ceph_assert(w_it->first >= bh->start());
		ceph_assert(w_it->first < bh->start() + bh->length());
	      }
	    }
	    offset = it->first + it->second->length();
	  }
	}
	
	/*
	 * map a range of extents on an object's buffer cache.
	 * - combine any bh's we're writing into one
	 * - break up bufferheads that don't fall completely within the range
	 * //no! - return a bh that includes the write.  may also include
	 * other dirty data to left and/or right.
	 */
	ObjectCacher::BufferHead *ObjectCacher::Object::map_write(ObjectExtent &ex,
								  ceph_tid_t tid)
	{
	  ceph_assert(ceph_mutex_is_locked(oc->lock));
	  BufferHead *final = 0;
	
	  ldout(oc->cct, 10) << "map_write oex " << ex.oid
	      	       << " " << ex.offset << "~" << ex.length << dendl;
	
	  loff_t cur = ex.offset;
	  loff_t left = ex.length;
	
	  map<loff_t, BufferHead*>::const_iterator p = data_lower_bound(ex.offset);
	  while (left > 0) {
	    loff_t max = left;
	
	    // at end ?
	    if (p == data.end()) {
	      if (final == NULL) {
	        final = new BufferHead(this);
	        replace_journal_tid(final, tid);
	        final->set_start( cur );
	        final->set_length( max );
	        oc->bh_add(this, final);
	        ldout(oc->cct, 10) << "map_write adding trailing bh " << *final << dendl;
	      } else {
	        oc->bh_stat_sub(final);
	        final->set_length(final->length() + max);
	        oc->bh_stat_add(final);
	      }
	      left -= max;
	      cur += max;
	      continue;
	    }
	
	    ldout(oc->cct, 10) << "cur is " << cur << ", p is " << *p->second << dendl;
	    //oc->verify_stats();
	
	    if (p->first <= cur) {
	      BufferHead *bh = p->second;
	      ldout(oc->cct, 10) << "map_write bh " << *bh << " intersected" << dendl;
	
	      if (p->first < cur) {
	        ceph_assert(final == 0);
	        if (cur + max >= bh->end()) {
	          // we want right bit (one splice)
	          final = split(bh, cur);   // just split it, take right half.
	          maybe_rebuild_buffer(bh);
	          replace_journal_tid(final, tid);
	          ++p;
	          ceph_assert(p->second == final);
	        } else {
	          // we want middle bit (two splices)
	          final = split(bh, cur);
	          maybe_rebuild_buffer(bh);
	          ++p;
	          ceph_assert(p->second == final);
	          auto right = split(final, cur+max);
	          maybe_rebuild_buffer(right);
	          replace_journal_tid(final, tid);
	        }
	      } else {
	        ceph_assert(p->first == cur);
	        if (bh->length() <= max) {
	          // whole bufferhead, piece of cake.
	        } else {
	          // we want left bit (one splice)
	          auto right = split(bh, cur + max);        // just split
	          maybe_rebuild_buffer(right);
	        }
	        if (final) {
	          oc->mark_dirty(bh);
	          oc->mark_dirty(final);
	          --p;  // move iterator back to final
	          ceph_assert(p->second == final);
	          replace_journal_tid(bh, tid);
	          merge_left(final, bh);
	        } else {
	          final = bh;
	          replace_journal_tid(final, tid);
	        }
	      }
	
	      // keep going.
	      loff_t lenfromcur = final->end() - cur;
	      cur += lenfromcur;
	      left -= lenfromcur;
	      ++p;
	      continue;
	    } else {
	      // gap!
	      loff_t next = p->first;
	      loff_t glen = std::min(next - cur, max);
	      ldout(oc->cct, 10) << "map_write gap " << cur << "~" << glen << dendl;
	      if (final) {
	        oc->bh_stat_sub(final);
	        final->set_length(final->length() + glen);
	        oc->bh_stat_add(final);
	      } else {
	        final = new BufferHead(this);
		replace_journal_tid(final, tid);
	        final->set_start( cur );
	        final->set_length( glen );
	        oc->bh_add(this, final);
	      }
	
	      cur += glen;
	      left -= glen;
	      continue;    // more?
	    }
	  }
	
	  // set version
	  ceph_assert(final);
	  ceph_assert(final->get_journal_tid() == tid);
	  ldout(oc->cct, 10) << "map_write final is " << *final << dendl;
	
	  return final;
	}
	
	void ObjectCacher::Object::replace_journal_tid(BufferHead *bh,
						       ceph_tid_t tid) {
	  ceph_tid_t bh_tid = bh->get_journal_tid();
	
	  ceph_assert(tid == 0 || bh_tid <= tid);
	  if (bh_tid != 0 && bh_tid != tid) {
	    // inform journal that it should not expect a writeback from this extent
	    oc->writeback_handler.overwrite_extent(get_oid(), bh->start(),
						   bh->length(), bh_tid, tid);
	  }
	  bh->set_journal_tid(tid);
	}
	
	void ObjectCacher::Object::truncate(loff_t s)
	{
	  ceph_assert(ceph_mutex_is_locked(oc->lock));
	  ldout(oc->cct, 10) << "truncate " << *this << " to " << s << dendl;
	
	  while (!data.empty()) {
	    BufferHead *bh = data.rbegin()->second;
	    if (bh->end() <= s)
	      break;
	
	    // split bh at truncation point?
	    if (bh->start() < s) {
	      split(bh, s);
	      maybe_rebuild_buffer(bh);
	      continue;
	    }
	
	    // remove bh entirely
	    ceph_assert(bh->start() >= s);
	    ceph_assert(bh->waitfor_read.empty());
	    replace_journal_tid(bh, 0);
	    oc->bh_remove(this, bh);
	    delete bh;
	  }
	}
	
	void ObjectCacher::Object::discard(loff_t off, loff_t len,
	                                   C_GatherBuilder* commit_gather)
	{
	  ceph_assert(ceph_mutex_is_locked(oc->lock));
	  ldout(oc->cct, 10) << "discard " << *this << " " << off << "~" << len
			     << dendl;
	
	  if (!exists) {
	    ldout(oc->cct, 10) << " setting exists on " << *this << dendl;
	    exists = true;
	  }
	  if (complete) {
	    ldout(oc->cct, 10) << " clearing complete on " << *this << dendl;
	    complete = false;
	  }
	
	  map<loff_t, BufferHead*>::const_iterator p = data_lower_bound(off);
	  while (p != data.end()) {
	    BufferHead *bh = p->second;
	    if (bh->start() >= off + len)
	      break;
	
	    // split bh at truncation point?
	    if (bh->start() < off) {
	      split(bh, off);
	      maybe_rebuild_buffer(bh);
	      ++p;
	      continue;
	    }
	
	    ceph_assert(bh->start() >= off);
	    if (bh->end() > off + len) {
	      auto right = split(bh, off + len);
	      maybe_rebuild_buffer(right);
	    }
	
	    ++p;
	    ldout(oc->cct, 10) << "discard " << *this << " bh " << *bh << dendl;
	    replace_journal_tid(bh, 0);
	
	    if (bh->is_tx() && commit_gather != nullptr) {
	      // wait for the writeback to commit
	      waitfor_commit[bh->last_write_tid].emplace_back(commit_gather->new_sub());
	    } else if (bh->is_rx()) {
	      // cannot remove bh with in-flight read, but we can ensure the
	      // read won't overwrite the discard
	      bh->last_read_tid = ++oc->last_read_tid;
	      bh->bl.clear();
	      bh->set_nocache(true);
	      oc->mark_zero(bh);
	      // we should mark all Rx bh to zero
	      continue;
	    } else {
	      ceph_assert(bh->waitfor_read.empty());
	    }
	
	    oc->bh_remove(this, bh);
	    delete bh;
	  }
	}
	
	
	
	/*** ObjectCacher ***/
	
	#undef dout_prefix
	#define dout_prefix *_dout << "objectcacher "
	
	
	ObjectCacher::ObjectCacher(CephContext *cct_, string name,
				   WritebackHandler& wb, ceph::mutex& l,
				   flush_set_callback_t flush_callback,
				   void *flush_callback_arg, uint64_t max_bytes,
				   uint64_t max_objects, uint64_t max_dirty,
				   uint64_t target_dirty, double max_dirty_age,
				   bool block_writes_upfront)
	  : perfcounter(NULL),
	    cct(cct_), writeback_handler(wb), name(name), lock(l),
	    max_dirty(max_dirty), target_dirty(target_dirty),
	    max_size(max_bytes), max_objects(max_objects),
	    max_dirty_age(ceph::make_timespan(max_dirty_age)),
	    block_writes_upfront(block_writes_upfront),
	    trace_endpoint("ObjectCacher"),
	    flush_set_callback(flush_callback),
	    flush_set_callback_arg(flush_callback_arg),
	    last_read_tid(0), flusher_stop(false), flusher_thread(this),finisher(cct),
	    stat_clean(0), stat_zero(0), stat_dirty(0), stat_rx(0), stat_tx(0),
	    stat_missing(0), stat_error(0), stat_dirty_waiting(0),
	    stat_nr_dirty_waiters(0), reads_outstanding(0)
	{
	  perf_start();
	  finisher.start();
	  scattered_write = writeback_handler.can_scattered_write();
	}
	

	ObjectCacher::~ObjectCacher()
	{
	  finisher.stop();
	  perf_stop();
	  // we should be empty.
	  for (vector<ceph::unordered_map<sobject_t, Object *> >::iterator i
		 = objects.begin();
	       i != objects.end();
	       ++i)

	    ceph_assert(i->empty());
	  ceph_assert(bh_lru_rest.lru_get_size() == 0);
	  ceph_assert(bh_lru_dirty.lru_get_size() == 0);
	  ceph_assert(ob_lru.lru_get_size() == 0);
	  ceph_assert(dirty_or_tx_bh.empty());
	}
	
	void ObjectCacher::perf_start()
	{
	  string n = "objectcacher-" + name;
	  PerfCountersBuilder plb(cct, n, l_objectcacher_first, l_objectcacher_last);
	
	  plb.add_u64_counter(l_objectcacher_cache_ops_hit,
			      "cache_ops_hit", "Hit operations");
	  plb.add_u64_counter(l_objectcacher_cache_ops_miss,
			      "cache_ops_miss", "Miss operations");
	  plb.add_u64_counter(l_objectcacher_cache_bytes_hit,
			      "cache_bytes_hit", "Hit data", NULL, 0, unit_t(UNIT_BYTES));
	  plb.add_u64_counter(l_objectcacher_cache_bytes_miss,
			      "cache_bytes_miss", "Miss data", NULL, 0, unit_t(UNIT_BYTES));
	  plb.add_u64_counter(l_objectcacher_data_read,
			      "data_read", "Read data");
	  plb.add_u64_counter(l_objectcacher_data_written,
			      "data_written", "Data written to cache");
	  plb.add_u64_counter(l_objectcacher_data_flushed,
			      "data_flushed", "Data flushed");
	  plb.add_u64_counter(l_objectcacher_overwritten_in_flush,
			      "data_overwritten_while_flushing",
			      "Data overwritten while flushing");
	  plb.add_u64_counter(l_objectcacher_write_ops_blocked, "write_ops_blocked",
			      "Write operations, delayed due to dirty limits");
	  plb.add_u64_counter(l_objectcacher_write_bytes_blocked,
			      "write_bytes_blocked",
			      "Write data blocked on dirty limit", NULL, 0, unit_t(UNIT_BYTES));
	  plb.add_time(l_objectcacher_write_time_blocked, "write_time_blocked",
		       "Time spent blocking a write due to dirty limits");
	
	  perfcounter = plb.create_perf_counters();
	  cct->get_perfcounters_collection()->add(perfcounter);
	}
	
	void ObjectCacher::perf_stop()
	{
	  ceph_assert(perfcounter);
	  cct->get_perfcounters_collection()->remove(perfcounter);
	  delete perfcounter;
	}
	
	/* private */
	ObjectCacher::Object *ObjectCacher::get_object(sobject_t oid,
						       uint64_t object_no,
						       ObjectSet *oset,
						       object_locator_t &l,
						       uint64_t truncate_size,
						       uint64_t truncate_seq)
	{
	  // XXX: Add handling of nspace in object_locator_t in cache
	  ceph_assert(ceph_mutex_is_locked(lock));
	  // have it?
	  if ((uint32_t)l.pool < objects.size()) {
	    if (objects[l.pool].count(oid)) {
	      Object *o = objects[l.pool][oid];
	      o->object_no = object_no;
	      o->truncate_size = truncate_size;
	      o->truncate_seq = truncate_seq;
	      return o;
	    }
	  } else {
	    objects.resize(l.pool+1);
	  }
	
	  // create it.
	  Object *o = new Object(this, oid, object_no, oset, l, truncate_size,
				 truncate_seq);
	  objects[l.pool][oid] = o;
	  ob_lru.lru_insert_top(o);
	  return o;
	}
	
	void ObjectCacher::close_object(Object *ob)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ldout(cct, 10) << "close_object " << *ob << dendl;
	  ceph_assert(ob->can_close());
	
	  // ok!
	  ob_lru.lru_remove(ob);
	  objects[ob->oloc.pool].erase(ob->get_soid());
	  ob->set_item.remove_myself();
	  delete ob;
	}
	
	void ObjectCacher::bh_read(BufferHead *bh, int op_flags,
	                           const ZTracer::Trace &parent_trace)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ldout(cct, 7) << "bh_read on " << *bh << " outstanding reads "
			<< reads_outstanding << dendl;
	
	  ZTracer::Trace trace;
	  if (parent_trace.valid()) {
	    trace.init("", &trace_endpoint, &parent_trace);
	    trace.copy_name("bh_read " + bh->ob->get_oid().name);
	    trace.event("start");
	  }
	
	  mark_rx(bh);
	  bh->last_read_tid = ++last_read_tid;
	
	  // finisher
	  C_ReadFinish *onfinish = new C_ReadFinish(this, bh->ob, bh->last_read_tid,
						    bh->start(), bh->length(), trace);
	  // go
	  writeback_handler.read(bh->ob->get_oid(), bh->ob->get_object_number(),
				 bh->ob->get_oloc(), bh->start(), bh->length(),
				 bh->ob->get_snap(), &onfinish->bl,
				 bh->ob->truncate_size, bh->ob->truncate_seq,
				 op_flags, trace, onfinish);
	
	  ++reads_outstanding;
	}
	
	void ObjectCacher::bh_read_finish(int64_t poolid, sobject_t oid,
					  ceph_tid_t tid, loff_t start,
					  uint64_t length, bufferlist &bl, int r,
					  bool trust_enoent)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ldout(cct, 7) << "bh_read_finish "
			<< oid
			<< " tid " << tid
			<< " " << start << "~" << length
			<< " (bl is " << bl.length() << ")"
			<< " returned " << r
			<< " outstanding reads " << reads_outstanding
			<< dendl;
	
	  if (r >= 0 && bl.length() < length) {
	    ldout(cct, 7) << "bh_read_finish " << oid << " padding " << start << "~"
			  << length << " with " << length - bl.length() << " bytes of zeroes"
			  << dendl;
	    bl.append_zero(length - bl.length());
	  }
	
	  list<Context*> ls;
	  int err = 0;
	
	  if (objects[poolid].count(oid) == 0) {
	    ldout(cct, 7) << "bh_read_finish no object cache" << dendl;
	  } else {
	    Object *ob = objects[poolid][oid];
	
	    if (r == -ENOENT && !ob->complete) {
	      // wake up *all* rx waiters, or else we risk reordering
	      // identical reads. e.g.
	      //   read 1~1
	      //   reply to unrelated 3~1 -> !exists
	      //   read 1~1 -> immediate ENOENT
	      //   reply to first 1~1 -> ooo ENOENT
	      bool allzero = true;
	      for (map<loff_t, BufferHead*>::iterator p = ob->data.begin();
		   p != ob->data.end(); ++p) {
		BufferHead *bh = p->second;
		for (map<loff_t, list<Context*> >::iterator p
		       = bh->waitfor_read.begin();
		     p != bh->waitfor_read.end();
		     ++p)
		  ls.splice(ls.end(), p->second);
		bh->waitfor_read.clear();
		if (!bh->is_zero() && !bh->is_rx())
		  allzero = false;
	      }
	
	      // just pass through and retry all waiters if we don't trust
	      // -ENOENT for this read
	      if (trust_enoent) {
		ldout(cct, 7)
		  << "bh_read_finish ENOENT, marking complete and !exists on " << *ob
		  << dendl;
		ob->complete = true;
		ob->exists = false;
	
		/* If all the bhs are effectively zero, get rid of them.  All
		 * the waiters will be retried and get -ENOENT immediately, so
		 * it's safe to clean up the unneeded bh's now. Since we know
		 * it's safe to remove them now, do so, so they aren't hanging
		 *around waiting for more -ENOENTs from rados while the cache
		 * is being shut down.
		 *
		 * Only do this when all the bhs are rx or clean, to match the
		 * condition in _readx(). If there are any non-rx or non-clean
		 * bhs, _readx() will wait for the final result instead of
		 * returning -ENOENT immediately.
		 */
		if (allzero) {
		  ldout(cct, 10)
		    << "bh_read_finish ENOENT and allzero, getting rid of "
		    << "bhs for " << *ob << dendl;
		  map<loff_t, BufferHead*>::iterator p = ob->data.begin();
		  while (p != ob->data.end()) {
		    BufferHead *bh = p->second;
		    // current iterator will be invalidated by bh_remove()
		    ++p;
		    bh_remove(ob, bh);
		    delete bh;
		  }
		}
	      }
	    }
	
	    // apply to bh's!
	    loff_t opos = start;
	    while (true) {
	      map<loff_t, BufferHead*>::const_iterator p = ob->data_lower_bound(opos);
	      if (p == ob->data.end())
		break;
	      if (opos >= start+(loff_t)length) {
		ldout(cct, 20) << "break due to opos " << opos << " >= start+length "
			       << start << "+" << length << "=" << start+(loff_t)length
			       << dendl;
		break;
	      }
	
	      BufferHead *bh = p->second;
	      ldout(cct, 20) << "checking bh " << *bh << dendl;
	
	      // finishers?
	      for (map<loff_t, list<Context*> >::iterator it
		     = bh->waitfor_read.begin();
		   it != bh->waitfor_read.end();
		   ++it)
		ls.splice(ls.end(), it->second);
	      bh->waitfor_read.clear();
	
	      if (bh->start() > opos) {
		ldout(cct, 1) << "bh_read_finish skipping gap "
			      << opos << "~" << bh->start() - opos
			      << dendl;
		opos = bh->start();
		continue;
	      }
	
	      if (!bh->is_rx()) {
		ldout(cct, 10) << "bh_read_finish skipping non-rx " << *bh << dendl;
		opos = bh->end();
		continue;
	      }
	
	      if (bh->last_read_tid != tid) {
		ldout(cct, 10) << "bh_read_finish bh->last_read_tid "
			       << bh->last_read_tid << " != tid " << tid
			       << ", skipping" << dendl;
		opos = bh->end();
		continue;
	      }
	
	      ceph_assert(opos >= bh->start());
	      ceph_assert(bh->start() == opos);   // we don't merge rx bh's... yet!
	      ceph_assert(bh->length() <= start+(loff_t)length-opos);
	
	      if (bh->error < 0)
		err = bh->error;
	
	      opos = bh->end();
	
	      if (r == -ENOENT) {
		if (trust_enoent) {
		  ldout(cct, 10) << "bh_read_finish removing " << *bh << dendl;
		  bh_remove(ob, bh);
		  delete bh;
		} else {
		  ldout(cct, 10) << "skipping unstrusted -ENOENT and will retry for "
				 << *bh << dendl;
		}
		continue;
	      }
	
	      if (r < 0) {
		bh->error = r;
		mark_error(bh);
	      } else {
		bh->bl.substr_of(bl,
				 bh->start() - start,
				 bh->length());
		mark_clean(bh);
	      }
	
	      ldout(cct, 10) << "bh_read_finish read " << *bh << dendl;
	
	      ob->try_merge_bh(bh);
	    }
	  }
	
	  // called with lock held.
	  ldout(cct, 20) << "finishing waiters " << ls << dendl;
	
	  finish_contexts(cct, ls, err);
	  retry_waiting_reads();
	
	  --reads_outstanding;
	  read_cond.notify_all();
	}
	
	void ObjectCacher::bh_write_adjacencies(BufferHead *bh, ceph::real_time cutoff,
						int64_t *max_amount, int *max_count)
	{
	  list<BufferHead*> blist;
	
	  int count = 0;
	  int64_t total_len = 0;
	  set<BufferHead*, BufferHead::ptr_lt>::iterator it = dirty_or_tx_bh.find(bh);
	  ceph_assert(it != dirty_or_tx_bh.end());
	  for (set<BufferHead*, BufferHead::ptr_lt>::iterator p = it;
	       p != dirty_or_tx_bh.end();
	       ++p) {
	    BufferHead *obh = *p;
	    if (obh->ob != bh->ob)
	      break;
	    if (obh->is_dirty() && obh->last_write <= cutoff) {
	      blist.push_back(obh);
	      ++count;
	      total_len += obh->length();
	      if ((max_count && count > *max_count) ||
		  (max_amount && total_len > *max_amount))
		break;
	    }
	  }
	
	  while (it != dirty_or_tx_bh.begin()) {
	    --it;
	    BufferHead *obh = *it;
	    if (obh->ob != bh->ob)
	      break;
	    if (obh->is_dirty() && obh->last_write <= cutoff) {
	      blist.push_front(obh);
	      ++count;
	      total_len += obh->length();
	      if ((max_count && count > *max_count) ||
		  (max_amount && total_len > *max_amount))
		break;
	    }
	  }
	  if (max_count)
	    *max_count -= count;
	  if (max_amount)
	    *max_amount -= total_len;
	
	  bh_write_scattered(blist);
	}
	
	class ObjectCacher::C_WriteCommit : public Context {
	  ObjectCacher *oc;
	  int64_t poolid;
	  sobject_t oid;
	  vector<pair<loff_t, uint64_t> > ranges;
	  ZTracer::Trace trace;
	public:
	  ceph_tid_t tid = 0;
	  C_WriteCommit(ObjectCacher *c, int64_t _poolid, sobject_t o, loff_t s,
			uint64_t l, const ZTracer::Trace &trace) :
	    oc(c), poolid(_poolid), oid(o), trace(trace) {
	      ranges.push_back(make_pair(s, l));
	    }
	  C_WriteCommit(ObjectCacher *c, int64_t _poolid, sobject_t o,
			vector<pair<loff_t, uint64_t> >& _ranges) :
	    oc(c), poolid(_poolid), oid(o), tid(0) {
	      ranges.swap(_ranges);
	    }
	  void finish(int r) override {
	    oc->bh_write_commit(poolid, oid, ranges, tid, r);
	    trace.event("finish");
	  }
	};
	void ObjectCacher::bh_write_scattered(list<BufferHead*>& blist)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	
	  Object *ob = blist.front()->ob;
	  ob->get();
	
	  ceph::real_time last_write;
	  SnapContext snapc;
	  vector<pair<loff_t, uint64_t> > ranges;
	  vector<pair<uint64_t, bufferlist> > io_vec;
	
	  ranges.reserve(blist.size());
	  io_vec.reserve(blist.size());
	
	  uint64_t total_len = 0;
	  for (list<BufferHead*>::iterator p = blist.begin(); p != blist.end(); ++p) {
	    BufferHead *bh = *p;
	    ldout(cct, 7) << "bh_write_scattered " << *bh << dendl;
	    ceph_assert(bh->ob == ob);
	    ceph_assert(bh->bl.length() == bh->length());
	    ranges.push_back(pair<loff_t, uint64_t>(bh->start(), bh->length()));
	
	    int n = io_vec.size();
	    io_vec.resize(n + 1);
	    io_vec[n].first = bh->start();
	    io_vec[n].second = bh->bl;
	
	    total_len += bh->length();
	    if (bh->snapc.seq > snapc.seq)
	      snapc = bh->snapc;
	    if (bh->last_write > last_write)
	      last_write = bh->last_write;
	  }
	
	  C_WriteCommit *oncommit = new C_WriteCommit(this, ob->oloc.pool, ob->get_soid(), ranges);
	
	  ceph_tid_t tid = writeback_handler.write(ob->get_oid(), ob->get_oloc(),
						   io_vec, snapc, last_write,
						   ob->truncate_size, ob->truncate_seq,
						   oncommit);
	  oncommit->tid = tid;
	  ob->last_write_tid = tid;
	  for (list<BufferHead*>::iterator p = blist.begin(); p != blist.end(); ++p) {
	    BufferHead *bh = *p;
	    bh->last_write_tid = tid;
	    mark_tx(bh);
	  }
	
	  if (perfcounter)
	    perfcounter->inc(l_objectcacher_data_flushed, total_len);
	}
	
	void ObjectCacher::bh_write(BufferHead *bh, const ZTracer::Trace &parent_trace)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ldout(cct, 7) << "bh_write " << *bh << dendl;
	
	  bh->ob->get();
	
	  ZTracer::Trace trace;
	  if (parent_trace.valid()) {
	    trace.init("", &trace_endpoint, &parent_trace);
	    trace.copy_name("bh_write " + bh->ob->get_oid().name);
	    trace.event("start");
	  }
	
	  // finishers
	  C_WriteCommit *oncommit = new C_WriteCommit(this, bh->ob->oloc.pool,
						      bh->ob->get_soid(), bh->start(),
						      bh->length(), trace);
	  // go
	  ceph_tid_t tid = writeback_handler.write(bh->ob->get_oid(),
						   bh->ob->get_oloc(),
						   bh->start(), bh->length(),
						   bh->snapc, bh->bl, bh->last_write,
						   bh->ob->truncate_size,
						   bh->ob->truncate_seq,
						   bh->journal_tid, trace, oncommit);
	  ldout(cct, 20) << " tid " << tid << " on " << bh->ob->get_oid() << dendl;
	
	  // set bh last_write_tid
	  oncommit->tid = tid;
	  bh->ob->last_write_tid = tid;
	  bh->last_write_tid = tid;
	
	  if (perfcounter) {
	    perfcounter->inc(l_objectcacher_data_flushed, bh->length());
	  }
	
	  mark_tx(bh);
	}
	
	void ObjectCacher::bh_write_commit(int64_t poolid, sobject_t oid,
					   vector<pair<loff_t, uint64_t> >& ranges,
					   ceph_tid_t tid, int r)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ldout(cct, 7) << "bh_write_commit " << oid << " tid " << tid
			<< " ranges " << ranges << " returned " << r << dendl;
	
	  if (objects[poolid].count(oid) == 0) {
	    ldout(cct, 7) << "bh_write_commit no object cache" << dendl;
	    return;
	  }
	
	  Object *ob = objects[poolid][oid];
	  int was_dirty_or_tx = ob->oset->dirty_or_tx;
	
	  for (vector<pair<loff_t, uint64_t> >::iterator p = ranges.begin();
	       p != ranges.end();
	       ++p) {
	    loff_t start = p->first;
	    uint64_t length = p->second;
	    if (!ob->exists) {
	      ldout(cct, 10) << "bh_write_commit marking exists on " << *ob << dendl;
	      ob->exists = true;
	
	      if (writeback_handler.may_copy_on_write(ob->get_oid(), start, length,
						      ob->get_snap())) {
		ldout(cct, 10) << "bh_write_commit may copy on write, clearing "
		  "complete on " << *ob << dendl;
		ob->complete = false;
	      }
	    }
	
	    vector<pair<loff_t, BufferHead*>> hit;
	    // apply to bh's!
	    for (map<loff_t, BufferHead*>::const_iterator p = ob->data_lower_bound(start);
		 p != ob->data.end();
		 ++p) {
	      BufferHead *bh = p->second;
	
	      if (bh->start() >= start+(loff_t)length)
		break;
	
	      // make sure bh is tx
	      if (!bh->is_tx()) {
		ldout(cct, 10) << "bh_write_commit skipping non-tx " << *bh << dendl;
		continue;
	      }
	
	      // make sure bh tid matches
	      if (bh->last_write_tid != tid) {
		ceph_assert(bh->last_write_tid > tid);
		ldout(cct, 10) << "bh_write_commit newer tid on " << *bh << dendl;
		continue;
	      }
	
	      // we don't merge tx buffers. tx buffer should be within the range
	      ceph_assert(bh->start() >= start);
	      ceph_assert(bh->end() <= start+(loff_t)length);
	
	      if (r >= 0) {
		// ok!  mark bh clean and error-free
		mark_clean(bh);
		bh->set_journal_tid(0);
		if (bh->get_nocache())
		  bh_lru_rest.lru_bottouch(bh);
		hit.push_back(make_pair(bh->start(), bh));
		ldout(cct, 10) << "bh_write_commit clean " << *bh << dendl;
	      } else {
		mark_dirty(bh);
		ldout(cct, 10) << "bh_write_commit marking dirty again due to error "
			       << *bh << " r = " << r << " " << cpp_strerror(-r)
			       << dendl;
	      }
	    }
	
	    for (auto& p : hit) {
	      //p.second maybe merged and deleted in merge_left
	      if (ob->data.count(p.first))
		ob->try_merge_bh(p.second);
	    }
	  }
	
	  // update last_commit.
	  ceph_assert(ob->last_commit_tid < tid);
	  ob->last_commit_tid = tid;
	
	  // waiters?
	  list<Context*> ls;
	  if (ob->waitfor_commit.count(tid)) {
	    ls.splice(ls.begin(), ob->waitfor_commit[tid]);
	    ob->waitfor_commit.erase(tid);
	  }
	
	  // 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
	    flush_set_callback(flush_set_callback_arg, oset);
	  }
	
	  if (!ls.empty())
	    finish_contexts(cct, ls, r);
	}
	
	void ObjectCacher::flush(ZTracer::Trace *trace, loff_t amount)
	{
	  ceph_assert(trace != nullptr);
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ceph::real_time cutoff = ceph::real_clock::now();
	
	  ldout(cct, 10) << "flush " << amount << dendl;
	
	  /*
	   * NOTE: we aren't actually pulling things off the LRU here, just
	   * looking at the tail item.  Then we call bh_write, which moves it
	   * to the other LRU, so that we can call
	   * lru_dirty.lru_get_next_expire() again.
	   */
	  int64_t left = amount;
	  while (amount == 0 || left > 0) {
	    BufferHead *bh = static_cast<BufferHead*>(
	      bh_lru_dirty.lru_get_next_expire());
	    if (!bh) break;
	    if (bh->last_write > cutoff) break;
	
	    if (scattered_write) {
	      bh_write_adjacencies(bh, cutoff, amount > 0 ? &left : NULL, NULL);
	    } else {
	      left -= bh->length();
	      bh_write(bh, *trace);
	    }
	  }
	}
	
	
	void ObjectCacher::trim()
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ldout(cct, 10) << "trim  start: bytes: max " << max_size << "  clean "
			 << get_stat_clean() << ", objects: max " << max_objects
			 << " current " << ob_lru.lru_get_size() << dendl;
	
	  uint64_t max_clean_bh = max_size >> BUFFER_MEMORY_WEIGHT;
	  uint64_t nr_clean_bh = bh_lru_rest.lru_get_size() - bh_lru_rest.lru_get_num_pinned();
	  while (get_stat_clean() > 0 &&
		 ((uint64_t)get_stat_clean() > max_size ||
		  nr_clean_bh > max_clean_bh)) {
	    BufferHead *bh = static_cast<BufferHead*>(bh_lru_rest.lru_expire());
	    if (!bh)
	      break;
	
	    ldout(cct, 10) << "trim trimming " << *bh << dendl;
	    ceph_assert(bh->is_clean() || bh->is_zero() || bh->is_error());
	
	    Object *ob = bh->ob;
	    bh_remove(ob, bh);
	    delete bh;
	
	    --nr_clean_bh;
	
	    if (ob->complete) {
	      ldout(cct, 10) << "trim clearing complete on " << *ob << dendl;
	      ob->complete = false;
	    }
	  }
	
	  while (ob_lru.lru_get_size() > max_objects) {
	    Object *ob = static_cast<Object*>(ob_lru.lru_expire());
	    if (!ob)
	      break;
	
	    ldout(cct, 10) << "trim trimming " << *ob << dendl;
	    close_object(ob);
	  }
	
	  ldout(cct, 10) << "trim finish:  max " << max_size << "  clean "
			 << get_stat_clean() << ", objects: max " << max_objects
			 << " current " << ob_lru.lru_get_size() << dendl;
	}
	
	
	
	/* public */
	
	bool ObjectCacher::is_cached(ObjectSet *oset, vector<ObjectExtent>& extents,
				     snapid_t snapid)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  for (vector<ObjectExtent>::iterator ex_it = extents.begin();
	       ex_it != extents.end();
	       ++ex_it) {
	    ldout(cct, 10) << "is_cached " << *ex_it << dendl;
	
	    // get Object cache
	    sobject_t soid(ex_it->oid, snapid);
	    Object *o = get_object_maybe(soid, ex_it->oloc);
	    if (!o)
	      return false;
	    if (!o->is_cached(ex_it->offset, ex_it->length))
	      return false;
	  }
	  return true;
	}
	
	
	/*
	 * returns # bytes read (if in cache).  onfinish is untouched (caller
	 *           must delete it)
	 * returns 0 if doing async read
	 */
	int ObjectCacher::readx(OSDRead *rd, ObjectSet *oset, Context *onfinish,
				ZTracer::Trace *parent_trace)
	{
	  ZTracer::Trace trace;
	  if (parent_trace != nullptr) {
	    trace.init("read", &trace_endpoint, parent_trace);
	    trace.event("start");
	  }
	
	  int r =_readx(rd, oset, onfinish, true, &trace);
	  if (r < 0) {
	    trace.event("finish");
	  }
	  return r;
	}
	
	int ObjectCacher::_readx(OSDRead *rd, ObjectSet *oset, Context *onfinish,
				 bool external_call, ZTracer::Trace *trace)
	{
	  ceph_assert(trace != nullptr);
	  ceph_assert(ceph_mutex_is_locked(lock));
	  bool success = true;
	  int error = 0;
	  uint64_t bytes_in_cache = 0;
	  uint64_t bytes_not_in_cache = 0;
	  uint64_t total_bytes_read = 0;
	  map<uint64_t, bufferlist> stripe_map;  // final buffer offset -> substring
	  bool dontneed = rd->fadvise_flags & LIBRADOS_OP_FLAG_FADVISE_DONTNEED;
	  bool nocache = rd->fadvise_flags & LIBRADOS_OP_FLAG_FADVISE_NOCACHE;
	
	  /*
	   * WARNING: we can only meaningfully return ENOENT if the read request
	   * passed in a single ObjectExtent.  Any caller who wants ENOENT instead of
	   * zeroed buffers needs to feed single extents into readx().
	   */
	  ceph_assert(!oset->return_enoent || rd->extents.size() == 1);
	
	  for (vector<ObjectExtent>::iterator ex_it = rd->extents.begin();
	       ex_it != rd->extents.end();
	       ++ex_it) {
	    ldout(cct, 10) << "readx " << *ex_it << dendl;
	
	    total_bytes_read += ex_it->length;
	
	    // get Object cache
	    sobject_t soid(ex_it->oid, rd->snap);
	    Object *o = get_object(soid, ex_it->objectno, oset, ex_it->oloc,
				   ex_it->truncate_size, oset->truncate_seq);
	    if (external_call)
	      touch_ob(o);
	
	    // does not exist and no hits?
	    if (oset->return_enoent && !o->exists) {
	      ldout(cct, 10) << "readx  object !exists, 1 extent..." << dendl;
	
	      // should we worry about COW underneath us?
	      if (writeback_handler.may_copy_on_write(soid.oid, ex_it->offset,
						      ex_it->length, soid.snap)) {
		ldout(cct, 20) << "readx  may copy on write" << dendl;
		bool wait = false;
		list<BufferHead*> blist;
		for (map<loff_t, BufferHead*>::iterator bh_it = o->data.begin();
		     bh_it != o->data.end();
		     ++bh_it) {
		  BufferHead *bh = bh_it->second;
		  if (bh->is_dirty() || bh->is_tx()) {
		    ldout(cct, 10) << "readx  flushing " << *bh << dendl;
		    wait = true;
		    if (bh->is_dirty()) {
		      if (scattered_write)
			blist.push_back(bh);
		      else
			bh_write(bh, *trace);
		    }
		  }
		}
		if (scattered_write && !blist.empty())
		  bh_write_scattered(blist);
		if (wait) {
		  ldout(cct, 10) << "readx  waiting on tid " << o->last_write_tid
				 << " on " << *o << dendl;
		  o->waitfor_commit[o->last_write_tid].push_back(
		    new C_RetryRead(this,rd, oset, onfinish, *trace));
		  // FIXME: perfcounter!
		  return 0;
		}
	      }
	
	      // can we return ENOENT?
	      bool allzero = true;
	      for (map<loff_t, BufferHead*>::iterator bh_it = o->data.begin();
		   bh_it != o->data.end();
		   ++bh_it) {
		ldout(cct, 20) << "readx  ob has bh " << *bh_it->second << dendl;
		if (!bh_it->second->is_zero() && !bh_it->second->is_rx()) {
		  allzero = false;
		  break;
		}
	      }
	      if (allzero) {
		ldout(cct, 10) << "readx  ob has all zero|rx, returning ENOENT"
			       << dendl;
		delete rd;
		if (dontneed)
		  bottouch_ob(o);
		return -ENOENT;
	      }
	    }
	
	    // map extent into bufferheads
	    map<loff_t, BufferHead*> hits, missing, rx, errors;
	    o->map_read(*ex_it, hits, missing, rx, errors);
	    if (external_call) {
	      // retry reading error buffers
	      missing.insert(errors.begin(), errors.end());
	    } else {
	      // some reads had errors, fail later so completions
	      // are cleaned up properly
	      // TODO: make read path not call _readx for every completion
	      hits.insert(errors.begin(), errors.end());
	    }
	
	    if (!missing.empty() || !rx.empty()) {
	      // read missing
	      map<loff_t, BufferHead*>::iterator last = missing.end();
	      for (map<loff_t, BufferHead*>::iterator bh_it = missing.begin();
		   bh_it != missing.end();
		   ++bh_it) {
		uint64_t rx_bytes = static_cast<uint64_t>(
		  stat_rx + bh_it->second->length());
		bytes_not_in_cache += bh_it->second->length();
		if (!waitfor_read.empty() || (stat_rx > 0 && rx_bytes > max_size)) {
		  // cache is full with concurrent reads -- wait for rx's to complete
		  // to constrain memory growth (especially during copy-ups)
		  if (success) {
		    ldout(cct, 10) << "readx missed, waiting on cache to complete "
				   << waitfor_read.size() << " blocked reads, "
				   << (std::max(rx_bytes, max_size) - max_size)
				   << " read bytes" << dendl;
		    waitfor_read.push_back(new C_RetryRead(this, rd, oset, onfinish,
							   *trace));
		  }
	
		  bh_remove(o, bh_it->second);
		  delete bh_it->second;
		} else {
		  bh_it->second->set_nocache(nocache);
		  bh_read(bh_it->second, rd->fadvise_flags, *trace);
		  if ((success && onfinish) || last != missing.end())
		    last = bh_it;
		}
		success = false;
	      }
	
	      //add wait in last bh avoid wakeup early. Because read is order
	      if (last != missing.end()) {
		ldout(cct, 10) << "readx missed, waiting on " << *last->second
		  << " off " << last->first << dendl;
		last->second->waitfor_read[last->first].push_back(
		  new C_RetryRead(this, rd, oset, onfinish, *trace) );
	
	      }
	
	      // bump rx
	      for (map<loff_t, BufferHead*>::iterator bh_it = rx.begin();
		   bh_it != rx.end();
		   ++bh_it) {
		touch_bh(bh_it->second); // bump in lru, so we don't lose it.
		if (success && onfinish) {
		  ldout(cct, 10) << "readx missed, waiting on " << *bh_it->second
				 << " off " << bh_it->first << dendl;
		  bh_it->second->waitfor_read[bh_it->first].push_back(
		    new C_RetryRead(this, rd, oset, onfinish, *trace) );
		}
		bytes_not_in_cache += bh_it->second->length();
		success = false;
	      }
	
	      for (map<loff_t, BufferHead*>::iterator bh_it = hits.begin();
		   bh_it != hits.end();  ++bh_it)
		//bump in lru, so we don't lose it when later read
		touch_bh(bh_it->second);
	
	    } else {
	      ceph_assert(!hits.empty());
	
	      // make a plain list
	      for (map<loff_t, BufferHead*>::iterator bh_it = hits.begin();
		   bh_it != hits.end();
		   ++bh_it) {
		BufferHead *bh = bh_it->second;
		ldout(cct, 10) << "readx hit bh " << *bh << dendl;
		if (bh->is_error() && bh->error)
		  error = bh->error;
		bytes_in_cache += bh->length();
	
		if (bh->get_nocache() && bh->is_clean())
		  bh_lru_rest.lru_bottouch(bh);
		else
		  touch_bh(bh);
		//must be after touch_bh because touch_bh set dontneed false
		if (dontneed &&
		    ((loff_t)ex_it->offset <= bh->start() &&
		     (bh->end() <=(loff_t)(ex_it->offset + ex_it->length)))) {
		  bh->set_dontneed(true); //if dirty
		  if (bh->is_clean())
		    bh_lru_rest.lru_bottouch(bh);
		}
	      }
	
	      if (!error) {
		// create reverse map of buffer offset -> object for the
		// eventual result.  this is over a single ObjectExtent, so we
		// know that
		//  - the bh's are contiguous
		//  - the buffer frags need not be (and almost certainly aren't)
		loff_t opos = ex_it->offset;
		map<loff_t, BufferHead*>::iterator bh_it = hits.begin();
		ceph_assert(bh_it->second->start() <= opos);
		uint64_t bhoff = opos - bh_it->second->start();
		vector<pair<uint64_t,uint64_t> >::iterator f_it
		  = ex_it->buffer_extents.begin();
		uint64_t foff = 0;
		while (1) {
		  BufferHead *bh = bh_it->second;
		  ceph_assert(opos == (loff_t)(bh->start() + bhoff));
	
		  uint64_t len = std::min(f_it->second - foff, bh->length() - bhoff);
		  ldout(cct, 10) << "readx rmap opos " << opos << ": " << *bh << " +"
				 << bhoff << " frag " << f_it->first << "~"
				 << f_it->second << " +" << foff << "~" << len
				 << dendl;
	
		  bufferlist bit;
		  // put substr here first, since substr_of clobbers, and we
		  // may get multiple bh's at this stripe_map position
		  if (bh->is_zero()) {
		    stripe_map[f_it->first].append_zero(len);
		  } else {
		    bit.substr_of(bh->bl,
			opos - bh->start(),
			len);
		    stripe_map[f_it->first].claim_append(bit);
		  }
	
		  opos += len;
		  bhoff += len;
		  foff += len;
		  if (opos == bh->end()) {
		    ++bh_it;
		    bhoff = 0;
		  }
		  if (foff == f_it->second) {
		    ++f_it;
		    foff = 0;
		  }
		  if (bh_it == hits.end()) break;
		  if (f_it == ex_it->buffer_extents.end())
		    break;
		}
		ceph_assert(f_it == ex_it->buffer_extents.end());
		ceph_assert(opos == (loff_t)ex_it->offset + (loff_t)ex_it->length);
	      }
	
	      if (dontneed && o->include_all_cached_data(ex_it->offset, ex_it->length))
		  bottouch_ob(o);
	    }
	  }
	
	  if (!success) {
	    if (perfcounter && external_call) {
	      perfcounter->inc(l_objectcacher_data_read, total_bytes_read);
	      perfcounter->inc(l_objectcacher_cache_bytes_miss, bytes_not_in_cache);
	      perfcounter->inc(l_objectcacher_cache_ops_miss);
	    }
	    if (onfinish) {
	      ldout(cct, 20) << "readx defer " << rd << dendl;
	    } else {
	      ldout(cct, 20) << "readx drop " << rd << " (no complete, but no waiter)"
			     << dendl;
	      delete rd;
	    }
	    return 0;  // wait!
	  }
	  if (perfcounter && external_call) {
	    perfcounter->inc(l_objectcacher_data_read, total_bytes_read);
	    perfcounter->inc(l_objectcacher_cache_bytes_hit, bytes_in_cache);
	    perfcounter->inc(l_objectcacher_cache_ops_hit);
	  }
	
	  // no misses... success!  do the read.
	  ldout(cct, 10) << "readx has all buffers" << dendl;
	
	  // ok, assemble into result buffer.
	  uint64_t pos = 0;
	  if (rd->bl && !error) {
	    rd->bl->clear();
	    for (map<uint64_t,bufferlist>::iterator i = stripe_map.begin();
		 i != stripe_map.end();
		 ++i) {
	      ceph_assert(pos == i->first);
	      ldout(cct, 10) << "readx  adding buffer len " << i->second.length()
			     << " at " << pos << dendl;
	      pos += i->second.length();
	      rd->bl->claim_append(i->second);
	      ceph_assert(rd->bl->length() == pos);
	    }
	    ldout(cct, 10) << "readx  result is " << rd->bl->length() << dendl;
	  } else if (!error) {
	    ldout(cct, 10) << "readx  no bufferlist ptr (readahead?), done." << dendl;
	    map<uint64_t,bufferlist>::reverse_iterator i = stripe_map.rbegin();
	    pos = i->first + i->second.length();
	  }
	
	  // done with read.
	  int ret = error ? error : pos;
	  ldout(cct, 20) << "readx done " << rd << " " << ret << dendl;
	  ceph_assert(pos <= (uint64_t) INT_MAX);
	
	  delete rd;
	
	  trim();
	
	  return ret;
	}
	
	void ObjectCacher::retry_waiting_reads()
	{
	  list<Context *> ls;
	  ls.swap(waitfor_read);
	
	  while (!ls.empty() && waitfor_read.empty()) {
	    Context *ctx = ls.front();
	    ls.pop_front();
	    ctx->complete(0);
	  }
	  waitfor_read.splice(waitfor_read.end(), ls);
	}
	
	int ObjectCacher::writex(OSDWrite *wr, ObjectSet *oset, Context *onfreespace,
				 ZTracer::Trace *parent_trace)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ceph::real_time now = ceph::real_clock::now();
	  uint64_t bytes_written = 0;
	  uint64_t bytes_written_in_flush = 0;
	  bool dontneed = wr->fadvise_flags & LIBRADOS_OP_FLAG_FADVISE_DONTNEED;
	  bool nocache = wr->fadvise_flags & LIBRADOS_OP_FLAG_FADVISE_NOCACHE;
	
	  ZTracer::Trace trace;
	  if (parent_trace != nullptr) {
	    trace.init("write", &trace_endpoint, parent_trace);
	    trace.event("start");
	  }
	
	  for (vector<ObjectExtent>::iterator ex_it = wr->extents.begin();
	       ex_it != wr->extents.end();
	       ++ex_it) {
	    // get object cache
	    sobject_t soid(ex_it->oid, CEPH_NOSNAP);
	    Object *o = get_object(soid, ex_it->objectno, oset, ex_it->oloc,
				   ex_it->truncate_size, oset->truncate_seq);
	
	    // map it all into a single bufferhead.
	    BufferHead *bh = o->map_write(*ex_it, wr->journal_tid);
	    bool missing = bh->is_missing();
	    bh->snapc = wr->snapc;
	
	    bytes_written += ex_it->length;
	    if (bh->is_tx()) {
	      bytes_written_in_flush += ex_it->length;
	    }
	
	    // adjust buffer pointers (ie "copy" data into my cache)
	    // this is over a single ObjectExtent, so we know that
	    //  - there is one contiguous bh
	    //  - the buffer frags need not be (and almost certainly aren't)
	    // note: i assume striping is monotonic... no jumps backwards, ever!
	    loff_t opos = ex_it->offset;
	    for (vector<pair<uint64_t, uint64_t> >::iterator f_it
		   = ex_it->buffer_extents.begin();
		 f_it != ex_it->buffer_extents.end();
		 ++f_it) {
	      ldout(cct, 10) << "writex writing " << f_it->first << "~"
			     << f_it->second << " into " << *bh << " at " << opos
			     << dendl;
	      uint64_t bhoff = opos - bh->start();
	      ceph_assert(f_it->second <= bh->length() - bhoff);
	
	      // get the frag we're mapping in
	      bufferlist frag;
	      frag.substr_of(wr->bl, f_it->first, f_it->second);
	
	      // keep anything left of bhoff
	      if (!bhoff)
	        bh->bl.swap(frag);
	      else
	        bh->bl.claim_append(frag);
	
	      opos += f_it->second;
	    }
	
	    // ok, now bh is dirty.
	    mark_dirty(bh);
	    if (dontneed)
	      bh->set_dontneed(true);
	    else if (nocache && missing)
	      bh->set_nocache(true);
	    else
	      touch_bh(bh);
	
	    bh->last_write = now;
	
	    o->try_merge_bh(bh);
	  }
	
	  if (perfcounter) {
	    perfcounter->inc(l_objectcacher_data_written, bytes_written);
	    if (bytes_written_in_flush) {
	      perfcounter->inc(l_objectcacher_overwritten_in_flush,
			       bytes_written_in_flush);
	    }
	  }
	
	  int r = _wait_for_write(wr, bytes_written, oset, &trace, onfreespace);
	  delete wr;
	
	  //verify_stats();
	  trim();
	  return r;
	}
	
	class ObjectCacher::C_WaitForWrite : public Context {
	public:
	  C_WaitForWrite(ObjectCacher *oc, uint64_t len,
	                 const ZTracer::Trace &trace, Context *onfinish) :
	    m_oc(oc), m_len(len), m_trace(trace), m_onfinish(onfinish) {}
	  void finish(int r) override;
	private:
	  ObjectCacher *m_oc;
	  uint64_t m_len;
	  ZTracer::Trace m_trace;
	  Context *m_onfinish;
	};
	
	void ObjectCacher::C_WaitForWrite::finish(int r)
	{
	  std::lock_guard l(m_oc->lock);
	  m_oc->_maybe_wait_for_writeback(m_len, &m_trace);
	  m_onfinish->complete(r);
	}
	
	void ObjectCacher::_maybe_wait_for_writeback(uint64_t len,
						     ZTracer::Trace *trace)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ceph::mono_time start = ceph::mono_clock::now();
	  int blocked = 0;
	  // wait for writeback?
	  //  - wait for dirty and tx bytes (relative to the max_dirty threshold)
	  //  - do not wait for bytes other waiters are waiting on.  this means that
	  //    threads do not wait for each other.  this effectively allows the cache
	  //    size to balloon proportional to the data that is in flight.
	
	  uint64_t max_dirty_bh = max_dirty >> BUFFER_MEMORY_WEIGHT;
	  while (get_stat_dirty() + get_stat_tx() > 0 &&
		 (((uint64_t)(get_stat_dirty() + get_stat_tx()) >=
		  max_dirty + get_stat_dirty_waiting()) ||
		 (dirty_or_tx_bh.size() >=
		  max_dirty_bh + get_stat_nr_dirty_waiters()))) {
	
	    if (blocked == 0) {
	      trace->event("start wait for writeback");
	    }
	    ldout(cct, 10) << __func__ << " waiting for dirty|tx "
			   << (get_stat_dirty() + get_stat_tx()) << " >= max "
			   << max_dirty << " + dirty_waiting "
			   << get_stat_dirty_waiting() << dendl;
	    flusher_cond.notify_all();
	    stat_dirty_waiting += len;
	    ++stat_nr_dirty_waiters;
	    std::unique_lock l{lock, std::adopt_lock};
	    stat_cond.wait(l);
	    l.release();
	    stat_dirty_waiting -= len;
	    --stat_nr_dirty_waiters;
	    ++blocked;
	    ldout(cct, 10) << __func__ << " woke up" << dendl;
	  }
	  if (blocked > 0) {
	    trace->event("finish wait for writeback");
	  }
	  if (blocked && perfcounter) {
	    perfcounter->inc(l_objectcacher_write_ops_blocked);
	    perfcounter->inc(l_objectcacher_write_bytes_blocked, len);
	    ceph::timespan blocked = ceph::mono_clock::now() - start;
	    perfcounter->tinc(l_objectcacher_write_time_blocked, blocked);
	  }
	}
	
	// blocking wait for write.
	int ObjectCacher::_wait_for_write(OSDWrite *wr, uint64_t len, ObjectSet *oset,
					  ZTracer::Trace *trace, Context *onfreespace)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ceph_assert(trace != nullptr);
	  int ret = 0;
	
	  if (max_dirty > 0 && !(wr->fadvise_flags & LIBRADOS_OP_FLAG_FADVISE_FUA)) {
	    if (block_writes_upfront) {
	      _maybe_wait_for_writeback(len, trace);
	      if (onfreespace)
		onfreespace->complete(0);
	    } else {
	      ceph_assert(onfreespace);
	      finisher.queue(new C_WaitForWrite(this, len, *trace, onfreespace));
	    }
	  } else {
	    // write-thru!  flush what we just wrote.
	    ceph::condition_variable cond;
	    bool done = false;
	    Context *fin = block_writes_upfront ?
	      new C_Cond(cond, &done, &ret) : onfreespace;
	    ceph_assert(fin);
	    bool flushed = flush_set(oset, wr->extents, trace, fin);
	    ceph_assert(!flushed);   // we just dirtied it, and didn't drop our lock!
	    ldout(cct, 10) << "wait_for_write waiting on write-thru of " << len
			   << " bytes" << dendl;
	    if (block_writes_upfront) {
	      std::unique_lock l{lock, std::adopt_lock};
	      cond.wait(l, [&done] { return done; });
	      l.release();
	      ldout(cct, 10) << "wait_for_write woke up, ret " << ret << dendl;
	      if (onfreespace)
		onfreespace->complete(ret);
	    }
	  }
	
	  // start writeback anyway?
	  if (get_stat_dirty() > 0 && (uint64_t) get_stat_dirty() > target_dirty) {
	    ldout(cct, 10) << "wait_for_write " << get_stat_dirty() << " > target "
			   << target_dirty << ", nudging flusher" << dendl;
	    flusher_cond.notify_all();
	  }
	  return ret;
	}
	
	void ObjectCacher::flusher_entry()
	{
	  ldout(cct, 10) << "flusher start" << dendl;
	  std::unique_lock l{lock};
	  while (!flusher_stop) {
	    loff_t all = get_stat_tx() + get_stat_rx() + get_stat_clean() +
	      get_stat_dirty();
	    ldout(cct, 11) << "flusher "
			   << all << " / " << max_size << ":  "
			   << get_stat_tx() << " tx, "
			   << get_stat_rx() << " rx, "
			   << get_stat_clean() << " clean, "
			   << get_stat_dirty() << " dirty ("
			   << target_dirty << " target, "
			   << max_dirty << " max)"
			   << dendl;
	    loff_t actual = get_stat_dirty() + get_stat_dirty_waiting();
	
	    ZTracer::Trace trace;
	    if (cct->_conf->osdc_blkin_trace_all) {
	      trace.init("flusher", &trace_endpoint);
	      trace.event("start");
	    }
	
	    if (actual > 0 && (uint64_t) actual > target_dirty) {
	      // flush some dirty pages
	      ldout(cct, 10) << "flusher " << get_stat_dirty() << " dirty + "
			     << get_stat_dirty_waiting() << " dirty_waiting > target "
			     << target_dirty << ", flushing some dirty bhs" << dendl;
	      flush(&trace, actual - target_dirty);
	    } else {
	      // check tail of lru for old dirty items
	      ceph::real_time cutoff = ceph::real_clock::now();
	      cutoff -= max_dirty_age;
	      BufferHead *bh = 0;
	      int max = MAX_FLUSH_UNDER_LOCK;
	      while ((bh = static_cast<BufferHead*>(bh_lru_dirty.
						    lru_get_next_expire())) != 0 &&
		     bh->last_write <= cutoff &&
		     max > 0) {
		ldout(cct, 10) << "flusher flushing aged dirty bh " << *bh << dendl;
		if (scattered_write) {
		  bh_write_adjacencies(bh, cutoff, NULL, &max);
	        } else {
		  bh_write(bh, trace);
		  --max;
		}
	      }
	      if (!max) {
		// back off the lock to avoid starving other threads
	        trace.event("backoff");
		l.unlock();
		l.lock();
		continue;
	      }
	    }
	
	    trace.event("finish");
	    if (flusher_stop)
	      break;
	
	    flusher_cond.wait_for(l, 1s);
	  }
	
	  /* Wait for reads to finish. This is only possible if handling
	   * -ENOENT made some read completions finish before their rados read
	   * came back. If we don't wait for them, and destroy the cache, when
	   * the rados reads do come back their callback will try to access the
	   * no-longer-valid ObjectCacher.
	   */
	  read_cond.wait(l, [this] {
	    if (reads_outstanding > 0) {
	      ldout(cct, 10) << "Waiting for all reads to complete. Number left: "
			     << reads_outstanding << dendl;
	      return false;
	    } else {
	      return true;
	    }
	  });
	  ldout(cct, 10) << "flusher finish" << dendl;
	}
	
	
	// -------------------------------------------------
	
	bool ObjectCacher::set_is_empty(ObjectSet *oset)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  if (oset->objects.empty())
	    return true;
	
	  for (xlist<Object*>::iterator p = oset->objects.begin(); !p.end(); ++p)
	    if (!(*p)->is_empty())
	      return false;
	
	  return true;
	}
	
	bool ObjectCacher::set_is_cached(ObjectSet *oset)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  if (oset->objects.empty())
	    return false;
	
	  for (xlist<Object*>::iterator p = oset->objects.begin();
	       !p.end(); ++p) {
	    Object *ob = *p;
	    for (map<loff_t,BufferHead*>::iterator q = ob->data.begin();
		 q != ob->data.end();
		 ++q) {
	      BufferHead *bh = q->second;
	      if (!bh->is_dirty() && !bh->is_tx())
		return true;
	    }
	  }
	
	  return false;
	}
	
	bool ObjectCacher::set_is_dirty_or_committing(ObjectSet *oset)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  if (oset->objects.empty())
	    return false;
	
	  for (xlist<Object*>::iterator i = oset->objects.begin();
	       !i.end(); ++i) {
	    Object *ob = *i;
	
	    for (map<loff_t,BufferHead*>::iterator p = ob->data.begin();
		 p != ob->data.end();
		 ++p) {
	      BufferHead *bh = p->second;
	      if (bh->is_dirty() || bh->is_tx())
		return true;
	    }
	  }
	
	  return false;
	}
	
	
	// purge.  non-blocking.  violently removes dirty buffers from cache.
	void ObjectCacher::purge(Object *ob)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ldout(cct, 10) << "purge " << *ob << dendl;
	
	  ob->truncate(0);
	}
	
	
	// flush.  non-blocking.  no callback.
	// true if clean, already flushed.
	// false if we wrote something.
	// be sloppy about the ranges and flush any buffer it touches
	bool ObjectCacher::flush(Object *ob, loff_t offset, loff_t length,
	                         ZTracer::Trace *trace)
	{
	  ceph_assert(trace != nullptr);
	  ceph_assert(ceph_mutex_is_locked(lock));
	  list<BufferHead*> blist;
	  bool clean = true;
	  ldout(cct, 10) << "flush " << *ob << " " << offset << "~" << length << dendl;
	  for (map<loff_t,BufferHead*>::const_iterator p = ob->data_lower_bound(offset);
	       p != ob->data.end();
	       ++p) {
	    BufferHead *bh = p->second;
	    ldout(cct, 20) << "flush  " << *bh << dendl;
	    if (length && bh->start() > offset+length) {
	      break;
	    }
	    if (bh->is_tx()) {
	      clean = false;
	      continue;
	    }
	    if (!bh->is_dirty()) {
	      continue;
	    }
	
	    if (scattered_write)
	      blist.push_back(bh);
	    else
	      bh_write(bh, *trace);
	    clean = false;
	  }
	  if (scattered_write && !blist.empty())
	    bh_write_scattered(blist);
	
	  return clean;
	}
	
	bool ObjectCacher::_flush_set_finish(C_GatherBuilder *gather,
					     Context *onfinish)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  if (gather->has_subs()) {
	    gather->set_finisher(onfinish);
	    gather->activate();
	    return false;
	  }
	
	  ldout(cct, 10) << "flush_set has no dirty|tx bhs" << dendl;
	  onfinish->complete(0);
	  return true;
	}
	
	// flush.  non-blocking, takes callback.
	// returns true if already flushed
	bool ObjectCacher::flush_set(ObjectSet *oset, Context *onfinish)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ceph_assert(onfinish != NULL);
	  if (oset->objects.empty()) {
	    ldout(cct, 10) << "flush_set on " << oset << " dne" << dendl;
	    onfinish->complete(0);
	    return true;
	  }
	
	  ldout(cct, 10) << "flush_set " << oset << dendl;
	
	  // we'll need to wait for all objects to flush!
	  C_GatherBuilder gather(cct);
	  set<Object*> waitfor_commit;
	
	  list<BufferHead*> blist;
	  Object *last_ob = NULL;
	  set<BufferHead*, BufferHead::ptr_lt>::const_iterator it, p, q;
	
	  // Buffer heads in dirty_or_tx_bh are sorted in ObjectSet/Object/offset
	  // order. But items in oset->objects are not sorted. So the iterator can
	  // point to any buffer head in the ObjectSet
	  BufferHead key(*oset->objects.begin());
	  it = dirty_or_tx_bh.lower_bound(&key);
	  p = q = it;
	
	  bool backwards = true;
	  if (it != dirty_or_tx_bh.begin())
	    --it;
	  else
	    backwards = false;
	
	  for (; p != dirty_or_tx_bh.end(); p = q) {
	    ++q;
	    BufferHead *bh = *p;
	    if (bh->ob->oset != oset)
	      break;
	    waitfor_commit.insert(bh->ob);
	    if (bh->is_dirty()) {
	      if (scattered_write) {
		if (last_ob != bh->ob) {
		  if (!blist.empty()) {
		    bh_write_scattered(blist);
		    blist.clear();
		  }
		  last_ob = bh->ob;
		}
		blist.push_back(bh);
	      } else {
		bh_write(bh, {});
	      }
	    }
	  }
	
	  if (backwards) {
	    for(p = q = it; true; p = q) {
	      if (q != dirty_or_tx_bh.begin())
		--q;
	      else
		backwards = false;
	      BufferHead *bh = *p;
	      if (bh->ob->oset != oset)
		break;
	      waitfor_commit.insert(bh->ob);
	      if (bh->is_dirty()) {
		if (scattered_write) {
		  if (last_ob != bh->ob) {
		    if (!blist.empty()) {
		      bh_write_scattered(blist);
		      blist.clear();
		    }
		    last_ob = bh->ob;
		  }
		  blist.push_front(bh);
		} else {
		  bh_write(bh, {});
		}
	      }
	      if (!backwards)
		break;
	    }
	  }
	
	  if (scattered_write && !blist.empty())
	    bh_write_scattered(blist);
	
	  for (set<Object*>::iterator i = waitfor_commit.begin();
	       i != waitfor_commit.end(); ++i) {
	    Object *ob = *i;
	
	    // we'll need to gather...
	    ldout(cct, 10) << "flush_set " << oset << " will wait for ack tid "
			   << ob->last_write_tid << " on " << *ob << dendl;
	    ob->waitfor_commit[ob->last_write_tid].push_back(gather.new_sub());
	  }
	
	  return _flush_set_finish(&gather, onfinish);
	}
	
	// flush.  non-blocking, takes callback.
	// returns true if already flushed
	bool ObjectCacher::flush_set(ObjectSet *oset, vector<ObjectExtent>& exv,
				     ZTracer::Trace *trace, Context *onfinish)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ceph_assert(trace != nullptr);
	  ceph_assert(onfinish != NULL);
	  if (oset->objects.empty()) {
	    ldout(cct, 10) << "flush_set on " << oset << " dne" << dendl;
	    onfinish->complete(0);
	    return true;
	  }
	
	  ldout(cct, 10) << "flush_set " << oset << " on " << exv.size()
			 << " ObjectExtents" << dendl;
	
	  // we'll need to wait for all objects to flush!
	  C_GatherBuilder gather(cct);
	
	  for (vector<ObjectExtent>::iterator p = exv.begin();
	       p != exv.end();
	       ++p) {
	    ObjectExtent &ex = *p;
	    sobject_t soid(ex.oid, CEPH_NOSNAP);
	    if (objects[oset->poolid].count(soid) == 0)
	      continue;
	    Object *ob = objects[oset->poolid][soid];
	
	    ldout(cct, 20) << "flush_set " << oset << " ex " << ex << " ob " << soid
			   << " " << ob << dendl;
	
	    if (!flush(ob, ex.offset, ex.length, trace)) {
	      // we'll need to gather...
	      ldout(cct, 10) << "flush_set " << oset << " will wait for ack tid "
			     << ob->last_write_tid << " on " << *ob << dendl;
	      ob->waitfor_commit[ob->last_write_tid].push_back(gather.new_sub());
	    }
	  }
	
	  return _flush_set_finish(&gather, onfinish);
	}
	
	// flush all dirty data.  non-blocking, takes callback.
	// returns true if already flushed
	bool ObjectCacher::flush_all(Context *onfinish)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ceph_assert(onfinish != NULL);
	
	  ldout(cct, 10) << "flush_all " << dendl;
	
	  // we'll need to wait for all objects to flush!
	  C_GatherBuilder gather(cct);
	  set<Object*> waitfor_commit;
	
	  list<BufferHead*> blist;
	  Object *last_ob = NULL;
	  set<BufferHead*, BufferHead::ptr_lt>::iterator next, it;
	  next = it = dirty_or_tx_bh.begin();
	  while (it != dirty_or_tx_bh.end()) {
	    ++next;
	    BufferHead *bh = *it;
	    waitfor_commit.insert(bh->ob);
	
	    if (bh->is_dirty()) {
	      if (scattered_write) {
		if (last_ob != bh->ob) {
		  if (!blist.empty()) {
		    bh_write_scattered(blist);
		    blist.clear();
		  }
		  last_ob = bh->ob;
		}
		blist.push_back(bh);
	      } else {
		bh_write(bh, {});
	      }
	    }
	
	    it = next;
	  }
	
	  if (scattered_write && !blist.empty())
	    bh_write_scattered(blist);
	
	  for (set<Object*>::iterator i = waitfor_commit.begin();
	       i != waitfor_commit.end();
	       ++i) {
	    Object *ob = *i;
	
	    // we'll need to gather...
	    ldout(cct, 10) << "flush_all will wait for ack tid "
			   << ob->last_write_tid << " on " << *ob << dendl;
	    ob->waitfor_commit[ob->last_write_tid].push_back(gather.new_sub());
	  }
	
	  return _flush_set_finish(&gather, onfinish);
	}
	
	void ObjectCacher::purge_set(ObjectSet *oset)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  if (oset->objects.empty()) {
	    ldout(cct, 10) << "purge_set on " << oset << " dne" << dendl;
	    return;
	  }
	
	  ldout(cct, 10) << "purge_set " << oset << dendl;
	  const bool were_dirty = oset->dirty_or_tx > 0;
	
	  for (xlist<Object*>::iterator i = oset->objects.begin();
	       !i.end(); ++i) {
	    Object *ob = *i;
		purge(ob);
	  }
	
	  // Although we have purged rather than flushed, caller should still
	  // drop any resources associate with dirty data.
	  ceph_assert(oset->dirty_or_tx == 0);
	  if (flush_set_callback && were_dirty) {
	    flush_set_callback(flush_set_callback_arg, oset);
	  }
	}
	
	
	loff_t ObjectCacher::release(Object *ob)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  list<BufferHead*> clean;
	  loff_t o_unclean = 0;
	
	  for (map<loff_t,BufferHead*>::iterator p = ob->data.begin();
	       p != ob->data.end();
	       ++p) {
	    BufferHead *bh = p->second;
	    if (bh->is_clean() || bh->is_zero() || bh->is_error())
	      clean.push_back(bh);
	    else
	      o_unclean += bh->length();
	  }
	
	  for (list<BufferHead*>::iterator p = clean.begin();
	       p != clean.end();
	       ++p) {
	    bh_remove(ob, *p);
	    delete *p;
	  }
	
	  if (ob->can_close()) {
	    ldout(cct, 10) << "release trimming " << *ob << dendl;
	    close_object(ob);
	    ceph_assert(o_unclean == 0);
	    return 0;
	  }
	
	  if (ob->complete) {
	    ldout(cct, 10) << "release clearing complete on " << *ob << dendl;
	    ob->complete = false;
	  }
	  if (!ob->exists) {
	    ldout(cct, 10) << "release setting exists on " << *ob << dendl;
	    ob->exists = true;
	  }
	
	  return o_unclean;
	}
	
	loff_t ObjectCacher::release_set(ObjectSet *oset)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  // return # bytes not clean (and thus not released).
	  loff_t unclean = 0;
	
	  if (oset->objects.empty()) {
	    ldout(cct, 10) << "release_set on " << oset << " dne" << dendl;
	    return 0;
	  }
	
	  ldout(cct, 10) << "release_set " << oset << dendl;
	
	  xlist<Object*>::iterator q;
	  for (xlist<Object*>::iterator p = oset->objects.begin();
	       !p.end(); ) {
	    q = p;
	    ++q;
	    Object *ob = *p;
	
	    loff_t o_unclean = release(ob);
	    unclean += o_unclean;
	
	    if (o_unclean)
	      ldout(cct, 10) << "release_set " << oset << " " << *ob
			     << " has " << o_unclean << " bytes left"
			     << dendl;
	    p = q;
	  }
	
	  if (unclean) {
	    ldout(cct, 10) << "release_set " << oset
			   << ", " << unclean << " bytes left" << dendl;
	  }
	
	  return unclean;
	}
	
	
	uint64_t ObjectCacher::release_all()
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ldout(cct, 10) << "release_all" << dendl;
	  uint64_t unclean = 0;
	
	  vector<ceph::unordered_map<sobject_t, Object*> >::iterator i
	    = objects.begin();
	  while (i != objects.end()) {
	    ceph::unordered_map<sobject_t, Object*>::iterator p = i->begin();
	    while (p != i->end()) {
	      ceph::unordered_map<sobject_t, Object*>::iterator n = p;
	      ++n;
	
	      Object *ob = p->second;
	
	      loff_t o_unclean = release(ob);
	      unclean += o_unclean;
	
	      if (o_unclean)
		ldout(cct, 10) << "release_all " << *ob
			       << " has " << o_unclean << " bytes left"
			       << dendl;
	    p = n;
	    }
	    ++i;
	  }
	
	  if (unclean) {
	    ldout(cct, 10) << "release_all unclean " << unclean << " bytes left"
			   << dendl;
	  }
	
	  return unclean;
	}
	
	void ObjectCacher::clear_nonexistence(ObjectSet *oset)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ldout(cct, 10) << "clear_nonexistence() " << oset << dendl;
	
	  for (xlist<Object*>::iterator p = oset->objects.begin();
	       !p.end(); ++p) {
	    Object *ob = *p;
	    if (!ob->exists) {
	      ldout(cct, 10) << " setting exists and complete on " << *ob << dendl;
	      ob->exists = true;
	      ob->complete = false;
	    }
	    for (xlist<C_ReadFinish*>::iterator q = ob->reads.begin();
		 !q.end(); ++q) {
	      C_ReadFinish *comp = *q;
	      comp->distrust_enoent();
	    }
	  }
	}
	
	/**
	 * discard object extents from an ObjectSet by removing the objects in
	 * exls from the in-memory oset.
	 */
	void ObjectCacher::discard_set(ObjectSet *oset, const vector<ObjectExtent>& exls)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  bool was_dirty = oset->dirty_or_tx > 0;
	
	  _discard(oset, exls, nullptr);
	  _discard_finish(oset, was_dirty, nullptr);
	}
	
	/**
	 * discard object extents from an ObjectSet by removing the objects in
	 * exls from the in-memory oset. If the bh is in TX state, the discard
	 * will wait for the write to commit prior to invoking on_finish.
	 */
	void ObjectCacher::discard_writeback(ObjectSet *oset,
	                                     const vector<ObjectExtent>& exls,
	                                     Context* on_finish)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  bool was_dirty = oset->dirty_or_tx > 0;
	
	  C_GatherBuilder gather(cct);
	  _discard(oset, exls, &gather);
	
	  if (gather.has_subs()) {
	    bool flushed = was_dirty && oset->dirty_or_tx == 0;
	    gather.set_finisher(new LambdaContext(
	      [this, oset, flushed, on_finish](int) {
		ceph_assert(ceph_mutex_is_locked(lock));
		if (flushed && flush_set_callback)
		  flush_set_callback(flush_set_callback_arg, oset);
		if (on_finish)
		  on_finish->complete(0);
	      }));
	    gather.activate();
	    return;
	  }
	
	  _discard_finish(oset, was_dirty, on_finish);
	}
	
	void ObjectCacher::_discard(ObjectSet *oset, const vector<ObjectExtent>& exls,
	                            C_GatherBuilder* gather)
	{
	  if (oset->objects.empty()) {
	    ldout(cct, 10) << __func__ << " on " << oset << " dne" << dendl;
	    return;
	  }
	
	  ldout(cct, 10) << __func__ << " " << oset << dendl;
	
	  for (auto& ex : exls) {
	    ldout(cct, 10) << __func__ << " " << oset << " ex " << ex << dendl;
	    sobject_t soid(ex.oid, CEPH_NOSNAP);
	    if (objects[oset->poolid].count(soid) == 0)
	      continue;
	    Object *ob = objects[oset->poolid][soid];
	
	    ob->discard(ex.offset, ex.length, gather);
	  }
	}
	
	void ObjectCacher::_discard_finish(ObjectSet *oset, bool was_dirty,
	                                   Context* on_finish)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	
	  // did we truncate off dirty data?
	  if (flush_set_callback && was_dirty && oset->dirty_or_tx == 0) {
	    flush_set_callback(flush_set_callback_arg, oset);
	  }
	
	  // notify that in-flight writeback has completed
	  if (on_finish != nullptr) {
	    on_finish->complete(0);
	  }
	}
	
	void ObjectCacher::verify_stats() const
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ldout(cct, 10) << "verify_stats" << dendl;
	
	  loff_t clean = 0, zero = 0, dirty = 0, rx = 0, tx = 0, missing = 0,
	    error = 0;
	  for (vector<ceph::unordered_map<sobject_t, Object*> >::const_iterator i
		 = objects.begin();
	       i != objects.end();
	       ++i) {
	    for (ceph::unordered_map<sobject_t, Object*>::const_iterator p
		   = i->begin();
		 p != i->end();
		 ++p) {
	      Object *ob = p->second;
	      for (map<loff_t, BufferHead*>::const_iterator q = ob->data.begin();
		   q != ob->data.end();
		  ++q) {
		BufferHead *bh = q->second;
		switch (bh->get_state()) {
		case BufferHead::STATE_MISSING:
		  missing += bh->length();
		  break;
		case BufferHead::STATE_CLEAN:
		  clean += bh->length();
		  break;
		case BufferHead::STATE_ZERO:
		  zero += bh->length();
		  break;
		case BufferHead::STATE_DIRTY:
		  dirty += bh->length();
		  break;
		case BufferHead::STATE_TX:
		  tx += bh->length();
		  break;
		case BufferHead::STATE_RX:
		  rx += bh->length();
		  break;
		case BufferHead::STATE_ERROR:
		  error += bh->length();
		  break;
		default:
		  ceph_abort();
		}
	      }
	    }
	  }
	
	  ldout(cct, 10) << " clean " << clean << " rx " << rx << " tx " << tx
			 << " dirty " << dirty << " missing " << missing
			 << " error " << error << dendl;
	  ceph_assert(clean == stat_clean);
	  ceph_assert(rx == stat_rx);
	  ceph_assert(tx == stat_tx);
	  ceph_assert(dirty == stat_dirty);
	  ceph_assert(missing == stat_missing);
	  ceph_assert(zero == stat_zero);
	  ceph_assert(error == stat_error);
	}
	
	void ObjectCacher::bh_stat_add(BufferHead *bh)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  switch (bh->get_state()) {
	  case BufferHead::STATE_MISSING:
	    stat_missing += bh->length();
	    break;
	  case BufferHead::STATE_CLEAN:
	    stat_clean += bh->length();
	    break;
	  case BufferHead::STATE_ZERO:
	    stat_zero += bh->length();
	    break;
	  case BufferHead::STATE_DIRTY:
	    stat_dirty += bh->length();
	    bh->ob->dirty_or_tx += bh->length();
	    bh->ob->oset->dirty_or_tx += bh->length();
	    break;
	  case BufferHead::STATE_TX:
	    stat_tx += bh->length();
	    bh->ob->dirty_or_tx += bh->length();
	    bh->ob->oset->dirty_or_tx += bh->length();
	    break;
	  case BufferHead::STATE_RX:
	    stat_rx += bh->length();
	    break;
	  case BufferHead::STATE_ERROR:
	    stat_error += bh->length();
	    break;
	  default:
	    ceph_abort_msg("bh_stat_add: invalid bufferhead state");
	  }
	  if (get_stat_dirty_waiting() > 0)
	    stat_cond.notify_all();
	}
	
	void ObjectCacher::bh_stat_sub(BufferHead *bh)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  switch (bh->get_state()) {
	  case BufferHead::STATE_MISSING:
	    stat_missing -= bh->length();
	    break;
	  case BufferHead::STATE_CLEAN:
	    stat_clean -= bh->length();
	    break;
	  case BufferHead::STATE_ZERO:
	    stat_zero -= bh->length();
	    break;
	  case BufferHead::STATE_DIRTY:
	    stat_dirty -= bh->length();
	    bh->ob->dirty_or_tx -= bh->length();
	    bh->ob->oset->dirty_or_tx -= bh->length();
	    break;
	  case BufferHead::STATE_TX:
	    stat_tx -= bh->length();
	    bh->ob->dirty_or_tx -= bh->length();
	    bh->ob->oset->dirty_or_tx -= bh->length();
	    break;
	  case BufferHead::STATE_RX:
	    stat_rx -= bh->length();
	    break;
	  case BufferHead::STATE_ERROR:
	    stat_error -= bh->length();
	    break;
	  default:
	    ceph_abort_msg("bh_stat_sub: invalid bufferhead state");
	  }
	}
	
	void ObjectCacher::bh_set_state(BufferHead *bh, int s)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  int state = bh->get_state();
	  // move between lru lists?
	  if (s == BufferHead::STATE_DIRTY && state != BufferHead::STATE_DIRTY) {
	    bh_lru_rest.lru_remove(bh);
	    bh_lru_dirty.lru_insert_top(bh);
	  } else if (s != BufferHead::STATE_DIRTY &&state == BufferHead::STATE_DIRTY) {
	    bh_lru_dirty.lru_remove(bh);
	    if (bh->get_dontneed())
	      bh_lru_rest.lru_insert_bot(bh);
	    else
	      bh_lru_rest.lru_insert_top(bh);
	  }
	
	  if ((s == BufferHead::STATE_TX ||
	       s == BufferHead::STATE_DIRTY) &&
	      state != BufferHead::STATE_TX &&
	      state != BufferHead::STATE_DIRTY) {
	    dirty_or_tx_bh.insert(bh);
	  } else if ((state == BufferHead::STATE_TX ||
		      state == BufferHead::STATE_DIRTY) &&
		     s != BufferHead::STATE_TX &&
		     s != BufferHead::STATE_DIRTY) {
	    dirty_or_tx_bh.erase(bh);
	  }
	
	  if (s != BufferHead::STATE_ERROR &&
	      state == BufferHead::STATE_ERROR) {
	    bh->error = 0;
	  }
	
	  // set state
	  bh_stat_sub(bh);
	  bh->set_state(s);
	  bh_stat_add(bh);
	}
	
	void ObjectCacher::bh_add(Object *ob, BufferHead *bh)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ldout(cct, 30) << "bh_add " << *ob << " " << *bh << dendl;
	  ob->add_bh(bh);
	  if (bh->is_dirty()) {
	    bh_lru_dirty.lru_insert_top(bh);
	    dirty_or_tx_bh.insert(bh);
	  } else {
	    if (bh->get_dontneed())
	      bh_lru_rest.lru_insert_bot(bh);
	    else
	      bh_lru_rest.lru_insert_top(bh);
	  }
	
	  if (bh->is_tx()) {
	    dirty_or_tx_bh.insert(bh);
	  }
	  bh_stat_add(bh);
	}
	
	void ObjectCacher::bh_remove(Object *ob, BufferHead *bh)
	{
	  ceph_assert(ceph_mutex_is_locked(lock));
	  ceph_assert(bh->get_journal_tid() == 0);
	  ldout(cct, 30) << "bh_remove " << *ob << " " << *bh << dendl;
	  ob->remove_bh(bh);
	  if (bh->is_dirty()) {
	    bh_lru_dirty.lru_remove(bh);
	    dirty_or_tx_bh.erase(bh);
	  } else {
	    bh_lru_rest.lru_remove(bh);
	  }
	
	  if (bh->is_tx()) {
	    dirty_or_tx_bh.erase(bh);
	  }
	  bh_stat_sub(bh);
	  if (get_stat_dirty_waiting() > 0)
	    stat_cond.notify_all();
	}