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
);
89 pcb
.add_u64_counter(l_paxos_start_leader
, "start_leader", "Starts in leader role");
90 pcb
.add_u64_counter(l_paxos_start_peon
, "start_peon", "Starts in peon role");
91 pcb
.add_u64_counter(l_paxos_restart
, "restart", "Restarts");
92 pcb
.add_u64_counter(l_paxos_refresh
, "refresh", "Refreshes");
93 pcb
.add_time_avg(l_paxos_refresh_latency
, "refresh_latency", "Refresh latency");
94 pcb
.add_u64_counter(l_paxos_begin
, "begin", "Started and handled begins");
95 pcb
.add_u64_avg(l_paxos_begin_keys
, "begin_keys", "Keys in transaction on begin");
96 pcb
.add_u64_avg(l_paxos_begin_bytes
, "begin_bytes", "Data in transaction on begin");
97 pcb
.add_time_avg(l_paxos_begin_latency
, "begin_latency", "Latency of begin operation");
98 pcb
.add_u64_counter(l_paxos_commit
, "commit",
100 pcb
.add_u64_avg(l_paxos_commit_keys
, "commit_keys", "Keys in transaction on commit");
101 pcb
.add_u64_avg(l_paxos_commit_bytes
, "commit_bytes", "Data in transaction on commit");
102 pcb
.add_time_avg(l_paxos_commit_latency
, "commit_latency",
103 "Commit latency", "clat");
104 pcb
.add_u64_counter(l_paxos_collect
, "collect", "Peon collects");
105 pcb
.add_u64_avg(l_paxos_collect_keys
, "collect_keys", "Keys in transaction on peon collect");
106 pcb
.add_u64_avg(l_paxos_collect_bytes
, "collect_bytes", "Data in transaction on peon collect");
107 pcb
.add_time_avg(l_paxos_collect_latency
, "collect_latency", "Peon collect latency");
108 pcb
.add_u64_counter(l_paxos_collect_uncommitted
, "collect_uncommitted", "Uncommitted values in started and handled collects");
109 pcb
.add_u64_counter(l_paxos_collect_timeout
, "collect_timeout", "Collect timeouts");
110 pcb
.add_u64_counter(l_paxos_accept_timeout
, "accept_timeout", "Accept timeouts");
111 pcb
.add_u64_counter(l_paxos_lease_ack_timeout
, "lease_ack_timeout", "Lease acknowledgement timeouts");
112 pcb
.add_u64_counter(l_paxos_lease_timeout
, "lease_timeout", "Lease timeouts");
113 pcb
.add_u64_counter(l_paxos_store_state
, "store_state", "Store a shared state on disk");
114 pcb
.add_u64_avg(l_paxos_store_state_keys
, "store_state_keys", "Keys in transaction in stored state");
115 pcb
.add_u64_avg(l_paxos_store_state_bytes
, "store_state_bytes", "Data in transaction in stored state");
116 pcb
.add_time_avg(l_paxos_store_state_latency
, "store_state_latency", "Storing state latency");
117 pcb
.add_u64_counter(l_paxos_share_state
, "share_state", "Sharings of state");
118 pcb
.add_u64_avg(l_paxos_share_state_keys
, "share_state_keys", "Keys in shared state");
119 pcb
.add_u64_avg(l_paxos_share_state_bytes
, "share_state_bytes", "Data in shared state");
120 pcb
.add_u64_counter(l_paxos_new_pn
, "new_pn", "New proposal number queries");
121 pcb
.add_time_avg(l_paxos_new_pn_latency
, "new_pn_latency", "New proposal number getting latency");
122 logger
= pcb
.create_perf_counters();
123 g_ceph_context
->get_perfcounters_collection()->add(logger
);
126 void Paxos::dump_info(Formatter
*f
)
128 f
->open_object_section("paxos");
129 f
->dump_unsigned("first_committed", first_committed
);
130 f
->dump_unsigned("last_committed", last_committed
);
131 f
->dump_unsigned("last_pn", last_pn
);
132 f
->dump_unsigned("accepted_pn", accepted_pn
);
136 // ---------------------------------
141 void Paxos::collect(version_t oldpn
)
143 // we're recoverying, it seems!
144 state
= STATE_RECOVERING
;
145 assert(mon
->is_leader());
147 // reset the number of lasts received
150 uncommitted_value
.clear();
151 peer_first_committed
.clear();
152 peer_last_committed
.clear();
154 // look for uncommitted value
155 if (get_store()->exists(get_name(), last_committed
+1)) {
156 version_t v
= get_store()->get(get_name(), "pending_v");
157 version_t pn
= get_store()->get(get_name(), "pending_pn");
158 if (v
&& pn
&& v
== last_committed
+ 1) {
161 dout(10) << "WARNING: no pending_pn on disk, using previous accepted_pn " << accepted_pn
162 << " and crossing our fingers" << dendl
;
163 uncommitted_pn
= accepted_pn
;
165 uncommitted_v
= last_committed
+1;
167 get_store()->get(get_name(), last_committed
+1, uncommitted_value
);
168 assert(uncommitted_value
.length());
169 dout(10) << "learned uncommitted " << (last_committed
+1)
170 << " pn " << uncommitted_pn
171 << " (" << uncommitted_value
.length() << " bytes) from myself"
174 logger
->inc(l_paxos_collect_uncommitted
);
178 accepted_pn
= get_new_proposal_number(MAX(accepted_pn
, oldpn
));
179 accepted_pn_from
= last_committed
;
181 dout(10) << "collect with pn " << accepted_pn
<< dendl
;
184 for (set
<int>::const_iterator p
= mon
->get_quorum().begin();
185 p
!= mon
->get_quorum().end();
187 if (*p
== mon
->rank
) continue;
189 MMonPaxos
*collect
= new MMonPaxos(mon
->get_epoch(), MMonPaxos::OP_COLLECT
,
191 collect
->last_committed
= last_committed
;
192 collect
->first_committed
= first_committed
;
193 collect
->pn
= accepted_pn
;
194 mon
->messenger
->send_message(collect
, mon
->monmap
->get_inst(*p
));
198 collect_timeout_event
= new C_MonContext(mon
, [this](int r
) {
203 mon
->timer
.add_event_after(g_conf
->mon_accept_timeout_factor
*
205 collect_timeout_event
);
210 void Paxos::handle_collect(MonOpRequestRef op
)
213 op
->mark_paxos_event("handle_collect");
215 MMonPaxos
*collect
= static_cast<MMonPaxos
*>(op
->get_req());
216 dout(10) << "handle_collect " << *collect
<< dendl
;
218 assert(mon
->is_peon()); // mon epoch filter should catch strays
220 // we're recoverying, it seems!
221 state
= STATE_RECOVERING
;
223 //update the peon recovery timeout
224 reset_lease_timeout();
226 if (collect
->first_committed
> last_committed
+1) {
228 << " leader's lowest version is too high for our last committed"
229 << " (theirs: " << collect
->first_committed
230 << "; ours: " << last_committed
<< ") -- bootstrap!" << dendl
;
231 op
->mark_paxos_event("need to bootstrap");
237 MMonPaxos
*last
= new MMonPaxos(mon
->get_epoch(), MMonPaxos::OP_LAST
,
239 last
->last_committed
= last_committed
;
240 last
->first_committed
= first_committed
;
242 version_t previous_pn
= accepted_pn
;
244 // can we accept this pn?
245 if (collect
->pn
> accepted_pn
) {
247 accepted_pn
= collect
->pn
;
248 accepted_pn_from
= collect
->pn_from
;
249 dout(10) << "accepting pn " << accepted_pn
<< " from "
250 << accepted_pn_from
<< dendl
;
252 auto t(std::make_shared
<MonitorDBStore::Transaction
>());
253 t
->put(get_name(), "accepted_pn", accepted_pn
);
255 dout(30) << __func__
<< " transaction dump:\n";
256 JSONFormatter
f(true);
261 logger
->inc(l_paxos_collect
);
262 logger
->inc(l_paxos_collect_keys
, t
->get_keys());
263 logger
->inc(l_paxos_collect_bytes
, t
->get_bytes());
264 utime_t start
= ceph_clock_now();
266 get_store()->apply_transaction(t
);
268 utime_t end
= ceph_clock_now();
269 logger
->tinc(l_paxos_collect_latency
, end
- start
);
272 dout(10) << "NOT accepting pn " << collect
->pn
<< " from " << collect
->pn_from
273 << ", we already accepted " << accepted_pn
274 << " from " << accepted_pn_from
<< dendl
;
276 last
->pn
= accepted_pn
;
277 last
->pn_from
= accepted_pn_from
;
279 // share whatever committed values we have
280 if (collect
->last_committed
< last_committed
)
281 share_state(last
, collect
->first_committed
, collect
->last_committed
);
283 // do we have an accepted but uncommitted value?
284 // (it'll be at last_committed+1)
286 if (collect
->last_committed
<= last_committed
&&
287 get_store()->exists(get_name(), last_committed
+1)) {
288 get_store()->get(get_name(), last_committed
+1, bl
);
289 assert(bl
.length() > 0);
290 dout(10) << " sharing our accepted but uncommitted value for "
291 << last_committed
+1 << " (" << bl
.length() << " bytes)" << dendl
;
292 last
->values
[last_committed
+1] = bl
;
294 version_t v
= get_store()->get(get_name(), "pending_v");
295 version_t pn
= get_store()->get(get_name(), "pending_pn");
296 if (v
&& pn
&& v
== last_committed
+ 1) {
297 last
->uncommitted_pn
= pn
;
299 // previously we didn't record which pn a value was accepted
300 // under! use the pn value we just had... :(
301 dout(10) << "WARNING: no pending_pn on disk, using previous accepted_pn " << previous_pn
302 << " and crossing our fingers" << dendl
;
303 last
->uncommitted_pn
= previous_pn
;
306 logger
->inc(l_paxos_collect_uncommitted
);
310 collect
->get_connection()->send_message(last
);
314 * @note This is Okay. We share our versions between peer_last_committed and
315 * our last_committed (inclusive), and add their bufferlists to the
316 * message. It will be the peer's job to apply them to its store, as
317 * these bufferlists will contain raw transactions.
318 * This function is called by both the Peon and the Leader. The Peon will
319 * share the state with the Leader during handle_collect(), sharing any
320 * values the leader may be missing (i.e., the leader's last_committed is
321 * lower than the peon's last_committed). The Leader will share the state
322 * with the Peon during handle_last(), if the peon's last_committed is
323 * lower than the leader's last_committed.
325 void Paxos::share_state(MMonPaxos
*m
, version_t peer_first_committed
,
326 version_t peer_last_committed
)
328 assert(peer_last_committed
< last_committed
);
330 dout(10) << "share_state peer has fc " << peer_first_committed
331 << " lc " << peer_last_committed
<< dendl
;
332 version_t v
= peer_last_committed
+ 1;
334 // include incrementals
336 for ( ; v
<= last_committed
; v
++) {
337 if (get_store()->exists(get_name(), v
)) {
338 get_store()->get(get_name(), v
, m
->values
[v
]);
339 assert(m
->values
[v
].length());
340 dout(10) << " sharing " << v
<< " ("
341 << m
->values
[v
].length() << " bytes)" << dendl
;
342 bytes
+= m
->values
[v
].length() + 16; // paxos_ + 10 digits = 16
345 logger
->inc(l_paxos_share_state
);
346 logger
->inc(l_paxos_share_state_keys
, m
->values
.size());
347 logger
->inc(l_paxos_share_state_bytes
, bytes
);
349 m
->last_committed
= last_committed
;
353 * Store on disk a state that was shared with us
355 * Basically, we received a set of version. Or just one. It doesn't matter.
356 * What matters is that we have to stash it in the store. So, we will simply
357 * write every single bufferlist into their own versions on our side (i.e.,
358 * onto paxos-related keys), and then we will decode those same bufferlists
359 * we just wrote and apply the transactions they hold. We will also update
360 * our first and last committed values to point to the new values, if need
361 * be. All all this is done tightly wrapped in a transaction to ensure we
362 * enjoy the atomicity guarantees given by our awesome k/v store.
364 bool Paxos::store_state(MMonPaxos
*m
)
366 auto t(std::make_shared
<MonitorDBStore::Transaction
>());
367 map
<version_t
,bufferlist
>::iterator start
= m
->values
.begin();
368 bool changed
= false;
370 // build map of values to store
371 // we want to write the range [last_committed, m->last_committed] only.
372 if (start
!= m
->values
.end() &&
373 start
->first
> last_committed
+ 1) {
374 // ignore everything if values start in the future.
375 dout(10) << "store_state ignoring all values, they start at " << start
->first
376 << " > last_committed+1" << dendl
;
380 // push forward the start position on the message's values iterator, up until
381 // we run out of positions or we find a position matching 'last_committed'.
382 while (start
!= m
->values
.end() && start
->first
<= last_committed
) {
386 // make sure we get the right interval of values to apply by pushing forward
387 // the 'end' iterator until it matches the message's 'last_committed'.
388 map
<version_t
,bufferlist
>::iterator end
= start
;
389 while (end
!= m
->values
.end() && end
->first
<= m
->last_committed
) {
390 last_committed
= end
->first
;
395 dout(10) << "store_state nothing to commit" << dendl
;
397 dout(10) << "store_state [" << start
->first
<< ".."
398 << last_committed
<< "]" << dendl
;
399 t
->put(get_name(), "last_committed", last_committed
);
401 // we should apply the state here -- decode every single bufferlist in the
402 // map and append the transactions to 't'.
403 map
<version_t
,bufferlist
>::iterator it
;
404 for (it
= start
; it
!= end
; ++it
) {
405 // write the bufferlist as the version's value
406 t
->put(get_name(), it
->first
, it
->second
);
407 // decode the bufferlist and append it to the transaction we will shortly
409 decode_append_transaction(t
, it
->second
);
412 // discard obsolete uncommitted value?
413 if (uncommitted_v
&& uncommitted_v
<= last_committed
) {
414 dout(10) << " forgetting obsolete uncommitted value " << uncommitted_v
415 << " pn " << uncommitted_pn
<< dendl
;
418 uncommitted_value
.clear();
422 dout(30) << __func__
<< " transaction dump:\n";
423 JSONFormatter
f(true);
428 logger
->inc(l_paxos_store_state
);
429 logger
->inc(l_paxos_store_state_bytes
, t
->get_bytes());
430 logger
->inc(l_paxos_store_state_keys
, t
->get_keys());
431 utime_t start
= ceph_clock_now();
433 get_store()->apply_transaction(t
);
435 utime_t end
= ceph_clock_now();
436 logger
->tinc(l_paxos_store_state_latency
, end
- start
);
438 // refresh first_committed; this txn may have trimmed.
439 first_committed
= get_store()->get(get_name(), "first_committed");
441 _sanity_check_store();
448 void Paxos::_sanity_check_store()
450 version_t lc
= get_store()->get(get_name(), "last_committed");
451 assert(lc
== last_committed
);
456 void Paxos::handle_last(MonOpRequestRef op
)
458 op
->mark_paxos_event("handle_last");
459 MMonPaxos
*last
= static_cast<MMonPaxos
*>(op
->get_req());
460 bool need_refresh
= false;
461 int from
= last
->get_source().num();
463 dout(10) << "handle_last " << *last
<< dendl
;
465 if (!mon
->is_leader()) {
466 dout(10) << "not leader, dropping" << dendl
;
470 // note peer's first_ and last_committed, in case we learn a new
471 // commit and need to push it to them.
472 peer_first_committed
[from
] = last
->first_committed
;
473 peer_last_committed
[from
] = last
->last_committed
;
475 if (last
->first_committed
> last_committed
+ 1) {
478 << " lowest version is too high for our last committed"
479 << " (theirs: " << last
->first_committed
480 << "; ours: " << last_committed
<< ") -- bootstrap!" << dendl
;
481 op
->mark_paxos_event("need to bootstrap");
486 assert(g_conf
->paxos_kill_at
!= 1);
488 // store any committed values if any are specified in the message
489 need_refresh
= store_state(last
);
491 assert(g_conf
->paxos_kill_at
!= 2);
493 // is everyone contiguous and up to date?
494 for (map
<int,version_t
>::iterator p
= peer_last_committed
.begin();
495 p
!= peer_last_committed
.end();
497 if (p
->second
+ 1 < first_committed
&& first_committed
> 1) {
499 << " peon " << p
->first
500 << " last_committed (" << p
->second
501 << ") is too low for our first_committed (" << first_committed
502 << ") -- bootstrap!" << dendl
;
503 op
->mark_paxos_event("need to bootstrap");
507 if (p
->second
< last_committed
) {
508 // share committed values
509 dout(10) << " sending commit to mon." << p
->first
<< dendl
;
510 MMonPaxos
*commit
= new MMonPaxos(mon
->get_epoch(),
511 MMonPaxos::OP_COMMIT
,
513 share_state(commit
, peer_first_committed
[p
->first
], p
->second
);
514 mon
->messenger
->send_message(commit
, mon
->monmap
->get_inst(p
->first
));
518 // do they accept your pn?
519 if (last
->pn
> accepted_pn
) {
521 dout(10) << " they had a higher pn than us, picking a new one." << dendl
;
523 // cancel timeout event
524 mon
->timer
.cancel_event(collect_timeout_event
);
525 collect_timeout_event
= 0;
528 } else if (last
->pn
== accepted_pn
) {
529 // yes, they accepted our pn. great.
531 dout(10) << " they accepted our pn, we now have "
532 << num_last
<< " peons" << dendl
;
534 // did this person send back an accepted but uncommitted value?
535 if (last
->uncommitted_pn
) {
536 if (last
->uncommitted_pn
>= uncommitted_pn
&&
537 last
->last_committed
>= last_committed
&&
538 last
->last_committed
+ 1 >= uncommitted_v
) {
539 uncommitted_v
= last
->last_committed
+1;
540 uncommitted_pn
= last
->uncommitted_pn
;
541 uncommitted_value
= last
->values
[uncommitted_v
];
542 dout(10) << "we learned an uncommitted value for " << uncommitted_v
543 << " pn " << uncommitted_pn
544 << " " << uncommitted_value
.length() << " bytes"
547 dout(10) << "ignoring uncommitted value for " << (last
->last_committed
+1)
548 << " pn " << last
->uncommitted_pn
549 << " " << last
->values
[last
->last_committed
+1].length() << " bytes"
555 if (num_last
== mon
->get_quorum().size()) {
556 // cancel timeout event
557 mon
->timer
.cancel_event(collect_timeout_event
);
558 collect_timeout_event
= 0;
559 peer_first_committed
.clear();
560 peer_last_committed
.clear();
564 // did we learn an old value?
565 if (uncommitted_v
== last_committed
+1 &&
566 uncommitted_value
.length()) {
567 dout(10) << "that's everyone. begin on old learned value" << dendl
;
568 state
= STATE_UPDATING_PREVIOUS
;
569 begin(uncommitted_value
);
572 dout(10) << "that's everyone. active!" << dendl
;
575 need_refresh
= false;
582 // no, this is an old message, discard
583 dout(10) << "old pn, ignoring" << dendl
;
590 void Paxos::collect_timeout()
592 dout(1) << "collect timeout, calling fresh election" << dendl
;
593 collect_timeout_event
= 0;
594 logger
->inc(l_paxos_collect_timeout
);
595 assert(mon
->is_leader());
601 void Paxos::begin(bufferlist
& v
)
603 dout(10) << "begin for " << last_committed
+1 << " "
604 << v
.length() << " bytes"
607 assert(mon
->is_leader());
608 assert(is_updating() || is_updating_previous());
610 // we must already have a majority for this to work.
611 assert(mon
->get_quorum().size() == 1 ||
612 num_last
> (unsigned)mon
->monmap
->size()/2);
614 // and no value, yet.
615 assert(new_value
.length() == 0);
617 // accept it ourselves
619 accepted
.insert(mon
->rank
);
622 if (last_committed
== 0) {
623 auto t(std::make_shared
<MonitorDBStore::Transaction
>());
624 // initial base case; set first_committed too
625 t
->put(get_name(), "first_committed", 1);
626 decode_append_transaction(t
, new_value
);
634 // store the proposed value in the store. IF it is accepted, we will then
635 // have to decode it into a transaction and apply it.
636 auto t(std::make_shared
<MonitorDBStore::Transaction
>());
637 t
->put(get_name(), last_committed
+1, new_value
);
639 // note which pn this pending value is for.
640 t
->put(get_name(), "pending_v", last_committed
+ 1);
641 t
->put(get_name(), "pending_pn", accepted_pn
);
643 dout(30) << __func__
<< " transaction dump:\n";
644 JSONFormatter
f(true);
647 auto debug_tx(std::make_shared
<MonitorDBStore::Transaction
>());
648 bufferlist::iterator new_value_it
= new_value
.begin();
649 debug_tx
->decode(new_value_it
);
651 *_dout
<< "\nbl dump:\n";
655 logger
->inc(l_paxos_begin
);
656 logger
->inc(l_paxos_begin_keys
, t
->get_keys());
657 logger
->inc(l_paxos_begin_bytes
, t
->get_bytes());
658 utime_t start
= ceph_clock_now();
660 get_store()->apply_transaction(t
);
662 utime_t end
= ceph_clock_now();
663 logger
->tinc(l_paxos_begin_latency
, end
- start
);
665 assert(g_conf
->paxos_kill_at
!= 3);
667 if (mon
->get_quorum().size() == 1) {
668 // we're alone, take it easy
673 // ask others to accept it too!
674 for (set
<int>::const_iterator p
= mon
->get_quorum().begin();
675 p
!= mon
->get_quorum().end();
677 if (*p
== mon
->rank
) continue;
679 dout(10) << " sending begin to mon." << *p
<< dendl
;
680 MMonPaxos
*begin
= new MMonPaxos(mon
->get_epoch(), MMonPaxos::OP_BEGIN
,
682 begin
->values
[last_committed
+1] = new_value
;
683 begin
->last_committed
= last_committed
;
684 begin
->pn
= accepted_pn
;
686 mon
->messenger
->send_message(begin
, mon
->monmap
->get_inst(*p
));
690 accept_timeout_event
= new C_MonContext(mon
, [this](int r
) {
695 mon
->timer
.add_event_after(g_conf
->mon_accept_timeout_factor
*
697 accept_timeout_event
);
701 void Paxos::handle_begin(MonOpRequestRef op
)
703 op
->mark_paxos_event("handle_begin");
704 MMonPaxos
*begin
= static_cast<MMonPaxos
*>(op
->get_req());
705 dout(10) << "handle_begin " << *begin
<< dendl
;
707 // can we accept this?
708 if (begin
->pn
< accepted_pn
) {
709 dout(10) << " we accepted a higher pn " << accepted_pn
<< ", ignoring" << dendl
;
710 op
->mark_paxos_event("have higher pn, ignore");
713 assert(begin
->pn
== accepted_pn
);
714 assert(begin
->last_committed
== last_committed
);
716 assert(g_conf
->paxos_kill_at
!= 4);
718 logger
->inc(l_paxos_begin
);
721 state
= STATE_UPDATING
;
722 lease_expire
= utime_t(); // cancel lease
725 version_t v
= last_committed
+1;
726 dout(10) << "accepting value for " << v
<< " pn " << accepted_pn
<< dendl
;
727 // store the accepted value onto our store. We will have to decode it and
728 // apply its transaction once we receive permission to commit.
729 auto t(std::make_shared
<MonitorDBStore::Transaction
>());
730 t
->put(get_name(), v
, begin
->values
[v
]);
732 // note which pn this pending value is for.
733 t
->put(get_name(), "pending_v", v
);
734 t
->put(get_name(), "pending_pn", accepted_pn
);
736 dout(30) << __func__
<< " transaction dump:\n";
737 JSONFormatter
f(true);
742 logger
->inc(l_paxos_begin_bytes
, t
->get_bytes());
743 utime_t start
= ceph_clock_now();
745 get_store()->apply_transaction(t
);
747 utime_t end
= ceph_clock_now();
748 logger
->tinc(l_paxos_begin_latency
, end
- start
);
750 assert(g_conf
->paxos_kill_at
!= 5);
753 MMonPaxos
*accept
= new MMonPaxos(mon
->get_epoch(), MMonPaxos::OP_ACCEPT
,
755 accept
->pn
= accepted_pn
;
756 accept
->last_committed
= last_committed
;
757 begin
->get_connection()->send_message(accept
);
761 void Paxos::handle_accept(MonOpRequestRef op
)
763 op
->mark_paxos_event("handle_accept");
764 MMonPaxos
*accept
= static_cast<MMonPaxos
*>(op
->get_req());
765 dout(10) << "handle_accept " << *accept
<< dendl
;
766 int from
= accept
->get_source().num();
768 if (accept
->pn
!= accepted_pn
) {
769 // we accepted a higher pn, from some other leader
770 dout(10) << " we accepted a higher pn " << accepted_pn
<< ", ignoring" << dendl
;
771 op
->mark_paxos_event("have higher pn, ignore");
774 if (last_committed
> 0 &&
775 accept
->last_committed
< last_committed
-1) {
776 dout(10) << " this is from an old round, ignoring" << dendl
;
777 op
->mark_paxos_event("old round, ignore");
780 assert(accept
->last_committed
== last_committed
|| // not committed
781 accept
->last_committed
== last_committed
-1); // committed
783 assert(is_updating() || is_updating_previous());
784 assert(accepted
.count(from
) == 0);
785 accepted
.insert(from
);
786 dout(10) << " now " << accepted
<< " have accepted" << dendl
;
788 assert(g_conf
->paxos_kill_at
!= 6);
790 // only commit (and expose committed state) when we get *all* quorum
791 // members to accept. otherwise, they may still be sharing the now
793 // FIXME: we can improve this with an additional lease revocation message
794 // that doesn't block for the persist.
795 if (accepted
== mon
->get_quorum()) {
797 dout(10) << " got majority, committing, done with update" << dendl
;
798 op
->mark_paxos_event("commit_start");
803 void Paxos::accept_timeout()
805 dout(1) << "accept timeout, calling fresh election" << dendl
;
806 accept_timeout_event
= 0;
807 assert(mon
->is_leader());
808 assert(is_updating() || is_updating_previous() || is_writing() ||
809 is_writing_previous());
810 logger
->inc(l_paxos_accept_timeout
);
814 struct C_Committed
: public Context
{
816 explicit C_Committed(Paxos
*p
) : paxos(p
) {}
817 void finish(int r
) override
{
819 Mutex::Locker
l(paxos
->mon
->lock
);
820 paxos
->commit_finish();
824 void Paxos::commit_start()
826 dout(10) << __func__
<< " " << (last_committed
+1) << dendl
;
828 assert(g_conf
->paxos_kill_at
!= 7);
830 auto t(std::make_shared
<MonitorDBStore::Transaction
>());
833 t
->put(get_name(), "last_committed", last_committed
+ 1);
835 // decode the value and apply its transaction to the store.
836 // this value can now be read from last_committed.
837 decode_append_transaction(t
, new_value
);
839 dout(30) << __func__
<< " transaction dump:\n";
840 JSONFormatter
f(true);
845 logger
->inc(l_paxos_commit
);
846 logger
->inc(l_paxos_commit_keys
, t
->get_keys());
847 logger
->inc(l_paxos_commit_bytes
, t
->get_bytes());
848 commit_start_stamp
= ceph_clock_now();
850 get_store()->queue_transaction(t
, new C_Committed(this));
852 if (is_updating_previous())
853 state
= STATE_WRITING_PREVIOUS
;
854 else if (is_updating())
855 state
= STATE_WRITING
;
859 if (mon
->get_quorum().size() > 1) {
860 // cancel timeout event
861 mon
->timer
.cancel_event(accept_timeout_event
);
862 accept_timeout_event
= 0;
866 void Paxos::commit_finish()
868 dout(20) << __func__
<< " " << (last_committed
+1) << dendl
;
869 utime_t end
= ceph_clock_now();
870 logger
->tinc(l_paxos_commit_latency
, end
- commit_start_stamp
);
872 assert(g_conf
->paxos_kill_at
!= 8);
874 // cancel lease - it was for the old value.
875 // (this would only happen if message layer lost the 'begin', but
876 // leader still got a majority and committed with out us.)
877 lease_expire
= utime_t(); // cancel lease
880 last_commit_time
= ceph_clock_now();
882 // refresh first_committed; this txn may have trimmed.
883 first_committed
= get_store()->get(get_name(), "first_committed");
885 _sanity_check_store();
888 for (set
<int>::const_iterator p
= mon
->get_quorum().begin();
889 p
!= mon
->get_quorum().end();
891 if (*p
== mon
->rank
) continue;
893 dout(10) << " sending commit to mon." << *p
<< dendl
;
894 MMonPaxos
*commit
= new MMonPaxos(mon
->get_epoch(), MMonPaxos::OP_COMMIT
,
896 commit
->values
[last_committed
] = new_value
;
897 commit
->pn
= accepted_pn
;
898 commit
->last_committed
= last_committed
;
900 mon
->messenger
->send_message(commit
, mon
->monmap
->get_inst(*p
));
903 assert(g_conf
->paxos_kill_at
!= 9);
905 // get ready for a new round.
908 // WRITING -> REFRESH
909 // among other things, this lets do_refresh() -> mon->bootstrap() know
910 // it doesn't need to flush the store queue
911 assert(is_writing() || is_writing_previous());
912 state
= STATE_REFRESH
;
916 if (mon
->get_quorum().size() > 1) {
920 finish_contexts(g_ceph_context
, waiting_for_commit
);
922 assert(g_conf
->paxos_kill_at
!= 10);
929 void Paxos::handle_commit(MonOpRequestRef op
)
931 op
->mark_paxos_event("handle_commit");
932 MMonPaxos
*commit
= static_cast<MMonPaxos
*>(op
->get_req());
933 dout(10) << "handle_commit on " << commit
->last_committed
<< dendl
;
935 logger
->inc(l_paxos_commit
);
937 if (!mon
->is_peon()) {
938 dout(10) << "not a peon, dropping" << dendl
;
943 op
->mark_paxos_event("store_state");
947 finish_contexts(g_ceph_context
, waiting_for_commit
);
951 void Paxos::extend_lease()
953 assert(mon
->is_leader());
954 //assert(is_active());
956 lease_expire
= ceph_clock_now();
957 lease_expire
+= g_conf
->mon_lease
;
959 acked_lease
.insert(mon
->rank
);
961 dout(7) << "extend_lease now+" << g_conf
->mon_lease
962 << " (" << lease_expire
<< ")" << dendl
;
965 for (set
<int>::const_iterator p
= mon
->get_quorum().begin();
966 p
!= mon
->get_quorum().end(); ++p
) {
968 if (*p
== mon
->rank
) continue;
969 MMonPaxos
*lease
= new MMonPaxos(mon
->get_epoch(), MMonPaxos::OP_LEASE
,
971 lease
->last_committed
= last_committed
;
972 lease
->lease_timestamp
= lease_expire
;
973 lease
->first_committed
= first_committed
;
974 mon
->messenger
->send_message(lease
, mon
->monmap
->get_inst(*p
));
977 // set timeout event.
978 // if old timeout is still in place, leave it.
979 if (!lease_ack_timeout_event
) {
980 lease_ack_timeout_event
= new C_MonContext(mon
, [this](int r
) {
985 mon
->timer
.add_event_after(g_conf
->mon_lease_ack_timeout_factor
*
987 lease_ack_timeout_event
);
991 lease_renew_event
= new C_MonContext(mon
, [this](int r
) {
994 lease_renew_timeout();
996 utime_t at
= lease_expire
;
997 at
-= g_conf
->mon_lease
;
998 at
+= g_conf
->mon_lease_renew_interval_factor
* g_conf
->mon_lease
;
999 mon
->timer
.add_event_at(at
, lease_renew_event
);
1002 void Paxos::warn_on_future_time(utime_t t
, entity_name_t from
)
1004 utime_t now
= ceph_clock_now();
1006 utime_t diff
= t
- now
;
1007 if (diff
> g_conf
->mon_clock_drift_allowed
) {
1008 utime_t warn_diff
= now
- last_clock_drift_warn
;
1010 pow(g_conf
->mon_clock_drift_warn_backoff
, clock_drift_warned
)) {
1011 mon
->clog
->warn() << "message from " << from
<< " was stamped " << diff
1012 << "s in the future, clocks not synchronized";
1013 last_clock_drift_warn
= ceph_clock_now();
1014 ++clock_drift_warned
;
1021 bool Paxos::do_refresh()
1023 bool need_bootstrap
= false;
1025 utime_t start
= ceph_clock_now();
1027 // make sure we have the latest state loaded up
1028 mon
->refresh_from_paxos(&need_bootstrap
);
1030 utime_t end
= ceph_clock_now();
1031 logger
->inc(l_paxos_refresh
);
1032 logger
->tinc(l_paxos_refresh_latency
, end
- start
);
1034 if (need_bootstrap
) {
1035 dout(10) << " doing requested bootstrap" << dendl
;
1043 void Paxos::commit_proposal()
1045 dout(10) << __func__
<< dendl
;
1046 assert(mon
->is_leader());
1047 assert(is_refresh());
1049 finish_contexts(g_ceph_context
, committing_finishers
);
1052 void Paxos::finish_round()
1054 dout(10) << __func__
<< dendl
;
1055 assert(mon
->is_leader());
1057 // ok, now go active!
1058 state
= STATE_ACTIVE
;
1060 dout(20) << __func__
<< " waiting_for_acting" << dendl
;
1061 finish_contexts(g_ceph_context
, waiting_for_active
);
1062 dout(20) << __func__
<< " waiting_for_readable" << dendl
;
1063 finish_contexts(g_ceph_context
, waiting_for_readable
);
1064 dout(20) << __func__
<< " waiting_for_writeable" << dendl
;
1065 finish_contexts(g_ceph_context
, waiting_for_writeable
);
1067 dout(10) << __func__
<< " done w/ waiters, state " << get_statename(state
) << dendl
;
1069 if (should_trim()) {
1073 if (is_active() && pending_proposal
) {
1080 void Paxos::handle_lease(MonOpRequestRef op
)
1082 op
->mark_paxos_event("handle_lease");
1083 MMonPaxos
*lease
= static_cast<MMonPaxos
*>(op
->get_req());
1085 if (!mon
->is_peon() ||
1086 last_committed
!= lease
->last_committed
) {
1087 dout(10) << "handle_lease i'm not a peon, or they're not the leader,"
1088 << " or the last_committed doesn't match, dropping" << dendl
;
1089 op
->mark_paxos_event("invalid lease, ignore");
1093 warn_on_future_time(lease
->sent_timestamp
, lease
->get_source());
1096 if (lease_expire
< lease
->lease_timestamp
) {
1097 lease_expire
= lease
->lease_timestamp
;
1099 utime_t now
= ceph_clock_now();
1100 if (lease_expire
< now
) {
1101 utime_t diff
= now
- lease_expire
;
1102 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
;
1106 state
= STATE_ACTIVE
;
1108 dout(10) << "handle_lease on " << lease
->last_committed
1109 << " now " << lease_expire
<< dendl
;
1112 MMonPaxos
*ack
= new MMonPaxos(mon
->get_epoch(), MMonPaxos::OP_LEASE_ACK
,
1114 ack
->last_committed
= last_committed
;
1115 ack
->first_committed
= first_committed
;
1116 ack
->lease_timestamp
= ceph_clock_now();
1117 ::encode(mon
->session_map
.feature_map
, ack
->feature_map
);
1118 lease
->get_connection()->send_message(ack
);
1120 // (re)set timeout event.
1121 reset_lease_timeout();
1124 finish_contexts(g_ceph_context
, waiting_for_active
);
1126 finish_contexts(g_ceph_context
, waiting_for_readable
);
1129 void Paxos::handle_lease_ack(MonOpRequestRef op
)
1131 op
->mark_paxos_event("handle_lease_ack");
1132 MMonPaxos
*ack
= static_cast<MMonPaxos
*>(op
->get_req());
1133 int from
= ack
->get_source().num();
1135 if (!lease_ack_timeout_event
) {
1136 dout(10) << "handle_lease_ack from " << ack
->get_source()
1137 << " -- stray (probably since revoked)" << dendl
;
1139 else if (acked_lease
.count(from
) == 0) {
1140 acked_lease
.insert(from
);
1141 if (ack
->feature_map
.length()) {
1142 auto p
= ack
->feature_map
.begin();
1143 FeatureMap
& t
= mon
->quorum_feature_map
[from
];
1146 if (acked_lease
== mon
->get_quorum()) {
1148 dout(10) << "handle_lease_ack from " << ack
->get_source()
1149 << " -- got everyone" << dendl
;
1150 mon
->timer
.cancel_event(lease_ack_timeout_event
);
1151 lease_ack_timeout_event
= 0;
1155 dout(10) << "handle_lease_ack from " << ack
->get_source()
1156 << " -- still need "
1157 << mon
->get_quorum().size() - acked_lease
.size()
1158 << " more" << dendl
;
1161 dout(10) << "handle_lease_ack from " << ack
->get_source()
1162 << " dup (lagging!), ignoring" << dendl
;
1165 warn_on_future_time(ack
->sent_timestamp
, ack
->get_source());
1168 void Paxos::lease_ack_timeout()
1170 dout(1) << "lease_ack_timeout -- calling new election" << dendl
;
1171 assert(mon
->is_leader());
1172 assert(is_active());
1173 logger
->inc(l_paxos_lease_ack_timeout
);
1174 lease_ack_timeout_event
= 0;
1178 void Paxos::reset_lease_timeout()
1180 dout(20) << "reset_lease_timeout - setting timeout event" << dendl
;
1181 if (lease_timeout_event
)
1182 mon
->timer
.cancel_event(lease_timeout_event
);
1183 lease_timeout_event
= new C_MonContext(mon
, [this](int r
) {
1184 if (r
== -ECANCELED
)
1188 mon
->timer
.add_event_after(g_conf
->mon_lease_ack_timeout_factor
*
1190 lease_timeout_event
);
1193 void Paxos::lease_timeout()
1195 dout(1) << "lease_timeout -- calling new election" << dendl
;
1196 assert(mon
->is_peon());
1197 logger
->inc(l_paxos_lease_timeout
);
1198 lease_timeout_event
= 0;
1202 void Paxos::lease_renew_timeout()
1204 lease_renew_event
= 0;
1214 assert(should_trim());
1215 version_t end
= MIN(get_version() - g_conf
->paxos_min
,
1216 get_first_committed() + g_conf
->paxos_trim_max
);
1218 if (first_committed
>= end
)
1221 dout(10) << "trim to " << end
<< " (was " << first_committed
<< ")" << dendl
;
1223 MonitorDBStore::TransactionRef t
= get_pending_transaction();
1225 for (version_t v
= first_committed
; v
< end
; ++v
) {
1226 dout(10) << "trim " << v
<< dendl
;
1227 t
->erase(get_name(), v
);
1229 t
->put(get_name(), "first_committed", end
);
1230 if (g_conf
->mon_compact_on_trim
) {
1231 dout(10) << " compacting trimmed range" << dendl
;
1232 t
->compact_range(get_name(), stringify(first_committed
- 1), stringify(end
));
1236 queue_pending_finisher(new C_Trimmed(this));
1240 * return a globally unique, monotonically increasing proposal number
1242 version_t
Paxos::get_new_proposal_number(version_t gt
)
1247 // update. make it unique among all monitors.
1251 last_pn
+= (version_t
)mon
->rank
;
1254 auto t(std::make_shared
<MonitorDBStore::Transaction
>());
1255 t
->put(get_name(), "last_pn", last_pn
);
1257 dout(30) << __func__
<< " transaction dump:\n";
1258 JSONFormatter
f(true);
1263 logger
->inc(l_paxos_new_pn
);
1264 utime_t start
= ceph_clock_now();
1266 get_store()->apply_transaction(t
);
1268 utime_t end
= ceph_clock_now();
1269 logger
->tinc(l_paxos_new_pn_latency
, end
- start
);
1271 dout(10) << "get_new_proposal_number = " << last_pn
<< dendl
;
1276 void Paxos::cancel_events()
1278 if (collect_timeout_event
) {
1279 mon
->timer
.cancel_event(collect_timeout_event
);
1280 collect_timeout_event
= 0;
1282 if (accept_timeout_event
) {
1283 mon
->timer
.cancel_event(accept_timeout_event
);
1284 accept_timeout_event
= 0;
1286 if (lease_renew_event
) {
1287 mon
->timer
.cancel_event(lease_renew_event
);
1288 lease_renew_event
= 0;
1290 if (lease_ack_timeout_event
) {
1291 mon
->timer
.cancel_event(lease_ack_timeout_event
);
1292 lease_ack_timeout_event
= 0;
1294 if (lease_timeout_event
) {
1295 mon
->timer
.cancel_event(lease_timeout_event
);
1296 lease_timeout_event
= 0;
1300 void Paxos::shutdown()
1302 dout(10) << __func__
<< " cancel all contexts" << dendl
;
1304 // discard pending transaction
1305 pending_proposal
.reset();
1307 finish_contexts(g_ceph_context
, waiting_for_writeable
, -ECANCELED
);
1308 finish_contexts(g_ceph_context
, waiting_for_commit
, -ECANCELED
);
1309 finish_contexts(g_ceph_context
, waiting_for_readable
, -ECANCELED
);
1310 finish_contexts(g_ceph_context
, waiting_for_active
, -ECANCELED
);
1311 finish_contexts(g_ceph_context
, pending_finishers
, -ECANCELED
);
1312 finish_contexts(g_ceph_context
, committing_finishers
, -ECANCELED
);
1314 g_ceph_context
->get_perfcounters_collection()->remove(logger
);
1318 void Paxos::leader_init()
1323 // discard pending transaction
1324 pending_proposal
.reset();
1326 finish_contexts(g_ceph_context
, pending_finishers
, -EAGAIN
);
1327 finish_contexts(g_ceph_context
, committing_finishers
, -EAGAIN
);
1329 logger
->inc(l_paxos_start_leader
);
1331 if (mon
->get_quorum().size() == 1) {
1332 state
= STATE_ACTIVE
;
1336 state
= STATE_RECOVERING
;
1337 lease_expire
= utime_t();
1338 dout(10) << "leader_init -- starting paxos recovery" << dendl
;
1342 void Paxos::peon_init()
1347 state
= STATE_RECOVERING
;
1348 lease_expire
= utime_t();
1349 dout(10) << "peon_init -- i am a peon" << dendl
;
1351 // start a timer, in case the leader never manages to issue a lease
1352 reset_lease_timeout();
1354 // discard pending transaction
1355 pending_proposal
.reset();
1357 // no chance to write now!
1358 finish_contexts(g_ceph_context
, waiting_for_writeable
, -EAGAIN
);
1359 finish_contexts(g_ceph_context
, waiting_for_commit
, -EAGAIN
);
1360 finish_contexts(g_ceph_context
, pending_finishers
, -EAGAIN
);
1361 finish_contexts(g_ceph_context
, committing_finishers
, -EAGAIN
);
1363 logger
->inc(l_paxos_start_peon
);
1366 void Paxos::restart()
1368 dout(10) << "restart -- canceling timeouts" << dendl
;
1372 if (is_writing() || is_writing_previous()) {
1373 dout(10) << __func__
<< " flushing" << dendl
;
1375 mon
->store
->flush();
1377 dout(10) << __func__
<< " flushed" << dendl
;
1379 state
= STATE_RECOVERING
;
1381 // discard pending transaction
1382 pending_proposal
.reset();
1384 finish_contexts(g_ceph_context
, committing_finishers
, -EAGAIN
);
1385 finish_contexts(g_ceph_context
, pending_finishers
, -EAGAIN
);
1386 finish_contexts(g_ceph_context
, waiting_for_commit
, -EAGAIN
);
1387 finish_contexts(g_ceph_context
, waiting_for_active
, -EAGAIN
);
1389 logger
->inc(l_paxos_restart
);
1393 void Paxos::dispatch(MonOpRequestRef op
)
1395 assert(op
->is_type_paxos());
1396 op
->mark_paxos_event("dispatch");
1397 PaxosServiceMessage
*m
= static_cast<PaxosServiceMessage
*>(op
->get_req());
1398 // election in progress?
1399 if (!mon
->is_leader() && !mon
->is_peon()) {
1400 dout(5) << "election in progress, dropping " << *m
<< dendl
;
1405 assert(mon
->is_leader() ||
1406 (mon
->is_peon() && m
->get_source().num() == mon
->get_leader()));
1408 switch (m
->get_type()) {
1412 MMonPaxos
*pm
= reinterpret_cast<MMonPaxos
*>(m
);
1414 // NOTE: these ops are defined in messages/MMonPaxos.h
1417 case MMonPaxos::OP_COLLECT
:
1420 case MMonPaxos::OP_LAST
:
1423 case MMonPaxos::OP_BEGIN
:
1426 case MMonPaxos::OP_ACCEPT
:
1429 case MMonPaxos::OP_COMMIT
:
1432 case MMonPaxos::OP_LEASE
:
1435 case MMonPaxos::OP_LEASE_ACK
:
1436 handle_lease_ack(op
);
1450 // -----------------
1451 // service interface
1455 bool Paxos::is_readable(version_t v
)
1458 if (v
> last_committed
)
1462 (mon
->is_peon() || mon
->is_leader()) &&
1463 (is_active() || is_updating() || is_writing()) &&
1464 last_committed
> 0 && is_lease_valid(); // must have a value alone, or have lease
1465 dout(5) << __func__
<< " = " << (int)ret
1466 << " - now=" << ceph_clock_now()
1467 << " lease_expire=" << lease_expire
1468 << " has v" << v
<< " lc " << last_committed
1473 bool Paxos::read(version_t v
, bufferlist
&bl
)
1475 if (!get_store()->get(get_name(), v
, bl
))
1480 version_t
Paxos::read_current(bufferlist
&bl
)
1482 if (read(last_committed
, bl
))
1483 return last_committed
;
1488 bool Paxos::is_lease_valid()
1490 return ((mon
->get_quorum().size() == 1)
1491 || (ceph_clock_now() < lease_expire
));
1496 bool Paxos::is_writeable()
1504 void Paxos::propose_pending()
1506 assert(is_active());
1507 assert(pending_proposal
);
1512 pending_proposal
->encode(bl
);
1514 dout(10) << __func__
<< " " << (last_committed
+ 1)
1515 << " " << bl
.length() << " bytes" << dendl
;
1516 dout(30) << __func__
<< " transaction dump:\n";
1517 JSONFormatter
f(true);
1518 pending_proposal
->dump(&f
);
1522 pending_proposal
.reset();
1524 committing_finishers
.swap(pending_finishers
);
1525 state
= STATE_UPDATING
;
1529 void Paxos::queue_pending_finisher(Context
*onfinished
)
1531 dout(5) << __func__
<< " " << onfinished
<< dendl
;
1533 pending_finishers
.push_back(onfinished
);
1536 MonitorDBStore::TransactionRef
Paxos::get_pending_transaction()
1538 assert(mon
->is_leader());
1539 if (!pending_proposal
) {
1540 pending_proposal
.reset(new MonitorDBStore::Transaction
);
1541 assert(pending_finishers
.empty());
1543 return pending_proposal
;
1546 bool Paxos::trigger_propose()
1549 dout(10) << __func__
<< " plugged, not proposing now" << dendl
;
1551 } else if (is_active()) {
1552 dout(10) << __func__
<< " active, proposing now" << dendl
;
1556 dout(10) << __func__
<< " not active, will propose later" << dendl
;
1561 bool Paxos::is_consistent()
1563 return (first_committed
<= last_committed
);