diff options
Diffstat (limited to 'wsrep-lib/src/server_state.cpp')
| -rw-r--r-- | wsrep-lib/src/server_state.cpp | 1654 |
1 files changed, 1654 insertions, 0 deletions
diff --git a/wsrep-lib/src/server_state.cpp b/wsrep-lib/src/server_state.cpp new file mode 100644 index 00000000..2fc9b199 --- /dev/null +++ b/wsrep-lib/src/server_state.cpp @@ -0,0 +1,1654 @@ +/* + * Copyright (C) 2018-2023 Codership Oy <info@codership.com> + * + * This file is part of wsrep-lib. + * + * Wsrep-lib is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 2 of the License, or + * (at your option) any later version. + * + * Wsrep-lib is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with wsrep-lib. If not, see <https://www.gnu.org/licenses/>. + */ + +#include "wsrep/server_state.hpp" +#include "wsrep/client_state.hpp" +#include "wsrep/server_service.hpp" +#include "wsrep/client_service.hpp" +#include "wsrep/high_priority_service.hpp" +#include "wsrep/transaction.hpp" +#include "wsrep/view.hpp" +#include "wsrep/logger.hpp" +#include "wsrep/compiler.hpp" +#include "wsrep/id.hpp" + +#include <cassert> +#include <sstream> +#include <algorithm> + + +////////////////////////////////////////////////////////////////////////////// +// Helpers // +////////////////////////////////////////////////////////////////////////////// + + +// +// This method is used to deal with historical burden of several +// ways to bootstrap the cluster. Bootstrap happens if +// +// * bootstrap option is given +// * cluster_address is "gcomm://" (Galera provider) +// +static bool is_bootstrap(const std::string& cluster_address, bool bootstrap) +{ + return (bootstrap || cluster_address == "gcomm://"); +} + +// Helper method to provide detailed error message if transaction +// adopt for fragment removal fails. +static void log_adopt_error(const wsrep::transaction& transaction) +{ + wsrep::log_warning() << "Adopting a transaction (" + << transaction.server_id() << "," << transaction.id() + << ") for rollback failed, " + << "this may leave stale entries to streaming log " + << "which may need to be removed manually."; +} + +// resolve which of the two errors return to caller +static inline int resolve_return_error(bool const vote, + int const vote_err, + int const apply_err) +{ + if (vote) return vote_err; + return vote_err != 0 ? vote_err : apply_err; +} + +static void +discard_streaming_applier(wsrep::server_state& server_state, + wsrep::high_priority_service& high_priority_service, + wsrep::high_priority_service* streaming_applier, + const wsrep::ws_meta& ws_meta) +{ + server_state.stop_streaming_applier( + ws_meta.server_id(), ws_meta.transaction_id()); + server_state.server_service().release_high_priority_service( + streaming_applier); + high_priority_service.store_globals(); +} + +static int apply_fragment(wsrep::server_state& server_state, + wsrep::high_priority_service& high_priority_service, + wsrep::high_priority_service* streaming_applier, + const wsrep::ws_handle& ws_handle, + const wsrep::ws_meta& ws_meta, + const wsrep::const_buffer& data) +{ + int ret(0); + int apply_err; + wsrep::mutable_buffer err; + { + wsrep::high_priority_switch sw(high_priority_service, + *streaming_applier); + apply_err = streaming_applier->apply_write_set(ws_meta, data, err); + if (!apply_err) + { + assert(err.size() == 0); + streaming_applier->after_apply(); + } + else + { + bool const remove_fragments(streaming_applier->transaction( + ).streaming_context().fragments().size() > 0); + ret = streaming_applier->rollback(ws_handle, ws_meta); + ret = ret || (streaming_applier->after_apply(), 0); + + if (remove_fragments) + { + ret = ret || streaming_applier->start_transaction(ws_handle, + ws_meta); + ret = ret || (streaming_applier->adopt_apply_error(err), 0); + ret = ret || streaming_applier->remove_fragments(ws_meta); + ret = ret || streaming_applier->commit(ws_handle, ws_meta); + ret = ret || (streaming_applier->after_apply(), 0); + } + else + { + ret = streaming_applier->log_dummy_write_set(ws_handle, + ws_meta, err); + } + } + } + + if (!ret) + { + if (!apply_err) + { + high_priority_service.debug_crash("crash_apply_cb_before_append_frag"); + const wsrep::xid xid(streaming_applier->transaction().xid()); + ret = high_priority_service.append_fragment_and_commit( + ws_handle, ws_meta, data, xid); + high_priority_service.debug_crash("crash_apply_cb_after_append_frag"); + ret = ret || (high_priority_service.after_apply(), 0); + } + else + { + discard_streaming_applier(server_state, + high_priority_service, + streaming_applier, + ws_meta); + ret = resolve_return_error(err.size() > 0, ret, apply_err); + } + } + + return ret; +} + +static int commit_fragment(wsrep::server_state& server_state, + wsrep::high_priority_service& high_priority_service, + wsrep::high_priority_service* streaming_applier, + const wsrep::ws_handle& ws_handle, + const wsrep::ws_meta& ws_meta, + const wsrep::const_buffer& data) +{ + int ret(0); + { + wsrep::high_priority_switch sw( + high_priority_service, *streaming_applier); + wsrep::mutable_buffer err; + int const apply_err( + streaming_applier->apply_write_set(ws_meta, data, err)); + if (apply_err) + { + assert(streaming_applier->transaction( + ).streaming_context().fragments().size() > 0); + ret = streaming_applier->rollback(ws_handle, ws_meta); + ret = ret || (streaming_applier->after_apply(), 0); + ret = ret || streaming_applier->start_transaction( + ws_handle, ws_meta); + ret = ret || (streaming_applier->adopt_apply_error(err),0); + } + else + { + assert(err.size() == 0); + } + + const wsrep::transaction& trx(streaming_applier->transaction()); + // Fragment removal for XA is going to happen in after_commit + if (trx.state() != wsrep::transaction::s_prepared) + { + streaming_applier->debug_crash( + "crash_apply_cb_before_fragment_removal"); + + ret = ret || streaming_applier->remove_fragments(ws_meta); + + streaming_applier->debug_crash( + "crash_apply_cb_after_fragment_removal"); + } + + streaming_applier->debug_crash( + "crash_commit_cb_before_last_fragment_commit"); + ret = ret || streaming_applier->commit(ws_handle, ws_meta); + streaming_applier->debug_crash( + "crash_commit_cb_last_fragment_commit_success"); + ret = ret || (streaming_applier->after_apply(), 0); + ret = resolve_return_error(err.size() > 0, ret, apply_err); + } + + if (!ret) + { + discard_streaming_applier(server_state, high_priority_service, + streaming_applier, ws_meta); + } + + return ret; +} + +static int rollback_fragment(wsrep::server_state& server_state, + wsrep::high_priority_service& high_priority_service, + wsrep::high_priority_service* streaming_applier, + const wsrep::ws_handle& ws_handle, + const wsrep::ws_meta& ws_meta) +{ + int ret(0); + int adopt_error(0); + bool const remove_fragments(streaming_applier->transaction(). + streaming_context().fragments().size() > 0); + // If fragment removal is needed, adopt transaction state + // and start a transaction for it. + if (remove_fragments && + (adopt_error = high_priority_service.adopt_transaction( + streaming_applier->transaction()))) + { + log_adopt_error(streaming_applier->transaction()); + } + // Even if the adopt above fails we roll back the streaming transaction. + // Adopt failure will leave stale entries in streaming log which can + // be removed manually. + wsrep::const_buffer no_error; + { + wsrep::high_priority_switch ws( + high_priority_service, *streaming_applier); + // Streaming applier rolls back out of order. Fragment + // removal grabs commit order below. + ret = streaming_applier->rollback(wsrep::ws_handle(), wsrep::ws_meta()); + ret = ret || (streaming_applier->after_apply(), 0); + } + + if (!ret) + { + discard_streaming_applier(server_state, high_priority_service, + streaming_applier, ws_meta); + + if (adopt_error == 0) + { + if (remove_fragments) + { + ret = high_priority_service.remove_fragments(ws_meta); + ret = ret || high_priority_service.commit(ws_handle, ws_meta); + if (ret) + { + high_priority_service.rollback(ws_handle, ws_meta); + } + high_priority_service.after_apply(); + } + else + { + if (ws_meta.ordered()) + { + wsrep::mutable_buffer no_error; + ret = high_priority_service.log_dummy_write_set( + ws_handle, ws_meta, no_error); + } + } + } + } + return ret; +} + +static int apply_write_set(wsrep::server_state& server_state, + wsrep::high_priority_service& high_priority_service, + const wsrep::ws_handle& ws_handle, + const wsrep::ws_meta& ws_meta, + const wsrep::const_buffer& data) +{ + int ret(0); + if (wsrep::rolls_back_transaction(ws_meta.flags())) + { + wsrep::mutable_buffer no_error; + if (wsrep::starts_transaction(ws_meta.flags())) + { + // No transaction existed before, log a dummy write set + ret = high_priority_service.log_dummy_write_set( + ws_handle, ws_meta, no_error); + } + else + { + wsrep::high_priority_service* sa( + server_state.find_streaming_applier( + ws_meta.server_id(), ws_meta.transaction_id())); + if (sa == 0) + { + // It is a known limitation that galera provider + // cannot always determine if certification test + // for interrupted transaction will pass or fail + // (see comments in transaction::certify_fragment()). + // As a consequence, unnecessary rollback fragments + // may be delivered here. The message below has + // been intentionally turned into a debug message, + // rather than warning. + WSREP_LOG_DEBUG(wsrep::log::debug_log_level(), + wsrep::log::debug_level_server_state, + "Could not find applier context for " + << ws_meta.server_id() + << ": " << ws_meta.transaction_id()); + ret = high_priority_service.log_dummy_write_set( + ws_handle, ws_meta, no_error); + } + else + { + // rollback_fragment() consumes sa + ret = rollback_fragment(server_state, + high_priority_service, + sa, + ws_handle, + ws_meta); + } + } + } + else if (wsrep::starts_transaction(ws_meta.flags()) && + wsrep::commits_transaction(ws_meta.flags())) + { + ret = high_priority_service.start_transaction(ws_handle, ws_meta); + if (!ret) + { + wsrep::mutable_buffer err; + int const apply_err(high_priority_service.apply_write_set( + ws_meta, data, err)); + if (!apply_err) + { + assert(err.size() == 0); + ret = high_priority_service.commit(ws_handle, ws_meta); + ret = ret || (high_priority_service.after_apply(), 0); + } + else + { + ret = high_priority_service.rollback(ws_handle, ws_meta); + ret = ret || (high_priority_service.after_apply(), 0); + ret = ret || high_priority_service.log_dummy_write_set( + ws_handle, ws_meta, err); + ret = resolve_return_error(err.size() > 0, ret, apply_err); + } + } + } + else if (wsrep::starts_transaction(ws_meta.flags())) + { + assert(server_state.find_streaming_applier( + ws_meta.server_id(), ws_meta.transaction_id()) == 0); + wsrep::high_priority_service* sa( + server_state.server_service().streaming_applier_service( + high_priority_service)); + server_state.start_streaming_applier( + ws_meta.server_id(), ws_meta.transaction_id(), sa); + sa->start_transaction(ws_handle, ws_meta); + ret = apply_fragment(server_state, + high_priority_service, + sa, + ws_handle, + ws_meta, + data); + } + else if (ws_meta.flags() == 0 || ws_meta.flags() == wsrep::provider::flag::pa_unsafe || + wsrep::prepares_transaction(ws_meta.flags())) + { + wsrep::high_priority_service* sa( + server_state.find_streaming_applier( + ws_meta.server_id(), ws_meta.transaction_id())); + if (sa == 0) + { + // It is possible that rapid group membership changes + // may cause streaming transaction be rolled back before + // commit fragment comes in. Although this is a valid + // situation, log a warning if a sac cannot be found as + // it may be an indication of a bug too. + wsrep::log_warning() << "Could not find applier context for " + << ws_meta.server_id() + << ": " << ws_meta.transaction_id(); + wsrep::mutable_buffer no_error; + ret = high_priority_service.log_dummy_write_set( + ws_handle, ws_meta, no_error); + } + else + { + sa->next_fragment(ws_meta); + ret = apply_fragment(server_state, + high_priority_service, + sa, + ws_handle, + ws_meta, + data); + } + } + else if (wsrep::commits_transaction(ws_meta.flags())) + { + if (high_priority_service.is_replaying()) + { + wsrep::mutable_buffer unused; + ret = high_priority_service.start_transaction( + ws_handle, ws_meta) || + high_priority_service.apply_write_set(ws_meta, data, unused) || + high_priority_service.commit(ws_handle, ws_meta); + } + else + { + wsrep::high_priority_service* sa( + server_state.find_streaming_applier( + ws_meta.server_id(), ws_meta.transaction_id())); + if (sa == 0) + { + // It is possible that rapid group membership changes + // may cause streaming transaction be rolled back before + // commit fragment comes in. Although this is a valid + // situation, log a warning if a sac cannot be found as + // it may be an indication of a bug too. + wsrep::log_warning() + << "Could not find applier context for " + << ws_meta.server_id() + << ": " << ws_meta.transaction_id(); + wsrep::mutable_buffer no_error; + ret = high_priority_service.log_dummy_write_set( + ws_handle, ws_meta, no_error); + } + else + { + // Commit fragment consumes sa + sa->next_fragment(ws_meta); + ret = commit_fragment(server_state, + high_priority_service, + sa, + ws_handle, + ws_meta, + data); + } + } + } + else + { + assert(0); + } + if (ret) + { + wsrep::log_error() << "Failed to apply write set: " << ws_meta; + } + return ret; +} + +static int apply_toi(wsrep::provider& provider, + wsrep::high_priority_service& high_priority_service, + const wsrep::ws_handle& ws_handle, + const wsrep::ws_meta& ws_meta, + const wsrep::const_buffer& data) +{ + if (wsrep::starts_transaction(ws_meta.flags()) && + wsrep::commits_transaction(ws_meta.flags())) + { + // + // Regular TOI. + // + provider.commit_order_enter(ws_handle, ws_meta); + wsrep::mutable_buffer err; + int const apply_err(high_priority_service.apply_toi(ws_meta,data,err)); + int const vote_err(provider.commit_order_leave(ws_handle, ws_meta,err)); + return resolve_return_error(err.size() > 0, vote_err, apply_err); + } + else if (wsrep::starts_transaction(ws_meta.flags())) + { + provider.commit_order_enter(ws_handle, ws_meta); + wsrep::mutable_buffer err; + int const apply_err(high_priority_service.apply_nbo_begin(ws_meta, data, err)); + int const vote_err(provider.commit_order_leave(ws_handle, ws_meta, err)); + return resolve_return_error(err.size() > 0, vote_err, apply_err); + } + else if (wsrep::commits_transaction(ws_meta.flags())) + { + // NBO end event is ignored here, both local and applied + // have NBO end handled via local TOI calls. + provider.commit_order_enter(ws_handle, ws_meta); + wsrep::mutable_buffer err; + provider.commit_order_leave(ws_handle, ws_meta, err); + return 0; + } + else + { + assert(0); + return 0; + } +} + +////////////////////////////////////////////////////////////////////////////// +// Server State // +////////////////////////////////////////////////////////////////////////////// + +int wsrep::server_state::load_provider( + const std::string& provider_spec, const std::string& provider_options, + const wsrep::provider::services& services) +{ + wsrep::log_info() << "Loading provider " << provider_spec + << " initial position: " << initial_position_; + + provider_ = wsrep::provider::make_provider(*this, + provider_spec, + provider_options, + services); + return (provider_ ? 0 : 1); +} + +void wsrep::server_state::unload_provider() +{ + delete provider_; + provider_ = 0; +} + +int wsrep::server_state::connect(const std::string& cluster_name, + const std::string& cluster_address, + const std::string& state_donor, + bool bootstrap) +{ + bootstrap_ = is_bootstrap(cluster_address, bootstrap); + wsrep::log_info() << "Connecting with bootstrap option: " << bootstrap_; + return provider().connect(cluster_name, cluster_address, state_donor, + bootstrap_); +} + +int wsrep::server_state::disconnect() +{ + { + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + // In case of failure situations which are caused by provider + // being shut down some failing operation may also try to shut + // down the replication. Check the state here and + // return success if the provider disconnect is already in progress + // or has completed. + if (state(lock) == s_disconnecting || state(lock) == s_disconnected) + { + return 0; + } + state(lock, s_disconnecting); + interrupt_state_waiters(lock); + } + return provider().disconnect(); +} + +wsrep::server_state::~server_state() +{ + delete provider_; +} + +std::vector<wsrep::provider::status_variable> +wsrep::server_state::status() const +{ + return provider().status(); +} + + +wsrep::seqno wsrep::server_state::pause() +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + // Disallow concurrent calls to pause to in order to have non-concurrent + // access to desynced_on_pause_ which is checked in resume() call. + wsrep::log_info() << "pause"; + while (pause_count_ > 0) + { + cond_.wait(lock); + } + ++pause_count_; + assert(pause_seqno_.is_undefined()); + lock.unlock(); + pause_seqno_ = provider().pause(); + lock.lock(); + if (pause_seqno_.is_undefined()) + { + --pause_count_; + } + return pause_seqno_; +} + +void wsrep::server_state::resume() +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + wsrep::log_info() << "resume"; + assert(pause_seqno_.is_undefined() == false); + assert(pause_count_ == 1); + if (provider().resume()) + { + throw wsrep::runtime_error("Failed to resume provider"); + } + pause_seqno_ = wsrep::seqno::undefined(); + --pause_count_; + cond_.notify_all(); +} + +wsrep::seqno wsrep::server_state::desync_and_pause() +{ + wsrep::log_info() << "Desyncing and pausing the provider"; + // Temporary variable to store desync() return status. This will be + // assigned to desynced_on_pause_ after pause() call to prevent + // concurrent access to member variable desynced_on_pause_. + bool desync_successful; + if (desync()) + { + // Desync may give transient error if the provider cannot + // communicate with the rest of the cluster. However, this + // error can be tolerated because if the provider can be + // paused successfully below. + WSREP_LOG_DEBUG(wsrep::log::debug_log_level(), + wsrep::log::debug_level_server_state, + "Failed to desync server before pause"); + desync_successful = false; + } + else + { + desync_successful = true; + } + wsrep::seqno ret(pause()); + if (ret.is_undefined()) + { + wsrep::log_warning() << "Failed to pause provider"; + resync(); + return wsrep::seqno::undefined(); + } + else + { + desynced_on_pause_ = desync_successful; + } + wsrep::log_info() << "Provider paused at: " << ret; + return ret; +} + +void wsrep::server_state::resume_and_resync() +{ + wsrep::log_info() << "Resuming and resyncing the provider"; + try + { + // Assign desynced_on_pause_ to local variable before resuming + // in order to avoid concurrent access to desynced_on_pause_ member + // variable. + bool do_resync = desynced_on_pause_; + desynced_on_pause_ = false; + resume(); + if (do_resync) + { + resync(); + } + } + catch (const wsrep::runtime_error& e) + { + wsrep::log_warning() + << "Resume and resync failed, server may have to be restarted"; + } +} + +std::string wsrep::server_state::prepare_for_sst() +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + state(lock, s_joiner); + lock.unlock(); + return server_service_.sst_request(); +} + +int wsrep::server_state::start_sst(const std::string& sst_request, + const wsrep::gtid& gtid, + bool bypass) +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + state(lock, s_donor); + int ret(0); + lock.unlock(); + if (server_service_.start_sst(sst_request, gtid, bypass)) + { + lock.lock(); + wsrep::log_warning() << "SST preparation failed"; + return_from_donor_state(lock); + ret = 1; + } + return ret; +} + +void wsrep::server_state::sst_sent(const wsrep::gtid& gtid, int error) +{ + if (0 == error) + wsrep::log_info() << "SST sent: " << gtid; + else + wsrep::log_info() << "SST sending failed: " << error; + + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + + return_from_donor_state(lock); + + lock.unlock(); + enum provider::status const retval(provider().sst_sent(gtid, error)); + if (retval != provider::success) + { + std::string msg("wsrep::sst_sent() returned an error: "); + msg += wsrep::provider::to_string(retval); + server_service_.log_message(wsrep::log::warning, msg.c_str()); + } +} + +int wsrep::server_state::sst_received(wsrep::client_service& cs, + int const error) +try +{ + wsrep::log_info() << "SST received"; + wsrep::gtid gtid(wsrep::gtid::undefined()); + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + assert(state_ == s_joiner || state_ == s_initialized); + + // Run initialization only if the SST was successful. + // In case of SST failure the system is in undefined state + // may not be recoverable. + if (error == 0) + { + if (server_service_.sst_before_init()) + { + if (init_initialized_ == false) + { + state(lock, s_initializing); + lock.unlock(); + server_service_.debug_sync("on_view_wait_initialized"); + lock.lock(); + wait_until_state(lock, s_initialized); + assert(init_initialized_); + } + } + lock.unlock(); + + if (id_.is_undefined()) + { + assert(0); + throw wsrep::runtime_error( + "wsrep::sst_received() called before connection to cluster"); + } + + gtid = server_service_.get_position(cs); + wsrep::log_info() << "Recovered position from storage: " << gtid; + + lock.lock(); + if (gtid.seqno() >= connected_gtid().seqno()) + { + /* Now the node has all the data the cluster has: part in + * storage, part in replication event queue. */ + state(lock, s_joined); + } + lock.unlock(); + + wsrep::view const v(server_service_.get_view(cs, id_)); + wsrep::log_info() << "Recovered view from SST:\n" << v; + + /* + * If the state id from recovered view has undefined ID, we may + * be upgrading from earlier version which does not provide + * view stored in stable storage. In this case we skip + * sanity checks and assigning the current view and wait + * until the first view delivery. + */ + if (v.state_id().id().is_undefined() == false) + { + if (v.state_id().id() != gtid.id() || + v.state_id().seqno() > gtid.seqno()) + { + /* Since IN GENERAL we may not be able to recover SST GTID from + * the state data, we have to rely on SST script passing the + * GTID value explicitly. + * Here we check if the passed GTID makes any sense: it should + * have the same UUID and greater or equal seqno than the last + * logged view. */ + std::ostringstream msg; + msg << "SST script passed bogus GTID: " << gtid + << ". Preceding view GTID: " << v.state_id(); + throw wsrep::runtime_error(msg.str()); + } + + if (current_view_.status() == wsrep::view::primary) + { + previous_primary_view_ = current_view_; + } + current_view_ = v; + server_service_.log_view(NULL /* this view is stored already */, v); + } + else + { + wsrep::log_warning() + << "View recovered from stable storage was empty. If the " + << "server is doing rolling upgrade from previous version " + << "which does not support storing view info into stable " + << "storage, this is ok. Otherwise this may be a sign of " + << "malfunction."; + } + lock.lock(); + recover_streaming_appliers_if_not_recovered(lock, cs); + lock.unlock(); + } + + enum provider::status const retval(provider().sst_received(gtid, error)); + if (retval != provider::success) + { + wsrep::log_error() << "provider.sst_received() failed: " + << wsrep::provider::to_string(retval); + return 1; + } + return 0; +} +catch (const wsrep::runtime_error& e) +{ + wsrep::log_error() << "sst_received failed: " << e.what(); + if (provider_) + { + provider_->sst_received(wsrep::gtid::undefined(), -EINTR); + } + return 1; +} + +void wsrep::server_state::initialized() +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + wsrep::log_info() << "Server initialized"; + init_initialized_ = true; + if (server_service_.sst_before_init()) + { + state(lock, s_initialized); + } + else + { + state(lock, s_initializing); + state(lock, s_initialized); + } +} + +enum wsrep::provider::status +wsrep::server_state::wait_for_gtid(const wsrep::gtid& gtid, int timeout) + const +{ + return provider().wait_for_gtid(gtid, timeout); +} + +int +wsrep::server_state::set_encryption_key(std::vector<unsigned char>& key) +{ + encryption_key_ = key; + if (provider_) + { + wsrep::const_buffer const key(encryption_key_.data(), + encryption_key_.size()); + enum provider::status const retval(provider_->enc_set_key(key)); + if (retval != provider::success) + { + wsrep::log_error() << "Failed to set encryption key: " + << provider::to_string(retval); + return 1; + } + } + return 0; +} + +std::pair<wsrep::gtid, enum wsrep::provider::status> +wsrep::server_state::causal_read(int timeout) const +{ + return provider().causal_read(timeout); +} + +void wsrep::server_state::on_connect(const wsrep::view& view) +{ + // Sanity checks + if (view.own_index() < 0 || + size_t(view.own_index()) >= view.members().size()) + { + std::ostringstream os; + os << "Invalid view on connect: own index out of range: " << view; +#ifndef NDEBUG + wsrep::log_error() << os.str(); + assert(0); +#endif + throw wsrep::runtime_error(os.str()); + } + + const size_t own_index(static_cast<size_t>(view.own_index())); + if (id_.is_undefined() == false && id_ != view.members()[own_index].id()) + { + std::ostringstream os; + os << "Connection in connected state.\n" + << "Connected view:\n" << view + << "Previous view:\n" << current_view_ + << "Current own ID: " << id_; +#ifndef NDEBUG + wsrep::log_error() << os.str(); + assert(0); +#endif + throw wsrep::runtime_error(os.str()); + } + else + { + id_ = view.members()[own_index].id(); + } + + wsrep::log_info() << "Server " + << name_ + << " connected to cluster at position " + << view.state_id() + << " with ID " + << id_; + + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + connected_gtid_ = view.state_id(); + state(lock, s_connected); +} + +void wsrep::server_state::on_primary_view( + const wsrep::view& view, + wsrep::high_priority_service* high_priority_service) +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + assert(view.final() == false); + // + // Reached primary from connected state. This may mean the following + // + // 1) Server was joined to the cluster and got SST transfer + // 2) Server was partitioned from the cluster and got back + // 3) A new cluster was bootstrapped from non-prim cluster + // + // There is no enough information here what was the cause + // of the primary component, so we need to walk through + // all states leading to joined to notify possible state + // waiters in other threads. + // + if (server_service_.sst_before_init()) + { + if (state_ == s_connected) + { + state(lock, s_joiner); + // We need to assign init_initialized_ here to local + // variable. If the value here was false, we need to skip + // the initializing -> initialized -> joined state cycle + // below. However, if we don't assign the value to + // local, it is possible that the main thread gets control + // between changing the state to initializing and checking + // initialized flag, which may cause the initialzing -> initialized + // state change to be executed even if it should not be. + const bool was_initialized(init_initialized_); + state(lock, s_initializing); + if (was_initialized) + { + // If server side has already been initialized, + // skip directly to s_joined. + state(lock, s_initialized); + } + } + } + else + { + if (state_ == s_connected) + { + state(lock, s_joiner); + } + } + if (init_initialized_ == false) + { + lock.unlock(); + server_service_.debug_sync("on_view_wait_initialized"); + lock.lock(); + wait_until_state(lock, s_initialized); + } + assert(init_initialized_); + + if (bootstrap_) + { + server_service_.bootstrap(); + bootstrap_ = false; + } + + assert(high_priority_service); + + if (high_priority_service) + { + recover_streaming_appliers_if_not_recovered(lock, + *high_priority_service); + close_orphaned_sr_transactions(lock, *high_priority_service); + } + + if (state(lock) < s_joined && + view.state_id().seqno() >= connected_gtid().seqno()) + { + // If we progressed beyond connected seqno, it means we have full state + state(lock, s_joined); + } +} + +void wsrep::server_state::on_non_primary_view( + const wsrep::view& view, + wsrep::high_priority_service* high_priority_service) +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + wsrep::log_info() << "Non-primary view"; + if (view.final()) + { + go_final(lock, view, high_priority_service); + } + else if (state_ != s_disconnecting) + { + state(lock, s_connected); + } +} + +void wsrep::server_state::go_final(wsrep::unique_lock<wsrep::mutex>& lock, + const wsrep::view& view, + wsrep::high_priority_service* hps) +{ + (void)view; // avoid compiler warning "unused parameter 'view'" + assert(view.final()); + assert(hps); + if (hps) + { + close_transactions_at_disconnect(*hps); + } + state(lock, s_disconnected); + id_ = wsrep::id::undefined(); +} + +void wsrep::server_state::on_view(const wsrep::view& view, + wsrep::high_priority_service* high_priority_service) +{ + wsrep::log_info() + << "================================================\nView:\n" + << view + << "================================================="; + if (current_view_.status() == wsrep::view::primary) + { + previous_primary_view_ = current_view_; + } + current_view_ = view; + switch (view.status()) + { + case wsrep::view::primary: + on_primary_view(view, high_priority_service); + break; + case wsrep::view::non_primary: + on_non_primary_view(view, high_priority_service); + break; + case wsrep::view::disconnected: + { + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + go_final(lock, view, high_priority_service); + break; + } + default: + wsrep::log_warning() << "Unrecognized view status: " << view.status(); + assert(0); + } + + server_service_.log_view(high_priority_service, view); +} + +void wsrep::server_state::on_sync() +{ + wsrep::log_info() << "Server " << name_ << " synced with group"; + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + + // Initial sync + if (server_service_.sst_before_init() && init_synced_ == false) + { + switch (state_) + { + case s_synced: + break; + case s_connected: // Seed node path: provider becomes + state(lock, s_joiner); // synced with itself before anything + WSREP_FALLTHROUGH; // else. Then goes DB initialization. + case s_joiner: // | + state(lock, s_initializing); // V + break; + case s_donor: + assert(false); // this should never happen + state(lock, s_joined); + state(lock, s_synced); + break; + case s_initialized: + state(lock, s_joined); + WSREP_FALLTHROUGH; + default: + /* State */ + state(lock, s_synced); + }; + } + else + { + // Calls to on_sync() in synced state are possible if + // server desyncs itself from the group. Provider does not + // inform about this through callbacks. + if (state_ != s_synced) + { + state(lock, s_synced); + } + } + init_synced_ = true; + + enum wsrep::provider::status status(send_pending_rollback_events(lock)); + if (status) + { + // TODO should be retried? + wsrep::log_warning() + << "Failed to flush rollback event cache: " << status; + } +} + +int wsrep::server_state::on_apply( + wsrep::high_priority_service& high_priority_service, + const wsrep::ws_handle& ws_handle, + const wsrep::ws_meta& ws_meta, + const wsrep::const_buffer& data) +{ + if (is_toi(ws_meta.flags())) + { + return apply_toi(provider(), high_priority_service, + ws_handle, ws_meta, data); + } + else if (is_commutative(ws_meta.flags()) || is_native(ws_meta.flags())) + { + // Not implemented yet. + assert(0); + return 0; + } + else + { + return apply_write_set(*this, high_priority_service, + ws_handle, ws_meta, data); + } +} + +enum wsrep::server_state::state wsrep::server_state::state( + wsrep::unique_lock<wsrep::mutex>& lock WSREP_UNUSED) const +{ + assert(lock.owns_lock()); + return state_; +} + +void wsrep::server_state::start_streaming_client( + wsrep::client_state* client_state) +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + WSREP_LOG_DEBUG(wsrep::log::debug_log_level(), + wsrep::log::debug_level_server_state, + "Start streaming client: " << client_state->id()); + if (streaming_clients_.insert( + std::make_pair(client_state->id(), client_state)).second == false) + { + wsrep::log_warning() << "Failed to insert streaming client " + << client_state->id(); + assert(0); + } +} + +void wsrep::server_state::convert_streaming_client_to_applier( + wsrep::client_state* client_state) +{ + // create streaming_applier beforehand as server_state lock should + // not be held when calling server_service methods + wsrep::high_priority_service* streaming_applier( + server_service_.streaming_applier_service( + client_state->client_service())); + + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + WSREP_LOG_DEBUG(wsrep::log::debug_log_level(), + wsrep::log::debug_level_server_state, + "Convert streaming client to applier " + << client_state->id()); + streaming_clients_map::iterator i( + streaming_clients_.find(client_state->id())); + assert(i != streaming_clients_.end()); + if (i == streaming_clients_.end()) + { + wsrep::log_warning() << "Unable to find streaming client " + << client_state->id(); + assert(0); + } + else + { + streaming_clients_.erase(i); + } + + // Convert to applier only if the state is not disconnected. In + // disconnected state the applier map is supposed to be empty + // and it will be reconstructed from fragment storage when + // joining back to cluster. + if (state(lock) != s_disconnected) + { + if (streaming_applier->adopt_transaction(client_state->transaction())) + { + log_adopt_error(client_state->transaction()); + streaming_applier->after_apply(); + server_service_.release_high_priority_service(streaming_applier); + return; + } + if (streaming_appliers_.insert( + std::make_pair( + std::make_pair(client_state->transaction().server_id(), + client_state->transaction().id()), + streaming_applier)).second == false) + { + wsrep::log_warning() << "Could not insert streaming applier " + << id_ + << ", " + << client_state->transaction().id(); + assert(0); + } + } + else + { + server_service_.release_high_priority_service(streaming_applier); + client_state->client_service().store_globals(); + } +} + + +void wsrep::server_state::stop_streaming_client( + wsrep::client_state* client_state) +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + WSREP_LOG_DEBUG(wsrep::log::debug_log_level(), + wsrep::log::debug_level_server_state, + "Stop streaming client: " << client_state->id()); + streaming_clients_map::iterator i( + streaming_clients_.find(client_state->id())); + assert(i != streaming_clients_.end()); + if (i == streaming_clients_.end()) + { + wsrep::log_warning() << "Unable to find streaming client " + << client_state->id(); + assert(0); + return; + } + else + { + streaming_clients_.erase(i); + cond_.notify_all(); + } +} + +void wsrep::server_state::start_streaming_applier( + const wsrep::id& server_id, + const wsrep::transaction_id& transaction_id, + wsrep::high_priority_service* sa) +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + if (streaming_appliers_.insert( + std::make_pair(std::make_pair(server_id, transaction_id), + sa)).second == false) + { + wsrep::log_error() << "Could not insert streaming applier"; + throw wsrep::fatal_error(); + } +} + +void wsrep::server_state::stop_streaming_applier( + const wsrep::id& server_id, + const wsrep::transaction_id& transaction_id) +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + streaming_appliers_map::iterator i( + streaming_appliers_.find(std::make_pair(server_id, transaction_id))); + assert(i != streaming_appliers_.end()); + if (i == streaming_appliers_.end()) + { + wsrep::log_warning() << "Could not find streaming applier for " + << server_id << ":" << transaction_id; + } + else + { + streaming_appliers_.erase(i); + cond_.notify_all(); + } +} + +wsrep::high_priority_service* wsrep::server_state::find_streaming_applier( + const wsrep::id& server_id, + const wsrep::transaction_id& transaction_id) const +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + streaming_appliers_map::const_iterator i( + streaming_appliers_.find(std::make_pair(server_id, transaction_id))); + return (i == streaming_appliers_.end() ? 0 : i->second); +} + +wsrep::high_priority_service* wsrep::server_state::find_streaming_applier( + const wsrep::xid& xid) const +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + streaming_appliers_map::const_iterator i(streaming_appliers_.begin()); + while (i != streaming_appliers_.end()) + { + wsrep::high_priority_service* sa(i->second); + if (sa->transaction().xid() == xid) + { + return sa; + } + i++; + } + return NULL; +} + +////////////////////////////////////////////////////////////////////////////// +// Private // +////////////////////////////////////////////////////////////////////////////// + +int wsrep::server_state::desync(wsrep::unique_lock<wsrep::mutex>& lock) +{ + assert(lock.owns_lock()); + ++desync_count_; + lock.unlock(); + int ret(provider().desync()); + lock.lock(); + if (ret) + { + --desync_count_; + } + return ret; +} + +void wsrep::server_state::resync(wsrep::unique_lock<wsrep::mutex>& + lock WSREP_UNUSED) +{ + assert(lock.owns_lock()); + assert(desync_count_ > 0); + if (desync_count_ > 0) + { + --desync_count_; + if (provider().resync()) + { + throw wsrep::runtime_error("Failed to resync"); + } + } + else + { + wsrep::log_warning() << "desync_count " << desync_count_ + << " on resync"; + } +} + + +void wsrep::server_state::state( + wsrep::unique_lock<wsrep::mutex>& lock WSREP_UNUSED, + enum wsrep::server_state::state state) +{ + assert(lock.owns_lock()); + static const char allowed[n_states_][n_states_] = + { + /* dis, ing, ized, cted, jer, jed, dor, sed, ding to/from */ + { 0, 1, 0, 1, 0, 0, 0, 0, 0}, /* dis */ + { 1, 0, 1, 0, 0, 0, 0, 0, 1}, /* ing */ + { 1, 0, 0, 1, 0, 1, 0, 0, 1}, /* ized */ + { 1, 0, 0, 1, 1, 0, 0, 1, 1}, /* cted */ + { 1, 1, 0, 0, 0, 1, 0, 0, 1}, /* jer */ + { 1, 0, 0, 1, 0, 0, 1, 1, 1}, /* jed */ + { 1, 0, 0, 1, 0, 1, 0, 0, 1}, /* dor */ + { 1, 0, 0, 1, 0, 1, 1, 0, 1}, /* sed */ + { 1, 0, 0, 0, 0, 0, 0, 0, 0} /* ding */ + }; + + if (allowed[state_][state] == false) + { + std::ostringstream os; + os << "server: " << name_ << " unallowed state transition: " + << wsrep::to_string(state_) << " -> " << wsrep::to_string(state); + wsrep::log_warning() << os.str() << "\n"; + assert(0); + } + + WSREP_LOG_DEBUG(wsrep::log::debug_log_level(), + wsrep::log::debug_level_server_state, + "server " << name_ << " state change: " + << to_c_string(state_) << " -> " + << to_c_string(state)); + state_hist_.push_back(state_); + server_service_.log_state_change(state_, state); + state_ = state; + cond_.notify_all(); + while (state_waiters_[state_]) + { + cond_.wait(lock); + } +} + +void wsrep::server_state::wait_until_state( + wsrep::unique_lock<wsrep::mutex>& lock, + enum wsrep::server_state::state state) const +{ + ++state_waiters_[state]; + while (state_ != state) + { + cond_.wait(lock); + // If the waiter waits for any other state than disconnecting + // or disconnected and the state has been changed to disconnecting, + // this usually means that some error was encountered + if (state != s_disconnecting && state != s_disconnected + && state_ == s_disconnecting) + { + throw wsrep::runtime_error("State wait was interrupted"); + } + } + --state_waiters_[state]; + cond_.notify_all(); +} + +int wsrep::server_state::wait_until_state(enum state state) const +try +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + wait_until_state(lock, state); + return 0; +} +catch (...) +{ + return 1; +} + +void wsrep::server_state::interrupt_state_waiters( + wsrep::unique_lock<wsrep::mutex>& lock WSREP_UNUSED) +{ + assert(lock.owns_lock()); + cond_.notify_all(); +} + +template <class C> +void wsrep::server_state::recover_streaming_appliers_if_not_recovered( + wsrep::unique_lock<wsrep::mutex>& lock, C& c) +{ + assert(lock.owns_lock()); + if (streaming_appliers_recovered_ == false) + { + lock.unlock(); + server_service_.recover_streaming_appliers(c); + lock.lock(); + } + streaming_appliers_recovered_ = true; +} + +class transaction_state_cmp +{ +public: + transaction_state_cmp(const enum wsrep::transaction::state s) + : state_(s) { } + bool operator()(const std::pair<wsrep::client_id, + wsrep::client_state*>& vt) const + { + return vt.second->transaction().state() == state_; + } +private: + enum wsrep::transaction::state state_; +}; + +void wsrep::server_state::close_orphaned_sr_transactions( + wsrep::unique_lock<wsrep::mutex>& lock, + wsrep::high_priority_service& high_priority_service) +{ + assert(lock.owns_lock()); + + // When the originator of an SR transaction leaves the primary + // component of the cluster, that SR must be rolled back. When two + // consecutive primary views have the same membership, the system + // may have been in a state with no primary components. + // Example with 2 node cluster: + // - (1,2 primary) + // - (1 non-primary) and (2 non-primary) + // - (1,2 primary) + // We need to rollback SRs owned by both 1 and 2. + // Notice that since the introduction of rollback_event_queue_, + // checking for equal consecutive views is no longer needed. + // However, we must keep it here for the time being, for backwards + // compatibility. + const bool equal_consecutive_views = + current_view_.equal_membership(previous_primary_view_); + + if (current_view_.own_index() == -1 || equal_consecutive_views) + { + streaming_clients_map::iterator i; + transaction_state_cmp prepared_state_cmp(wsrep::transaction::s_prepared); + while ((i = std::find_if_not(streaming_clients_.begin(), + streaming_clients_.end(), + prepared_state_cmp)) + != streaming_clients_.end()) + { + wsrep::client_id client_id(i->first); + wsrep::transaction_id transaction_id(i->second->transaction().id()); + // It is safe to unlock the server state temporarily here. + // The processing happens inside view handler which is + // protected by the provider commit ordering critical + // section. The lock must be unlocked temporarily to + // allow converting the current client to streaming + // applier in transaction::streaming_rollback(). + // The iterator i may be invalidated when the server state + // remains unlocked, so it should not be accessed after + // the bf abort call. + lock.unlock(); + i->second->total_order_bf_abort(current_view_.view_seqno()); + lock.lock(); + streaming_clients_map::const_iterator found_i; + while ((found_i = streaming_clients_.find(client_id)) != + streaming_clients_.end() && + found_i->second->transaction().id() == transaction_id) + { + cond_.wait(lock); + } + } + } + + streaming_appliers_map::iterator i(streaming_appliers_.begin()); + while (i != streaming_appliers_.end()) + { + wsrep::high_priority_service* streaming_applier(i->second); + + // Rollback SR on equal consecutive primary views or if its + // originator is not in the current view. + // Transactions in prepared state must be committed or + // rolled back explicitly, those are never rolled back here. + if ((streaming_applier->transaction().state() != + wsrep::transaction::s_prepared) && + (equal_consecutive_views || + not current_view_.is_member( + streaming_applier->transaction().server_id()))) + { + WSREP_LOG_DEBUG(wsrep::log::debug_log_level(), + wsrep::log::debug_level_server_state, + "Removing SR fragments for " + << i->first.first + << ", " << i->first.second); + wsrep::id server_id(i->first.first); + wsrep::transaction_id transaction_id(i->first.second); + int adopt_error; + if ((adopt_error = high_priority_service.adopt_transaction( + streaming_applier->transaction()))) + { + log_adopt_error(streaming_applier->transaction()); + } + // Even if the transaction adopt above fails, we roll back + // the transaction. Adopt error will leave stale entries + // in the streaming log which can be removed manually. + { + wsrep::high_priority_switch sw(high_priority_service, + *streaming_applier); + streaming_applier->rollback( + wsrep::ws_handle(), wsrep::ws_meta()); + streaming_applier->after_apply(); + } + + streaming_appliers_.erase(i++); + server_service_.release_high_priority_service(streaming_applier); + high_priority_service.store_globals(); + wsrep::ws_meta ws_meta( + wsrep::gtid(), + wsrep::stid(server_id, transaction_id, wsrep::client_id()), + wsrep::seqno::undefined(), 0); + lock.unlock(); + if (adopt_error == 0) + { + high_priority_service.remove_fragments(ws_meta); + high_priority_service.commit(wsrep::ws_handle(transaction_id, 0), + ws_meta); + } + high_priority_service.after_apply(); + lock.lock(); + } + else + { + ++i; + } + } +} + +void wsrep::server_state::close_transactions_at_disconnect( + wsrep::high_priority_service& high_priority_service) +{ + // Close streaming applier without removing fragments + // from fragment storage. When the server is started again, + // it must be able to recover ongoing streaming transactions. + streaming_appliers_map::iterator i(streaming_appliers_.begin()); + while (i != streaming_appliers_.end()) + { + wsrep::high_priority_service* streaming_applier(i->second); + { + wsrep::high_priority_switch sw(high_priority_service, + *streaming_applier); + streaming_applier->rollback( + wsrep::ws_handle(), wsrep::ws_meta()); + streaming_applier->after_apply(); + } + streaming_appliers_.erase(i++); + server_service_.release_high_priority_service(streaming_applier); + high_priority_service.store_globals(); + } + streaming_appliers_recovered_ = false; +} + +// +// Rollback event queue +// + +void wsrep::server_state::queue_rollback_event( + const wsrep::transaction_id& id) +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); +#ifndef NDEBUG + // Make sure we don't have duplicate + // transaction ids in rollback event queue. + // There is no need to do this in release + // build given that caller (streaming_rollback()) + // should avoid duplicates. + for (auto i : rollback_event_queue_) + { + assert(id != i); + } +#endif + rollback_event_queue_.push_back(id); +} + +enum wsrep::provider::status +wsrep::server_state::send_pending_rollback_events( + wsrep::unique_lock<wsrep::mutex>& lock WSREP_UNUSED) +{ + assert(lock.owns_lock()); + while (not rollback_event_queue_.empty()) + { + const wsrep::transaction_id& id(rollback_event_queue_.front()); + const enum wsrep::provider::status status(provider().rollback(id)); + if (status) + { + return status; + } + rollback_event_queue_.pop_front(); + } + return wsrep::provider::success; +} + +enum wsrep::provider::status +wsrep::server_state::send_pending_rollback_events() +{ + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + return send_pending_rollback_events(lock); +} + +void wsrep::server_state::return_from_donor_state( + wsrep::unique_lock<wsrep::mutex>& lock) +{ + assert(lock.owns_lock()); + // v26 API does not have JOINED event, so in anticipation of SYNCED + // we must do it here. Do not modify the state if donor lost the + // donor state e.g. due to cluster partitioning. + if (state(lock) == s_donor) + { + state(lock, s_joined); + } +} |