1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 * Ceph - scalable distributed file system
6 * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
8 * This is free software; you can redistribute it and/or
9 * modify it under the terms of the GNU Lesser General Public
10 * License version 2.1, as published by the Free Software
11 * Foundation. See file COPYING.
18 #include "messages/MMonPaxos.h"
20 #include "mon/mon_types.h"
21 #include "common/config.h"
22 #include "include/assert.h"
23 #include "include/stringify.h"
24 #include "common/Timer.h"
25 #include "messages/PaxosServiceMessage.h"
27 #define dout_subsys ceph_subsys_paxos
29 #define dout_prefix _prefix(_dout, mon, mon->name, mon->rank, paxos_name, state, first_committed, last_committed)
30 static ostream& _prefix(std::ostream *_dout, Monitor *mon, const string& name,
31 int rank, const string& paxos_name, int state,
32 version_t first_committed, version_t last_committed)
34 return *_dout << "mon." << name << "@" << rank
35 << "(" << mon->get_state_name() << ")"
36 << ".paxos(" << paxos_name << " " << Paxos::get_statename(state)
37 << " c " << first_committed << ".." << last_committed
41 class Paxos::C_Trimmed : public Context {
44 explicit C_Trimmed(Paxos *p) : paxos(p) { }
45 void finish(int r) override {
46 paxos->trimming = false;
50 MonitorDBStore *Paxos::get_store()
55 void Paxos::read_and_prepare_transactions(MonitorDBStore::TransactionRef tx,
56 version_t first, version_t last)
58 dout(10) << __func__ << " first " << first << " last " << last << dendl;
59 for (version_t v = first; v <= last; ++v) {
60 dout(30) << __func__ << " apply version " << v << dendl;
62 int err = get_store()->get(get_name(), v, bl);
65 decode_append_transaction(tx, bl);
67 dout(15) << __func__ << " total versions " << (last-first) << dendl;
72 // load paxos variables from stable storage
73 last_pn = get_store()->get(get_name(), "last_pn");
74 accepted_pn = get_store()->get(get_name(), "accepted_pn");
75 last_committed = get_store()->get(get_name(), "last_committed");
76 first_committed = get_store()->get(get_name(), "first_committed");
78 dout(10) << __func__ << " last_pn: " << last_pn << " accepted_pn: "
79 << accepted_pn << " last_committed: " << last_committed
80 << " first_committed: " << first_committed << dendl;
82 dout(10) << "init" << dendl;
83 assert(is_consistent());
86 void Paxos::init_logger()
88 PerfCountersBuilder pcb(g_ceph_context, "paxos", l_paxos_first, l_paxos_last);
90 // Because monitors are so few in number, the resource cost of capturing
91 // almost all their perf counters at USEFUL is trivial.
92 pcb.set_prio_default(PerfCountersBuilder::PRIO_USEFUL);
94 pcb.add_u64_counter(l_paxos_start_leader, "start_leader", "Starts in leader role");
95 pcb.add_u64_counter(l_paxos_start_peon, "start_peon", "Starts in peon role");
96 pcb.add_u64_counter(l_paxos_restart, "restart", "Restarts");
97 pcb.add_u64_counter(l_paxos_refresh, "refresh", "Refreshes");
98 pcb.add_time_avg(l_paxos_refresh_latency, "refresh_latency", "Refresh latency");
99 pcb.add_u64_counter(l_paxos_begin, "begin", "Started and handled begins");
100 pcb.add_u64_avg(l_paxos_begin_keys, "begin_keys", "Keys in transaction on begin");
101 pcb.add_u64_avg(l_paxos_begin_bytes, "begin_bytes", "Data in transaction on begin");
102 pcb.add_time_avg(l_paxos_begin_latency, "begin_latency", "Latency of begin operation");
103 pcb.add_u64_counter(l_paxos_commit, "commit",
105 pcb.add_u64_avg(l_paxos_commit_keys, "commit_keys", "Keys in transaction on commit");
106 pcb.add_u64_avg(l_paxos_commit_bytes, "commit_bytes", "Data in transaction on commit");
107 pcb.add_time_avg(l_paxos_commit_latency, "commit_latency",
108 "Commit latency", "clat");
109 pcb.add_u64_counter(l_paxos_collect, "collect", "Peon collects");
110 pcb.add_u64_avg(l_paxos_collect_keys, "collect_keys", "Keys in transaction on peon collect");
111 pcb.add_u64_avg(l_paxos_collect_bytes, "collect_bytes", "Data in transaction on peon collect");
112 pcb.add_time_avg(l_paxos_collect_latency, "collect_latency", "Peon collect latency");
113 pcb.add_u64_counter(l_paxos_collect_uncommitted, "collect_uncommitted", "Uncommitted values in started and handled collects");
114 pcb.add_u64_counter(l_paxos_collect_timeout, "collect_timeout", "Collect timeouts");
115 pcb.add_u64_counter(l_paxos_accept_timeout, "accept_timeout", "Accept timeouts");
116 pcb.add_u64_counter(l_paxos_lease_ack_timeout, "lease_ack_timeout", "Lease acknowledgement timeouts");
117 pcb.add_u64_counter(l_paxos_lease_timeout, "lease_timeout", "Lease timeouts");
118 pcb.add_u64_counter(l_paxos_store_state, "store_state", "Store a shared state on disk");
119 pcb.add_u64_avg(l_paxos_store_state_keys, "store_state_keys", "Keys in transaction in stored state");
120 pcb.add_u64_avg(l_paxos_store_state_bytes, "store_state_bytes", "Data in transaction in stored state");
121 pcb.add_time_avg(l_paxos_store_state_latency, "store_state_latency", "Storing state latency");
122 pcb.add_u64_counter(l_paxos_share_state, "share_state", "Sharings of state");
123 pcb.add_u64_avg(l_paxos_share_state_keys, "share_state_keys", "Keys in shared state");
124 pcb.add_u64_avg(l_paxos_share_state_bytes, "share_state_bytes", "Data in shared state");
125 pcb.add_u64_counter(l_paxos_new_pn, "new_pn", "New proposal number queries");
126 pcb.add_time_avg(l_paxos_new_pn_latency, "new_pn_latency", "New proposal number getting latency");
127 logger = pcb.create_perf_counters();
128 g_ceph_context->get_perfcounters_collection()->add(logger);
131 void Paxos::dump_info(Formatter *f)
133 f->open_object_section("paxos");
134 f->dump_unsigned("first_committed", first_committed);
135 f->dump_unsigned("last_committed", last_committed);
136 f->dump_unsigned("last_pn", last_pn);
137 f->dump_unsigned("accepted_pn", accepted_pn);
141 // ---------------------------------
146 void Paxos::collect(version_t oldpn)
148 // we're recoverying, it seems!
149 state = STATE_RECOVERING;
150 assert(mon->is_leader());
152 // reset the number of lasts received
155 uncommitted_value.clear();
156 peer_first_committed.clear();
157 peer_last_committed.clear();
159 // look for uncommitted value
160 if (get_store()->exists(get_name(), last_committed+1)) {
161 version_t v = get_store()->get(get_name(), "pending_v");
162 version_t pn = get_store()->get(get_name(), "pending_pn");
163 if (v && pn && v == last_committed + 1) {
166 dout(10) << "WARNING: no pending_pn on disk, using previous accepted_pn " << accepted_pn
167 << " and crossing our fingers" << dendl;
168 uncommitted_pn = accepted_pn;
170 uncommitted_v = last_committed+1;
172 get_store()->get(get_name(), last_committed+1, uncommitted_value);
173 assert(uncommitted_value.length());
174 dout(10) << "learned uncommitted " << (last_committed+1)
175 << " pn " << uncommitted_pn
176 << " (" << uncommitted_value.length() << " bytes) from myself"
179 logger->inc(l_paxos_collect_uncommitted);
183 accepted_pn = get_new_proposal_number(MAX(accepted_pn, oldpn));
184 accepted_pn_from = last_committed;
186 dout(10) << "collect with pn " << accepted_pn << dendl;
189 for (set<int>::const_iterator p = mon->get_quorum().begin();
190 p != mon->get_quorum().end();
192 if (*p == mon->rank) continue;
194 MMonPaxos *collect = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_COLLECT,
196 collect->last_committed = last_committed;
197 collect->first_committed = first_committed;
198 collect->pn = accepted_pn;
199 mon->messenger->send_message(collect, mon->monmap->get_inst(*p));
203 collect_timeout_event = mon->timer.add_event_after(
204 g_conf->mon_accept_timeout_factor *
206 new C_MonContext(mon, [this](int r) {
215 void Paxos::handle_collect(MonOpRequestRef op)
218 op->mark_paxos_event("handle_collect");
220 MMonPaxos *collect = static_cast<MMonPaxos*>(op->get_req());
221 dout(10) << "handle_collect " << *collect << dendl;
223 assert(mon->is_peon()); // mon epoch filter should catch strays
225 // we're recoverying, it seems!
226 state = STATE_RECOVERING;
228 //update the peon recovery timeout
229 reset_lease_timeout();
231 if (collect->first_committed > last_committed+1) {
233 << " leader's lowest version is too high for our last committed"
234 << " (theirs: " << collect->first_committed
235 << "; ours: " << last_committed << ") -- bootstrap!" << dendl;
236 op->mark_paxos_event("need to bootstrap");
242 MMonPaxos *last = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_LAST,
244 last->last_committed = last_committed;
245 last->first_committed = first_committed;
247 version_t previous_pn = accepted_pn;
249 // can we accept this pn?
250 if (collect->pn > accepted_pn) {
252 accepted_pn = collect->pn;
253 accepted_pn_from = collect->pn_from;
254 dout(10) << "accepting pn " << accepted_pn << " from "
255 << accepted_pn_from << dendl;
257 auto t(std::make_shared<MonitorDBStore::Transaction>());
258 t->put(get_name(), "accepted_pn", accepted_pn);
260 dout(30) << __func__ << " transaction dump:\n";
261 JSONFormatter f(true);
266 logger->inc(l_paxos_collect);
267 logger->inc(l_paxos_collect_keys, t->get_keys());
268 logger->inc(l_paxos_collect_bytes, t->get_bytes());
269 utime_t start = ceph_clock_now();
271 get_store()->apply_transaction(t);
273 utime_t end = ceph_clock_now();
274 logger->tinc(l_paxos_collect_latency, end - start);
277 dout(10) << "NOT accepting pn " << collect->pn << " from " << collect->pn_from
278 << ", we already accepted " << accepted_pn
279 << " from " << accepted_pn_from << dendl;
281 last->pn = accepted_pn;
282 last->pn_from = accepted_pn_from;
284 // share whatever committed values we have
285 if (collect->last_committed < last_committed)
286 share_state(last, collect->first_committed, collect->last_committed);
288 // do we have an accepted but uncommitted value?
289 // (it'll be at last_committed+1)
291 if (collect->last_committed <= last_committed &&
292 get_store()->exists(get_name(), last_committed+1)) {
293 get_store()->get(get_name(), last_committed+1, bl);
294 assert(bl.length() > 0);
295 dout(10) << " sharing our accepted but uncommitted value for "
296 << last_committed+1 << " (" << bl.length() << " bytes)" << dendl;
297 last->values[last_committed+1] = bl;
299 version_t v = get_store()->get(get_name(), "pending_v");
300 version_t pn = get_store()->get(get_name(), "pending_pn");
301 if (v && pn && v == last_committed + 1) {
302 last->uncommitted_pn = pn;
304 // previously we didn't record which pn a value was accepted
305 // under! use the pn value we just had... :(
306 dout(10) << "WARNING: no pending_pn on disk, using previous accepted_pn " << previous_pn
307 << " and crossing our fingers" << dendl;
308 last->uncommitted_pn = previous_pn;
311 logger->inc(l_paxos_collect_uncommitted);
315 collect->get_connection()->send_message(last);
319 * @note This is Okay. We share our versions between peer_last_committed and
320 * our last_committed (inclusive), and add their bufferlists to the
321 * message. It will be the peer's job to apply them to its store, as
322 * these bufferlists will contain raw transactions.
323 * This function is called by both the Peon and the Leader. The Peon will
324 * share the state with the Leader during handle_collect(), sharing any
325 * values the leader may be missing (i.e., the leader's last_committed is
326 * lower than the peon's last_committed). The Leader will share the state
327 * with the Peon during handle_last(), if the peon's last_committed is
328 * lower than the leader's last_committed.
330 void Paxos::share_state(MMonPaxos *m, version_t peer_first_committed,
331 version_t peer_last_committed)
333 assert(peer_last_committed < last_committed);
335 dout(10) << "share_state peer has fc " << peer_first_committed
336 << " lc " << peer_last_committed << dendl;
337 version_t v = peer_last_committed + 1;
339 // include incrementals
341 for ( ; v <= last_committed; v++) {
342 if (get_store()->exists(get_name(), v)) {
343 get_store()->get(get_name(), v, m->values[v]);
344 assert(m->values[v].length());
345 dout(10) << " sharing " << v << " ("
346 << m->values[v].length() << " bytes)" << dendl;
347 bytes += m->values[v].length() + 16; // paxos_ + 10 digits = 16
350 logger->inc(l_paxos_share_state);
351 logger->inc(l_paxos_share_state_keys, m->values.size());
352 logger->inc(l_paxos_share_state_bytes, bytes);
354 m->last_committed = last_committed;
358 * Store on disk a state that was shared with us
360 * Basically, we received a set of version. Or just one. It doesn't matter.
361 * What matters is that we have to stash it in the store. So, we will simply
362 * write every single bufferlist into their own versions on our side (i.e.,
363 * onto paxos-related keys), and then we will decode those same bufferlists
364 * we just wrote and apply the transactions they hold. We will also update
365 * our first and last committed values to point to the new values, if need
366 * be. All all this is done tightly wrapped in a transaction to ensure we
367 * enjoy the atomicity guarantees given by our awesome k/v store.
369 bool Paxos::store_state(MMonPaxos *m)
371 auto t(std::make_shared<MonitorDBStore::Transaction>());
372 map<version_t,bufferlist>::iterator start = m->values.begin();
373 bool changed = false;
375 // build map of values to store
376 // we want to write the range [last_committed, m->last_committed] only.
377 if (start != m->values.end() &&
378 start->first > last_committed + 1) {
379 // ignore everything if values start in the future.
380 dout(10) << "store_state ignoring all values, they start at " << start->first
381 << " > last_committed+1" << dendl;
385 // push forward the start position on the message's values iterator, up until
386 // we run out of positions or we find a position matching 'last_committed'.
387 while (start != m->values.end() && start->first <= last_committed) {
391 // make sure we get the right interval of values to apply by pushing forward
392 // the 'end' iterator until it matches the message's 'last_committed'.
393 map<version_t,bufferlist>::iterator end = start;
394 while (end != m->values.end() && end->first <= m->last_committed) {
395 last_committed = end->first;
400 dout(10) << "store_state nothing to commit" << dendl;
402 dout(10) << "store_state [" << start->first << ".."
403 << last_committed << "]" << dendl;
404 t->put(get_name(), "last_committed", last_committed);
406 // we should apply the state here -- decode every single bufferlist in the
407 // map and append the transactions to 't'.
408 map<version_t,bufferlist>::iterator it;
409 for (it = start; it != end; ++it) {
410 // write the bufferlist as the version's value
411 t->put(get_name(), it->first, it->second);
412 // decode the bufferlist and append it to the transaction we will shortly
414 decode_append_transaction(t, it->second);
417 // discard obsolete uncommitted value?
418 if (uncommitted_v && uncommitted_v <= last_committed) {
419 dout(10) << " forgetting obsolete uncommitted value " << uncommitted_v
420 << " pn " << uncommitted_pn << dendl;
423 uncommitted_value.clear();
427 dout(30) << __func__ << " transaction dump:\n";
428 JSONFormatter f(true);
433 logger->inc(l_paxos_store_state);
434 logger->inc(l_paxos_store_state_bytes, t->get_bytes());
435 logger->inc(l_paxos_store_state_keys, t->get_keys());
436 utime_t start = ceph_clock_now();
438 get_store()->apply_transaction(t);
440 utime_t end = ceph_clock_now();
441 logger->tinc(l_paxos_store_state_latency, end - start);
443 // refresh first_committed; this txn may have trimmed.
444 first_committed = get_store()->get(get_name(), "first_committed");
446 _sanity_check_store();
453 void Paxos::_sanity_check_store()
455 version_t lc = get_store()->get(get_name(), "last_committed");
456 assert(lc == last_committed);
461 void Paxos::handle_last(MonOpRequestRef op)
463 op->mark_paxos_event("handle_last");
464 MMonPaxos *last = static_cast<MMonPaxos*>(op->get_req());
465 bool need_refresh = false;
466 int from = last->get_source().num();
468 dout(10) << "handle_last " << *last << dendl;
470 if (!mon->is_leader()) {
471 dout(10) << "not leader, dropping" << dendl;
475 // note peer's first_ and last_committed, in case we learn a new
476 // commit and need to push it to them.
477 peer_first_committed[from] = last->first_committed;
478 peer_last_committed[from] = last->last_committed;
480 if (last->first_committed > last_committed + 1) {
483 << " lowest version is too high for our last committed"
484 << " (theirs: " << last->first_committed
485 << "; ours: " << last_committed << ") -- bootstrap!" << dendl;
486 op->mark_paxos_event("need to bootstrap");
491 assert(g_conf->paxos_kill_at != 1);
493 // store any committed values if any are specified in the message
494 need_refresh = store_state(last);
496 assert(g_conf->paxos_kill_at != 2);
498 // is everyone contiguous and up to date?
499 for (map<int,version_t>::iterator p = peer_last_committed.begin();
500 p != peer_last_committed.end();
502 if (p->second + 1 < first_committed && first_committed > 1) {
504 << " peon " << p->first
505 << " last_committed (" << p->second
506 << ") is too low for our first_committed (" << first_committed
507 << ") -- bootstrap!" << dendl;
508 op->mark_paxos_event("need to bootstrap");
512 if (p->second < last_committed) {
513 // share committed values
514 dout(10) << " sending commit to mon." << p->first << dendl;
515 MMonPaxos *commit = new MMonPaxos(mon->get_epoch(),
516 MMonPaxos::OP_COMMIT,
518 share_state(commit, peer_first_committed[p->first], p->second);
519 mon->messenger->send_message(commit, mon->monmap->get_inst(p->first));
523 // do they accept your pn?
524 if (last->pn > accepted_pn) {
526 dout(10) << " they had a higher pn than us, picking a new one." << dendl;
528 // cancel timeout event
529 mon->timer.cancel_event(collect_timeout_event);
530 collect_timeout_event = 0;
533 } else if (last->pn == accepted_pn) {
534 // yes, they accepted our pn. great.
536 dout(10) << " they accepted our pn, we now have "
537 << num_last << " peons" << dendl;
539 // did this person send back an accepted but uncommitted value?
540 if (last->uncommitted_pn) {
541 if (last->uncommitted_pn >= uncommitted_pn &&
542 last->last_committed >= last_committed &&
543 last->last_committed + 1 >= uncommitted_v) {
544 uncommitted_v = last->last_committed+1;
545 uncommitted_pn = last->uncommitted_pn;
546 uncommitted_value = last->values[uncommitted_v];
547 dout(10) << "we learned an uncommitted value for " << uncommitted_v
548 << " pn " << uncommitted_pn
549 << " " << uncommitted_value.length() << " bytes"
552 dout(10) << "ignoring uncommitted value for " << (last->last_committed+1)
553 << " pn " << last->uncommitted_pn
554 << " " << last->values[last->last_committed+1].length() << " bytes"
560 if (num_last == mon->get_quorum().size()) {
561 // cancel timeout event
562 mon->timer.cancel_event(collect_timeout_event);
563 collect_timeout_event = 0;
564 peer_first_committed.clear();
565 peer_last_committed.clear();
569 // did we learn an old value?
570 if (uncommitted_v == last_committed+1 &&
571 uncommitted_value.length()) {
572 dout(10) << "that's everyone. begin on old learned value" << dendl;
573 state = STATE_UPDATING_PREVIOUS;
574 begin(uncommitted_value);
577 dout(10) << "that's everyone. active!" << dendl;
580 need_refresh = false;
587 // no, this is an old message, discard
588 dout(10) << "old pn, ignoring" << dendl;
595 void Paxos::collect_timeout()
597 dout(1) << "collect timeout, calling fresh election" << dendl;
598 collect_timeout_event = 0;
599 logger->inc(l_paxos_collect_timeout);
600 assert(mon->is_leader());
606 void Paxos::begin(bufferlist& v)
608 dout(10) << "begin for " << last_committed+1 << " "
609 << v.length() << " bytes"
612 assert(mon->is_leader());
613 assert(is_updating() || is_updating_previous());
615 // we must already have a majority for this to work.
616 assert(mon->get_quorum().size() == 1 ||
617 num_last > (unsigned)mon->monmap->size()/2);
619 // and no value, yet.
620 assert(new_value.length() == 0);
622 // accept it ourselves
624 accepted.insert(mon->rank);
627 if (last_committed == 0) {
628 auto t(std::make_shared<MonitorDBStore::Transaction>());
629 // initial base case; set first_committed too
630 t->put(get_name(), "first_committed", 1);
631 decode_append_transaction(t, new_value);
639 // store the proposed value in the store. IF it is accepted, we will then
640 // have to decode it into a transaction and apply it.
641 auto t(std::make_shared<MonitorDBStore::Transaction>());
642 t->put(get_name(), last_committed+1, new_value);
644 // note which pn this pending value is for.
645 t->put(get_name(), "pending_v", last_committed + 1);
646 t->put(get_name(), "pending_pn", accepted_pn);
648 dout(30) << __func__ << " transaction dump:\n";
649 JSONFormatter f(true);
652 auto debug_tx(std::make_shared<MonitorDBStore::Transaction>());
653 bufferlist::iterator new_value_it = new_value.begin();
654 debug_tx->decode(new_value_it);
656 *_dout << "\nbl dump:\n";
660 logger->inc(l_paxos_begin);
661 logger->inc(l_paxos_begin_keys, t->get_keys());
662 logger->inc(l_paxos_begin_bytes, t->get_bytes());
663 utime_t start = ceph_clock_now();
665 get_store()->apply_transaction(t);
667 utime_t end = ceph_clock_now();
668 logger->tinc(l_paxos_begin_latency, end - start);
670 assert(g_conf->paxos_kill_at != 3);
672 if (mon->get_quorum().size() == 1) {
673 // we're alone, take it easy
678 // ask others to accept it too!
679 for (set<int>::const_iterator p = mon->get_quorum().begin();
680 p != mon->get_quorum().end();
682 if (*p == mon->rank) continue;
684 dout(10) << " sending begin to mon." << *p << dendl;
685 MMonPaxos *begin = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_BEGIN,
687 begin->values[last_committed+1] = new_value;
688 begin->last_committed = last_committed;
689 begin->pn = accepted_pn;
691 mon->messenger->send_message(begin, mon->monmap->get_inst(*p));
695 accept_timeout_event = mon->timer.add_event_after(
696 g_conf->mon_accept_timeout_factor * g_conf->mon_lease,
697 new C_MonContext(mon, [this](int r) {
705 void Paxos::handle_begin(MonOpRequestRef op)
707 op->mark_paxos_event("handle_begin");
708 MMonPaxos *begin = static_cast<MMonPaxos*>(op->get_req());
709 dout(10) << "handle_begin " << *begin << dendl;
711 // can we accept this?
712 if (begin->pn < accepted_pn) {
713 dout(10) << " we accepted a higher pn " << accepted_pn << ", ignoring" << dendl;
714 op->mark_paxos_event("have higher pn, ignore");
717 assert(begin->pn == accepted_pn);
718 assert(begin->last_committed == last_committed);
720 assert(g_conf->paxos_kill_at != 4);
722 logger->inc(l_paxos_begin);
725 state = STATE_UPDATING;
726 lease_expire = utime_t(); // cancel lease
729 version_t v = last_committed+1;
730 dout(10) << "accepting value for " << v << " pn " << accepted_pn << dendl;
731 // store the accepted value onto our store. We will have to decode it and
732 // apply its transaction once we receive permission to commit.
733 auto t(std::make_shared<MonitorDBStore::Transaction>());
734 t->put(get_name(), v, begin->values[v]);
736 // note which pn this pending value is for.
737 t->put(get_name(), "pending_v", v);
738 t->put(get_name(), "pending_pn", accepted_pn);
740 dout(30) << __func__ << " transaction dump:\n";
741 JSONFormatter f(true);
746 logger->inc(l_paxos_begin_bytes, t->get_bytes());
747 utime_t start = ceph_clock_now();
749 get_store()->apply_transaction(t);
751 utime_t end = ceph_clock_now();
752 logger->tinc(l_paxos_begin_latency, end - start);
754 assert(g_conf->paxos_kill_at != 5);
757 MMonPaxos *accept = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_ACCEPT,
759 accept->pn = accepted_pn;
760 accept->last_committed = last_committed;
761 begin->get_connection()->send_message(accept);
765 void Paxos::handle_accept(MonOpRequestRef op)
767 op->mark_paxos_event("handle_accept");
768 MMonPaxos *accept = static_cast<MMonPaxos*>(op->get_req());
769 dout(10) << "handle_accept " << *accept << dendl;
770 int from = accept->get_source().num();
772 if (accept->pn != accepted_pn) {
773 // we accepted a higher pn, from some other leader
774 dout(10) << " we accepted a higher pn " << accepted_pn << ", ignoring" << dendl;
775 op->mark_paxos_event("have higher pn, ignore");
778 if (last_committed > 0 &&
779 accept->last_committed < last_committed-1) {
780 dout(10) << " this is from an old round, ignoring" << dendl;
781 op->mark_paxos_event("old round, ignore");
784 assert(accept->last_committed == last_committed || // not committed
785 accept->last_committed == last_committed-1); // committed
787 assert(is_updating() || is_updating_previous());
788 assert(accepted.count(from) == 0);
789 accepted.insert(from);
790 dout(10) << " now " << accepted << " have accepted" << dendl;
792 assert(g_conf->paxos_kill_at != 6);
794 // only commit (and expose committed state) when we get *all* quorum
795 // members to accept. otherwise, they may still be sharing the now
797 // FIXME: we can improve this with an additional lease revocation message
798 // that doesn't block for the persist.
799 if (accepted == mon->get_quorum()) {
801 dout(10) << " got majority, committing, done with update" << dendl;
802 op->mark_paxos_event("commit_start");
807 void Paxos::accept_timeout()
809 dout(1) << "accept timeout, calling fresh election" << dendl;
810 accept_timeout_event = 0;
811 assert(mon->is_leader());
812 assert(is_updating() || is_updating_previous() || is_writing() ||
813 is_writing_previous());
814 logger->inc(l_paxos_accept_timeout);
818 struct C_Committed : public Context {
820 explicit C_Committed(Paxos *p) : paxos(p) {}
821 void finish(int r) override {
823 Mutex::Locker l(paxos->mon->lock);
824 if (paxos->is_shutdown()) {
825 paxos->abort_commit();
828 paxos->commit_finish();
832 void Paxos::abort_commit()
834 assert(commits_started > 0);
836 if (commits_started == 0)
837 shutdown_cond.Signal();
840 void Paxos::commit_start()
842 dout(10) << __func__ << " " << (last_committed+1) << dendl;
844 assert(g_conf->paxos_kill_at != 7);
846 auto t(std::make_shared<MonitorDBStore::Transaction>());
849 t->put(get_name(), "last_committed", last_committed + 1);
851 // decode the value and apply its transaction to the store.
852 // this value can now be read from last_committed.
853 decode_append_transaction(t, new_value);
855 dout(30) << __func__ << " transaction dump:\n";
856 JSONFormatter f(true);
861 logger->inc(l_paxos_commit);
862 logger->inc(l_paxos_commit_keys, t->get_keys());
863 logger->inc(l_paxos_commit_bytes, t->get_bytes());
864 commit_start_stamp = ceph_clock_now();
866 get_store()->queue_transaction(t, new C_Committed(this));
868 if (is_updating_previous())
869 state = STATE_WRITING_PREVIOUS;
870 else if (is_updating())
871 state = STATE_WRITING;
876 if (mon->get_quorum().size() > 1) {
877 // cancel timeout event
878 mon->timer.cancel_event(accept_timeout_event);
879 accept_timeout_event = 0;
883 void Paxos::commit_finish()
885 dout(20) << __func__ << " " << (last_committed+1) << dendl;
886 utime_t end = ceph_clock_now();
887 logger->tinc(l_paxos_commit_latency, end - commit_start_stamp);
889 assert(g_conf->paxos_kill_at != 8);
891 // cancel lease - it was for the old value.
892 // (this would only happen if message layer lost the 'begin', but
893 // leader still got a majority and committed with out us.)
894 lease_expire = utime_t(); // cancel lease
897 last_commit_time = ceph_clock_now();
899 // refresh first_committed; this txn may have trimmed.
900 first_committed = get_store()->get(get_name(), "first_committed");
902 _sanity_check_store();
905 for (set<int>::const_iterator p = mon->get_quorum().begin();
906 p != mon->get_quorum().end();
908 if (*p == mon->rank) continue;
910 dout(10) << " sending commit to mon." << *p << dendl;
911 MMonPaxos *commit = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_COMMIT,
913 commit->values[last_committed] = new_value;
914 commit->pn = accepted_pn;
915 commit->last_committed = last_committed;
917 mon->messenger->send_message(commit, mon->monmap->get_inst(*p));
920 assert(g_conf->paxos_kill_at != 9);
922 // get ready for a new round.
925 // WRITING -> REFRESH
926 // among other things, this lets do_refresh() -> mon->bootstrap() know
927 // it doesn't need to flush the store queue
928 assert(is_writing() || is_writing_previous());
929 state = STATE_REFRESH;
930 assert(commits_started > 0);
935 if (mon->get_quorum().size() > 1) {
939 finish_contexts(g_ceph_context, waiting_for_commit);
941 assert(g_conf->paxos_kill_at != 10);
948 void Paxos::handle_commit(MonOpRequestRef op)
950 op->mark_paxos_event("handle_commit");
951 MMonPaxos *commit = static_cast<MMonPaxos*>(op->get_req());
952 dout(10) << "handle_commit on " << commit->last_committed << dendl;
954 logger->inc(l_paxos_commit);
956 if (!mon->is_peon()) {
957 dout(10) << "not a peon, dropping" << dendl;
962 op->mark_paxos_event("store_state");
966 finish_contexts(g_ceph_context, waiting_for_commit);
970 void Paxos::extend_lease()
972 assert(mon->is_leader());
973 //assert(is_active());
975 lease_expire = ceph_clock_now();
976 lease_expire += g_conf->mon_lease;
978 acked_lease.insert(mon->rank);
980 dout(7) << "extend_lease now+" << g_conf->mon_lease
981 << " (" << lease_expire << ")" << dendl;
984 for (set<int>::const_iterator p = mon->get_quorum().begin();
985 p != mon->get_quorum().end(); ++p) {
987 if (*p == mon->rank) continue;
988 MMonPaxos *lease = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_LEASE,
990 lease->last_committed = last_committed;
991 lease->lease_timestamp = lease_expire;
992 lease->first_committed = first_committed;
993 mon->messenger->send_message(lease, mon->monmap->get_inst(*p));
996 // set timeout event.
997 // if old timeout is still in place, leave it.
998 if (!lease_ack_timeout_event) {
999 lease_ack_timeout_event = mon->timer.add_event_after(
1000 g_conf->mon_lease_ack_timeout_factor * g_conf->mon_lease,
1001 new C_MonContext(mon, [this](int r) {
1002 if (r == -ECANCELED)
1004 lease_ack_timeout();
1009 utime_t at = lease_expire;
1010 at -= g_conf->mon_lease;
1011 at += g_conf->mon_lease_renew_interval_factor * g_conf->mon_lease;
1012 lease_renew_event = mon->timer.add_event_at(
1013 at, new C_MonContext(mon, [this](int r) {
1014 if (r == -ECANCELED)
1016 lease_renew_timeout();
1020 void Paxos::warn_on_future_time(utime_t t, entity_name_t from)
1022 utime_t now = ceph_clock_now();
1024 utime_t diff = t - now;
1025 if (diff > g_conf->mon_clock_drift_allowed) {
1026 utime_t warn_diff = now - last_clock_drift_warn;
1028 pow(g_conf->mon_clock_drift_warn_backoff, clock_drift_warned)) {
1029 mon->clog->warn() << "message from " << from << " was stamped " << diff
1030 << "s in the future, clocks not synchronized";
1031 last_clock_drift_warn = ceph_clock_now();
1032 ++clock_drift_warned;
1039 bool Paxos::do_refresh()
1041 bool need_bootstrap = false;
1043 utime_t start = ceph_clock_now();
1045 // make sure we have the latest state loaded up
1046 mon->refresh_from_paxos(&need_bootstrap);
1048 utime_t end = ceph_clock_now();
1049 logger->inc(l_paxos_refresh);
1050 logger->tinc(l_paxos_refresh_latency, end - start);
1052 if (need_bootstrap) {
1053 dout(10) << " doing requested bootstrap" << dendl;
1061 void Paxos::commit_proposal()
1063 dout(10) << __func__ << dendl;
1064 assert(mon->is_leader());
1065 assert(is_refresh());
1067 finish_contexts(g_ceph_context, committing_finishers);
1070 void Paxos::finish_round()
1072 dout(10) << __func__ << dendl;
1073 assert(mon->is_leader());
1075 // ok, now go active!
1076 state = STATE_ACTIVE;
1078 dout(20) << __func__ << " waiting_for_acting" << dendl;
1079 finish_contexts(g_ceph_context, waiting_for_active);
1080 dout(20) << __func__ << " waiting_for_readable" << dendl;
1081 finish_contexts(g_ceph_context, waiting_for_readable);
1082 dout(20) << __func__ << " waiting_for_writeable" << dendl;
1083 finish_contexts(g_ceph_context, waiting_for_writeable);
1085 dout(10) << __func__ << " done w/ waiters, state " << get_statename(state) << dendl;
1087 if (should_trim()) {
1091 if (is_active() && pending_proposal) {
1098 void Paxos::handle_lease(MonOpRequestRef op)
1100 op->mark_paxos_event("handle_lease");
1101 MMonPaxos *lease = static_cast<MMonPaxos*>(op->get_req());
1103 if (!mon->is_peon() ||
1104 last_committed != lease->last_committed) {
1105 dout(10) << "handle_lease i'm not a peon, or they're not the leader,"
1106 << " or the last_committed doesn't match, dropping" << dendl;
1107 op->mark_paxos_event("invalid lease, ignore");
1111 warn_on_future_time(lease->sent_timestamp, lease->get_source());
1114 if (lease_expire < lease->lease_timestamp) {
1115 lease_expire = lease->lease_timestamp;
1117 utime_t now = ceph_clock_now();
1118 if (lease_expire < now) {
1119 utime_t diff = now - lease_expire;
1120 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;
1124 state = STATE_ACTIVE;
1126 dout(10) << "handle_lease on " << lease->last_committed
1127 << " now " << lease_expire << dendl;
1130 MMonPaxos *ack = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_LEASE_ACK,
1132 ack->last_committed = last_committed;
1133 ack->first_committed = first_committed;
1134 ack->lease_timestamp = ceph_clock_now();
1135 ::encode(mon->session_map.feature_map, ack->feature_map);
1136 lease->get_connection()->send_message(ack);
1138 // (re)set timeout event.
1139 reset_lease_timeout();
1142 finish_contexts(g_ceph_context, waiting_for_active);
1144 finish_contexts(g_ceph_context, waiting_for_readable);
1147 void Paxos::handle_lease_ack(MonOpRequestRef op)
1149 op->mark_paxos_event("handle_lease_ack");
1150 MMonPaxos *ack = static_cast<MMonPaxos*>(op->get_req());
1151 int from = ack->get_source().num();
1153 if (!lease_ack_timeout_event) {
1154 dout(10) << "handle_lease_ack from " << ack->get_source()
1155 << " -- stray (probably since revoked)" << dendl;
1157 } else if (acked_lease.count(from) == 0) {
1158 acked_lease.insert(from);
1159 if (ack->feature_map.length()) {
1160 auto p = ack->feature_map.begin();
1161 FeatureMap& t = mon->quorum_feature_map[from];
1164 if (acked_lease == mon->get_quorum()) {
1166 dout(10) << "handle_lease_ack from " << ack->get_source()
1167 << " -- got everyone" << dendl;
1168 mon->timer.cancel_event(lease_ack_timeout_event);
1169 lease_ack_timeout_event = 0;
1173 dout(10) << "handle_lease_ack from " << ack->get_source()
1174 << " -- still need "
1175 << mon->get_quorum().size() - acked_lease.size()
1176 << " more" << dendl;
1179 dout(10) << "handle_lease_ack from " << ack->get_source()
1180 << " dup (lagging!), ignoring" << dendl;
1183 warn_on_future_time(ack->sent_timestamp, ack->get_source());
1186 void Paxos::lease_ack_timeout()
1188 dout(1) << "lease_ack_timeout -- calling new election" << dendl;
1189 assert(mon->is_leader());
1190 assert(is_active());
1191 logger->inc(l_paxos_lease_ack_timeout);
1192 lease_ack_timeout_event = 0;
1196 void Paxos::reset_lease_timeout()
1198 dout(20) << "reset_lease_timeout - setting timeout event" << dendl;
1199 if (lease_timeout_event)
1200 mon->timer.cancel_event(lease_timeout_event);
1201 lease_timeout_event = mon->timer.add_event_after(
1202 g_conf->mon_lease_ack_timeout_factor * g_conf->mon_lease,
1203 new C_MonContext(mon, [this](int r) {
1204 if (r == -ECANCELED)
1210 void Paxos::lease_timeout()
1212 dout(1) << "lease_timeout -- calling new election" << dendl;
1213 assert(mon->is_peon());
1214 logger->inc(l_paxos_lease_timeout);
1215 lease_timeout_event = 0;
1219 void Paxos::lease_renew_timeout()
1221 lease_renew_event = 0;
1231 assert(should_trim());
1232 version_t end = MIN(get_version() - g_conf->paxos_min,
1233 get_first_committed() + g_conf->paxos_trim_max);
1235 if (first_committed >= end)
1238 dout(10) << "trim to " << end << " (was " << first_committed << ")" << dendl;
1240 MonitorDBStore::TransactionRef t = get_pending_transaction();
1242 for (version_t v = first_committed; v < end; ++v) {
1243 dout(10) << "trim " << v << dendl;
1244 t->erase(get_name(), v);
1246 t->put(get_name(), "first_committed", end);
1247 if (g_conf->mon_compact_on_trim) {
1248 dout(10) << " compacting trimmed range" << dendl;
1249 t->compact_range(get_name(), stringify(first_committed - 1), stringify(end));
1253 queue_pending_finisher(new C_Trimmed(this));
1257 * return a globally unique, monotonically increasing proposal number
1259 version_t Paxos::get_new_proposal_number(version_t gt)
1264 // update. make it unique among all monitors.
1268 last_pn += (version_t)mon->rank;
1271 auto t(std::make_shared<MonitorDBStore::Transaction>());
1272 t->put(get_name(), "last_pn", last_pn);
1274 dout(30) << __func__ << " transaction dump:\n";
1275 JSONFormatter f(true);
1280 logger->inc(l_paxos_new_pn);
1281 utime_t start = ceph_clock_now();
1283 get_store()->apply_transaction(t);
1285 utime_t end = ceph_clock_now();
1286 logger->tinc(l_paxos_new_pn_latency, end - start);
1288 dout(10) << "get_new_proposal_number = " << last_pn << dendl;
1293 void Paxos::cancel_events()
1295 if (collect_timeout_event) {
1296 mon->timer.cancel_event(collect_timeout_event);
1297 collect_timeout_event = 0;
1299 if (accept_timeout_event) {
1300 mon->timer.cancel_event(accept_timeout_event);
1301 accept_timeout_event = 0;
1303 if (lease_renew_event) {
1304 mon->timer.cancel_event(lease_renew_event);
1305 lease_renew_event = 0;
1307 if (lease_ack_timeout_event) {
1308 mon->timer.cancel_event(lease_ack_timeout_event);
1309 lease_ack_timeout_event = 0;
1311 if (lease_timeout_event) {
1312 mon->timer.cancel_event(lease_timeout_event);
1313 lease_timeout_event = 0;
1317 void Paxos::shutdown()
1319 dout(10) << __func__ << " cancel all contexts" << dendl;
1321 state = STATE_SHUTDOWN;
1323 // discard pending transaction
1324 pending_proposal.reset();
1326 // Let store finish commits in progress
1327 // XXX: I assume I can't use finish_contexts() because the store
1328 // is going to trigger
1329 while(commits_started > 0)
1330 shutdown_cond.Wait(mon->lock);
1332 finish_contexts(g_ceph_context, waiting_for_writeable, -ECANCELED);
1333 finish_contexts(g_ceph_context, waiting_for_commit, -ECANCELED);
1334 finish_contexts(g_ceph_context, waiting_for_readable, -ECANCELED);
1335 finish_contexts(g_ceph_context, waiting_for_active, -ECANCELED);
1336 finish_contexts(g_ceph_context, pending_finishers, -ECANCELED);
1337 finish_contexts(g_ceph_context, committing_finishers, -ECANCELED);
1339 g_ceph_context->get_perfcounters_collection()->remove(logger);
1343 void Paxos::leader_init()
1348 // discard pending transaction
1349 pending_proposal.reset();
1351 finish_contexts(g_ceph_context, pending_finishers, -EAGAIN);
1352 finish_contexts(g_ceph_context, committing_finishers, -EAGAIN);
1354 logger->inc(l_paxos_start_leader);
1356 if (mon->get_quorum().size() == 1) {
1357 state = STATE_ACTIVE;
1361 state = STATE_RECOVERING;
1362 lease_expire = utime_t();
1363 dout(10) << "leader_init -- starting paxos recovery" << dendl;
1367 void Paxos::peon_init()
1372 state = STATE_RECOVERING;
1373 lease_expire = utime_t();
1374 dout(10) << "peon_init -- i am a peon" << dendl;
1376 // start a timer, in case the leader never manages to issue a lease
1377 reset_lease_timeout();
1379 // discard pending transaction
1380 pending_proposal.reset();
1382 // no chance to write now!
1383 finish_contexts(g_ceph_context, waiting_for_writeable, -EAGAIN);
1384 finish_contexts(g_ceph_context, waiting_for_commit, -EAGAIN);
1385 finish_contexts(g_ceph_context, pending_finishers, -EAGAIN);
1386 finish_contexts(g_ceph_context, committing_finishers, -EAGAIN);
1388 logger->inc(l_paxos_start_peon);
1391 void Paxos::restart()
1393 dout(10) << "restart -- canceling timeouts" << dendl;
1397 if (is_writing() || is_writing_previous()) {
1398 dout(10) << __func__ << " flushing" << dendl;
1400 mon->store->flush();
1402 dout(10) << __func__ << " flushed" << dendl;
1404 state = STATE_RECOVERING;
1406 // discard pending transaction
1407 pending_proposal.reset();
1409 finish_contexts(g_ceph_context, committing_finishers, -EAGAIN);
1410 finish_contexts(g_ceph_context, pending_finishers, -EAGAIN);
1411 finish_contexts(g_ceph_context, waiting_for_commit, -EAGAIN);
1412 finish_contexts(g_ceph_context, waiting_for_active, -EAGAIN);
1414 logger->inc(l_paxos_restart);
1418 void Paxos::dispatch(MonOpRequestRef op)
1420 assert(op->is_type_paxos());
1421 op->mark_paxos_event("dispatch");
1422 PaxosServiceMessage *m = static_cast<PaxosServiceMessage*>(op->get_req());
1423 // election in progress?
1424 if (!mon->is_leader() && !mon->is_peon()) {
1425 dout(5) << "election in progress, dropping " << *m << dendl;
1430 assert(mon->is_leader() ||
1431 (mon->is_peon() && m->get_source().num() == mon->get_leader()));
1433 switch (m->get_type()) {
1437 MMonPaxos *pm = reinterpret_cast<MMonPaxos*>(m);
1439 // NOTE: these ops are defined in messages/MMonPaxos.h
1442 case MMonPaxos::OP_COLLECT:
1445 case MMonPaxos::OP_LAST:
1448 case MMonPaxos::OP_BEGIN:
1451 case MMonPaxos::OP_ACCEPT:
1454 case MMonPaxos::OP_COMMIT:
1457 case MMonPaxos::OP_LEASE:
1460 case MMonPaxos::OP_LEASE_ACK:
1461 handle_lease_ack(op);
1475 // -----------------
1476 // service interface
1480 bool Paxos::is_readable(version_t v)
1483 if (v > last_committed)
1487 (mon->is_peon() || mon->is_leader()) &&
1488 (is_active() || is_updating() || is_writing()) &&
1489 last_committed > 0 && is_lease_valid(); // must have a value alone, or have lease
1490 dout(5) << __func__ << " = " << (int)ret
1491 << " - now=" << ceph_clock_now()
1492 << " lease_expire=" << lease_expire
1493 << " has v" << v << " lc " << last_committed
1498 bool Paxos::read(version_t v, bufferlist &bl)
1500 if (!get_store()->get(get_name(), v, bl))
1505 version_t Paxos::read_current(bufferlist &bl)
1507 if (read(last_committed, bl))
1508 return last_committed;
1513 bool Paxos::is_lease_valid()
1515 return ((mon->get_quorum().size() == 1)
1516 || (ceph_clock_now() < lease_expire));
1521 bool Paxos::is_writeable()
1529 void Paxos::propose_pending()
1531 assert(is_active());
1532 assert(pending_proposal);
1537 pending_proposal->encode(bl);
1539 dout(10) << __func__ << " " << (last_committed + 1)
1540 << " " << bl.length() << " bytes" << dendl;
1541 dout(30) << __func__ << " transaction dump:\n";
1542 JSONFormatter f(true);
1543 pending_proposal->dump(&f);
1547 pending_proposal.reset();
1549 committing_finishers.swap(pending_finishers);
1550 state = STATE_UPDATING;
1554 void Paxos::queue_pending_finisher(Context *onfinished)
1556 dout(5) << __func__ << " " << onfinished << dendl;
1558 pending_finishers.push_back(onfinished);
1561 MonitorDBStore::TransactionRef Paxos::get_pending_transaction()
1563 assert(mon->is_leader());
1564 if (!pending_proposal) {
1565 pending_proposal.reset(new MonitorDBStore::Transaction);
1566 assert(pending_finishers.empty());
1568 return pending_proposal;
1571 bool Paxos::trigger_propose()
1574 dout(10) << __func__ << " plugged, not proposing now" << dendl;
1576 } else if (is_active()) {
1577 dout(10) << __func__ << " active, proposing now" << dendl;
1581 dout(10) << __func__ << " not active, will propose later" << dendl;
1586 bool Paxos::is_consistent()
1588 return (first_committed <= last_committed);