diff options
Diffstat (limited to 'wsrep-lib/include/wsrep/server_state.hpp')
| -rw-r--r-- | wsrep-lib/include/wsrep/server_state.hpp | 745 |
1 files changed, 745 insertions, 0 deletions
diff --git a/wsrep-lib/include/wsrep/server_state.hpp b/wsrep-lib/include/wsrep/server_state.hpp new file mode 100644 index 00000000..0fa6420b --- /dev/null +++ b/wsrep-lib/include/wsrep/server_state.hpp @@ -0,0 +1,745 @@ +/* + * Copyright (C) 2018 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/>. + */ + +/** @file server_state.hpp + * + * Server State Abstraction + * ========================== + * + * This file defines an interface for WSREP Server State. + * The Server State will encapsulate server identification, + * server state and server capabilities. The class also + * defines an interface for manipulating server state, applying + * of remote transaction write sets, processing SST requests, + * creating local client connections for local storage access + * operations. + * + * Concepts + * ======== + * + * State Snapshot Transfer + * ----------------------- + * + * Depending on SST type (physical or logical), the server storage + * engine initialization must be done before or after SST happens. + * In case of physical SST method (typically rsync, filesystem snapshot) + * the SST happens before the storage engine is initialized, in case + * of logical backup typically after the storage engine initialization. + * + * Rollback Mode + * ------------- + * + * When High Priority Transaction (HPT) write set is applied, it + * may be required that the HPT Brute Force Aborts (BFA) locally + * executing transaction. As HPT must be able to apply all its + * write sets without interruption, the locally executing transaction + * must yield immediately, otherwise a transaction processing + * may stop or even deadlock. Depending on DBMS implementation, + * the local transaction may need to be rolled back immediately + * (synchronous mode) or the rollback may happen later on + * (asynchronous mode). The Server Context implementation + * which derives from Server Context base class must provide + * the base class for the rollback mode which server operates on. + * + * ### Synchronous + * + * If the DBMS server implementation does not allow asynchronous rollback, + * the victim transaction must be rolled back immediately in order to + * allow transaction processing to proceed. Depending on DBMS process model, + * there may be either background thread which processes the rollback + * or the rollback can be done by the HTP applier. + * + * ### Asynchronous + * + * In asynchronous mode the BFA victim transaction is just marked + * to be aborted or in case of fully optimistic concurrency control, + * the conflict is detected at commit. + * + * + * # Return value conventions + * + * The calls which are proxies to corresponding provider functionality + * will return wsrep::provider::status enum as a result. Otherwise + * the return value is generally zero on success, non zero on failure. + */ + +#ifndef WSREP_SERVER_STATE_HPP +#define WSREP_SERVER_STATE_HPP + +#include "mutex.hpp" +#include "condition_variable.hpp" +#include "id.hpp" +#include "view.hpp" +#include "transaction_id.hpp" +#include "logger.hpp" +#include "provider.hpp" +#include "compiler.hpp" +#include "xid.hpp" + +#include <deque> +#include <vector> +#include <string> +#include <map> + +/** + * Magic string to tell provider to engage into trivial (empty) + * state transfer. No data will be passed, but the node shall be + * considered joined. + */ +#define WSREP_LIB_SST_TRIVIAL "trivial" + +namespace wsrep +{ + // Forward declarations + class ws_handle; + class ws_meta; + class client_state; + class transaction; + class const_buffer; + class server_service; + class client_service; + class encryption_service; + + /** @class Server Context + * + * + */ + class server_state + { + public: + /** + * Server state enumeration. + * + * @todo Fix UML generation + * + * Server state diagram if initialization happens before SST. + * + * [*] --> disconnected + * disconnected --> initializing + * initializing --> initialized + * initialized --> connected + * connected --> joiner + * joiner --> joined + * joined --> synced + * synced --> donor + * donor --> joined + * + * Server state diagram if SST happens before initialization. + * + * [*] --> disconnected + * disconnected --> connected + * connected --> joiner + * joiner --> initializing + * initializing --> initialized + * initialized --> joined + * joined --> synced + * synced --> donor + * donor --> joined + */ + enum state + { + /** Server is in disconnected state. */ + s_disconnected, + /** Server is initializing */ + s_initializing, + /** Server has been initialized */ + s_initialized, + /** Server is connected to the cluster */ + s_connected, + /** Server is receiving SST */ + s_joiner, + /** Server has received SST successfully but has not synced + with rest of the cluster yet. */ + s_joined, + /** Server is donating state snapshot transfer */ + s_donor, + /** Server has synced with the cluster */ + s_synced, + /** Server is disconnecting from group */ + s_disconnecting + }; + + static const int n_states_ = s_disconnecting + 1; + + /** + * Rollback Mode enumeration + */ + enum rollback_mode + { + /** Asynchronous rollback mode */ + rm_async, + /** Synchronous rollback mode */ + rm_sync + }; + + virtual ~server_state(); + + wsrep::encryption_service* encryption_service() + { return encryption_service_; } + + wsrep::server_service& server_service() { return server_service_; } + + /** + * Return human readable server name. + * + * @return Human readable server name string. + */ + const std::string& name() const { return name_; } + + /** + * Return Server identifier. + * + * @return Server identifier. + */ + const wsrep::id& id() const { return id_; } + + const std::string& incoming_address() const + { return incoming_address_; } + /** + * Return server group communication address. + * + * @return Return server group communication address. + */ + const std::string& address() const { return address_; } + + /** + * Return working directory + * + * @return String containing path to working directory. + */ + const std::string& working_dir() const { return working_dir_; } + + /** + * Return initial position for server. + */ + const wsrep::gtid& initial_position() const + { return initial_position_; } + /** + * Return maximum protocol version. + */ + int max_protocol_version() const { return max_protocol_version_;} + + /** + * Get the rollback mode which server is operating in. + * + * @return Rollback mode. + */ + enum rollback_mode rollback_mode() const { return rollback_mode_; } + + /** + * Registers a streaming client. + */ + void start_streaming_client(wsrep::client_state* client_state); + + void convert_streaming_client_to_applier( + wsrep::client_state* client_state); + void stop_streaming_client(wsrep::client_state* client_state); + + void start_streaming_applier( + const wsrep::id&, + const wsrep::transaction_id&, + wsrep::high_priority_service*); + + void stop_streaming_applier( + const wsrep::id&, const wsrep::transaction_id&); + + /** + * Find a streaming applier matching server and transaction ids + */ + wsrep::high_priority_service* find_streaming_applier( + const wsrep::id&, + const wsrep::transaction_id&) const; + + /** + * Find a streaming applier matching xid + */ + wsrep::high_priority_service* find_streaming_applier( + const wsrep::xid& xid) const; + + /** + * Queue a rollback fragment the transaction with given id + */ + void queue_rollback_event(const wsrep::transaction_id& id); + + /** + * Send rollback fragments for previously queued events via + * queue_rollback_event() + */ + enum wsrep::provider::status send_pending_rollback_events(); + + /** + * Load WSRep provider. + * + * @param provider WSRep provider library to be loaded. + * @param provider_options Provider specific options string + * to be passed for provider during initialization. + * @param services Application defined services passed to + * the provider. + * + * @return Zero on success, non-zero on error. + */ + int load_provider(const std::string& provider, + const std::string& provider_options, + const wsrep::provider::services& services + = wsrep::provider::services()); + + void unload_provider(); + + bool is_provider_loaded() const { return provider_ != 0; } + + /** + * Return reference to provider. + * + * @return Reference to provider + * + * @throw wsrep::runtime_error if provider has not been loaded + * + * @todo This should not be virtual. However, currently there + * is no mechanism for tests and integrations to provide + * their own provider implementations, so this is kept virtual + * for time being. + */ + virtual wsrep::provider& provider() const + { + if (provider_ == 0) + { + throw wsrep::runtime_error("provider not loaded"); + } + return *provider_; + } + + /** + * Initialize connection to cluster. + * + * @param cluster_name A string containing the name of the cluster + * @param cluster_address Cluster address string + * @param state_donor String containing a list of desired donors + * @param bootstrap Bootstrap option + * + * @return Zero in case of success, non-zero on error. + */ + int connect(const std::string& cluster_name, + const std::string& cluster_address, + const std::string& state_donor, + bool bootstrap); + + int disconnect(); + + /** + * A method which will be called when the server + * has been joined to the cluster + */ + void on_connect(const wsrep::view& view); + + /** + * A method which will be called when a view + * notification event has been delivered by the + * provider. + * + * @params view wsrep::view object which holds the new view + * information. + */ + void on_view(const wsrep::view& view, + wsrep::high_priority_service*); + + /** + * A method which will be called when the server + * has been synchronized with the cluster. + * + * This will have a side effect of changing the Server Context + * state to s_synced. + */ + void on_sync(); + + /** + * Wait until server reaches given state. + */ + void wait_until_state(enum state state) const + { + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + wait_until_state(lock, state); + } + + /** + * Return GTID at the position when server connected to + * the cluster. + */ + wsrep::gtid connected_gtid() const { return connected_gtid_; } + + /** + * Return current view + */ + const wsrep::view& current_view() const { return current_view_; } + + /** + * Wait until all the write sets up to given GTID have been + * committed. + * + * @return Zero on success, non-zero on failure. + */ + enum wsrep::provider::status + wait_for_gtid(const wsrep::gtid&, int timeout) const; + + /** + * Set encryption key + * + * @param key Encryption key + * + * @return Zero on success, non-zero on failure. + */ + int set_encryption_key(std::vector<unsigned char>& key); + + /** + * Return encryption key. + */ + const std::vector<unsigned char>& get_encryption_key() const + { return encryption_key_; } + + /** + * Perform a causal read in the cluster. After the call returns, + * all the causally preceding write sets have been committed + * or the error is returned. + * + * This operation may require communication with other processes + * in the DBMS cluster, so it may be relatively heavy operation. + * Method wait_for_gtid() should be used whenever possible. + * + * @param timeout Timeout in seconds + * + * @return Pair of GTID and result status from provider. + */ + + std::pair<wsrep::gtid, enum wsrep::provider::status> + causal_read(int timeout) const; + + /** + * Desynchronize the server. + * + * If the server state is synced, this call will desynchronize + * the server from the cluster. + */ + int desync() + { + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + return desync(lock); + } + + /** + * Resynchronize the server. + */ + void resync() + { + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + resync(lock); + } + + wsrep::seqno pause(); + + wsrep::seqno pause_seqno() const { return pause_seqno_; } + + void resume(); + + /** + * Desync and pause the provider on one go. Will return + * pause seqno if successful. In case of failure, + * undefined seqno will be returned. + */ + wsrep::seqno desync_and_pause(); + + /** + * Resume and resync the provider on one go. Prior this + * call the provider must have been both desynced and paused, + * by either desync_and_pause() or separate calls to desync() + * and pause(). + */ + void resume_and_resync(); + + /** + * True if server has issued and active desync and pause in one go, + * false otherwise. + */ + bool desynced_on_pause() const { return desynced_on_pause_; } + + /** + * Prepares server state for SST. + * + * @return String containing a SST request + */ + std::string prepare_for_sst(); + + /** + * Start a state snapshot transfer. + * + * @param sst_requets SST request string + * @param gtid Current GTID + * @param bypass Bypass flag + * + * @return Zero in case of success, non-zero otherwise + */ + int start_sst(const std::string& sst_request, + const wsrep::gtid& gtid, + bool bypass); + + /** + * + */ + void sst_sent(const wsrep::gtid& gtid, int error); + + /** + * This method must be called by the joiner after the SST + * transfer has been received. If the DBMS state has not been + * initialized, the call will shift the state to initializing + * and will wait until the initialization is complete. + * + * @param client_service + * @param error code of the SST operation + */ + void sst_received(wsrep::client_service& cs, int error); + + /** + * This method must be called after the server initialization + * has been completed. The call has a side effect of changing + * the Server Context state to s_initialized. + */ + void initialized(); + + /** + * Return true if the server has been initialized. + */ + bool is_initialized() const + { + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + return init_initialized_; + } + + /** + * This method will be called by the provider when + * a remote write set is being applied. It is the responsibility + * of the caller to set up transaction context and data properly. + * + * @todo Make this private, allow calls for provider implementations + * only. + * @param high_priority_service High priority applier service. + * @param transaction Transaction context. + * @param data Write set data + * + * @return Zero on success, non-zero on failure. + */ + int 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); + + enum state state() const + { + wsrep::unique_lock<wsrep::mutex> lock(mutex_); + return state(lock); + } + + enum state state(wsrep::unique_lock<wsrep::mutex>& + lock WSREP_UNUSED) const + { + assert(lock.owns_lock()); + return state_; + } + + /** + * Get provider status variables. + */ + std::vector<wsrep::provider::status_variable> status() const; + + /** + * Set server wide wsrep debug logging level. + * + * Log levels are + * - 0 - No debug logging. + * - 1..n - Debug logging with increasing verbosity. + */ + void debug_log_level(int level) + { + wsrep::log::debug_log_level(level); + } + + wsrep::mutex& mutex() { return mutex_; } + + protected: + /** Server state constructor + * + * @param mutex Mutex provided by the DBMS implementation. + * @param name Human Readable Server Name. + * @param id Server Identifier String, UUID or some unique + * identifier. + * @param address Server address in form of IPv4 address, IPv6 address + * or hostname. + * @param working_dir Working directory for replication specific + * data files. + * @param rollback_mode Rollback mode which server operates on. + */ + server_state(wsrep::mutex& mutex, + wsrep::condition_variable& cond, + wsrep::server_service& server_service, + wsrep::encryption_service* encryption_service, + const std::string& name, + const std::string& incoming_address, + const std::string& address, + const std::string& working_dir, + const wsrep::gtid& initial_position, + int max_protocol_version, + enum rollback_mode rollback_mode) + : mutex_(mutex) + , cond_(cond) + , server_service_(server_service) + , encryption_service_(encryption_service) + , state_(s_disconnected) + , state_hist_() + , state_waiters_(n_states_) + , bootstrap_() + , initial_position_(initial_position) + , init_initialized_() + , init_synced_() + , sst_gtid_() + , desync_count_() + , desynced_on_pause_() + , pause_count_() + , pause_seqno_() + , streaming_clients_() + , streaming_appliers_() + , streaming_appliers_recovered_() + , provider_() + , name_(name) + , id_(wsrep::id::undefined()) + , incoming_address_(incoming_address) + , address_(address) + , working_dir_(working_dir) + , encryption_key_() + , max_protocol_version_(max_protocol_version) + , rollback_mode_(rollback_mode) + , connected_gtid_() + , previous_primary_view_() + , current_view_() + , rollback_event_queue_() + { } + + private: + + server_state(const server_state&); + server_state& operator=(const server_state&); + + int desync(wsrep::unique_lock<wsrep::mutex>&); + void resync(wsrep::unique_lock<wsrep::mutex>&); + void state(wsrep::unique_lock<wsrep::mutex>&, enum state); + void wait_until_state(wsrep::unique_lock<wsrep::mutex>&, enum state) const; + // Interrupt all threads which are waiting for state + void interrupt_state_waiters(wsrep::unique_lock<wsrep::mutex>&); + + // Recover streaming appliers if not already recoverd + template <class C> + void recover_streaming_appliers_if_not_recovered( + wsrep::unique_lock<wsrep::mutex>&, C&); + + // Close SR transcations whose origin is outside of current + // cluster view. + void close_orphaned_sr_transactions( + wsrep::unique_lock<wsrep::mutex>&, + wsrep::high_priority_service&); + + // Close transactions when handling disconnect from the group. + void close_transactions_at_disconnect(wsrep::high_priority_service&); + + // Handle primary view + void on_primary_view(const wsrep::view&, + wsrep::high_priority_service*); + // Handle non-primary view + void on_non_primary_view(const wsrep::view&, + wsrep::high_priority_service*); + // Common actions on final view + void go_final(wsrep::unique_lock<wsrep::mutex>&, + const wsrep::view&, wsrep::high_priority_service*); + + // Send rollback fragments for all transactions in + // rollback_event_queue_ + enum wsrep::provider::status send_pending_rollback_events( + wsrep::unique_lock<wsrep::mutex>& lock); + + wsrep::mutex& mutex_; + wsrep::condition_variable& cond_; + wsrep::server_service& server_service_; + wsrep::encryption_service* encryption_service_; + enum state state_; + std::vector<enum state> state_hist_; + mutable std::vector<int> state_waiters_; + bool bootstrap_; + const wsrep::gtid initial_position_; + bool init_initialized_; + bool init_synced_; + wsrep::gtid sst_gtid_; + size_t desync_count_; + // Boolean to denote if desync was succesfull when desyncing + // and pausing the provider on one go. + bool desynced_on_pause_; + size_t pause_count_; + wsrep::seqno pause_seqno_; + typedef std::map<wsrep::client_id, wsrep::client_state*> + streaming_clients_map; + streaming_clients_map streaming_clients_; + typedef std::map<std::pair<wsrep::id, wsrep::transaction_id>, + wsrep::high_priority_service*> streaming_appliers_map; + streaming_appliers_map streaming_appliers_; + bool streaming_appliers_recovered_; + wsrep::provider* provider_; + std::string name_; + wsrep::id id_; + std::string incoming_address_; + std::string address_; + std::string working_dir_; + std::vector<unsigned char> encryption_key_; + int max_protocol_version_; + enum rollback_mode rollback_mode_; + wsrep::gtid connected_gtid_; + wsrep::view previous_primary_view_; + wsrep::view current_view_; + std::deque<wsrep::transaction_id> rollback_event_queue_; + }; + + static inline const char* to_c_string( + enum wsrep::server_state::state state) + { + switch (state) + { + case wsrep::server_state::s_disconnected: return "disconnected"; + case wsrep::server_state::s_initializing: return "initializing"; + case wsrep::server_state::s_initialized: return "initialized"; + case wsrep::server_state::s_connected: return "connected"; + case wsrep::server_state::s_joiner: return "joiner"; + case wsrep::server_state::s_joined: return "joined"; + case wsrep::server_state::s_donor: return "donor"; + case wsrep::server_state::s_synced: return "synced"; + case wsrep::server_state::s_disconnecting: return "disconnecting"; + } + return "unknown"; + } + + static inline std::string to_string(enum wsrep::server_state::state state) + { + return (to_c_string(state)); + } + +} + +#endif // WSREP_SERVER_STATE_HPP |