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
);