+++ /dev/null
-// -*- 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/assert.h"
-
-#define MAX_FLUSH_UNDER_LOCK 20 ///< max bh's we start writeback on
-#define BUFFER_MEMORY_WEIGHT 12 // 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)
-{
- assert(oc->lock.is_locked());
- 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()) {
- 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 );
- 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)
-{
- assert(oc->lock.is_locked());
- assert(left->end() == right->start());
- assert(left->get_state() == right->get_state());
- assert(left->can_merge_journal(right));
-
- 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 = MAX( left->last_write_tid, right->last_write_tid );
- left->last_write = 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;
-}
-
-void ObjectCacher::Object::try_merge_bh(BufferHead *bh)
-{
- assert(oc->lock.is_locked());
- 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());
- assert(p->second == bh);
- if (p != data.begin()) {
- --p;
- if (p->second->end() == bh->start() &&
- p->second->get_state() == bh->get_state() &&
- p->second->can_merge_journal(bh)) {
- merge_left(p->second, bh);
- bh = p->second;
- } else {
- ++p;
- }
- }
- // to the right?
- assert(p->second == bh);
- ++p;
- if (p != data.end() &&
- p->second->start() == bh->end() &&
- p->second->get_state() == bh->get_state() &&
- p->second->can_merge_journal(bh))
- merge_left(bh, p->second);
-}
-
-/*
- * count bytes we have cached in given range
- */
-bool ObjectCacher::Object::is_cached(loff_t cur, loff_t left) const
-{
- assert(oc->lock.is_locked());
- 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 = 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)
-{
- assert(oc->lock.is_locked());
- 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)
-{
- assert(oc->lock.is_locked());
- 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;
- 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 = 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 = 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 += MIN(left, n->length());
- left -= 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;
- 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;
- 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;
- assert(w_it->first >= bh->start());
- 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)
-{
- assert(oc->lock.is_locked());
- 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) {
- assert(final == 0);
- if (cur + max >= bh->end()) {
- // we want right bit (one splice)
- final = split(bh, cur); // just split it, take right half.
- replace_journal_tid(final, tid);
- ++p;
- assert(p->second == final);
- } else {
- // we want middle bit (two splices)
- final = split(bh, cur);
- ++p;
- assert(p->second == final);
- split(final, cur+max);
- replace_journal_tid(final, tid);
- }
- } else {
- assert(p->first == cur);
- if (bh->length() <= max) {
- // whole bufferhead, piece of cake.
- } else {
- // we want left bit (one splice)
- split(bh, cur + max); // just split
- }
- if (final) {
- oc->mark_dirty(bh);
- oc->mark_dirty(final);
- --p; // move iterator back to final
- 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 = 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
- assert(final);
- 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();
-
- 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)
-{
- assert(oc->lock.is_locked());
- 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);
- continue;
- }
-
- // remove bh entirely
- assert(bh->start() >= s);
- 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)
-{
- assert(oc->lock.is_locked());
- 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);
- ++p;
- continue;
- }
-
- assert(bh->start() >= off);
- if (bh->end() > off + len) {
- split(bh, off + len);
- }
-
- ++p;
- ldout(oc->cct, 10) << "discard " << *this << " bh " << *bh << dendl;
- assert(bh->waitfor_read.empty());
- replace_journal_tid(bh, 0);
- oc->bh_remove(this, bh);
- delete bh;
- }
-}
-
-
-
-/*** ObjectCacher ***/
-
-#undef dout_prefix
-#define dout_prefix *_dout << "objectcacher "
-
-
-ObjectCacher::ObjectCacher(CephContext *cct_, string name,
- WritebackHandler& wb, 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)
- assert(i->empty());
- assert(bh_lru_rest.lru_get_size() == 0);
- assert(bh_lru_dirty.lru_get_size() == 0);
- assert(ob_lru.lru_get_size() == 0);
- 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");
- plb.add_u64_counter(l_objectcacher_cache_bytes_miss,
- "cache_bytes_miss", "Miss data");
- 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");
- 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()
-{
- 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
- assert(lock.is_locked());
- // 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)
-{
- assert(lock.is_locked());
- ldout(cct, 10) << "close_object " << *ob << dendl;
- 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)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- 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;
- }
-
- assert(opos >= bh->start());
- assert(bh->start() == opos); // we don't merge rx bh's... yet!
- 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.Signal();
-}
-
-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);
- 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)
-{
- assert(lock.is_locked());
-
- 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;
- assert(bh->ob == ob);
- 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)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- 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;
-
- if (bh->start() < start &&
- bh->end() > start+(loff_t)length) {
- ldout(cct, 20) << "bh_write_commit skipping " << *bh << dendl;
- continue;
- }
-
- // 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) {
- assert(bh->last_write_tid > tid);
- ldout(cct, 10) << "bh_write_commit newer tid on " << *bh << dendl;
- continue;
- }
-
- 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.
- 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)
-{
- assert(trace != nullptr);
- assert(lock.is_locked());
- 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()
-{
- assert(lock.is_locked());
- 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;
- 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)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(trace != nullptr);
- assert(lock.is_locked());
- 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().
- */
- 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, "
- << (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 {
- 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();
- 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;
- assert(opos == (loff_t)(bh->start() + bhoff));
-
- uint64_t len = 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;
- }
- assert(f_it == ex_it->buffer_extents.end());
- 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) {
- 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);
- 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;
- 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)
-{
- assert(lock.is_locked());
- 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 = bh->start() - opos;
- 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
- bufferlist newbl;
- if (bhoff)
- newbl.substr_of(bh->bl, 0, bhoff);
- newbl.claim_append(frag);
- bh->bl.swap(newbl);
-
- 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)
-{
- Mutex::Locker 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)
-{
- assert(lock.is_locked());
- 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.Signal();
- stat_dirty_waiting += len;
- ++stat_nr_dirty_waiters;
- stat_cond.Wait(lock);
- 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)
-{
- assert(lock.is_locked());
- assert(trace != nullptr);
- int ret = 0;
-
- if (max_dirty > 0) {
- if (block_writes_upfront) {
- maybe_wait_for_writeback(len, trace);
- if (onfreespace)
- onfreespace->complete(0);
- } else {
- assert(onfreespace);
- finisher.queue(new C_WaitForWrite(this, len, *trace, onfreespace));
- }
- } else {
- // write-thru! flush what we just wrote.
- Cond cond;
- bool done = false;
- Context *fin = block_writes_upfront ?
- new C_Cond(&cond, &done, &ret) : onfreespace;
- assert(fin);
- bool flushed = flush_set(oset, wr->extents, trace, fin);
- 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) {
- while (!done)
- cond.Wait(lock);
- 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.Signal();
- }
- return ret;
-}
-
-void ObjectCacher::flusher_entry()
-{
- ldout(cct, 10) << "flusher start" << dendl;
- lock.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");
- lock.Unlock();
- lock.Lock();
- continue;
- }
- }
-
- trace.event("finish");
- if (flusher_stop)
- break;
-
- flusher_cond.WaitInterval(lock, seconds(1));
- }
-
- /* 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.
- */
- while (reads_outstanding > 0) {
- ldout(cct, 10) << "Waiting for all reads to complete. Number left: "
- << reads_outstanding << dendl;
- read_cond.Wait(lock);
- }
-
- lock.Unlock();
- ldout(cct, 10) << "flusher finish" << dendl;
-}
-
-
-// -------------------------------------------------
-
-bool ObjectCacher::set_is_empty(ObjectSet *oset)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(trace != nullptr);
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- assert(trace != nullptr);
- 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)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- 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.
- 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)
-{
- assert(lock.is_locked());
- 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);
- 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)
-{
- assert(lock.is_locked());
- // 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()
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- if (oset->objects.empty()) {
- ldout(cct, 10) << "discard_set on " << oset << " dne" << dendl;
- return;
- }
-
- ldout(cct, 10) << "discard_set " << oset << dendl;
-
- bool were_dirty = oset->dirty_or_tx > 0;
-
- for (vector<ObjectExtent>::const_iterator p = exls.begin();
- p != exls.end();
- ++p) {
- ldout(cct, 10) << "discard_set " << oset << " ex " << *p << dendl;
- const ObjectExtent &ex = *p;
- 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);
- }
-
- // did we truncate off dirty data?
- if (flush_set_callback &&
- were_dirty && oset->dirty_or_tx == 0)
- flush_set_callback(flush_set_callback_arg, oset);
-}
-
-void ObjectCacher::verify_stats() const
-{
- assert(lock.is_locked());
- 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;
- assert(clean == stat_clean);
- assert(rx == stat_rx);
- assert(tx == stat_tx);
- assert(dirty == stat_dirty);
- assert(missing == stat_missing);
- assert(zero == stat_zero);
- assert(error == stat_error);
-}
-
-void ObjectCacher::bh_stat_add(BufferHead *bh)
-{
- assert(lock.is_locked());
- 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:
- assert(0 == "bh_stat_add: invalid bufferhead state");
- }
- if (get_stat_dirty_waiting() > 0)
- stat_cond.Signal();
-}
-
-void ObjectCacher::bh_stat_sub(BufferHead *bh)
-{
- assert(lock.is_locked());
- 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:
- assert(0 == "bh_stat_sub: invalid bufferhead state");
- }
-}
-
-void ObjectCacher::bh_set_state(BufferHead *bh, int s)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- 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)
-{
- assert(lock.is_locked());
- 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.Signal();
-}
-
Code Review / stor4nfv.git / blobdiff