+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-/*
- * Ceph - scalable distributed file system
- *
- * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
- *
- * This is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License version 2.1, as published by the Free Software
- * Foundation. See file COPYING.
- *
- */
-
-#include "PaxosService.h"
-#include "common/Clock.h"
-#include "common/config.h"
-#include "include/stringify.h"
-#include "include/assert.h"
-#include "mon/MonOpRequest.h"
-
-#define dout_subsys ceph_subsys_paxos
-#undef dout_prefix
-#define dout_prefix _prefix(_dout, mon, paxos, service_name, get_first_committed(), get_last_committed())
-static ostream& _prefix(std::ostream *_dout, Monitor *mon, Paxos *paxos, string service_name,
- version_t fc, version_t lc) {
- return *_dout << "mon." << mon->name << "@" << mon->rank
- << "(" << mon->get_state_name()
- << ").paxosservice(" << service_name << " " << fc << ".." << lc << ") ";
-}
-
-bool PaxosService::dispatch(MonOpRequestRef op)
-{
- assert(op->is_type_service() || op->is_type_command());
- PaxosServiceMessage *m = static_cast<PaxosServiceMessage*>(op->get_req());
- op->mark_event("psvc:dispatch");
-
- dout(10) << __func__ << " " << m << " " << *m
- << " from " << m->get_orig_source_inst()
- << " con " << m->get_connection() << dendl;
-
- if (mon->is_shutdown()) {
- return true;
- }
-
- // make sure this message isn't forwarded from a previous election epoch
- if (m->rx_election_epoch &&
- m->rx_election_epoch < mon->get_epoch()) {
- dout(10) << " discarding forwarded message from previous election epoch "
- << m->rx_election_epoch << " < " << mon->get_epoch() << dendl;
- return true;
- }
-
- // make sure the client is still connected. note that a proxied
- // connection will be disconnected with a null message; don't drop
- // those. also ignore loopback (e.g., log) messages.
- if (m->get_connection() &&
- !m->get_connection()->is_connected() &&
- m->get_connection() != mon->con_self &&
- m->get_connection()->get_messenger() != NULL) {
- dout(10) << " discarding message from disconnected client "
- << m->get_source_inst() << " " << *m << dendl;
- return true;
- }
-
- // make sure our map is readable and up to date
- if (!is_readable(m->version)) {
- dout(10) << " waiting for paxos -> readable (v" << m->version << ")" << dendl;
- wait_for_readable(op, new C_RetryMessage(this, op), m->version);
- return true;
- }
-
- // preprocess
- if (preprocess_query(op))
- return true; // easy!
-
- // leader?
- if (!mon->is_leader()) {
- mon->forward_request_leader(op);
- return true;
- }
-
- // writeable?
- if (!is_writeable()) {
- dout(10) << " waiting for paxos -> writeable" << dendl;
- wait_for_writeable(op, new C_RetryMessage(this, op));
- return true;
- }
-
- // update
- if (!prepare_update(op)) {
- // no changes made.
- return true;
- }
-
- if (need_immediate_propose) {
- dout(10) << __func__ << " forced immediate propose" << dendl;
- need_immediate_propose = false;
- propose_pending();
- return true;
- }
-
- double delay = 0.0;
- if (!should_propose(delay)) {
- dout(10) << " not proposing" << dendl;
- return true;
- }
-
- if (delay == 0.0) {
- propose_pending();
- return true;
- }
-
- // delay a bit
- if (!proposal_timer) {
- /**
- * Callback class used to propose the pending value once the proposal_timer
- * fires up.
- */
- auto do_propose = new C_MonContext(mon, [this](int r) {
- proposal_timer = 0;
- if (r >= 0) {
- propose_pending();
- } else if (r == -ECANCELED || r == -EAGAIN) {
- return;
- } else {
- assert(0 == "bad return value for proposal_timer");
- }
- });
- dout(10) << " setting proposal_timer " << do_propose
- << " with delay of " << delay << dendl;
- proposal_timer = mon->timer.add_event_after(delay, do_propose);
- } else {
- dout(10) << " proposal_timer already set" << dendl;
- }
- return true;
-}
-
-void PaxosService::refresh(bool *need_bootstrap)
-{
- // update cached versions
- cached_first_committed = mon->store->get(get_service_name(), first_committed_name);
- cached_last_committed = mon->store->get(get_service_name(), last_committed_name);
-
- version_t new_format = get_value("format_version");
- if (new_format != format_version) {
- dout(1) << __func__ << " upgraded, format " << format_version << " -> " << new_format << dendl;
- on_upgrade();
- }
- format_version = new_format;
-
- dout(10) << __func__ << dendl;
-
- update_from_paxos(need_bootstrap);
-}
-
-void PaxosService::post_refresh()
-{
- dout(10) << __func__ << dendl;
-
- post_paxos_update();
-
- if (mon->is_peon() && !waiting_for_finished_proposal.empty()) {
- finish_contexts(g_ceph_context, waiting_for_finished_proposal, -EAGAIN);
- }
-}
-
-bool PaxosService::should_propose(double& delay)
-{
- // simple default policy: quick startup, then some damping.
- if (get_last_committed() <= 1) {
- delay = 0.0;
- } else {
- utime_t now = ceph_clock_now();
- if ((now - paxos->last_commit_time) > g_conf->paxos_propose_interval)
- delay = (double)g_conf->paxos_min_wait;
- else
- delay = (double)(g_conf->paxos_propose_interval + paxos->last_commit_time
- - now);
- }
- return true;
-}
-
-
-void PaxosService::propose_pending()
-{
- dout(10) << __func__ << dendl;
- assert(have_pending);
- assert(!proposing);
- assert(mon->is_leader());
- assert(is_active());
-
- if (proposal_timer) {
- dout(10) << " canceling proposal_timer " << proposal_timer << dendl;
- mon->timer.cancel_event(proposal_timer);
- proposal_timer = NULL;
- }
-
- /**
- * @note What we contribute to the pending Paxos transaction is
- * obtained by calling a function that must be implemented by
- * the class implementing us. I.e., the function
- * encode_pending will be the one responsible to encode
- * whatever is pending on the implementation class into a
- * bufferlist, so we can then propose that as a value through
- * Paxos.
- */
- MonitorDBStore::TransactionRef t = paxos->get_pending_transaction();
-
- if (should_stash_full())
- encode_full(t);
-
- encode_pending(t);
- have_pending = false;
-
- if (format_version > 0) {
- t->put(get_service_name(), "format_version", format_version);
- }
-
- // apply to paxos
- proposing = true;
- /**
- * Callback class used to mark us as active once a proposal finishes going
- * through Paxos.
- *
- * We should wake people up *only* *after* we inform the service we
- * just went active. And we should wake people up only once we finish
- * going active. This is why we first go active, avoiding to wake up the
- * wrong people at the wrong time, such as waking up a C_RetryMessage
- * before waking up a C_Active, thus ending up without a pending value.
- */
- class C_Committed : public Context {
- PaxosService *ps;
- public:
- explicit C_Committed(PaxosService *p) : ps(p) { }
- void finish(int r) override {
- ps->proposing = false;
- if (r >= 0)
- ps->_active();
- else if (r == -ECANCELED || r == -EAGAIN)
- return;
- else
- assert(0 == "bad return value for C_Committed");
- }
- };
- paxos->queue_pending_finisher(new C_Committed(this));
- paxos->trigger_propose();
-}
-
-bool PaxosService::should_stash_full()
-{
- version_t latest_full = get_version_latest_full();
- /* @note The first member of the condition is moot and it is here just for
- * clarity's sake. The second member would end up returing true
- * nonetheless because, in that event,
- * latest_full == get_trim_to() == 0.
- */
- return (!latest_full ||
- (latest_full <= get_trim_to()) ||
- (get_last_committed() - latest_full > (version_t)g_conf->paxos_stash_full_interval));
-}
-
-void PaxosService::restart()
-{
- dout(10) << __func__ << dendl;
- if (proposal_timer) {
- dout(10) << " canceling proposal_timer " << proposal_timer << dendl;
- mon->timer.cancel_event(proposal_timer);
- proposal_timer = 0;
- }
-
- finish_contexts(g_ceph_context, waiting_for_finished_proposal, -EAGAIN);
-
- if (have_pending) {
- discard_pending();
- have_pending = false;
- }
- proposing = false;
-
- on_restart();
-}
-
-void PaxosService::election_finished()
-{
- dout(10) << __func__ << dendl;
-
- finish_contexts(g_ceph_context, waiting_for_finished_proposal, -EAGAIN);
-
- // make sure we update our state
- _active();
-}
-
-void PaxosService::_active()
-{
- if (is_proposing()) {
- dout(10) << __func__ << " - proposing" << dendl;
- return;
- }
- if (!is_active()) {
- dout(10) << __func__ << " - not active" << dendl;
- /**
- * Callback used to make sure we call the PaxosService::_active function
- * whenever a condition is fulfilled.
- *
- * This is used in multiple situations, from waiting for the Paxos to commit
- * our proposed value, to waiting for the Paxos to become active once an
- * election is finished.
- */
- class C_Active : public Context {
- PaxosService *svc;
- public:
- explicit C_Active(PaxosService *s) : svc(s) {}
- void finish(int r) override {
- if (r >= 0)
- svc->_active();
- }
- };
- wait_for_active_ctx(new C_Active(this));
- return;
- }
- dout(10) << __func__ << dendl;
-
- // create pending state?
- if (mon->is_leader()) {
- dout(7) << __func__ << " creating new pending" << dendl;
- if (!have_pending) {
- create_pending();
- have_pending = true;
- }
-
- if (get_last_committed() == 0) {
- // create initial state
- create_initial();
- propose_pending();
- return;
- }
- } else {
- if (!mon->is_leader()) {
- dout(7) << __func__ << " we are not the leader, hence we propose nothing!" << dendl;
- }
- }
-
- // wake up anyone who came in while we were proposing. note that
- // anyone waiting for the previous proposal to commit is no longer
- // on this list; it is on Paxos's.
- finish_contexts(g_ceph_context, waiting_for_finished_proposal, 0);
-
- if (mon->is_leader())
- upgrade_format();
-
- // NOTE: it's possible that this will get called twice if we commit
- // an old paxos value. Implementations should be mindful of that.
- on_active();
-}
-
-
-void PaxosService::shutdown()
-{
- cancel_events();
-
- if (proposal_timer) {
- dout(10) << " canceling proposal_timer " << proposal_timer << dendl;
- mon->timer.cancel_event(proposal_timer);
- proposal_timer = 0;
- }
-
- finish_contexts(g_ceph_context, waiting_for_finished_proposal, -EAGAIN);
-
- on_shutdown();
-}
-
-void PaxosService::maybe_trim()
-{
- if (!is_writeable())
- return;
-
- version_t trim_to = get_trim_to();
- if (trim_to < get_first_committed())
- return;
-
- version_t to_remove = trim_to - get_first_committed();
- if (g_conf->paxos_service_trim_min > 0 &&
- to_remove < (version_t)g_conf->paxos_service_trim_min) {
- dout(10) << __func__ << " trim_to " << trim_to << " would only trim " << to_remove
- << " < paxos_service_trim_min " << g_conf->paxos_service_trim_min << dendl;
- return;
- }
-
- if (g_conf->paxos_service_trim_max > 0 &&
- to_remove > (version_t)g_conf->paxos_service_trim_max) {
- dout(10) << __func__ << " trim_to " << trim_to << " would only trim " << to_remove
- << " > paxos_service_trim_max, limiting to " << g_conf->paxos_service_trim_max
- << dendl;
- trim_to = get_first_committed() + g_conf->paxos_service_trim_max;
- to_remove = g_conf->paxos_service_trim_max;
- }
-
- dout(10) << __func__ << " trimming to " << trim_to << ", " << to_remove << " states" << dendl;
- MonitorDBStore::TransactionRef t = paxos->get_pending_transaction();
- trim(t, get_first_committed(), trim_to);
- put_first_committed(t, trim_to);
-
- // let the service add any extra stuff
- encode_trim_extra(t, trim_to);
-
- paxos->trigger_propose();
-}
-
-void PaxosService::trim(MonitorDBStore::TransactionRef t,
- version_t from, version_t to)
-{
- dout(10) << __func__ << " from " << from << " to " << to << dendl;
- assert(from != to);
-
- for (version_t v = from; v < to; ++v) {
- dout(20) << __func__ << " " << v << dendl;
- t->erase(get_service_name(), v);
-
- string full_key = mon->store->combine_strings("full", v);
- if (mon->store->exists(get_service_name(), full_key)) {
- dout(20) << __func__ << " " << full_key << dendl;
- t->erase(get_service_name(), full_key);
- }
- }
- if (g_conf->mon_compact_on_trim) {
- dout(20) << " compacting prefix " << get_service_name() << dendl;
- t->compact_range(get_service_name(), stringify(from - 1), stringify(to));
- t->compact_range(get_service_name(),
- mon->store->combine_strings(full_prefix_name, from - 1),
- mon->store->combine_strings(full_prefix_name, to));
- }
-}
-
-void PaxosService::load_health()
-{
- bufferlist bl;
- mon->store->get("health", service_name, bl);
- if (bl.length()) {
- auto p = bl.begin();
- ::decode(health_checks, p);
- }
-}
Code Review / stor4nfv.git / blobdiff