+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-/*
- * Ceph - scalable distributed file system
- *
- * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
- *
- * This is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License version 2.1, as published by the Free Software
- * Foundation. See file COPYING.
- *
- */
-
-#include <sstream>
-#include "Paxos.h"
-#include "Monitor.h"
-#include "messages/MMonPaxos.h"
-
-#include "mon/mon_types.h"
-#include "common/config.h"
-#include "include/assert.h"
-#include "include/stringify.h"
-#include "common/Timer.h"
-#include "messages/PaxosServiceMessage.h"
-
-#define dout_subsys ceph_subsys_paxos
-#undef dout_prefix
-#define dout_prefix _prefix(_dout, mon, mon->name, mon->rank, paxos_name, state, first_committed, last_committed)
-static ostream& _prefix(std::ostream *_dout, Monitor *mon, const string& name,
- int rank, const string& paxos_name, int state,
- version_t first_committed, version_t last_committed)
-{
- return *_dout << "mon." << name << "@" << rank
- << "(" << mon->get_state_name() << ")"
- << ".paxos(" << paxos_name << " " << Paxos::get_statename(state)
- << " c " << first_committed << ".." << last_committed
- << ") ";
-}
-
-class Paxos::C_Trimmed : public Context {
- Paxos *paxos;
-public:
- explicit C_Trimmed(Paxos *p) : paxos(p) { }
- void finish(int r) override {
- paxos->trimming = false;
- }
-};
-
-MonitorDBStore *Paxos::get_store()
-{
- return mon->store;
-}
-
-void Paxos::read_and_prepare_transactions(MonitorDBStore::TransactionRef tx,
- version_t first, version_t last)
-{
- dout(10) << __func__ << " first " << first << " last " << last << dendl;
- for (version_t v = first; v <= last; ++v) {
- dout(30) << __func__ << " apply version " << v << dendl;
- bufferlist bl;
- int err = get_store()->get(get_name(), v, bl);
- assert(err == 0);
- assert(bl.length());
- decode_append_transaction(tx, bl);
- }
- dout(15) << __func__ << " total versions " << (last-first) << dendl;
-}
-
-void Paxos::init()
-{
- // load paxos variables from stable storage
- last_pn = get_store()->get(get_name(), "last_pn");
- accepted_pn = get_store()->get(get_name(), "accepted_pn");
- last_committed = get_store()->get(get_name(), "last_committed");
- first_committed = get_store()->get(get_name(), "first_committed");
-
- dout(10) << __func__ << " last_pn: " << last_pn << " accepted_pn: "
- << accepted_pn << " last_committed: " << last_committed
- << " first_committed: " << first_committed << dendl;
-
- dout(10) << "init" << dendl;
- assert(is_consistent());
-}
-
-void Paxos::init_logger()
-{
- PerfCountersBuilder pcb(g_ceph_context, "paxos", l_paxos_first, l_paxos_last);
-
- // Because monitors are so few in number, the resource cost of capturing
- // almost all their perf counters at USEFUL is trivial.
- pcb.set_prio_default(PerfCountersBuilder::PRIO_USEFUL);
-
- pcb.add_u64_counter(l_paxos_start_leader, "start_leader", "Starts in leader role");
- pcb.add_u64_counter(l_paxos_start_peon, "start_peon", "Starts in peon role");
- pcb.add_u64_counter(l_paxos_restart, "restart", "Restarts");
- pcb.add_u64_counter(l_paxos_refresh, "refresh", "Refreshes");
- pcb.add_time_avg(l_paxos_refresh_latency, "refresh_latency", "Refresh latency");
- pcb.add_u64_counter(l_paxos_begin, "begin", "Started and handled begins");
- pcb.add_u64_avg(l_paxos_begin_keys, "begin_keys", "Keys in transaction on begin");
- pcb.add_u64_avg(l_paxos_begin_bytes, "begin_bytes", "Data in transaction on begin");
- pcb.add_time_avg(l_paxos_begin_latency, "begin_latency", "Latency of begin operation");
- pcb.add_u64_counter(l_paxos_commit, "commit",
- "Commits", "cmt");
- pcb.add_u64_avg(l_paxos_commit_keys, "commit_keys", "Keys in transaction on commit");
- pcb.add_u64_avg(l_paxos_commit_bytes, "commit_bytes", "Data in transaction on commit");
- pcb.add_time_avg(l_paxos_commit_latency, "commit_latency",
- "Commit latency", "clat");
- pcb.add_u64_counter(l_paxos_collect, "collect", "Peon collects");
- pcb.add_u64_avg(l_paxos_collect_keys, "collect_keys", "Keys in transaction on peon collect");
- pcb.add_u64_avg(l_paxos_collect_bytes, "collect_bytes", "Data in transaction on peon collect");
- pcb.add_time_avg(l_paxos_collect_latency, "collect_latency", "Peon collect latency");
- pcb.add_u64_counter(l_paxos_collect_uncommitted, "collect_uncommitted", "Uncommitted values in started and handled collects");
- pcb.add_u64_counter(l_paxos_collect_timeout, "collect_timeout", "Collect timeouts");
- pcb.add_u64_counter(l_paxos_accept_timeout, "accept_timeout", "Accept timeouts");
- pcb.add_u64_counter(l_paxos_lease_ack_timeout, "lease_ack_timeout", "Lease acknowledgement timeouts");
- pcb.add_u64_counter(l_paxos_lease_timeout, "lease_timeout", "Lease timeouts");
- pcb.add_u64_counter(l_paxos_store_state, "store_state", "Store a shared state on disk");
- pcb.add_u64_avg(l_paxos_store_state_keys, "store_state_keys", "Keys in transaction in stored state");
- pcb.add_u64_avg(l_paxos_store_state_bytes, "store_state_bytes", "Data in transaction in stored state");
- pcb.add_time_avg(l_paxos_store_state_latency, "store_state_latency", "Storing state latency");
- pcb.add_u64_counter(l_paxos_share_state, "share_state", "Sharings of state");
- pcb.add_u64_avg(l_paxos_share_state_keys, "share_state_keys", "Keys in shared state");
- pcb.add_u64_avg(l_paxos_share_state_bytes, "share_state_bytes", "Data in shared state");
- pcb.add_u64_counter(l_paxos_new_pn, "new_pn", "New proposal number queries");
- pcb.add_time_avg(l_paxos_new_pn_latency, "new_pn_latency", "New proposal number getting latency");
- logger = pcb.create_perf_counters();
- g_ceph_context->get_perfcounters_collection()->add(logger);
-}
-
-void Paxos::dump_info(Formatter *f)
-{
- f->open_object_section("paxos");
- f->dump_unsigned("first_committed", first_committed);
- f->dump_unsigned("last_committed", last_committed);
- f->dump_unsigned("last_pn", last_pn);
- f->dump_unsigned("accepted_pn", accepted_pn);
- f->close_section();
-}
-
-// ---------------------------------
-
-// PHASE 1
-
-// leader
-void Paxos::collect(version_t oldpn)
-{
- // we're recoverying, it seems!
- state = STATE_RECOVERING;
- assert(mon->is_leader());
-
- // reset the number of lasts received
- uncommitted_v = 0;
- uncommitted_pn = 0;
- uncommitted_value.clear();
- peer_first_committed.clear();
- peer_last_committed.clear();
-
- // look for uncommitted value
- if (get_store()->exists(get_name(), last_committed+1)) {
- version_t v = get_store()->get(get_name(), "pending_v");
- version_t pn = get_store()->get(get_name(), "pending_pn");
- if (v && pn && v == last_committed + 1) {
- uncommitted_pn = pn;
- } else {
- dout(10) << "WARNING: no pending_pn on disk, using previous accepted_pn " << accepted_pn
- << " and crossing our fingers" << dendl;
- uncommitted_pn = accepted_pn;
- }
- uncommitted_v = last_committed+1;
-
- get_store()->get(get_name(), last_committed+1, uncommitted_value);
- assert(uncommitted_value.length());
- dout(10) << "learned uncommitted " << (last_committed+1)
- << " pn " << uncommitted_pn
- << " (" << uncommitted_value.length() << " bytes) from myself"
- << dendl;
-
- logger->inc(l_paxos_collect_uncommitted);
- }
-
- // pick new pn
- accepted_pn = get_new_proposal_number(MAX(accepted_pn, oldpn));
- accepted_pn_from = last_committed;
- num_last = 1;
- dout(10) << "collect with pn " << accepted_pn << dendl;
-
- // send collect
- for (set<int>::const_iterator p = mon->get_quorum().begin();
- p != mon->get_quorum().end();
- ++p) {
- if (*p == mon->rank) continue;
-
- MMonPaxos *collect = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_COLLECT,
- ceph_clock_now());
- collect->last_committed = last_committed;
- collect->first_committed = first_committed;
- collect->pn = accepted_pn;
- mon->messenger->send_message(collect, mon->monmap->get_inst(*p));
- }
-
- // set timeout event
- collect_timeout_event = mon->timer.add_event_after(
- g_conf->mon_accept_timeout_factor *
- g_conf->mon_lease,
- new C_MonContext(mon, [this](int r) {
- if (r == -ECANCELED)
- return;
- collect_timeout();
- }));
-}
-
-
-// peon
-void Paxos::handle_collect(MonOpRequestRef op)
-{
-
- op->mark_paxos_event("handle_collect");
-
- MMonPaxos *collect = static_cast<MMonPaxos*>(op->get_req());
- dout(10) << "handle_collect " << *collect << dendl;
-
- assert(mon->is_peon()); // mon epoch filter should catch strays
-
- // we're recoverying, it seems!
- state = STATE_RECOVERING;
-
- //update the peon recovery timeout
- reset_lease_timeout();
-
- if (collect->first_committed > last_committed+1) {
- dout(2) << __func__
- << " leader's lowest version is too high for our last committed"
- << " (theirs: " << collect->first_committed
- << "; ours: " << last_committed << ") -- bootstrap!" << dendl;
- op->mark_paxos_event("need to bootstrap");
- mon->bootstrap();
- return;
- }
-
- // reply
- MMonPaxos *last = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_LAST,
- ceph_clock_now());
- last->last_committed = last_committed;
- last->first_committed = first_committed;
-
- version_t previous_pn = accepted_pn;
-
- // can we accept this pn?
- if (collect->pn > accepted_pn) {
- // ok, accept it
- accepted_pn = collect->pn;
- accepted_pn_from = collect->pn_from;
- dout(10) << "accepting pn " << accepted_pn << " from "
- << accepted_pn_from << dendl;
-
- auto t(std::make_shared<MonitorDBStore::Transaction>());
- t->put(get_name(), "accepted_pn", accepted_pn);
-
- dout(30) << __func__ << " transaction dump:\n";
- JSONFormatter f(true);
- t->dump(&f);
- f.flush(*_dout);
- *_dout << dendl;
-
- logger->inc(l_paxos_collect);
- logger->inc(l_paxos_collect_keys, t->get_keys());
- logger->inc(l_paxos_collect_bytes, t->get_bytes());
- utime_t start = ceph_clock_now();
-
- get_store()->apply_transaction(t);
-
- utime_t end = ceph_clock_now();
- logger->tinc(l_paxos_collect_latency, end - start);
- } else {
- // don't accept!
- dout(10) << "NOT accepting pn " << collect->pn << " from " << collect->pn_from
- << ", we already accepted " << accepted_pn
- << " from " << accepted_pn_from << dendl;
- }
- last->pn = accepted_pn;
- last->pn_from = accepted_pn_from;
-
- // share whatever committed values we have
- if (collect->last_committed < last_committed)
- share_state(last, collect->first_committed, collect->last_committed);
-
- // do we have an accepted but uncommitted value?
- // (it'll be at last_committed+1)
- bufferlist bl;
- if (collect->last_committed <= last_committed &&
- get_store()->exists(get_name(), last_committed+1)) {
- get_store()->get(get_name(), last_committed+1, bl);
- assert(bl.length() > 0);
- dout(10) << " sharing our accepted but uncommitted value for "
- << last_committed+1 << " (" << bl.length() << " bytes)" << dendl;
- last->values[last_committed+1] = bl;
-
- version_t v = get_store()->get(get_name(), "pending_v");
- version_t pn = get_store()->get(get_name(), "pending_pn");
- if (v && pn && v == last_committed + 1) {
- last->uncommitted_pn = pn;
- } else {
- // previously we didn't record which pn a value was accepted
- // under! use the pn value we just had... :(
- dout(10) << "WARNING: no pending_pn on disk, using previous accepted_pn " << previous_pn
- << " and crossing our fingers" << dendl;
- last->uncommitted_pn = previous_pn;
- }
-
- logger->inc(l_paxos_collect_uncommitted);
- }
-
- // send reply
- collect->get_connection()->send_message(last);
-}
-
-/**
- * @note This is Okay. We share our versions between peer_last_committed and
- * our last_committed (inclusive), and add their bufferlists to the
- * message. It will be the peer's job to apply them to its store, as
- * these bufferlists will contain raw transactions.
- * This function is called by both the Peon and the Leader. The Peon will
- * share the state with the Leader during handle_collect(), sharing any
- * values the leader may be missing (i.e., the leader's last_committed is
- * lower than the peon's last_committed). The Leader will share the state
- * with the Peon during handle_last(), if the peon's last_committed is
- * lower than the leader's last_committed.
- */
-void Paxos::share_state(MMonPaxos *m, version_t peer_first_committed,
- version_t peer_last_committed)
-{
- assert(peer_last_committed < last_committed);
-
- dout(10) << "share_state peer has fc " << peer_first_committed
- << " lc " << peer_last_committed << dendl;
- version_t v = peer_last_committed + 1;
-
- // include incrementals
- uint64_t bytes = 0;
- for ( ; v <= last_committed; v++) {
- if (get_store()->exists(get_name(), v)) {
- get_store()->get(get_name(), v, m->values[v]);
- assert(m->values[v].length());
- dout(10) << " sharing " << v << " ("
- << m->values[v].length() << " bytes)" << dendl;
- bytes += m->values[v].length() + 16; // paxos_ + 10 digits = 16
- }
- }
- logger->inc(l_paxos_share_state);
- logger->inc(l_paxos_share_state_keys, m->values.size());
- logger->inc(l_paxos_share_state_bytes, bytes);
-
- m->last_committed = last_committed;
-}
-
-/**
- * Store on disk a state that was shared with us
- *
- * Basically, we received a set of version. Or just one. It doesn't matter.
- * What matters is that we have to stash it in the store. So, we will simply
- * write every single bufferlist into their own versions on our side (i.e.,
- * onto paxos-related keys), and then we will decode those same bufferlists
- * we just wrote and apply the transactions they hold. We will also update
- * our first and last committed values to point to the new values, if need
- * be. All all this is done tightly wrapped in a transaction to ensure we
- * enjoy the atomicity guarantees given by our awesome k/v store.
- */
-bool Paxos::store_state(MMonPaxos *m)
-{
- auto t(std::make_shared<MonitorDBStore::Transaction>());
- map<version_t,bufferlist>::iterator start = m->values.begin();
- bool changed = false;
-
- // build map of values to store
- // we want to write the range [last_committed, m->last_committed] only.
- if (start != m->values.end() &&
- start->first > last_committed + 1) {
- // ignore everything if values start in the future.
- dout(10) << "store_state ignoring all values, they start at " << start->first
- << " > last_committed+1" << dendl;
- return false;
- }
-
- // push forward the start position on the message's values iterator, up until
- // we run out of positions or we find a position matching 'last_committed'.
- while (start != m->values.end() && start->first <= last_committed) {
- ++start;
- }
-
- // make sure we get the right interval of values to apply by pushing forward
- // the 'end' iterator until it matches the message's 'last_committed'.
- map<version_t,bufferlist>::iterator end = start;
- while (end != m->values.end() && end->first <= m->last_committed) {
- last_committed = end->first;
- ++end;
- }
-
- if (start == end) {
- dout(10) << "store_state nothing to commit" << dendl;
- } else {
- dout(10) << "store_state [" << start->first << ".."
- << last_committed << "]" << dendl;
- t->put(get_name(), "last_committed", last_committed);
-
- // we should apply the state here -- decode every single bufferlist in the
- // map and append the transactions to 't'.
- map<version_t,bufferlist>::iterator it;
- for (it = start; it != end; ++it) {
- // write the bufferlist as the version's value
- t->put(get_name(), it->first, it->second);
- // decode the bufferlist and append it to the transaction we will shortly
- // apply.
- decode_append_transaction(t, it->second);
- }
-
- // discard obsolete uncommitted value?
- if (uncommitted_v && uncommitted_v <= last_committed) {
- dout(10) << " forgetting obsolete uncommitted value " << uncommitted_v
- << " pn " << uncommitted_pn << dendl;
- uncommitted_v = 0;
- uncommitted_pn = 0;
- uncommitted_value.clear();
- }
- }
- if (!t->empty()) {
- dout(30) << __func__ << " transaction dump:\n";
- JSONFormatter f(true);
- t->dump(&f);
- f.flush(*_dout);
- *_dout << dendl;
-
- logger->inc(l_paxos_store_state);
- logger->inc(l_paxos_store_state_bytes, t->get_bytes());
- logger->inc(l_paxos_store_state_keys, t->get_keys());
- utime_t start = ceph_clock_now();
-
- get_store()->apply_transaction(t);
-
- utime_t end = ceph_clock_now();
- logger->tinc(l_paxos_store_state_latency, end - start);
-
- // refresh first_committed; this txn may have trimmed.
- first_committed = get_store()->get(get_name(), "first_committed");
-
- _sanity_check_store();
- changed = true;
- }
-
- return changed;
-}
-
-void Paxos::_sanity_check_store()
-{
- version_t lc = get_store()->get(get_name(), "last_committed");
- assert(lc == last_committed);
-}
-
-
-// leader
-void Paxos::handle_last(MonOpRequestRef op)
-{
- op->mark_paxos_event("handle_last");
- MMonPaxos *last = static_cast<MMonPaxos*>(op->get_req());
- bool need_refresh = false;
- int from = last->get_source().num();
-
- dout(10) << "handle_last " << *last << dendl;
-
- if (!mon->is_leader()) {
- dout(10) << "not leader, dropping" << dendl;
- return;
- }
-
- // note peer's first_ and last_committed, in case we learn a new
- // commit and need to push it to them.
- peer_first_committed[from] = last->first_committed;
- peer_last_committed[from] = last->last_committed;
-
- if (last->first_committed > last_committed + 1) {
- dout(5) << __func__
- << " mon." << from
- << " lowest version is too high for our last committed"
- << " (theirs: " << last->first_committed
- << "; ours: " << last_committed << ") -- bootstrap!" << dendl;
- op->mark_paxos_event("need to bootstrap");
- mon->bootstrap();
- return;
- }
-
- assert(g_conf->paxos_kill_at != 1);
-
- // store any committed values if any are specified in the message
- need_refresh = store_state(last);
-
- assert(g_conf->paxos_kill_at != 2);
-
- // is everyone contiguous and up to date?
- for (map<int,version_t>::iterator p = peer_last_committed.begin();
- p != peer_last_committed.end();
- ++p) {
- if (p->second + 1 < first_committed && first_committed > 1) {
- dout(5) << __func__
- << " peon " << p->first
- << " last_committed (" << p->second
- << ") is too low for our first_committed (" << first_committed
- << ") -- bootstrap!" << dendl;
- op->mark_paxos_event("need to bootstrap");
- mon->bootstrap();
- return;
- }
- if (p->second < last_committed) {
- // share committed values
- dout(10) << " sending commit to mon." << p->first << dendl;
- MMonPaxos *commit = new MMonPaxos(mon->get_epoch(),
- MMonPaxos::OP_COMMIT,
- ceph_clock_now());
- share_state(commit, peer_first_committed[p->first], p->second);
- mon->messenger->send_message(commit, mon->monmap->get_inst(p->first));
- }
- }
-
- // do they accept your pn?
- if (last->pn > accepted_pn) {
- // no, try again.
- dout(10) << " they had a higher pn than us, picking a new one." << dendl;
-
- // cancel timeout event
- mon->timer.cancel_event(collect_timeout_event);
- collect_timeout_event = 0;
-
- collect(last->pn);
- } else if (last->pn == accepted_pn) {
- // yes, they accepted our pn. great.
- num_last++;
- dout(10) << " they accepted our pn, we now have "
- << num_last << " peons" << dendl;
-
- // did this person send back an accepted but uncommitted value?
- if (last->uncommitted_pn) {
- if (last->uncommitted_pn >= uncommitted_pn &&
- last->last_committed >= last_committed &&
- last->last_committed + 1 >= uncommitted_v) {
- uncommitted_v = last->last_committed+1;
- uncommitted_pn = last->uncommitted_pn;
- uncommitted_value = last->values[uncommitted_v];
- dout(10) << "we learned an uncommitted value for " << uncommitted_v
- << " pn " << uncommitted_pn
- << " " << uncommitted_value.length() << " bytes"
- << dendl;
- } else {
- dout(10) << "ignoring uncommitted value for " << (last->last_committed+1)
- << " pn " << last->uncommitted_pn
- << " " << last->values[last->last_committed+1].length() << " bytes"
- << dendl;
- }
- }
-
- // is that everyone?
- if (num_last == mon->get_quorum().size()) {
- // cancel timeout event
- mon->timer.cancel_event(collect_timeout_event);
- collect_timeout_event = 0;
- peer_first_committed.clear();
- peer_last_committed.clear();
-
- // almost...
-
- // did we learn an old value?
- if (uncommitted_v == last_committed+1 &&
- uncommitted_value.length()) {
- dout(10) << "that's everyone. begin on old learned value" << dendl;
- state = STATE_UPDATING_PREVIOUS;
- begin(uncommitted_value);
- } else {
- // active!
- dout(10) << "that's everyone. active!" << dendl;
- extend_lease();
-
- need_refresh = false;
- if (do_refresh()) {
- finish_round();
- }
- }
- }
- } else {
- // no, this is an old message, discard
- dout(10) << "old pn, ignoring" << dendl;
- }
-
- if (need_refresh)
- (void)do_refresh();
-}
-
-void Paxos::collect_timeout()
-{
- dout(1) << "collect timeout, calling fresh election" << dendl;
- collect_timeout_event = 0;
- logger->inc(l_paxos_collect_timeout);
- assert(mon->is_leader());
- mon->bootstrap();
-}
-
-
-// leader
-void Paxos::begin(bufferlist& v)
-{
- dout(10) << "begin for " << last_committed+1 << " "
- << v.length() << " bytes"
- << dendl;
-
- assert(mon->is_leader());
- assert(is_updating() || is_updating_previous());
-
- // we must already have a majority for this to work.
- assert(mon->get_quorum().size() == 1 ||
- num_last > (unsigned)mon->monmap->size()/2);
-
- // and no value, yet.
- assert(new_value.length() == 0);
-
- // accept it ourselves
- accepted.clear();
- accepted.insert(mon->rank);
- new_value = v;
-
- if (last_committed == 0) {
- auto t(std::make_shared<MonitorDBStore::Transaction>());
- // initial base case; set first_committed too
- t->put(get_name(), "first_committed", 1);
- decode_append_transaction(t, new_value);
-
- bufferlist tx_bl;
- t->encode(tx_bl);
-
- new_value = tx_bl;
- }
-
- // store the proposed value in the store. IF it is accepted, we will then
- // have to decode it into a transaction and apply it.
- auto t(std::make_shared<MonitorDBStore::Transaction>());
- t->put(get_name(), last_committed+1, new_value);
-
- // note which pn this pending value is for.
- t->put(get_name(), "pending_v", last_committed + 1);
- t->put(get_name(), "pending_pn", accepted_pn);
-
- dout(30) << __func__ << " transaction dump:\n";
- JSONFormatter f(true);
- t->dump(&f);
- f.flush(*_dout);
- auto debug_tx(std::make_shared<MonitorDBStore::Transaction>());
- bufferlist::iterator new_value_it = new_value.begin();
- debug_tx->decode(new_value_it);
- debug_tx->dump(&f);
- *_dout << "\nbl dump:\n";
- f.flush(*_dout);
- *_dout << dendl;
-
- logger->inc(l_paxos_begin);
- logger->inc(l_paxos_begin_keys, t->get_keys());
- logger->inc(l_paxos_begin_bytes, t->get_bytes());
- utime_t start = ceph_clock_now();
-
- get_store()->apply_transaction(t);
-
- utime_t end = ceph_clock_now();
- logger->tinc(l_paxos_begin_latency, end - start);
-
- assert(g_conf->paxos_kill_at != 3);
-
- if (mon->get_quorum().size() == 1) {
- // we're alone, take it easy
- commit_start();
- return;
- }
-
- // ask others to accept it too!
- for (set<int>::const_iterator p = mon->get_quorum().begin();
- p != mon->get_quorum().end();
- ++p) {
- if (*p == mon->rank) continue;
-
- dout(10) << " sending begin to mon." << *p << dendl;
- MMonPaxos *begin = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_BEGIN,
- ceph_clock_now());
- begin->values[last_committed+1] = new_value;
- begin->last_committed = last_committed;
- begin->pn = accepted_pn;
-
- mon->messenger->send_message(begin, mon->monmap->get_inst(*p));
- }
-
- // set timeout event
- accept_timeout_event = mon->timer.add_event_after(
- g_conf->mon_accept_timeout_factor * g_conf->mon_lease,
- new C_MonContext(mon, [this](int r) {
- if (r == -ECANCELED)
- return;
- accept_timeout();
- }));
-}
-
-// peon
-void Paxos::handle_begin(MonOpRequestRef op)
-{
- op->mark_paxos_event("handle_begin");
- MMonPaxos *begin = static_cast<MMonPaxos*>(op->get_req());
- dout(10) << "handle_begin " << *begin << dendl;
-
- // can we accept this?
- if (begin->pn < accepted_pn) {
- dout(10) << " we accepted a higher pn " << accepted_pn << ", ignoring" << dendl;
- op->mark_paxos_event("have higher pn, ignore");
- return;
- }
- assert(begin->pn == accepted_pn);
- assert(begin->last_committed == last_committed);
-
- assert(g_conf->paxos_kill_at != 4);
-
- logger->inc(l_paxos_begin);
-
- // set state.
- state = STATE_UPDATING;
- lease_expire = utime_t(); // cancel lease
-
- // yes.
- version_t v = last_committed+1;
- dout(10) << "accepting value for " << v << " pn " << accepted_pn << dendl;
- // store the accepted value onto our store. We will have to decode it and
- // apply its transaction once we receive permission to commit.
- auto t(std::make_shared<MonitorDBStore::Transaction>());
- t->put(get_name(), v, begin->values[v]);
-
- // note which pn this pending value is for.
- t->put(get_name(), "pending_v", v);
- t->put(get_name(), "pending_pn", accepted_pn);
-
- dout(30) << __func__ << " transaction dump:\n";
- JSONFormatter f(true);
- t->dump(&f);
- f.flush(*_dout);
- *_dout << dendl;
-
- logger->inc(l_paxos_begin_bytes, t->get_bytes());
- utime_t start = ceph_clock_now();
-
- get_store()->apply_transaction(t);
-
- utime_t end = ceph_clock_now();
- logger->tinc(l_paxos_begin_latency, end - start);
-
- assert(g_conf->paxos_kill_at != 5);
-
- // reply
- MMonPaxos *accept = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_ACCEPT,
- ceph_clock_now());
- accept->pn = accepted_pn;
- accept->last_committed = last_committed;
- begin->get_connection()->send_message(accept);
-}
-
-// leader
-void Paxos::handle_accept(MonOpRequestRef op)
-{
- op->mark_paxos_event("handle_accept");
- MMonPaxos *accept = static_cast<MMonPaxos*>(op->get_req());
- dout(10) << "handle_accept " << *accept << dendl;
- int from = accept->get_source().num();
-
- if (accept->pn != accepted_pn) {
- // we accepted a higher pn, from some other leader
- dout(10) << " we accepted a higher pn " << accepted_pn << ", ignoring" << dendl;
- op->mark_paxos_event("have higher pn, ignore");
- return;
- }
- if (last_committed > 0 &&
- accept->last_committed < last_committed-1) {
- dout(10) << " this is from an old round, ignoring" << dendl;
- op->mark_paxos_event("old round, ignore");
- return;
- }
- assert(accept->last_committed == last_committed || // not committed
- accept->last_committed == last_committed-1); // committed
-
- assert(is_updating() || is_updating_previous());
- assert(accepted.count(from) == 0);
- accepted.insert(from);
- dout(10) << " now " << accepted << " have accepted" << dendl;
-
- assert(g_conf->paxos_kill_at != 6);
-
- // only commit (and expose committed state) when we get *all* quorum
- // members to accept. otherwise, they may still be sharing the now
- // stale state.
- // FIXME: we can improve this with an additional lease revocation message
- // that doesn't block for the persist.
- if (accepted == mon->get_quorum()) {
- // yay, commit!
- dout(10) << " got majority, committing, done with update" << dendl;
- op->mark_paxos_event("commit_start");
- commit_start();
- }
-}
-
-void Paxos::accept_timeout()
-{
- dout(1) << "accept timeout, calling fresh election" << dendl;
- accept_timeout_event = 0;
- assert(mon->is_leader());
- assert(is_updating() || is_updating_previous() || is_writing() ||
- is_writing_previous());
- logger->inc(l_paxos_accept_timeout);
- mon->bootstrap();
-}
-
-struct C_Committed : public Context {
- Paxos *paxos;
- explicit C_Committed(Paxos *p) : paxos(p) {}
- void finish(int r) override {
- assert(r >= 0);
- Mutex::Locker l(paxos->mon->lock);
- if (paxos->is_shutdown()) {
- paxos->abort_commit();
- return;
- }
- paxos->commit_finish();
- }
-};
-
-void Paxos::abort_commit()
-{
- assert(commits_started > 0);
- --commits_started;
- if (commits_started == 0)
- shutdown_cond.Signal();
-}
-
-void Paxos::commit_start()
-{
- dout(10) << __func__ << " " << (last_committed+1) << dendl;
-
- assert(g_conf->paxos_kill_at != 7);
-
- auto t(std::make_shared<MonitorDBStore::Transaction>());
-
- // commit locally
- t->put(get_name(), "last_committed", last_committed + 1);
-
- // decode the value and apply its transaction to the store.
- // this value can now be read from last_committed.
- decode_append_transaction(t, new_value);
-
- dout(30) << __func__ << " transaction dump:\n";
- JSONFormatter f(true);
- t->dump(&f);
- f.flush(*_dout);
- *_dout << dendl;
-
- logger->inc(l_paxos_commit);
- logger->inc(l_paxos_commit_keys, t->get_keys());
- logger->inc(l_paxos_commit_bytes, t->get_bytes());
- commit_start_stamp = ceph_clock_now();
-
- get_store()->queue_transaction(t, new C_Committed(this));
-
- if (is_updating_previous())
- state = STATE_WRITING_PREVIOUS;
- else if (is_updating())
- state = STATE_WRITING;
- else
- ceph_abort();
- ++commits_started;
-
- if (mon->get_quorum().size() > 1) {
- // cancel timeout event
- mon->timer.cancel_event(accept_timeout_event);
- accept_timeout_event = 0;
- }
-}
-
-void Paxos::commit_finish()
-{
- dout(20) << __func__ << " " << (last_committed+1) << dendl;
- utime_t end = ceph_clock_now();
- logger->tinc(l_paxos_commit_latency, end - commit_start_stamp);
-
- assert(g_conf->paxos_kill_at != 8);
-
- // cancel lease - it was for the old value.
- // (this would only happen if message layer lost the 'begin', but
- // leader still got a majority and committed with out us.)
- lease_expire = utime_t(); // cancel lease
-
- last_committed++;
- last_commit_time = ceph_clock_now();
-
- // refresh first_committed; this txn may have trimmed.
- first_committed = get_store()->get(get_name(), "first_committed");
-
- _sanity_check_store();
-
- // tell everyone
- for (set<int>::const_iterator p = mon->get_quorum().begin();
- p != mon->get_quorum().end();
- ++p) {
- if (*p == mon->rank) continue;
-
- dout(10) << " sending commit to mon." << *p << dendl;
- MMonPaxos *commit = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_COMMIT,
- ceph_clock_now());
- commit->values[last_committed] = new_value;
- commit->pn = accepted_pn;
- commit->last_committed = last_committed;
-
- mon->messenger->send_message(commit, mon->monmap->get_inst(*p));
- }
-
- assert(g_conf->paxos_kill_at != 9);
-
- // get ready for a new round.
- new_value.clear();
-
- // WRITING -> REFRESH
- // among other things, this lets do_refresh() -> mon->bootstrap() know
- // it doesn't need to flush the store queue
- assert(is_writing() || is_writing_previous());
- state = STATE_REFRESH;
- assert(commits_started > 0);
- --commits_started;
-
- if (do_refresh()) {
- commit_proposal();
- if (mon->get_quorum().size() > 1) {
- extend_lease();
- }
-
- finish_contexts(g_ceph_context, waiting_for_commit);
-
- assert(g_conf->paxos_kill_at != 10);
-
- finish_round();
- }
-}
-
-
-void Paxos::handle_commit(MonOpRequestRef op)
-{
- op->mark_paxos_event("handle_commit");
- MMonPaxos *commit = static_cast<MMonPaxos*>(op->get_req());
- dout(10) << "handle_commit on " << commit->last_committed << dendl;
-
- logger->inc(l_paxos_commit);
-
- if (!mon->is_peon()) {
- dout(10) << "not a peon, dropping" << dendl;
- ceph_abort();
- return;
- }
-
- op->mark_paxos_event("store_state");
- store_state(commit);
-
- if (do_refresh()) {
- finish_contexts(g_ceph_context, waiting_for_commit);
- }
-}
-
-void Paxos::extend_lease()
-{
- assert(mon->is_leader());
- //assert(is_active());
-
- lease_expire = ceph_clock_now();
- lease_expire += g_conf->mon_lease;
- acked_lease.clear();
- acked_lease.insert(mon->rank);
-
- dout(7) << "extend_lease now+" << g_conf->mon_lease
- << " (" << lease_expire << ")" << dendl;
-
- // bcast
- for (set<int>::const_iterator p = mon->get_quorum().begin();
- p != mon->get_quorum().end(); ++p) {
-
- if (*p == mon->rank) continue;
- MMonPaxos *lease = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_LEASE,
- ceph_clock_now());
- lease->last_committed = last_committed;
- lease->lease_timestamp = lease_expire;
- lease->first_committed = first_committed;
- mon->messenger->send_message(lease, mon->monmap->get_inst(*p));
- }
-
- // set timeout event.
- // if old timeout is still in place, leave it.
- if (!lease_ack_timeout_event) {
- lease_ack_timeout_event = mon->timer.add_event_after(
- g_conf->mon_lease_ack_timeout_factor * g_conf->mon_lease,
- new C_MonContext(mon, [this](int r) {
- if (r == -ECANCELED)
- return;
- lease_ack_timeout();
- }));
- }
-
- // set renew event
- utime_t at = lease_expire;
- at -= g_conf->mon_lease;
- at += g_conf->mon_lease_renew_interval_factor * g_conf->mon_lease;
- lease_renew_event = mon->timer.add_event_at(
- at, new C_MonContext(mon, [this](int r) {
- if (r == -ECANCELED)
- return;
- lease_renew_timeout();
- }));
-}
-
-void Paxos::warn_on_future_time(utime_t t, entity_name_t from)
-{
- utime_t now = ceph_clock_now();
- if (t > now) {
- utime_t diff = t - now;
- if (diff > g_conf->mon_clock_drift_allowed) {
- utime_t warn_diff = now - last_clock_drift_warn;
- if (warn_diff >
- pow(g_conf->mon_clock_drift_warn_backoff, clock_drift_warned)) {
- mon->clog->warn() << "message from " << from << " was stamped " << diff
- << "s in the future, clocks not synchronized";
- last_clock_drift_warn = ceph_clock_now();
- ++clock_drift_warned;
- }
- }
- }
-
-}
-
-bool Paxos::do_refresh()
-{
- bool need_bootstrap = false;
-
- utime_t start = ceph_clock_now();
-
- // make sure we have the latest state loaded up
- mon->refresh_from_paxos(&need_bootstrap);
-
- utime_t end = ceph_clock_now();
- logger->inc(l_paxos_refresh);
- logger->tinc(l_paxos_refresh_latency, end - start);
-
- if (need_bootstrap) {
- dout(10) << " doing requested bootstrap" << dendl;
- mon->bootstrap();
- return false;
- }
-
- return true;
-}
-
-void Paxos::commit_proposal()
-{
- dout(10) << __func__ << dendl;
- assert(mon->is_leader());
- assert(is_refresh());
-
- finish_contexts(g_ceph_context, committing_finishers);
-}
-
-void Paxos::finish_round()
-{
- dout(10) << __func__ << dendl;
- assert(mon->is_leader());
-
- // ok, now go active!
- state = STATE_ACTIVE;
-
- dout(20) << __func__ << " waiting_for_acting" << dendl;
- finish_contexts(g_ceph_context, waiting_for_active);
- dout(20) << __func__ << " waiting_for_readable" << dendl;
- finish_contexts(g_ceph_context, waiting_for_readable);
- dout(20) << __func__ << " waiting_for_writeable" << dendl;
- finish_contexts(g_ceph_context, waiting_for_writeable);
-
- dout(10) << __func__ << " done w/ waiters, state " << get_statename(state) << dendl;
-
- if (should_trim()) {
- trim();
- }
-
- if (is_active() && pending_proposal) {
- propose_pending();
- }
-}
-
-
-// peon
-void Paxos::handle_lease(MonOpRequestRef op)
-{
- op->mark_paxos_event("handle_lease");
- MMonPaxos *lease = static_cast<MMonPaxos*>(op->get_req());
- // sanity
- if (!mon->is_peon() ||
- last_committed != lease->last_committed) {
- dout(10) << "handle_lease i'm not a peon, or they're not the leader,"
- << " or the last_committed doesn't match, dropping" << dendl;
- op->mark_paxos_event("invalid lease, ignore");
- return;
- }
-
- warn_on_future_time(lease->sent_timestamp, lease->get_source());
-
- // extend lease
- if (lease_expire < lease->lease_timestamp) {
- lease_expire = lease->lease_timestamp;
-
- utime_t now = ceph_clock_now();
- if (lease_expire < now) {
- utime_t diff = now - lease_expire;
- derr << "lease_expire from " << lease->get_source_inst() << " is " << diff << " seconds in the past; mons are probably laggy (or possibly clocks are too skewed)" << dendl;
- }
- }
-
- state = STATE_ACTIVE;
-
- dout(10) << "handle_lease on " << lease->last_committed
- << " now " << lease_expire << dendl;
-
- // ack
- MMonPaxos *ack = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_LEASE_ACK,
- ceph_clock_now());
- ack->last_committed = last_committed;
- ack->first_committed = first_committed;
- ack->lease_timestamp = ceph_clock_now();
- ::encode(mon->session_map.feature_map, ack->feature_map);
- lease->get_connection()->send_message(ack);
-
- // (re)set timeout event.
- reset_lease_timeout();
-
- // kick waiters
- finish_contexts(g_ceph_context, waiting_for_active);
- if (is_readable())
- finish_contexts(g_ceph_context, waiting_for_readable);
-}
-
-void Paxos::handle_lease_ack(MonOpRequestRef op)
-{
- op->mark_paxos_event("handle_lease_ack");
- MMonPaxos *ack = static_cast<MMonPaxos*>(op->get_req());
- int from = ack->get_source().num();
-
- if (!lease_ack_timeout_event) {
- dout(10) << "handle_lease_ack from " << ack->get_source()
- << " -- stray (probably since revoked)" << dendl;
-
- } else if (acked_lease.count(from) == 0) {
- acked_lease.insert(from);
- if (ack->feature_map.length()) {
- auto p = ack->feature_map.begin();
- FeatureMap& t = mon->quorum_feature_map[from];
- ::decode(t, p);
- }
- if (acked_lease == mon->get_quorum()) {
- // yay!
- dout(10) << "handle_lease_ack from " << ack->get_source()
- << " -- got everyone" << dendl;
- mon->timer.cancel_event(lease_ack_timeout_event);
- lease_ack_timeout_event = 0;
-
-
- } else {
- dout(10) << "handle_lease_ack from " << ack->get_source()
- << " -- still need "
- << mon->get_quorum().size() - acked_lease.size()
- << " more" << dendl;
- }
- } else {
- dout(10) << "handle_lease_ack from " << ack->get_source()
- << " dup (lagging!), ignoring" << dendl;
- }
-
- warn_on_future_time(ack->sent_timestamp, ack->get_source());
-}
-
-void Paxos::lease_ack_timeout()
-{
- dout(1) << "lease_ack_timeout -- calling new election" << dendl;
- assert(mon->is_leader());
- assert(is_active());
- logger->inc(l_paxos_lease_ack_timeout);
- lease_ack_timeout_event = 0;
- mon->bootstrap();
-}
-
-void Paxos::reset_lease_timeout()
-{
- dout(20) << "reset_lease_timeout - setting timeout event" << dendl;
- if (lease_timeout_event)
- mon->timer.cancel_event(lease_timeout_event);
- lease_timeout_event = mon->timer.add_event_after(
- g_conf->mon_lease_ack_timeout_factor * g_conf->mon_lease,
- new C_MonContext(mon, [this](int r) {
- if (r == -ECANCELED)
- return;
- lease_timeout();
- }));
-}
-
-void Paxos::lease_timeout()
-{
- dout(1) << "lease_timeout -- calling new election" << dendl;
- assert(mon->is_peon());
- logger->inc(l_paxos_lease_timeout);
- lease_timeout_event = 0;
- mon->bootstrap();
-}
-
-void Paxos::lease_renew_timeout()
-{
- lease_renew_event = 0;
- extend_lease();
-}
-
-
-/*
- * trim old states
- */
-void Paxos::trim()
-{
- assert(should_trim());
- version_t end = MIN(get_version() - g_conf->paxos_min,
- get_first_committed() + g_conf->paxos_trim_max);
-
- if (first_committed >= end)
- return;
-
- dout(10) << "trim to " << end << " (was " << first_committed << ")" << dendl;
-
- MonitorDBStore::TransactionRef t = get_pending_transaction();
-
- for (version_t v = first_committed; v < end; ++v) {
- dout(10) << "trim " << v << dendl;
- t->erase(get_name(), v);
- }
- t->put(get_name(), "first_committed", end);
- if (g_conf->mon_compact_on_trim) {
- dout(10) << " compacting trimmed range" << dendl;
- t->compact_range(get_name(), stringify(first_committed - 1), stringify(end));
- }
-
- trimming = true;
- queue_pending_finisher(new C_Trimmed(this));
-}
-
-/*
- * return a globally unique, monotonically increasing proposal number
- */
-version_t Paxos::get_new_proposal_number(version_t gt)
-{
- if (last_pn < gt)
- last_pn = gt;
-
- // update. make it unique among all monitors.
- last_pn /= 100;
- last_pn++;
- last_pn *= 100;
- last_pn += (version_t)mon->rank;
-
- // write
- auto t(std::make_shared<MonitorDBStore::Transaction>());
- t->put(get_name(), "last_pn", last_pn);
-
- dout(30) << __func__ << " transaction dump:\n";
- JSONFormatter f(true);
- t->dump(&f);
- f.flush(*_dout);
- *_dout << dendl;
-
- logger->inc(l_paxos_new_pn);
- utime_t start = ceph_clock_now();
-
- get_store()->apply_transaction(t);
-
- utime_t end = ceph_clock_now();
- logger->tinc(l_paxos_new_pn_latency, end - start);
-
- dout(10) << "get_new_proposal_number = " << last_pn << dendl;
- return last_pn;
-}
-
-
-void Paxos::cancel_events()
-{
- if (collect_timeout_event) {
- mon->timer.cancel_event(collect_timeout_event);
- collect_timeout_event = 0;
- }
- if (accept_timeout_event) {
- mon->timer.cancel_event(accept_timeout_event);
- accept_timeout_event = 0;
- }
- if (lease_renew_event) {
- mon->timer.cancel_event(lease_renew_event);
- lease_renew_event = 0;
- }
- if (lease_ack_timeout_event) {
- mon->timer.cancel_event(lease_ack_timeout_event);
- lease_ack_timeout_event = 0;
- }
- if (lease_timeout_event) {
- mon->timer.cancel_event(lease_timeout_event);
- lease_timeout_event = 0;
- }
-}
-
-void Paxos::shutdown()
-{
- dout(10) << __func__ << " cancel all contexts" << dendl;
-
- state = STATE_SHUTDOWN;
-
- // discard pending transaction
- pending_proposal.reset();
-
- // Let store finish commits in progress
- // XXX: I assume I can't use finish_contexts() because the store
- // is going to trigger
- while(commits_started > 0)
- shutdown_cond.Wait(mon->lock);
-
- finish_contexts(g_ceph_context, waiting_for_writeable, -ECANCELED);
- finish_contexts(g_ceph_context, waiting_for_commit, -ECANCELED);
- finish_contexts(g_ceph_context, waiting_for_readable, -ECANCELED);
- finish_contexts(g_ceph_context, waiting_for_active, -ECANCELED);
- finish_contexts(g_ceph_context, pending_finishers, -ECANCELED);
- finish_contexts(g_ceph_context, committing_finishers, -ECANCELED);
- if (logger)
- g_ceph_context->get_perfcounters_collection()->remove(logger);
- delete logger;
-}
-
-void Paxos::leader_init()
-{
- cancel_events();
- new_value.clear();
-
- // discard pending transaction
- pending_proposal.reset();
-
- finish_contexts(g_ceph_context, pending_finishers, -EAGAIN);
- finish_contexts(g_ceph_context, committing_finishers, -EAGAIN);
-
- logger->inc(l_paxos_start_leader);
-
- if (mon->get_quorum().size() == 1) {
- state = STATE_ACTIVE;
- return;
- }
-
- state = STATE_RECOVERING;
- lease_expire = utime_t();
- dout(10) << "leader_init -- starting paxos recovery" << dendl;
- collect(0);
-}
-
-void Paxos::peon_init()
-{
- cancel_events();
- new_value.clear();
-
- state = STATE_RECOVERING;
- lease_expire = utime_t();
- dout(10) << "peon_init -- i am a peon" << dendl;
-
- // start a timer, in case the leader never manages to issue a lease
- reset_lease_timeout();
-
- // discard pending transaction
- pending_proposal.reset();
-
- // no chance to write now!
- finish_contexts(g_ceph_context, waiting_for_writeable, -EAGAIN);
- finish_contexts(g_ceph_context, waiting_for_commit, -EAGAIN);
- finish_contexts(g_ceph_context, pending_finishers, -EAGAIN);
- finish_contexts(g_ceph_context, committing_finishers, -EAGAIN);
-
- logger->inc(l_paxos_start_peon);
-}
-
-void Paxos::restart()
-{
- dout(10) << "restart -- canceling timeouts" << dendl;
- cancel_events();
- new_value.clear();
-
- if (is_writing() || is_writing_previous()) {
- dout(10) << __func__ << " flushing" << dendl;
- mon->lock.Unlock();
- mon->store->flush();
- mon->lock.Lock();
- dout(10) << __func__ << " flushed" << dendl;
- }
- state = STATE_RECOVERING;
-
- // discard pending transaction
- pending_proposal.reset();
-
- finish_contexts(g_ceph_context, committing_finishers, -EAGAIN);
- finish_contexts(g_ceph_context, pending_finishers, -EAGAIN);
- finish_contexts(g_ceph_context, waiting_for_commit, -EAGAIN);
- finish_contexts(g_ceph_context, waiting_for_active, -EAGAIN);
-
- logger->inc(l_paxos_restart);
-}
-
-
-void Paxos::dispatch(MonOpRequestRef op)
-{
- assert(op->is_type_paxos());
- op->mark_paxos_event("dispatch");
- PaxosServiceMessage *m = static_cast<PaxosServiceMessage*>(op->get_req());
- // election in progress?
- if (!mon->is_leader() && !mon->is_peon()) {
- dout(5) << "election in progress, dropping " << *m << dendl;
- return;
- }
-
- // check sanity
- assert(mon->is_leader() ||
- (mon->is_peon() && m->get_source().num() == mon->get_leader()));
-
- switch (m->get_type()) {
-
- case MSG_MON_PAXOS:
- {
- MMonPaxos *pm = reinterpret_cast<MMonPaxos*>(m);
-
- // NOTE: these ops are defined in messages/MMonPaxos.h
- switch (pm->op) {
- // learner
- case MMonPaxos::OP_COLLECT:
- handle_collect(op);
- break;
- case MMonPaxos::OP_LAST:
- handle_last(op);
- break;
- case MMonPaxos::OP_BEGIN:
- handle_begin(op);
- break;
- case MMonPaxos::OP_ACCEPT:
- handle_accept(op);
- break;
- case MMonPaxos::OP_COMMIT:
- handle_commit(op);
- break;
- case MMonPaxos::OP_LEASE:
- handle_lease(op);
- break;
- case MMonPaxos::OP_LEASE_ACK:
- handle_lease_ack(op);
- break;
- default:
- ceph_abort();
- }
- }
- break;
-
- default:
- ceph_abort();
- }
-}
-
-
-// -----------------
-// service interface
-
-// -- READ --
-
-bool Paxos::is_readable(version_t v)
-{
- bool ret;
- if (v > last_committed)
- ret = false;
- else
- ret =
- (mon->is_peon() || mon->is_leader()) &&
- (is_active() || is_updating() || is_writing()) &&
- last_committed > 0 && is_lease_valid(); // must have a value alone, or have lease
- dout(5) << __func__ << " = " << (int)ret
- << " - now=" << ceph_clock_now()
- << " lease_expire=" << lease_expire
- << " has v" << v << " lc " << last_committed
- << dendl;
- return ret;
-}
-
-bool Paxos::read(version_t v, bufferlist &bl)
-{
- if (!get_store()->get(get_name(), v, bl))
- return false;
- return true;
-}
-
-version_t Paxos::read_current(bufferlist &bl)
-{
- if (read(last_committed, bl))
- return last_committed;
- return 0;
-}
-
-
-bool Paxos::is_lease_valid()
-{
- return ((mon->get_quorum().size() == 1)
- || (ceph_clock_now() < lease_expire));
-}
-
-// -- WRITE --
-
-bool Paxos::is_writeable()
-{
- return
- mon->is_leader() &&
- is_active() &&
- is_lease_valid();
-}
-
-void Paxos::propose_pending()
-{
- assert(is_active());
- assert(pending_proposal);
-
- cancel_events();
-
- bufferlist bl;
- pending_proposal->encode(bl);
-
- dout(10) << __func__ << " " << (last_committed + 1)
- << " " << bl.length() << " bytes" << dendl;
- dout(30) << __func__ << " transaction dump:\n";
- JSONFormatter f(true);
- pending_proposal->dump(&f);
- f.flush(*_dout);
- *_dout << dendl;
-
- pending_proposal.reset();
-
- committing_finishers.swap(pending_finishers);
- state = STATE_UPDATING;
- begin(bl);
-}
-
-void Paxos::queue_pending_finisher(Context *onfinished)
-{
- dout(5) << __func__ << " " << onfinished << dendl;
- assert(onfinished);
- pending_finishers.push_back(onfinished);
-}
-
-MonitorDBStore::TransactionRef Paxos::get_pending_transaction()
-{
- assert(mon->is_leader());
- if (!pending_proposal) {
- pending_proposal.reset(new MonitorDBStore::Transaction);
- assert(pending_finishers.empty());
- }
- return pending_proposal;
-}
-
-bool Paxos::trigger_propose()
-{
- if (plugged) {
- dout(10) << __func__ << " plugged, not proposing now" << dendl;
- return false;
- } else if (is_active()) {
- dout(10) << __func__ << " active, proposing now" << dendl;
- propose_pending();
- return true;
- } else {
- dout(10) << __func__ << " not active, will propose later" << dendl;
- return false;
- }
-}
-
-bool Paxos::is_consistent()
-{
- return (first_committed <= last_committed);
-}
-
Code Review / stor4nfv.git / blobdiff