pg_whoami(recovery_state.pg_whoami),
info(recovery_state.info),
pg_log(recovery_state.pg_log),
- last_update_ondisk(recovery_state.last_update_ondisk),
- last_complete_ondisk(recovery_state.last_complete_ondisk),
- last_update_applied(recovery_state.last_update_applied),
peer_info(recovery_state.peer_info),
- peer_last_complete_ondisk(recovery_state.peer_last_complete_ondisk),
- min_last_complete_ondisk(recovery_state.min_last_complete_ondisk),
- pg_trim_to(recovery_state.pg_trim_to),
missing_loc(recovery_state.missing_loc),
pg_id(p),
coll(p),
/******* PG ***********/
void PG::clear_primary_state()
{
- last_update_ondisk = eversion_t();
-
projected_log = PGLog::IndexedLog();
snap_trimq.clear();
break;
case PG::Scrubber::WAIT_LAST_UPDATE:
- if (last_update_applied < scrubber.subset_last_update) {
+ if (recovery_state.get_last_update_applied() <
+ scrubber.subset_last_update) {
// will be requeued by op_applied
dout(15) << "wait for EC read/modify/writes to queue" << dendl;
done = true;
break;
case PG::Scrubber::BUILD_MAP:
- ceph_assert(last_update_applied >= scrubber.subset_last_update);
+ ceph_assert(recovery_state.get_last_update_applied() >=
+ scrubber.subset_last_update);
// build my own scrub map
if (scrub_preempted) {
break;
case PG::Scrubber::COMPARE_MAPS:
- ceph_assert(last_update_applied >= scrubber.subset_last_update);
+ ceph_assert(recovery_state.get_last_update_applied() >=
+ scrubber.subset_last_update);
ceph_assert(scrubber.waiting_on_whom.empty());
scrub_compare_maps();
pg_shard_t pg_whoami;
pg_info_t &info;
PGLog &pg_log;
- eversion_t &last_update_ondisk;
- eversion_t &last_complete_ondisk;
- eversion_t &last_update_applied;
map<pg_shard_t, pg_info_t> &peer_info;
- map<pg_shard_t,eversion_t> &peer_last_complete_ondisk;
- eversion_t &min_last_complete_ondisk;
- eversion_t &pg_trim_to;
MissingLoc &missing_loc;
public:
bool all_unfound_are_queried_or_lost(const OSDMapRef osdmap) const;
- virtual void calc_trim_to() = 0;
-
- virtual void calc_trim_to_aggressive() = 0;
-
struct PGLogEntryHandler : public PGLog::LogEntryHandler {
PG *pg;
ObjectStore::Transaction *t;
bool delete_needs_sleep = false;
protected:
- bool hard_limit_pglog() const {
- return (get_osdmap()->test_flag(CEPH_OSDMAP_PGLOG_HARDLIMIT));
- }
-
bool state_test(uint64_t m) const { return recovery_state.state_test(m); }
void state_set(uint64_t m) { recovery_state.state_set(m); }
void state_clear(uint64_t m) { recovery_state.state_clear(m); }
#include "messages/MRecoveryReserve.h"
#include "messages/MOSDScrubReserve.h"
#include "messages/MOSDPGInfo.h"
+#include "messages/MOSDPGTrim.h"
#define dout_context cct
#define dout_subsys ceph_subsys_osd
clear_recovery_state();
+ last_update_ondisk = eversion_t();
pl->clear_primary_state();
}
}
}
+void PeeringState::recovery_committed_to(eversion_t version)
+{
+ psdout(10) << __func__ << " version " << version
+ << " now ondisk" << dendl;
+ last_complete_ondisk = version;
+
+ if (last_complete_ondisk == info.last_update) {
+ if (!is_primary()) {
+ // Either we are a replica or backfill target.
+ // we are fully up to date. tell the primary!
+ pl->send_cluster_message(
+ get_primary().osd,
+ new MOSDPGTrim(
+ get_osdmap_epoch(),
+ spg_t(info.pgid.pgid, primary.shard),
+ last_complete_ondisk),
+ get_osdmap_epoch());
+ } else {
+ calc_min_last_complete_ondisk();
+ }
+ }
+}
+
+void PeeringState::complete_write(eversion_t v, eversion_t lc)
+{
+ last_update_ondisk = v;
+ last_complete_ondisk = lc;
+ calc_min_last_complete_ondisk();
+}
+
+void PeeringState::calc_trim_to()
+{
+ size_t target = cct->_conf->osd_min_pg_log_entries;
+ if (is_degraded() ||
+ state_test(PG_STATE_RECOVERING |
+ PG_STATE_RECOVERY_WAIT |
+ PG_STATE_BACKFILLING |
+ PG_STATE_BACKFILL_WAIT |
+ PG_STATE_BACKFILL_TOOFULL)) {
+ target = cct->_conf->osd_max_pg_log_entries;
+ }
+
+ eversion_t limit = std::min(
+ min_last_complete_ondisk,
+ pg_log.get_can_rollback_to());
+ if (limit != eversion_t() &&
+ limit != pg_trim_to &&
+ pg_log.get_log().approx_size() > target) {
+ size_t num_to_trim = std::min(pg_log.get_log().approx_size() - target,
+ cct->_conf->osd_pg_log_trim_max);
+ if (num_to_trim < cct->_conf->osd_pg_log_trim_min &&
+ cct->_conf->osd_pg_log_trim_max >= cct->_conf->osd_pg_log_trim_min) {
+ return;
+ }
+ list<pg_log_entry_t>::const_iterator it = pg_log.get_log().log.begin();
+ eversion_t new_trim_to;
+ for (size_t i = 0; i < num_to_trim; ++i) {
+ new_trim_to = it->version;
+ ++it;
+ if (new_trim_to > limit) {
+ new_trim_to = limit;
+ psdout(10) << "calc_trim_to trimming to min_last_complete_ondisk" << dendl;
+ break;
+ }
+ }
+ psdout(10) << "calc_trim_to " << pg_trim_to << " -> " << new_trim_to << dendl;
+ pg_trim_to = new_trim_to;
+ assert(pg_trim_to <= pg_log.get_head());
+ assert(pg_trim_to <= min_last_complete_ondisk);
+ }
+}
+
+void PeeringState::calc_trim_to_aggressive()
+{
+ size_t target = cct->_conf->osd_min_pg_log_entries;
+ if (is_degraded() ||
+ state_test(PG_STATE_RECOVERING |
+ PG_STATE_RECOVERY_WAIT |
+ PG_STATE_BACKFILLING |
+ PG_STATE_BACKFILL_WAIT |
+ PG_STATE_BACKFILL_TOOFULL)) {
+ target = cct->_conf->osd_max_pg_log_entries;
+ }
+ // limit pg log trimming up to the can_rollback_to value
+ eversion_t limit = std::min(
+ pg_log.get_head(),
+ pg_log.get_can_rollback_to());
+ psdout(10) << __func__ << " limit = " << limit << dendl;
+
+ if (limit != eversion_t() &&
+ limit != pg_trim_to &&
+ pg_log.get_log().approx_size() > target) {
+ psdout(10) << __func__ << " approx pg log length = "
+ << pg_log.get_log().approx_size() << dendl;
+ uint64_t num_to_trim = std::min<uint64_t>(pg_log.get_log().approx_size() - target,
+ cct->_conf->osd_pg_log_trim_max);
+ psdout(10) << __func__ << " num_to_trim = " << num_to_trim << dendl;
+ if (num_to_trim < cct->_conf->osd_pg_log_trim_min &&
+ cct->_conf->osd_pg_log_trim_max >= cct->_conf->osd_pg_log_trim_min) {
+ return;
+ }
+ auto it = pg_log.get_log().log.begin(); // oldest log entry
+ auto rit = pg_log.get_log().log.rbegin();
+ eversion_t by_n_to_keep; // start from tail
+ eversion_t by_n_to_trim = eversion_t::max(); // start from head
+ for (size_t i = 0; it != pg_log.get_log().log.end(); ++it, ++rit) {
+ i++;
+ if (i > target && by_n_to_keep == eversion_t()) {
+ by_n_to_keep = rit->version;
+ }
+ if (i >= num_to_trim && by_n_to_trim == eversion_t::max()) {
+ by_n_to_trim = it->version;
+ }
+ if (by_n_to_keep != eversion_t() &&
+ by_n_to_trim != eversion_t::max()) {
+ break;
+ }
+ }
+
+ if (by_n_to_keep == eversion_t()) {
+ return;
+ }
+
+ pg_trim_to = std::min({by_n_to_keep, by_n_to_trim, limit});
+ psdout(10) << __func__ << " pg_trim_to now " << pg_trim_to << dendl;
+ ceph_assert(pg_trim_to <= pg_log.get_head());
+ }
+}
+
+
/*------------ Peering State Machine----------------*/
#undef dout_prefix
#define dout_prefix (context< PeeringMachine >().dpp->gen_prefix(*_dout) \
void add_log_entry(const pg_log_entry_t& e, bool applied);
+ void calc_trim_to();
+ void calc_trim_to_aggressive();
+
public:
PeeringState(
CephContext *cct,
bool transaction_applied,
bool async);
+ void update_trim_to() {
+ bool hard_limit = (get_osdmap()->test_flag(CEPH_OSDMAP_PGLOG_HARDLIMIT));
+ if (hard_limit)
+ calc_trim_to_aggressive();
+ else
+ calc_trim_to();
+ }
+
void pre_submit_op(
const hobject_t &hoid,
const vector<pg_log_entry_t>& logv,
const hobject_t &oid,
eversion_t version);
+ void update_peer_last_complete_ondisk(
+ pg_shard_t fromosd,
+ eversion_t lcod) {
+ peer_last_complete_ondisk[fromosd] = lcod;
+ }
+
+ void update_last_complete_ondisk(
+ eversion_t lcod) {
+ last_complete_ondisk = lcod;
+ }
+
+ void recovery_committed_to(eversion_t version);
+
+ void complete_write(eversion_t v, eversion_t lc);
+ void local_write_applied(eversion_t v) {
+ last_update_applied = v;
+ }
+
void dump_history(Formatter *f) const {
state_history.dump(f);
}
return min_last_complete_ondisk;
}
+ eversion_t get_pg_trim_to() const {
+ return pg_trim_to;
+ }
+
+ eversion_t get_last_update_applied() const {
+ return last_update_applied;
+ }
+
bool debug_has_dirty_state() const {
return dirty_info || dirty_big_info;
}
return r;
}
-void PrimaryLogPG::calc_trim_to()
-{
- size_t target = cct->_conf->osd_min_pg_log_entries;
- if (is_degraded() ||
- state_test(PG_STATE_RECOVERING |
- PG_STATE_RECOVERY_WAIT |
- PG_STATE_BACKFILLING |
- PG_STATE_BACKFILL_WAIT |
- PG_STATE_BACKFILL_TOOFULL)) {
- target = cct->_conf->osd_max_pg_log_entries;
- }
-
- eversion_t limit = std::min(
- min_last_complete_ondisk,
- pg_log.get_can_rollback_to());
- if (limit != eversion_t() &&
- limit != pg_trim_to &&
- pg_log.get_log().approx_size() > target) {
- size_t num_to_trim = std::min(pg_log.get_log().approx_size() - target,
- cct->_conf->osd_pg_log_trim_max);
- if (num_to_trim < cct->_conf->osd_pg_log_trim_min &&
- cct->_conf->osd_pg_log_trim_max >= cct->_conf->osd_pg_log_trim_min) {
- return;
- }
- list<pg_log_entry_t>::const_iterator it = pg_log.get_log().log.begin();
- eversion_t new_trim_to;
- for (size_t i = 0; i < num_to_trim; ++i) {
- new_trim_to = it->version;
- ++it;
- if (new_trim_to > limit) {
- new_trim_to = limit;
- dout(10) << "calc_trim_to trimming to min_last_complete_ondisk" << dendl;
- break;
- }
- }
- dout(10) << "calc_trim_to " << pg_trim_to << " -> " << new_trim_to << dendl;
- pg_trim_to = new_trim_to;
- assert(pg_trim_to <= pg_log.get_head());
- assert(pg_trim_to <= min_last_complete_ondisk);
- }
-}
-
-void PrimaryLogPG::calc_trim_to_aggressive()
-{
- size_t target = cct->_conf->osd_min_pg_log_entries;
- if (is_degraded() ||
- state_test(PG_STATE_RECOVERING |
- PG_STATE_RECOVERY_WAIT |
- PG_STATE_BACKFILLING |
- PG_STATE_BACKFILL_WAIT |
- PG_STATE_BACKFILL_TOOFULL)) {
- target = cct->_conf->osd_max_pg_log_entries;
- }
- // limit pg log trimming up to the can_rollback_to value
- eversion_t limit = std::min(
- pg_log.get_head(),
- pg_log.get_can_rollback_to());
- dout(10) << __func__ << " limit = " << limit << dendl;
-
- if (limit != eversion_t() &&
- limit != pg_trim_to &&
- pg_log.get_log().approx_size() > target) {
- dout(10) << __func__ << " approx pg log length = "
- << pg_log.get_log().approx_size() << dendl;
- uint64_t num_to_trim = std::min<uint64_t>(pg_log.get_log().approx_size() - target,
- cct->_conf->osd_pg_log_trim_max);
- dout(10) << __func__ << " num_to_trim = " << num_to_trim << dendl;
- if (num_to_trim < cct->_conf->osd_pg_log_trim_min &&
- cct->_conf->osd_pg_log_trim_max >= cct->_conf->osd_pg_log_trim_min) {
- return;
- }
- auto it = pg_log.get_log().log.begin(); // oldest log entry
- auto rit = pg_log.get_log().log.rbegin();
- eversion_t by_n_to_keep; // start from tail
- eversion_t by_n_to_trim = eversion_t::max(); // start from head
- for (size_t i = 0; it != pg_log.get_log().log.end(); ++it, ++rit) {
- i++;
- if (i > target && by_n_to_keep == eversion_t()) {
- by_n_to_keep = rit->version;
- }
- if (i >= num_to_trim && by_n_to_trim == eversion_t::max()) {
- by_n_to_trim = it->version;
- }
- if (by_n_to_keep != eversion_t() &&
- by_n_to_trim != eversion_t::max()) {
- break;
- }
- }
-
- if (by_n_to_keep == eversion_t()) {
- return;
- }
-
- pg_trim_to = std::min({by_n_to_keep, by_n_to_trim, limit});
- dout(10) << __func__ << " pg_trim_to now " << pg_trim_to << dendl;
- ceph_assert(pg_trim_to <= pg_log.get_head());
- }
-}
-
PrimaryLogPG::PrimaryLogPG(OSDService *o, OSDMapRef curmap,
const PGPool &_pool,
const map<string,string>& ec_profile, spg_t p) :
ceph_assert(op->may_write() || op->may_cache());
// trim log?
- if (hard_limit_pglog())
- calc_trim_to_aggressive();
- else
- calc_trim_to();
+ recovery_state.update_trim_to();
// verify that we are doing this in order?
if (cct->_conf->osd_debug_op_order && m->get_source().is_client() &&
repop->all_committed = true;
if (!repop->rep_aborted) {
if (repop->v != eversion_t()) {
- last_update_ondisk = repop->v;
- last_complete_ondisk = repop->pg_local_last_complete;
+ recovery_state.complete_write(repop->v, repop->pg_local_last_complete);
}
eval_repop(repop);
}
dout(10) << "op_applied version " << applied_version << dendl;
ceph_assert(applied_version != eversion_t());
ceph_assert(applied_version <= info.last_update);
- last_update_applied = applied_version;
+ recovery_state.local_write_applied(applied_version);
if (is_primary()) {
if (scrubber.active) {
- if (last_update_applied >= scrubber.subset_last_update) {
+ if (recovery_state.get_last_update_applied() >=
+ scrubber.subset_last_update) {
requeue_scrub(ops_blocked_by_scrub());
}
} else {
}
publish_stats_to_osd();
- recovery_state.calc_min_last_complete_ondisk();
dout(10) << " removing " << *repop << dendl;
ceph_assert(!repop_queue.empty());
ctx->delta_stats,
ctx->at_version,
std::move(ctx->op_t),
- pg_trim_to,
- min_last_complete_ondisk,
+ recovery_state.get_pg_trim_to(),
+ recovery_state.get_min_last_complete_ondisk(),
ctx->log,
ctx->updated_hset_history,
on_all_commit,
dout(20) << __func__ << " " << repop << dendl;
issue_repop(repop, ctx.get());
eval_repop(repop);
- if (hard_limit_pglog())
- calc_trim_to_aggressive();
- else
- calc_trim_to();
+ recovery_state.update_trim_to();
repop->put();
}
ObjectStore::Transaction t;
eversion_t old_last_update = info.last_update;
recovery_state.merge_new_log_entries(
- entries, t, pg_trim_to, min_last_complete_ondisk);
-
+ entries, t, recovery_state.get_pg_trim_to(),
+ recovery_state.get_min_last_complete_ondisk());
set<pg_shard_t> waiting_on;
for (set<pg_shard_t>::const_iterator i = get_acting_recovery_backfill().begin();
get_osdmap_epoch(),
get_last_peering_reset(),
repop->rep_tid,
- pg_trim_to,
- min_last_complete_ondisk);
+ recovery_state.get_pg_trim_to(),
+ recovery_state.get_min_last_complete_ondisk());
osd->send_message_osd_cluster(
peer.osd, m, get_osdmap_epoch());
waiting_on.insert(peer);
op_applied(info.last_update);
});
- if (hard_limit_pglog())
- calc_trim_to_aggressive();
- else
- calc_trim_to();
+ recovery_state.update_trim_to();
}
void PrimaryLogPG::cancel_log_updates()
{
lock();
if (!pg_has_reset_since(epoch)) {
- dout(10) << __func__ << " last_complete " << last_complete << " now ondisk" << dendl;
- last_complete_ondisk = last_complete;
-
- if (last_complete_ondisk == info.last_update) {
- if (!is_primary()) {
- // Either we are a replica or backfill target.
- // we are fully up to date. tell the primary!
- osd->send_message_osd_cluster(
- get_primary().osd,
- new MOSDPGTrim(
- get_osdmap_epoch(),
- spg_t(info.pgid.pgid, get_primary().shard),
- last_complete_ondisk),
- get_osdmap_epoch());
- } else {
- recovery_state.calc_min_last_complete_ondisk();
- }
- }
-
+ recovery_state.recovery_committed_to(last_complete);
} else {
- dout(10) << __func__ << " pg has changed, not touching last_complete_ondisk" << dendl;
+ dout(10) << __func__
+ << " pg has changed, not touching last_complete_ondisk" << dendl;
}
unlock();
void update_peer_last_complete_ondisk(
pg_shard_t fromosd,
eversion_t lcod) override {
- peer_last_complete_ondisk[fromosd] = lcod;
+ recovery_state.update_peer_last_complete_ondisk(fromosd, lcod);
}
void update_last_complete_ondisk(
eversion_t lcod) override {
- last_complete_ondisk = lcod;
+ recovery_state.update_last_complete_ondisk(lcod);
}
void update_stats(
unsigned split_bits) override;
void apply_and_flush_repops(bool requeue);
- void calc_trim_to() override;
- void calc_trim_to_aggressive() override;
int do_xattr_cmp_u64(int op, __u64 v1, bufferlist& xattr);
int do_xattr_cmp_str(int op, string& v1s, bufferlist& xattr);
projects / ceph.git / commitdiff