database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-08-01 03:47:19 +03:00
Code Activity

re-merging wsrep files from lp:codership-mysql

Browse Source
This commit is contained in:
Seppo Jaakola
2013-02-05 15:48:54 +02:00
parent db74958229
commit e0c6a87b99
22 changed files with 0 additions and 6399 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
BUILD/compile-amd64-debug-wsrep
View File

@ -1,11 +0,0 @@
#! /bin/sh
path=`dirname $0`
. "$path/SETUP.sh"
extra_flags="$amd64_cflags $debug_cflags -g -O0 $wsrep_cflags"
c_warnings="$c_warnings $debug_extra_warnings"
cxx_warnings="$cxx_warnings $debug_extra_warnings"
extra_configs="$amd64_configs $debug_configs $wsrep_configs --with-wsrep"
. "$path/FINISH.sh"

9
BUILD/compile-amd64-wsrep
View File

@ -1,9 +0,0 @@
#! /bin/sh
path=`dirname $0`
. "$path/SETUP.sh"
extra_flags="$amd64_cflags $fast_cflags -g $wsrep_cflags"
extra_configs="$amd64_configs $wsrep_configs --with-wsrep"
. "$path/FINISH.sh"

12
BUILD/compile-pentium-debug-wsrep
View File

@ -1,12 +0,0 @@
#! /bin/sh -x
path=`dirname $0`
set -- "$@" --with-debug=full
. "$path/SETUP.sh"
extra_flags="$pentium_cflags $debug_cflags -g -O0 $wsrep_cflags"
c_warnings="$c_warnings $debug_extra_warnings"
cxx_warnings="$cxx_warnings $debug_extra_warnings"
extra_configs="$pentium_configs $debug_configs $wsrep_configs --with-wsrep"
. "$path/FINISH.sh"

11
BUILD/compile-pentium-wsrep
View File

@ -1,11 +0,0 @@
#! /bin/sh
path=`dirname $0`
. "$path/SETUP.sh"
extra_flags="$pentium_cflags $fast_cflags $wsrep_cflags"
extra_configs="$pentium_configs $wsrep_configs --with-wsrep"
#strip=yes
. "$path/FINISH.sh"

28
BUILD/compile-pentium64-wsrep
View File

@ -1,28 +0,0 @@
#! /bin/sh
# Copyright (C) 2006, 2007 MySQL AB
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Library General Public
# License as published by the Free Software Foundation; version 2
# of the License.
#
# This program 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
# Library General Public License for more details.
#
# You should have received a copy of the GNU Library General Public
# License along with this library; if not, write to the Free
# Software Foundation, Inc., 59 Temple Place - Suite 330, Boston,
# MA 02111-1307, USA
path=`dirname $0`
. "$path/SETUP.sh"
extra_flags="$pentium64_cflags $fast_cflags -g $wsrep_cflags"
extra_configs="$pentium_configs $static_link $wsrep_configs --with-wsrep"
CC="$CC --pipe"
strip=yes
. "$path/FINISH.sh"

487
Docs/README-wsrep
View File

@ -1,487 +0,0 @@
Codership Oy
http://www.codership.com
<info@codership.com>
DISCLAIMER
THIS SOFTWARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND, EITHER
EXPRESSED OR IMPLIED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.
IN NO EVENT SHALL CODERSHIP OY BE HELD LIABLE TO ANY PARTY FOR ANY DAMAGES
RESULTING DIRECTLY OR INDIRECTLY FROM THE USE OF THIS SOFTWARE.
Trademark Information.
MySQL is a trademark or registered trademark of Oracle and/or its affiliates.
Other trademarks are the property of their respective owners.
Licensing Information.
Please see file COPYING that came with this distribution
Source code can be found at
wsrep API: https://launchpad.net/wsrep
MySQL patch: https://launchpad.net/codership-mysql
ABOUT THIS DOCUMENT
This document covers installation and configuration issues specific to this
wsrep-patched MySQL distribution by Codership. It does not cover the use or
administration of MySQL server per se. The reader is assumed to know how to
install, configure, administer and use standard MySQL server version 5.1.xx.
MYSQL-5.5.x/wsrep-23.x
CONTENTS:
=========
1. WHAT IS WSREP PATCH FOR MYSQL
2. INSTALLATION
3. FIRST TIME SETUP
3.1 CONFIGURATION FILES
3.2 DATABASE PRIVILEGES
3.3 CHECK AND CORRECT FIREWALL SETTINGS
3.4 SELINUX
3.5 APPARMOR
3.6 CONNECT TO CLUSTER
4. UPGRADING FROM MySQL 5.1.x
5. CONFIGURATION OPTIONS
5.1 MANDATORY MYSQL OPTIONS
5.2 WSREP OPTIONS
6. ONLINE SCHEMA UPGRADE
6.1 TOTAL ORDER ISOLATION (TOI)
6.2 ROLLING SCHEMA UPGRADE (RSU)
7. LIMITATIONS
1. WHAT IS WSREP PATCH FOR MYSQL/INNODB
Wsrep API developed by Codership Oy is a modern generic (database-agnostic)
replication API for transactional databases with a goal to make database
replication/logging subsystem completely modular and pluggable. It is developed
with flexibility and completeness in mind to satisfy broad range of modern
replication scenarios. It is equally suitable for synchronous and asynchronous,
master-slave and multi-master replication.
wsrep stands for Write Set REPlication.
Wsrep patch for MySQL/InnoDB allows MySQL server to load and use various wsrep
API implementations ("wsrep providers") with different qualities of service.
Without wsrep provider MySQL-wsrep server will function like a regular
standalone server.
2. INSTALLATION
In the examples below mysql authentication options are omitted for brevity.
2.1 Download and install mysql-wsrep package.
Download binary package for your Linux distribution from
https://launchpad.net/codership-mysql/
2.1.1 On Debian and Debian-derived distributions.
Upgrade from mysql-server-5.0 to mysql-wsrep is not supported yet, please
upgrade to mysql-server-5.1 first.
If you're installing over an existing mysql installation, mysql-server-wsrep
will conflict with mysql-server-5.1 package, so remove it first:
$ sudo apt-get remove mysql-server-5.1 mysql-server-core-5.1
mysql-server-wsrep requires psmisc and mysql-client-5.1.47 (or later).
MySQL 5.1 packages can be found from backports repositories.
For further information about configuring and using Debian or Ubuntu
backports, see:
* http://backports.debian.org
* https://help.ubuntu.com/community/UbuntuBackports
For example, installation of required packages on Debian Lenny:
$ sudo apt-get install psmisc
$ sudo apt-get -t lenny-backports install mysql-client-5.1
Now you should be able to install mysql-wsrep package:
$ sudo dpkg -i <mysql-server-wsrep DEB>
2.1.2 On CentOS and similar RPM-based distributions.
If you're migrating from existing MySQL installation, there are two variants:
a) If you're already using official MySQL-server-community 5.1.x RPM from
Oracle:
# rpm -e mysql-server
b) If you're upgrading from the stock mysql-5.0.77 on CentOS:
1) Make sure that the following packages are not installed:
# rpm --nodeps --allmatches -e mysql-server mysql-test mysql-bench
2) Install *official* MySQL-shared-compat-5.1.x from
http://dev.mysql.com/downloads/mysql/5.1.html
Actual installation:
# rpm -Uvh <MySQL-server-wsrep RPM>
If this fails due to unsatisfied dependencies, install missing packages
(e.g. yum install perl-DBI) and retry.
Additional packages to consider (if not yet installed):
* galera (multi-master replication provider, https://launchpad.net/galera)
* MySQL-client-community (for connecting to server and mysqldump-based SST)
* rsync (for rsync-based SST)
* xtrabackup and nc (for xtrabackup-based SST)
2.2 Upgrade system tables.
If you're upgrading a previous MySQL installation, it might be advisable to
upgrade system tables. To do that start mysqld and run mysql_upgrade command.
Consult MySQL documentation in case of errors. Normally they are not critical
and can be ignored unless specific functionality is needed.
3. FIRST TIME SETUP
Unless you're upgrading an already installed mysql-wsrep package, you will need
to set up a few things to prepare server for operation.
3.1 CONFIGURATION FILES
* Make sure system-wide my.cnf does not bind mysqld to 127.0.0.1. That is, if
you have the following line in [mysqld] section, comment it out:
#bind-address = 127.0.0.1
* Make sure system-wide my.cnf contains "!includedir /etc/mysql/conf.d/" line.
* Edit /etc/mysql/conf.d/wsrep.cnf and set wsrep_provider option by specifying
a path to provider library. If you don't have a provider, leave it as it is.
* When a new node joins the cluster it'll have to receive a state snapshot from
one of the peers. This requires a privileged MySQL account with access from
the rest of the cluster. Edit /etc/mysql/conf.d/wsrep.cnf and set mysql
login/password pair for SST, for example:
wsrep_sst_auth=wsrep_sst:wspass
* See CONFIGURATION section below about other configuration parameters that you
might want to change at this point.
3.2 DATABASE PRIVILEGES
Restart MySQL server and connect to it as root to grant privileges to SST
account (empty users confuse MySQL authentication matching rules, we need to
delete them too):
$ mysql -e "SET wsrep_on=OFF; DELETE FROM mysql.user WHERE user='';"
$ mysql -e "SET wsrep_on=OFF; GRANT ALL ON *.* TO wsrep_sst@'%' IDENTIFIED BY 'wspass'";
3.3 CHECK AND CORRECT FIREWALL SETTINGS.
MySQL-wsrep server needs to be accessible from other cluster members through
its client listening socket and through wsrep provider socket. See your
distribution and wsrep provider documentation for details. For example on
CentOS you might need to do something along these lines:
# iptables --insert RH-Firewall-1-INPUT 1 --proto tcp --source <my IP>/24 --destination <my IP>/32 --dport 3306 -j ACCEPT
# iptables --insert RH-Firewall-1-INPUT 1 --proto tcp --source <my IP>/24 --destination <my IP>/32 --dport 4567 -j ACCEPT
If there is a NAT firewall between the nodes, it must be configured to allow
direct connections between the nodes (e.g. via port forwarding).
3.4 SELINUX
If you have SELinux enabled, it may block mysqld from doing required operations.
You'll need to either disable it or configure to allow mysqld to run external
programs and open listen sockets at unprivileged ports (i.e. things that
an unprivileged user can do). See SELinux documentation about it.
To quickly disable SELinux:
1) run 'setenforce 0' as root.
2) set 'SELINUX=permissive' in /etc/selinux/config
3.5 APPARMOR
AppArmor automatically comes with Ubuntu and may also prevent mysqld to from
opening additional ports or run scripts. See AppArmor documentation about its
configuration. To disable AppArmor for mysqld:
$ cd /etc/apparmor.d/disable/
$ sudo ln -s /etc/apparmor.d/usr.sbin.mysqld
$ sudo service apparmor restart
3.6 CONNECT TO CLUSTER
Now you're ready to connect to cluster by setting wsrep_cluster_address variable
and monitor status of wsrep provider:
mysql> SET GLOBAL wsrep_cluster_address='<cluster address string>';
mysql> SHOW STATUS LIKE 'wsrep%';
4 UPGRADING FROM MySQL 5.1.x
!!! THESE INSTRUCTIONS ARE PRELIMINARY AND INCOMPLETE !!!
1) BEFORE UPGRADE (while running 5.1.x):
- comment out 'wsrep_provider' setting from configuration files
(my.cnf and/or wsrep.cnf)
- If performing a rolling upgrade on a running cluster, set
wsrep_sst_method=mysqldump.
You might also need to configure wsrep_sst_receive_address and
wsrep_sst_auth appropriately. mysqldump is the only way to transfer data
from 5.1.x to 5.5.x reliably.
- remove innodb_plugin settings from configuration files.
2) Perform upgrade as usual:
http://dev.mysql.com/doc/refman/5.5/en/upgrading-from-previous-series.html
Don't forget to run 'mysql_upgrade' command.
3) AFTER UPGRADING individual node:
- uncomment 'wsrep_provider' line in configuration file.
- restart the server and join the cluster.
4) AFTER UPGRADING the whole cluster:
- revert to usual wsrep SST settings if not 'mysqldump'.
5. CONFIGURATION OPTIONS
5.1 MANDATORY MYSQL OPTIONS
binlog_format=ROW
This option is required to use row-level replication as opposed to
statement-level. For performance and consistency considerations don't change
that. As a side effect, binlog, if turned on, can be ROW only. In future this
option won't have special meaning.
innodb_autoinc_lock_mode=2
This is a required parameter. Without it INSERTs into tables with
AUTO_INCREMENT column may fail.
autoinc lock modes 0 and 1 can cause unresolved deadlock, and make
system unresponsive.
innodb_locks_unsafe_for_binlog=1
This option is required for parallel applying.
5.2 WSREP OPTIONS
All options are optional except for wsrep_provider, wsrep_cluster_address, and
wsrep_sst_auth.
wsrep_provider=none
A full path to the library that implements WSREP interface. If none is
specified, the server behaves like a regular mysqld.
wsrep_provider_options=
Provider-specific option string. Check wsrep provider documentation or
http://www.codership.com/wiki
wsrep_cluster_address=
Provider-specific cluster address string. This is used to connect a node to
the desired cluster. This option can be given either on mysqld startup or set
during runtime. See wsrep provider documentation for possible values.
wsrep_cluster_name="my_wsrep_cluster"
Logical cluster name, must be the same for all nodes of the cluster.
wsrep_node_address=
An option to explicitly specify the network address of the node in the form
<address>[:port] if autoguessing for some reason does not produce desirable
results (multiple network interfaces, NAT, etc.)
If not explicitly overridden by wsrep_sst_receive_address, the <address> part
will be used to listen for SST (see below). And the whole <address>[:port]
will be passed to wsrep provider to be used as a base address in its
communications.
wsrep_node_name=
Human readable node name (for easier log reading only). Defaults to hostname.
wsrep_slave_threads=1
Number of threads dedicated to processing of writesets from other nodes.
For best performance should be few per CPU core.
wsrep_dbug_option
Options for the built-in DBUG library (independent from what MySQL uses).
Empty by default. Not currently in use.
wsrep_debug=0
Enable debug-level logging.
wsrep_convert_LOCK_to_trx=0
Implicitly convert locking sessions into transactions inside mysqld. By
itself it does not mean support for locking sessions, but it prevents the
database from going into logically inconsistent state. Note however, that
loading large database dump with LOCK statements might result in abnormally
large transactions and cause an out-of-memory condition
wsrep_retry_autocommit=1
Retry autocommit queries and single statement transactions should they fail
certification test. This is analogous to rescheduling an autocommit query
should it go into deadlock with other transactions in the database lock
manager.
wsrep_auto_increment_control=1
Automatically adjust auto_increment_increment and auto_increment_offset
variables based on the number of nodes in the cluster. Significantly reduces
certification conflict rate for INSERTS.
wsrep_drupal_282555_workaround=1
MySQL seems to have an obscure bug when INSERT into table with
AUTO_INCREMENT column with NULL value for that column can fail with a
duplicate key error. When this option is on, it retries such INSERTs.
Required for stable Drupal operation. Documented at:
http://bugs.mysql.com/bug.php?id=41984
http://drupal.org/node/282555
wsrep_causal_reads=0
Enforce strict READ COMMITTED semantics on reads and transactions. May
result in additional latencies. It is a session variable.
wsrep_OSU_method=TOI
Online Schema Upgrade (OSU) can be performed with two alternative methods:
Total Order Isolation (TOI) runs DDL statement in all cluster nodes in
same total order sequence locking the affected table for the duration of the
operation. This may result in the whole cluster being blocked for the
duration of the operation.
Rolling Schema Upgrade (RSU) executes the DDL statement only locally, thus
blocking only one cluster node. During the DDL processing, the node
is not replicating and may be unable to process replication events (due to
table lock). Once DDL operation is complete, the node will catch up and sync
with the cluster to become fully operational again. The DDL statement or
its effects are not replicated, so it is user's responsibility to manually
perform this operation on each of the nodes.
wsrep_forced_binlog_format=none
Force every transaction to use given binlog format. When this variable is
set to something else than NONE, all transactions will use the given forced
format, regardless of what the client session has specified in binlog_format.
Valid choices for wsrep_forced_binlog_format are: ROW, STATEMENT, MIXED and
special value NONE, meaning that there is no forced binlog format in effect.
This variable was intruduced to support STATEMENT format replication during
rolling schema upgrade processing. However, in most cases ROW replication
is valid for asymmetrict schema replication.
State snapshot transfer options.
When a new node joins the cluster it has to synchronize its initial state with
the other cluster members by transferring state snapshot from one of them.
The options below govern how this happens and should be set up before attempting
to join or start a cluster.
wsrep_sst_method=mysqldump
What method to use to copy database state to a newly joined node. Supported
methods:
- mysqldump: slow (except for small datasets) but most tested.
- rsync: much faster on large datasets.
- rsync_wan: same as rsync but with deltaxfer to minimize network traffic.
- xtrabackup: very fast and practically non-blocking SST method based on
Percona's xtrabackup tool.
(for xtrabackup to work the following settings must be present in my.cnf
on all nodes:
[mysqld]
wsrep_sst_auth=root:<root password>
datadir=<path to data dir>
[client]
socket=<path to socket>
)
wsrep_sst_receive_address=
Address (hostname:port) at which this node wants to receive state snapshot.
Defaults to mysqld bind address, and if that is not specified (0.0.0.0) -
to the first IP of eth0 + mysqld bind port.
NOTE: check that your firewall allows connections to this address from other
cluster nodes.
wsrep_sst_auth=
Authentication information needed for state transfer. Depends on the state
transfer method. For mysqldump-based SST it is
<mysql_root_user>:<mysql_root_password>
and should be the same on all nodes - it is used to authenticate with both
state snapshot receiver and state snapshot donor.
wsrep_sst_donor=
A name of the node which should serve as state snapshot donor. This allows
to control which node will serve state snapshot request. By default the
most suitable node is chosen by wsrep provider. This is the same as given in
wsrep_node_name.
6. ONLINE SCHEMA UPGRADE
Schema upgrades mean any data definition statements (DDL statemnents) run
for the database. They change the database structure and are non-
transactional.
Release 22.3 brings a new method for performing schema upgrades. User can
now choose whether to use the traditional total order isolation or new
rolling schema upgrade method. The OSU method choice is done by global
parameter: 'wsrep_OSU_method'.
6.1 Total Order Isolation (TOI)
With earlier releases, DDL processing happened always by Total Order
Isolation (TOI) method. With TOI, the DDL was scheduled to be processed in
same transaction seqeuncing 'slot' in each cluster node.
The processing is secured by locking the affected table from any other use.
With TOI method, the whole cluster has part of the database locked for the
duration of the DDL processing.
6.2 Rolling Schema Upgrade (RSU)
Rolling schema upgrade is new DDL processing method, where DDL will be
processed locally for the node. The node is disconnected of the replication
for the duration of the DDL processing, so that there is only DDL statement
processing in the node and it does not block the rest of the cluster. When
the DDL processing is complete, the node applies delayed replication events
and synchronizes back with the cluster.
The DDL can then be executed cluster-wide by running the same DDL statement
for each node in turn. When this rolling schema upgrade proceeds, part of
the cluster will have old schema structure and part of the cluster will have
new schema structure.
7. LIMITATIONS
1) Currently replication works only with InnoDB storage engine. Any writes to
tables of other types, including system (mysql.*) tables are not replicated.
However, DDL statements are replicated in statement level, and changes
to mysql.* tables will get replicated that way.
So, you can safely issue: CREATE USER...,
but issuing: INSERT INTO mysql.user..., will not be replicated.
2) DELETE operation is unsupported on tables without primary key. Also rows in
tables without primary key may appear in different order on different nodes.
As a result SELECT...LIMIT... may return slightly different sets.
3) Unsupported queries:
* LOCK/UNLOCK TABLES cannot be supported in multi-master setups.
* lock functions (GET_LOCK(), RELEASE_LOCK()... )
4) Query log cannot be directed to table. If you enable query logging,
you must forward the log to a file:
log_output = FILE
Use general_log and general_log_file to choose query logging and the
log file name
5) Maximum allowed transaction size is defined by wsrep_max_ws_rows and
wsrep_max_ws_size. Anything bigger (e.g. huge LOAD DATA) will be rejected.
6) Due to cluster level optimistic concurrency control, transaction issuing
COMMIT may still be aborted at that stage. There can be two transactions.
writing to same rows and committing in separate cluster nodes, and only one
of the them can successfully commit. The failing one will be aborted.
For cluster level aborts, MySQL/galera cluster gives back deadlock error.
code (Error: 1213 SQLSTATE: 40001 (ER_LOCK_DEADLOCK)).
7) XA transactions can not be supported due to possible rollback on commit.

59
cmake/wsrep.cmake
View File

@ -1,59 +0,0 @@
# Copyright (c) 2011, Codership Oy <info@codership.com>.
#
# This program 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; version 2 of the License.
#
# This program 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 this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
# We need to generate a proper spec file even without --with-wsrep flag,
# so WSREP_VERSION is produced regardless
# Set the patch version
SET(WSREP_PATCH_VERSION "7rc1")
# Obtain patch revision number
SET(WSREP_PATCH_REVNO $ENV{WSREP_REV})
IF(NOT WSREP_PATCH_REVNO)
EXECUTE_PROCESS(
COMMAND bzr revno
OUTPUT_VARIABLE WSREP_PATCH_REVNO
RESULT_VARIABLE RESULT
)
STRING(REGEX REPLACE "(\r?\n)+$" "" WSREP_PATCH_REVNO "${WSREP_PATCH_REVNO}")
#FILE(WRITE "wsrep_config" "Debug: WSREP_PATCH_REVNO result: ${RESULT}\n")
ENDIF()
IF(NOT WSREP_PATCH_REVNO)
SET(WSREP_PATCH_REVNO "XXXX")
ENDIF()
# Obtain wsrep API version
EXECUTE_PROCESS(
COMMAND sh -c "grep WSREP_INTERFACE_VERSION ${MySQL_SOURCE_DIR}/wsrep/wsrep_api.h | cut -d '\"' -f 2"
OUTPUT_VARIABLE WSREP_API_VERSION
RESULT_VARIABLE RESULT
)
#FILE(WRITE "wsrep_config" "Debug: WSREP_API_VERSION result: ${RESULT}\n")
STRING(REGEX REPLACE "(\r?\n)+$" "" WSREP_API_VERSION "${WSREP_API_VERSION}")
SET(WSREP_VERSION
"${WSREP_API_VERSION}.${WSREP_PATCH_VERSION}.r${WSREP_PATCH_REVNO}"
)
OPTION(WITH_WSREP "WSREP replication API (to use, e.g. Galera Replication library)" ON)
IF (WITH_WSREP)
SET(WSREP_C_FLAGS "-DWITH_WSREP -DWSREP_PROC_INFO -DMYSQL_MAX_VARIABLE_VALUE_LEN=2048")
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${WSREP_C_FLAGS}")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${WSREP_C_FLAGS}")
SET(COMPILATION_COMMENT "${COMPILATION_COMMENT}, wsrep_${WSREP_VERSION}")
SET(WITH_EMBEDDED_SERVER OFF CACHE INTERNAL "" FORCE)
ENDIF()
#

106
scripts/wsrep_sst_common.sh
View File

@ -1,106 +0,0 @@
# Copyright (C) 2010 Codership Oy
#
# This program 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; version 2 of the License.
#
# This program 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 this program; see the file COPYING. If not, write to the
# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston
# MA 02110-1301 USA.
# This is a common command line parser to be sourced by other SST scripts
set -u
WSREP_SST_OPT_BYPASS=0
while [ $# -gt 0 ]; do
case "$1" in
'--address')
readonly WSREP_SST_OPT_ADDR="$2"
shift
;;
'--auth')
readonly WSREP_SST_OPT_AUTH="$2"
shift
;;
'--bypass')
WSREP_SST_OPT_BYPASS=1
;;
'--datadir')
readonly WSREP_SST_OPT_DATA="$2"
shift
;;
'--defaults-file')
readonly WSREP_SST_OPT_CONF="$2"
shift
;;
'--host')
readonly WSREP_SST_OPT_HOST="$2"
shift
;;
'--local-port')
readonly WSREP_SST_OPT_LPORT="$2"
shift
;;
'--parent')
readonly WSREP_SST_OPT_PARENT="$2"
shift
;;
'--password')
readonly WSREP_SST_OPT_PSWD="$2"
shift
;;
'--port')
readonly WSREP_SST_OPT_PORT="$2"
shift
;;
'--role')
readonly WSREP_SST_OPT_ROLE="$2"
shift
;;
'--socket')
readonly WSREP_SST_OPT_SOCKET="$2"
shift
;;
'--user')
readonly WSREP_SST_OPT_USER="$2"
shift
;;
'--gtid')
readonly WSREP_SST_OPT_GTID="$2"
shift
;;
*) # must be command
# usage
# exit 1
;;
esac
shift
done
readonly WSREP_SST_OPT_BYPASS
wsrep_log()
{
# echo everything to stderr so that it gets into common error log
# deliberately made to look different from the rest of the log
local readonly tst="$(date +%Y%m%d\ %H:%M:%S.%N | cut -b -21)"
echo "WSREP_SST: $* ($tst)" >>/dev/stderr
}
wsrep_log_error()
{
wsrep_log "[ERROR] $*"
}
wsrep_log_info()
{
wsrep_log "[INFO] $*"
}

121
scripts/wsrep_sst_mysqldump.sh
View File

@ -1,121 +0,0 @@
#!/bin/sh -e
# Copyright (C) 2009 Codership Oy
#
# This program 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; version 2 of the License.
#
# This program 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 this program; see the file COPYING. If not, write to the
# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston
# MA 02110-1301 USA.
# This is a reference script for mysqldump-based state snapshot tansfer
. $(dirname $0)/wsrep_sst_common
EINVAL=22
local_ip()
{
PATH=$PATH:/usr/sbin:/usr/bin:/sbin:/bin
[ "$1" = "127.0.0.1" ] && return 0
[ "$1" = "localhost" ] && return 0
[ "$1" = "$(hostname -s)" ] && return 0
[ "$1" = "$(hostname -f)" ] && return 0
[ "$1" = "$(hostname -d)" ] && return 0
# Now if ip program is not found in the path, we can't return 0 since
# it would block any address. Thankfully grep should fail in this case
ip route get "$1" | grep local >/dev/null && return 0
return 1
}
if test -z "$WSREP_SST_OPT_USER"; then err "USER cannot be nil"; exit $EINVAL; fi
if test -z "$WSREP_SST_OPT_HOST"; then err "HOST cannot be nil"; exit $EINVAL; fi
if test -z "$WSREP_SST_OPT_PORT"; then err "PORT cannot be nil"; exit $EINVAL; fi
if test -z "$WSREP_SST_OPT_LPORT"; then err "LPORT cannot be nil"; exit $EINVAL; fi
if test -z "$WSREP_SST_OPT_SOCKET";then err "SOCKET cannot be nil";exit $EINVAL; fi
if test -z "$WSREP_SST_OPT_GTID"; then err "GTID cannot be nil"; exit $EINVAL; fi
if local_ip $WSREP_SST_OPT_HOST && \
[ "$WSREP_SST_OPT_PORT" = "$WSREP_SST_OPT_LPORT" ]
then
wsrep_log_error \
"destination address '$WSREP_SST_OPT_HOST:$WSREP_SST_OPT_PORT' matches source address."
exit $EINVAL
fi
# Check client version
if ! mysql --version | grep 'Distrib 5.5' >/dev/null
then
mysql --version >&2
err "this operation requires MySQL client version 5.5.x"
exit $EINVAL
fi
AUTH="-u$WSREP_SST_OPT_USER"
if test -n "$WSREP_SST_OPT_PSWD"; then AUTH="$AUTH -p$WSREP_SST_OPT_PSWD"; fi
STOP_WSREP="SET wsrep_on=OFF;"
# NOTE: we don't use --routines here because we're dumping mysql.proc table
MYSQLDUMP="mysqldump $AUTH -S$WSREP_SST_OPT_SOCKET \
--add-drop-database --add-drop-table --skip-add-locks --create-options \
--disable-keys --extended-insert --skip-lock-tables --quick --set-charset \
--skip-comments --flush-privileges --all-databases"
# mysqldump cannot restore CSV tables, fix this issue
CSV_TABLES_FIX="
set sql_mode='';
USE mysql;
SET @str = IF (@@have_csv = 'YES', 'CREATE TABLE IF NOT EXISTS general_log (event_time TIMESTAMP NOT NULL, user_host MEDIUMTEXT NOT NULL, thread_id INTEGER NOT NULL, server_id INTEGER UNSIGNED NOT NULL, command_type VARCHAR(64) NOT NULL,argument MEDIUMTEXT NOT NULL) engine=CSV CHARACTER SET utf8 comment=\"General log\"', 'SET @dummy = 0');
PREPARE stmt FROM @str;
EXECUTE stmt;
DROP PREPARE stmt;
SET @str = IF (@@have_csv = 'YES', 'CREATE TABLE IF NOT EXISTS slow_log (start_time TIMESTAMP NOT NULL, user_host MEDIUMTEXT NOT NULL, query_time TIME NOT NULL, lock_time TIME NOT NULL, rows_sent INTEGER NOT NULL, rows_examined INTEGER NOT NULL, db VARCHAR(512) NOT NULL, last_insert_id INTEGER NOT NULL, insert_id INTEGER NOT NULL, server_id INTEGER UNSIGNED NOT NULL, sql_text MEDIUMTEXT NOT NULL) engine=CSV CHARACTER SET utf8 comment=\"Slow log\"', 'SET @dummy = 0');
PREPARE stmt FROM @str;
EXECUTE stmt;
DROP PREPARE stmt;"
SET_START_POSITION="SET GLOBAL wsrep_start_position='$WSREP_SST_OPT_GTID';"
MYSQL="mysql $AUTH -h$WSREP_SST_OPT_HOST -P$WSREP_SST_OPT_PORT "\
"--disable-reconnect --connect_timeout=10"
# need to disable logging when loading the dump
# reason is that dump contains ALTER TABLE for log tables, and
# this causes an error if logging is enabled
GENERAL_LOG_OPT=`$MYSQL --skip-column-names -e"$STOP_WSREP SELECT @@GENERAL_LOG"`
SLOW_LOG_OPT=`$MYSQL --skip-column-names -e"$STOP_WSREP SELECT @@SLOW_QUERY_LOG"`
$MYSQL -e"$STOP_WSREP SET GLOBAL GENERAL_LOG=OFF"
$MYSQL -e"$STOP_WSREP SET GLOBAL SLOW_QUERY_LOG=OFF"
# commands to restore log settings
RESTORE_GENERAL_LOG="SET GLOBAL GENERAL_LOG=$GENERAL_LOG_OPT;"
RESTORE_SLOW_QUERY_LOG="SET GLOBAL SLOW_QUERY_LOG=$SLOW_LOG_OPT;"
if [ $WSREP_SST_OPT_BYPASS -eq 0 ]
then
(echo $STOP_WSREP && $MYSQLDUMP && echo $CSV_TABLES_FIX \
&& echo $RESTORE_GENERAL_LOG && echo $RESTORE_SLOW_QUERY_LOG \
&& echo $SET_START_POSITION \
|| echo "SST failed to complete;") | $MYSQL
else
wsrep_log_info "Bypassing state dump."
echo $SET_START_POSITION | $MYSQL
fi
#

210
scripts/wsrep_sst_rsync.sh
View File

@ -1,210 +0,0 @@
#!/bin/bash -ue
# Copyright (C) 2010 Codership Oy
#
# This program 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; version 2 of the License.
#
# This program 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 this program; see the file COPYING. If not, write to the
# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston
# MA 02110-1301 USA.
# This is a reference script for rsync-based state snapshot tansfer
RSYNC_PID=
RSYNC_CONF=
. $(dirname $0)/wsrep_sst_common
cleanup_joiner()
{
wsrep_log_info "Joiner cleanup."
local PID=$(cat "$RSYNC_PID" 2>/dev/null || echo 0)
[ "0" != "$PID" ] && kill $PID && sleep 0.5 && kill -9 $PID >/dev/null 2>&1 \
|| :
rm -rf "$RSYNC_CONF"
rm -rf "$MAGIC_FILE"
rm -rf "$RSYNC_PID"
echo " done." >&2
}
check_pid()
{
local pid_file=$1
[ -r "$pid_file" ] && ps -p $(cat $pid_file) >/dev/null 2>&1
}
check_pid_and_port()
{
local pid_file=$1
local rsync_pid=$(cat $pid_file)
local rsync_port=$2
check_pid $pid_file && \
netstat -anpt 2>/dev/null | \
grep LISTEN | grep \:$rsync_port | grep $rsync_pid/rsync >/dev/null
}
MAGIC_FILE="$WSREP_SST_OPT_DATA/rsync_sst_complete"
rm -rf "$MAGIC_FILE"
if [ "$WSREP_SST_OPT_ROLE" = "donor" ]
then
if [ $WSREP_SST_OPT_BYPASS -eq 0 ]
then
FLUSHED="$WSREP_SST_OPT_DATA/tables_flushed"
rm -rf "$FLUSHED"
# Use deltaxfer only for WAN
inv=$(basename $0)
[ "$inv" = "wsrep_sst_rsync_wan" ] && WHOLE_FILE_OPT="" \
|| WHOLE_FILE_OPT="--whole-file"
echo "flush tables"
# wait for tables flushed and state ID written to the file
while [ ! -r "$FLUSHED" ] && ! grep -q ':' "$FLUSHED" >/dev/null 2>&1
do
sleep 0.2
done
STATE="$(cat $FLUSHED)"
rm -rf "$FLUSHED"
sync
# Old filter - include everything except selected
# FILTER=(--exclude '*.err' --exclude '*.pid' --exclude '*.sock' \
# --exclude '*.conf' --exclude core --exclude 'galera.*' \
# --exclude grastate.txt --exclude '*.pem' \
# --exclude '*.[0-9][0-9][0-9][0-9][0-9][0-9]' --exclude '*.index')
# New filter - exclude everything except dirs (schemas) and innodb files
FILTER=(-f '+ /ibdata*' -f '+ /ib_logfile*' -f '+ */' -f '-! */*')
RC=0
rsync --archive --no-times --ignore-times --inplace --delete --quiet \
$WHOLE_FILE_OPT "${FILTER[@]}" "$WSREP_SST_OPT_DATA" \
rsync://$WSREP_SST_OPT_ADDR || RC=$?
[ $RC -ne 0 ] && echo "rsync returned code $RC:" >> /dev/stderr
case $RC in
0) RC=0 # Success
;;
12) RC=71 # EPROTO
wsrep_log_error \
"rsync server on the other end has incompatible protocol. " \
"Make sure you have the same version of rsync on all nodes."
;;
22) RC=12 # ENOMEM
;;
*) RC=255 # unknown error
;;
esac
[ $RC -ne 0 ] && exit $RC
else # BYPASS
wsrep_log_info "Bypassing state dump."
STATE="$WSREP_SST_OPT_GTID"
fi
echo "continue" # now server can resume updating data
echo "$STATE" > "$MAGIC_FILE"
rsync -aqc "$MAGIC_FILE" rsync://$WSREP_SST_OPT_ADDR
echo "done $STATE"
elif [ "$WSREP_SST_OPT_ROLE" = "joiner" ]
then
MYSQLD_PID=$WSREP_SST_OPT_PARENT
MODULE="rsync_sst"
RSYNC_PID="$WSREP_SST_OPT_DATA/$MODULE.pid"
if check_pid $RSYNC_PID
then
echo "rsync daemon already running."
exit 114 # EALREADY
fi
rm -rf "$RSYNC_PID"
ADDR=$WSREP_SST_OPT_ADDR
RSYNC_PORT=$(echo $ADDR | awk -F ':' '{ print $2 }')
if [ -z "$RSYNC_PORT" ]
then
RSYNC_PORT=4444
ADDR="$(echo $ADDR | awk -F ':' '{ print $1 }'):$RSYNC_PORT"
fi
trap "exit 32" HUP PIPE
trap "exit 3" INT TERM
trap cleanup_joiner EXIT
MYUID=$(id -u)
MYGID=$(id -g)
RSYNC_CONF="$WSREP_SST_OPT_DATA/$MODULE.conf"
cat << EOF > "$RSYNC_CONF"
pid file = $RSYNC_PID
use chroot = no
[$MODULE]
path = $WSREP_SST_OPT_DATA
read only = no
timeout = 300
uid = $MYUID
gid = $MYGID
EOF
# rm -rf "$DATA"/ib_logfile* # we don't want old logs around
# listen at all interfaces (for firewalled setups)
rsync --daemon --port $RSYNC_PORT --config "$RSYNC_CONF"
until check_pid_and_port $RSYNC_PID $RSYNC_PORT
do
sleep 0.2
done
echo "ready $ADDR/$MODULE"
# wait for SST to complete by monitoring magic file
while [ ! -r "$MAGIC_FILE" ] && check_pid "$RSYNC_PID" && \
ps -p $MYSQLD_PID >/dev/null
do
sleep 1
done
if ! ps -p $MYSQLD_PID >/dev/null
then
echo "Parent mysqld process (PID:$MYSQLD_PID) terminated unexpectedly." >&2
exit 32
fi
if [ -r "$MAGIC_FILE" ]
then
cat "$MAGIC_FILE" # output UUID:seqno
else
# this message should cause joiner to abort
echo "rsync process ended without creating '$MAGIC_FILE'"
fi
# cleanup_joiner
else
echo "Unrecognized role: '$WSREP_SST_OPT_ROLE'"
exit 22 # EINVAL
fi
exit 0

222
scripts/wsrep_sst_xtrabackup.sh
View File

@ -1,222 +0,0 @@
#!/bin/bash -ue
# Copyright (C) 2011 Percona Inc
#
# This program 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; version 2 of the License.
#
# This program 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 this program; see the file COPYING. If not, write to the
# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston
# MA 02110-1301 USA.
# This is a reference script for Percona XtraBackup-based state snapshot tansfer
TMPDIR="/tmp"
. $(dirname $0)/wsrep_sst_common
cleanup_joiner()
{
#set -x
local PID=$(ps -aef |grep nc| grep $NC_PORT | awk '{ print $2 }')
wsrep_log_info "Killing nc pid $PID"
[ -n "$PID" -a "0" != "$PID" ] && kill $PID && (kill $PID && kill -9 $PID) || :
rm -f "$MAGIC_FILE"
#set +x
}
check_pid()
{
local pid_file="$1"
[ -r "$pid_file" ] && ps -p $(cat "$pid_file") >/dev/null 2>&1
}
kill_xtrabackup()
{
#set -x
local PID=$(cat $XTRABACKUP_PID)
[ -n "$PID" -a "0" != "$PID" ] && kill $PID && (kill $PID && kill -9 $PID) || :
rm -f "$XTRABACKUP_PID"
#set +x
}
# waits ~10 seconds for nc to open the port and then reports ready
# (regardless of timeout)
wait_for_nc()
{
local PORT=$1
local ADDR=$2
local MODULE=$3
for i in $(seq 1 50)
do
netstat -nptl 2>/dev/null | grep '/nc\s*$' | awk '{ print $4 }' | \
sed 's/.*://' | grep \^${PORT}\$ >/dev/null && break
sleep 0.2
done
echo "ready ${ADDR}/${MODULE}"
}
INNOBACKUPEX_BIN=innobackupex
INNOBACKUPEX_ARGS=""
NC_BIN=nc
for TOOL_BIN in INNOBACKUPEX_BIN NC_BIN ; do
if ! which ${!TOOL_BIN} > /dev/null 2>&1
then
echo "Can't find ${!TOOL_BIN} in the path"
exit 22 # EINVAL
fi
done
#ROLE=$1
#ADDR=$2
readonly AUTH=(${WSREP_SST_OPT_AUTH//:/ })
readonly DATA="${WSREP_SST_OPT_DATA}"
#CONF=$5
INFO_FILE="xtrabackup_galera_info"
IST_FILE="xtrabackup_ist"
MAGIC_FILE="${DATA}/${INFO_FILE}"
rm -f "${MAGIC_FILE}"
if [ "$WSREP_SST_OPT_ROLE" = "donor" ]
then
# UUID=$6
# SEQNO=$7
# BYPASS=$8
NC_PORT=$(echo $WSREP_SST_OPT_ADDR | awk -F '[:/]' '{ print $2 }')
REMOTEIP=$(echo $WSREP_SST_OPT_ADDR | awk -F ':' '{ print $1 }')
if [ $WSREP_SST_OPT_BYPASS -eq 0 ]
then
INNOBACKUPEX_ARGS="--galera-info --tmpdir=${TMPDIR} --stream=tar
--defaults-file=${WSREP_SST_OPT_CONF}
--socket=${WSREP_SST_OPT_SOCKET}"
if [ "${AUTH[0]}" != "(null)" ]; then
INNOBACKUPEX_ARGS="${INNOBACKUPEX_ARGS} --user=${AUTH[0]}"
fi
if [ ${#AUTH[*]} -eq 2 ]; then
INNOBACKUPEX_ARGS="${INNOBACKUPEX_ARGS} --password=${AUTH[1]}"
fi
set +e
# This file and variable seems to have no effect and probably should be deleted
XTRABACKUP_PID=$(mktemp --tmpdir wsrep_sst_xtrabackupXXXX.pid)
${INNOBACKUPEX_BIN} ${INNOBACKUPEX_ARGS} ${TMPDIR} \
2> ${DATA}/innobackup.backup.log | \
${NC_BIN} ${REMOTEIP} ${NC_PORT}
RC=( "${PIPESTATUS[@]}" )
set -e
if [ ${RC[0]} -ne 0 ]; then
wsrep_log_error "${INNOBACKUPEX_BIN} finished with error: ${RC[0]}. " \
"Check ${DATA}/innobackup.backup.log"
exit 22
elif [ ${RC[1]} -ne 0 ]; then
wsrep_log_error "${NC_BIN} finished with error: ${RC[1]}"
exit 22
fi
if check_pid "${XTRABACKUP_PID}"
then
wsrep_log_error "xtrabackup process is still running. Killing... "
kill_xtrabackup
exit 22
fi
rm -f "${XTRABACKUP_PID}"
else # BYPASS
STATE="${WSREP_SST_OPT_GTID}"
echo "continue" # now server can resume updating data
echo "${STATE}" > "${MAGIC_FILE}"
echo "1" > "${DATA}/${IST_FILE}"
(cd ${DATA}; tar cf - ${INFO_FILE} ${IST_FILE}) | ${NC_BIN} ${REMOTEIP} ${NC_PORT}
rm -f ${DATA}/${IST_FILE}
fi
echo "done ${WSREP_SST_OPT_GTID}"
elif [ "${WSREP_SST_OPT_ROLE}" = "joiner" ]
then
MODULE="xtrabackup_sst"
rm -f ${DATA}/xtrabackup_*
ADDR=${WSREP_SST_OPT_ADDR}
NC_PORT=$(echo ${ADDR} | awk -F ':' '{ print $2 }')
if [ -z "${NC_PORT}" ]
then
NC_PORT=4444
ADDR="$(echo ${ADDR} | awk -F ':' '{ print $1 }'):${NC_PORT}"
fi
wait_for_nc ${NC_PORT} ${ADDR} ${MODULE} &
# trap "exit 32" HUP PIPE
# trap "exit 3" INT TERM
trap cleanup_joiner HUP PIPE INT TERM
set +e
${NC_BIN} -dl ${NC_PORT} | tar xfi - -C ${DATA} 1>&2
RC=( "${PIPESTATUS[@]}" )
set -e
wait %% # join wait_for_nc thread
if [ ${RC[0]} -ne 0 -o ${RC[1]} -ne 0 ];
then
wsrep_log_error "Error while getting st data from donor node: " \
"${RC[0]}, ${RC[1]}"
exit 32
fi
if [ ! -r "${MAGIC_FILE}" ]
then
# this message should cause joiner to abort
wsrep_log_error "xtrabackup process ended without creating '${MAGIC_FILE}'"
exit 32
fi
if ! ps -p ${WSREP_SST_OPT_PARENT} >/dev/null
then
wsrep_log_error "Parent mysqld process (PID:${WSREP_SST_OPT_PARENT}) terminated unexpectedly." >&2
exit 32
fi
if [ ! -r "${IST_FILE}" ]
then
rm -f ${DATA}/ib_logfile*
${INNOBACKUPEX_BIN} --defaults-file=${WSREP_SST_OPT_CONF} --apply-log \
--ibbackup=xtrabackup ${DATA} 1>&2 2> ${DATA}/innobackup.prepare.log
if [ $? -ne 0 ];
then
wsrep_log_error "${INNOBACKUPEX_BIN} finished with errors. Check ${DATA}/innobackup.prepare.log" >&2
exit 22
fi
fi
cat "${MAGIC_FILE}" # output UUID:seqno
else
wsrep_log_error "Unrecognized role: ${WSREP_SST_OPT_ROLE}"
exit 22 # EINVAL
fi
exit 0

392
sql/wsrep_check_opts.cc
View File

@ -1,392 +0,0 @@
/* Copyright 2011 Codership Oy <http://www.codership.com>
This program 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; version 2 of the License.
This program 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 this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
//#include <mysqld.h>
#include <sql_class.h>
//#include <sql_plugin.h>
//#include <set_var.h>
#include "wsrep_mysqld.h"
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <ctype.h>
/* This file is about checking for correctness of mysql configuration options */
struct opt
{
const char* const name;
const char* value;
};
/* A list of options to check.
* At first we assume default values and then see if they are changed on CLI or
* in my.cnf */
static struct opt opts[] =
{
{ "wsrep_slave_threads", "1" }, // mysqld.cc
{ "bind_address", "0.0.0.0" }, // mysqld.cc
{ "wsrep_sst_method","mysqldump" }, // mysqld.cc
{ "wsrep_sst_receive_address","AUTO"}, // mysqld.cc
{ "binlog_format", "ROW" }, // mysqld.cc
{ "wsrep_provider", "none" }, // mysqld.cc
{ "query_cache_type", "0" }, // mysqld.cc
{ "query_cache_size", "0" }, // mysqld.cc
{ "locked_in_memory", "0" }, // mysqld.cc
{ "wsrep_cluster_address", "0" }, // mysqld.cc
{ "locks_unsafe_for_binlog", "0" }, // ha_innodb.cc
{ "autoinc_lock_mode", "1" }, // ha_innodb.cc
{ 0, 0 }
};
enum
{
WSREP_SLAVE_THREADS,
BIND_ADDRESS,
WSREP_SST_METHOD,
WSREP_SST_RECEIVE_ADDRESS,
BINLOG_FORMAT,
WSREP_PROVIDER,
QUERY_CACHE_TYPE,
QUERY_CACHE_SIZE,
LOCKED_IN_MEMORY,
WSREP_CLUSTER_ADDRESS,
LOCKS_UNSAFE_FOR_BINLOG,
AUTOINC_LOCK_MODE
};
/* A class to make a copy of argv[] vector */
struct argv_copy
{
int const argc_;
char** argv_;
argv_copy (int const argc, const char* const argv[]) :
argc_ (argc),
argv_ (reinterpret_cast<char**>(calloc(argc_, sizeof(char*))))
{
if (argv_)
{
for (int i = 0; i < argc_; ++i)
{
argv_[i] = strdup(argv[i]);
if (!argv_[i])
{
argv_free (); // free whatever bee allocated
return;
}
}
}
}
~argv_copy () { argv_free (); }
private:
argv_copy (const argv_copy&);
argv_copy& operator= (const argv_copy&);
void argv_free()
{
if (argv_)
{
for (int i = 0; (i < argc_) && argv_[i] ; ++i) free (argv_[i]);
free (argv_);
argv_ = 0;
}
}
};
/* a short corresponding to '--' byte sequence */
static short const long_opt_prefix ('-' + ('-' << 8));
/* Normalizes long options to have '_' instead of '-' */
static int
normalize_opts (argv_copy& a)
{
if (a.argv_)
{
for (int i = 0; i < a.argc_; ++i)
{
char* ptr = a.argv_[i];
if (long_opt_prefix == *(short*)ptr) // long option
{
ptr += 2;
const char* end = strchr(ptr, '=');
if (!end) end = ptr + strlen(ptr);
for (; ptr != end; ++ptr) if ('-' == *ptr) *ptr = '_';
}
}
return 0;
}
return EINVAL;
}
/* Find required options in the argument list and change their values */
static int
find_opts (argv_copy& a, struct opt* const opts)
{
for (int i = 0; i < a.argc_; ++i)
{
char* ptr = a.argv_[i] + 2; // we're interested only in long options
struct opt* opt = opts;
for (; 0 != opt->name; ++opt)
{
if (!strstr(ptr, opt->name)) continue; // try next option
/* 1. try to find value after the '=' */
opt->value = strchr(ptr, '=') + 1;
/* 2. if no '=', try next element in the argument vector */
if (reinterpret_cast<void*>(1) == opt->value)
{
/* also check that the next element is not an option itself */
if (i + 1 < a.argc_ && *(a.argv_[i + 1]) != '-')
{
++i;
opt->value = a.argv_[i];
}
else opt->value = ""; // no value supplied (like boolean opt)
}
break; // option found, break inner loop
}
}
return 0;
}
/* Parses string for an integer. Returns 0 on success. */
int get_long_long (const struct opt& opt, long long* const val, int const base)
{
const char* const str = opt.value;
if ('\0' != *str)
{
char* endptr;
*val = strtoll (str, &endptr, base);
if ('k' == *endptr || 'K' == *endptr)
{
*val *= 1024L;
endptr++;
}
else if ('m' == *endptr || 'M' == *endptr)
{
*val *= 1024L * 1024L;
endptr++;
}
else if ('g' == *endptr || 'G' == *endptr)
{
*val *= 1024L * 1024L * 1024L;
endptr++;
}
if ('\0' == *endptr) return 0; // the whole string was a valid integer
}
WSREP_ERROR ("Bad value for *%s: '%s'. Should be integer.",
opt.name, opt.value);
return EINVAL;
}
/* This is flimzy coz hell knows how mysql interprets boolean strings...
* and, no, I'm not going to become versed in how mysql handles options -
* I'd rather sing.
Aha, http://dev.mysql.com/doc/refman/5.1/en/dynamic-system-variables.html:
Variables that have a type of “boolean” can be set to 0, 1, ON or OFF. (If you
set them on the command line or in an option file, use the numeric values.)
So it is '0' for FALSE, '1' or empty string for TRUE
*/
int get_bool (const struct opt& opt, bool* const val)
{
const char* str = opt.value;
while (isspace(*str)) ++str; // skip initial whitespaces
ssize_t str_len = strlen(str);
switch (str_len)
{
case 0:
*val = true;
return 0;
case 1:
if ('0' == *str || '1' == *str)
{
*val = ('1' == *str);
return 0;
}
}
WSREP_ERROR ("Bad value for *%s: '%s'. Should be '0', '1' or empty string.",
opt.name, opt.value);
return EINVAL;
}
static int
check_opts (int const argc, const char* const argv[], struct opt opts[])
{
/* First, make a copy of argv to be able to manipulate it */
argv_copy a(argc, argv);
if (!a.argv_)
{
WSREP_ERROR ("Could not copy argv vector: not enough memory.");
return ENOMEM;
}
int err = normalize_opts (a);
if (err)
{
WSREP_ERROR ("Failed to normalize options.");
return err;
}
err = find_opts (a, opts);
if (err)
{
WSREP_ERROR ("Failed to parse options.");
return err;
}
/* At this point we have updated default values in our option list to
what has been specified on the command line / my.cnf */
long long slave_threads;
err = get_long_long (opts[WSREP_SLAVE_THREADS], &slave_threads, 10);
if (err) return err;
int rcode = 0;
if (slave_threads > 1)
/* Need to check AUTOINC_LOCK_MODE and LOCKS_UNSAFE_FOR_BINLOG */
{
long long autoinc_lock_mode;
err = get_long_long (opts[AUTOINC_LOCK_MODE], &autoinc_lock_mode, 10);
if (err) return err;
bool locks_unsafe_for_binlog;
err = get_bool (opts[LOCKS_UNSAFE_FOR_BINLOG],&locks_unsafe_for_binlog);
if (err) return err;
if (autoinc_lock_mode != 2)
{
WSREP_ERROR ("Parallel applying (wsrep_slave_threads > 1) requires"
" innodb_autoinc_lock_mode = 2.");
rcode = EINVAL;
}
}
long long query_cache_size, query_cache_type;
if ((err = get_long_long (opts[QUERY_CACHE_SIZE], &query_cache_size, 10)))
return err;
if ((err = get_long_long (opts[QUERY_CACHE_TYPE], &query_cache_type, 10)))
return err;
if (0 != query_cache_size && 0 != query_cache_type)
{
WSREP_ERROR ("Query cache is not supported (size=%lld type=%lld)",
query_cache_size, query_cache_type);
rcode = EINVAL;
}
bool locked_in_memory;
err = get_bool (opts[LOCKED_IN_MEMORY], &locked_in_memory);
if (err) { WSREP_ERROR("get_bool error: %s", strerror(err)); return err; }
if (locked_in_memory)
{
WSREP_ERROR ("Memory locking is not supported (locked_in_memory=%s)",
locked_in_memory ? "ON" : "OFF");
rcode = EINVAL;
}
if (!strcasecmp(opts[WSREP_SST_METHOD].value,"mysqldump"))
{
if (!strcasecmp(opts[BIND_ADDRESS].value, "127.0.0.1") ||
!strcasecmp(opts[BIND_ADDRESS].value, "localhost"))
{
WSREP_ERROR ("wsrep_sst_method is set to 'mysqldump' yet "
"mysqld bind_address is set to '%s', which makes it "
"impossible to receive state transfer from another "
"node, since mysqld won't accept such connections. "
"If you wish to use mysqldump state transfer method, "
"set bind_address to allow mysql client connections "
"from other cluster members (e.g. 0.0.0.0).",
opts[BIND_ADDRESS].value);
rcode = EINVAL;
}
}
else
{
// non-mysqldump SST requires wsrep_cluster_address on startup
if (strlen(opts[WSREP_CLUSTER_ADDRESS].value) == 0)
{
WSREP_ERROR ("%s SST method requires wsrep_cluster_address to be "
"configured on startup.",opts[WSREP_SST_METHOD].value);
rcode = EINVAL;
}
}
if (strcasecmp(opts[WSREP_SST_RECEIVE_ADDRESS].value, "AUTO"))
{
if (!strncasecmp(opts[WSREP_SST_RECEIVE_ADDRESS].value,
"127.0.0.1", strlen("127.0.0.1")) ||
!strncasecmp(opts[WSREP_SST_RECEIVE_ADDRESS].value,
"localhost", strlen("localhost")))
{
WSREP_WARN ("wsrep_sst_receive_address is set to '%s' which "
"makes it impossible for another host to reach this "
"one. Please set it to the address which this node "
"can be connected at by other cluster members.",
opts[WSREP_SST_RECEIVE_ADDRESS].value);
// rcode = EINVAL;
}
}
if (strcasecmp(opts[WSREP_PROVIDER].value, "none"))
{
if (strcasecmp(opts[BINLOG_FORMAT].value, "ROW"))
{
WSREP_ERROR ("Only binlog_format = 'ROW' is currently supported. "
"Configured value: '%s'. Please adjust your "
"configuration.", opts[BINLOG_FORMAT].value);
rcode = EINVAL;
}
}
return rcode;
}
int
wsrep_check_opts (int const argc, char* const* const argv)
{
return check_opts (argc, argv, opts);
}

449
sql/wsrep_hton.cc
View File

@ -1,449 +0,0 @@
/* Copyright 2008 Codership Oy <http://www.codership.com>
This program 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; version 2 of the License.
This program 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 this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include <mysqld.h>
#include "sql_base.h"
#include "rpl_filter.h"
#include <sql_class.h>
#include "wsrep_mysqld.h"
#include "wsrep_priv.h"
#include <cstdio>
#include <cstdlib>
extern handlerton *binlog_hton;
extern int binlog_close_connection(handlerton *hton, THD *thd);
extern ulonglong thd_to_trx_id(THD *thd);
extern "C" int thd_binlog_format(const MYSQL_THD thd);
// todo: share interface with ha_innodb.c
enum wsrep_trx_status wsrep_run_wsrep_commit(THD *thd, handlerton *hton, bool all);
/*
a post-commit cleanup on behalf of wsrep. Can't be a part of hton struct.
Is called by THD::transactions.cleanup()
*/
void wsrep_cleanup_transaction(THD *thd)
{
if (thd->thread_id == 0) return;
if (thd->wsrep_exec_mode == LOCAL_COMMIT)
{
if (thd->variables.wsrep_on &&
thd->wsrep_conflict_state != MUST_REPLAY)
{
if (thd->wsrep_seqno_changed)
{
if (wsrep->post_commit(wsrep, &thd->wsrep_trx_handle))
{
DBUG_PRINT("wsrep", ("set committed fail"));
WSREP_WARN("set committed fail: %llu %d",
(long long)thd->real_id, thd->stmt_da->status());
}
}
//else
//WSREP_DEBUG("no trx handle for %s", thd->query());
thd_binlog_trx_reset(thd);
thd->wsrep_seqno_changed = false;
}
thd->wsrep_exec_mode= LOCAL_STATE;
}
}
/*
wsrep hton
*/
handlerton *wsrep_hton;
void wsrep_register_hton(THD* thd, bool all)
{
THD_TRANS *trans=all ? &thd->transaction.all : &thd->transaction.stmt;
for (Ha_trx_info *i= trans->ha_list; WSREP(thd) && i; i = i->next())
{
if (i->ht()->db_type == DB_TYPE_INNODB)
{
trans_register_ha(thd, all, wsrep_hton);
thd->ha_data[wsrep_hton->slot].ha_info[all].set_trx_read_write();
break;
}
}
}
/*
wsrep exploits binlog's caches even if binlogging itself is not
activated. In such case connection close needs calling
actual binlog's method.
Todo: split binlog hton from its caches to use ones by wsrep
without referring to binlog's stuff.
*/
static int
wsrep_close_connection(handlerton* hton, THD* thd)
{
DBUG_ENTER("wsrep_close_connection");
if (thd_get_ha_data(thd, binlog_hton) != NULL)
binlog_hton->close_connection (binlog_hton, thd);
DBUG_RETURN(0);
}
/*
prepare/wsrep_run_wsrep_commit can fail in two ways
- certification test or an equivalent. As a result,
the current transaction just rolls back
Error codes:
WSREP_TRX_ROLLBACK, WSREP_TRX_ERROR
- a post-certification failure makes this server unable to
commit its own WS and therefore the server must abort
*/
static int wsrep_prepare(handlerton *hton, THD *thd, bool all)
{
#ifndef DBUG_OFF
//wsrep_seqno_t old = thd->wsrep_trx_seqno;
#endif
DBUG_ENTER("wsrep_prepare");
if ((all ||
!thd_test_options(thd, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) &&
(thd->variables.wsrep_on && !wsrep_trans_cache_is_empty(thd)))
{
switch (wsrep_run_wsrep_commit(thd, hton, all))
{
case WSREP_TRX_OK:
// DBUG_ASSERT(thd->wsrep_trx_seqno > old ||
// thd->wsrep_exec_mode == REPL_RECV ||
// thd->wsrep_exec_mode == TOTAL_ORDER);
break;
case WSREP_TRX_ROLLBACK:
case WSREP_TRX_ERROR:
DBUG_RETURN(1);
}
}
DBUG_RETURN(0);
}
static int wsrep_savepoint_set(handlerton *hton, THD *thd, void *sv)
{
if (!wsrep_emulate_bin_log) return 0;
int rcode = binlog_hton->savepoint_set(binlog_hton, thd, sv);
return rcode;
}
static int wsrep_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
{
if (!wsrep_emulate_bin_log) return 0;
int rcode = binlog_hton->savepoint_rollback(binlog_hton, thd, sv);
return rcode;
}
static int wsrep_rollback(handlerton *hton, THD *thd, bool all)
{
DBUG_ENTER("wsrep_rollback");
mysql_mutex_lock(&thd->LOCK_wsrep_thd);
if ((all || !thd_test_options(thd, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) &&
(thd->variables.wsrep_on && thd->wsrep_conflict_state != MUST_REPLAY))
{
if (wsrep->post_rollback(wsrep, &thd->wsrep_trx_handle))
{
DBUG_PRINT("wsrep", ("setting rollback fail"));
WSREP_ERROR("settting rollback fail: thd: %llu SQL: %s",
(long long)thd->real_id, thd->query());
}
}
int rcode = 0;
if (!wsrep_emulate_bin_log)
{
if (all) thd_binlog_trx_reset(thd);
}
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
DBUG_RETURN(rcode);
}
int wsrep_commit(handlerton *hton, THD *thd, bool all)
{
DBUG_ENTER("wsrep_commit");
DBUG_RETURN(0);
}
extern Rpl_filter* binlog_filter;
extern my_bool opt_log_slave_updates;
enum wsrep_trx_status
wsrep_run_wsrep_commit(
THD *thd, handlerton *hton, bool all)
{
int rcode = -1;
uint data_len = 0;
uchar *rbr_data = NULL;
IO_CACHE *cache;
int replay_round= 0;
if (thd->stmt_da->is_error()) {
WSREP_ERROR("commit issue, error: %d %s",
thd->stmt_da->sql_errno(), thd->stmt_da->message());
}
DBUG_ENTER("wsrep_run_wsrep_commit");
if (thd->slave_thread && !opt_log_slave_updates) {
DBUG_RETURN(WSREP_TRX_OK);
}
if (thd->wsrep_exec_mode == REPL_RECV) {
mysql_mutex_lock(&thd->LOCK_wsrep_thd);
if (thd->wsrep_conflict_state == MUST_ABORT) {
if (wsrep_debug)
WSREP_INFO("WSREP: must abort for BF");
DBUG_PRINT("wsrep", ("BF apply commit fail"));
thd->wsrep_conflict_state = NO_CONFLICT;
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
//
// TODO: test all calls of the rollback.
// rollback must happen automagically innobase_rollback(hton, thd, 1);
//
DBUG_RETURN(WSREP_TRX_ERROR);
}
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
}
if (thd->wsrep_exec_mode != LOCAL_STATE) {
DBUG_RETURN(WSREP_TRX_OK);
}
if (thd->wsrep_consistency_check == CONSISTENCY_CHECK_RUNNING) {
WSREP_DEBUG("commit for consistency check: %s", thd->query());
DBUG_RETURN(WSREP_TRX_OK);
}
DBUG_PRINT("wsrep", ("replicating commit"));
mysql_mutex_lock(&thd->LOCK_wsrep_thd);
if (thd->wsrep_conflict_state == MUST_ABORT) {
DBUG_PRINT("wsrep", ("replicate commit fail"));
thd->wsrep_conflict_state = ABORTED;
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
if (wsrep_debug) {
WSREP_INFO("innobase_commit, abort %s",
(thd->query()) ? thd->query() : "void");
}
DBUG_RETURN(WSREP_TRX_ROLLBACK);
}
mysql_mutex_lock(&LOCK_wsrep_replaying);
while (wsrep_replaying > 0 &&
thd->wsrep_conflict_state == NO_CONFLICT &&
thd->killed == NOT_KILLED &&
!shutdown_in_progress)
{
mysql_mutex_unlock(&LOCK_wsrep_replaying);
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
mysql_mutex_lock(&thd->mysys_var->mutex);
thd_proc_info(thd, "wsrep waiting on replaying");
thd->mysys_var->current_mutex= &LOCK_wsrep_replaying;
thd->mysys_var->current_cond= &COND_wsrep_replaying;
mysql_mutex_unlock(&thd->mysys_var->mutex);
mysql_mutex_lock(&LOCK_wsrep_replaying);
// Using timedwait is a hack to avoid deadlock in case if BF victim
// misses the signal.
struct timespec wtime = {0, 1000000};
mysql_cond_timedwait(&COND_wsrep_replaying, &LOCK_wsrep_replaying,
&wtime);
if (replay_round++ % 100000 == 0)
WSREP_DEBUG("commit waiting for replaying: replayers %d, thd: (%lu) conflict: %d (round: %d)",
wsrep_replaying, thd->thread_id, thd->wsrep_conflict_state, replay_round);
mysql_mutex_unlock(&LOCK_wsrep_replaying);
mysql_mutex_lock(&thd->mysys_var->mutex);
thd->mysys_var->current_mutex= 0;
thd->mysys_var->current_cond= 0;
mysql_mutex_unlock(&thd->mysys_var->mutex);
mysql_mutex_lock(&thd->LOCK_wsrep_thd);
mysql_mutex_lock(&LOCK_wsrep_replaying);
}
mysql_mutex_unlock(&LOCK_wsrep_replaying);
if (thd->wsrep_conflict_state == MUST_ABORT) {
DBUG_PRINT("wsrep", ("replicate commit fail"));
thd->wsrep_conflict_state = ABORTED;
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
WSREP_DEBUG("innobase_commit abort after replaying wait %s",
(thd->query()) ? thd->query() : "void");
DBUG_RETURN(WSREP_TRX_ROLLBACK);
}
thd->wsrep_query_state = QUERY_COMMITTING;
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
cache = get_trans_log(thd);
rcode = 0;
if (cache) {
thd->binlog_flush_pending_rows_event(true);
rcode = wsrep_write_cache(cache, &rbr_data, &data_len);
if (rcode) {
WSREP_ERROR("rbr write fail, data_len: %d, %d", data_len, rcode);
if (data_len) my_free(rbr_data);
DBUG_RETURN(WSREP_TRX_ROLLBACK);
}
}
if (data_len == 0)
{
mysql_mutex_lock(&thd->LOCK_wsrep_thd);
thd->wsrep_exec_mode = LOCAL_COMMIT;
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
if (thd->stmt_da->is_ok() &&
thd->stmt_da->affected_rows() > 0 &&
!binlog_filter->is_on())
{
WSREP_DEBUG("empty rbr buffer, query: %s, "
"affected rows: %llu, "
"changed tables: %d, "
"sql_log_bin: %d, "
"wsrep status (%d %d %d)",
thd->query(), thd->stmt_da->affected_rows(),
stmt_has_updated_trans_table(thd), thd->variables.sql_log_bin,
thd->wsrep_exec_mode, thd->wsrep_query_state,
thd->wsrep_conflict_state);
}
else
{
WSREP_DEBUG("empty rbr buffer, query: %s", thd->query());
}
DBUG_RETURN(WSREP_TRX_OK);
}
if (!rcode) {
rcode = wsrep->pre_commit(
wsrep,
(wsrep_conn_id_t)thd->thread_id,
&thd->wsrep_trx_handle,
rbr_data,
data_len,
(thd->wsrep_PA_safe) ? WSREP_FLAG_PA_SAFE : 0ULL,
&thd->wsrep_trx_seqno);
if (rcode == WSREP_TRX_MISSING) {
rcode = WSREP_OK;
} else if (rcode == WSREP_BF_ABORT) {
mysql_mutex_lock(&thd->LOCK_wsrep_thd);
thd->wsrep_conflict_state = MUST_REPLAY;
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
mysql_mutex_lock(&LOCK_wsrep_replaying);
wsrep_replaying++;
WSREP_DEBUG("replaying increased: %d, thd: %lu",
wsrep_replaying, thd->thread_id);
mysql_mutex_unlock(&LOCK_wsrep_replaying);
}
thd->wsrep_seqno_changed = true;
} else {
WSREP_ERROR("I/O error reading from thd's binlog iocache: "
"errno=%d, io cache code=%d", my_errno, cache->error);
if (data_len) my_free(rbr_data);
DBUG_ASSERT(0); // failure like this can not normally happen
DBUG_RETURN(WSREP_TRX_ERROR);
}
if (data_len) {
my_free(rbr_data);
}
mysql_mutex_lock(&thd->LOCK_wsrep_thd);
switch(rcode) {
case 0:
thd->wsrep_exec_mode = LOCAL_COMMIT;
/* Override XID iff it was generated by mysql */
if (thd->transaction.xid_state.xid.get_my_xid())
{
wsrep_xid_init(&thd->transaction.xid_state.xid,
wsrep_cluster_uuid(),
thd->wsrep_trx_seqno);
}
DBUG_PRINT("wsrep", ("replicating commit success"));
break;
case WSREP_TRX_FAIL:
case WSREP_BF_ABORT:
WSREP_DEBUG("commit failed for reason: %d", rcode);
DBUG_PRINT("wsrep", ("replicating commit fail"));
thd->wsrep_query_state= QUERY_EXEC;
if (thd->wsrep_conflict_state == MUST_ABORT) {
thd->wsrep_conflict_state= ABORTED;
}
else
{
WSREP_DEBUG("conflict state: %d", thd->wsrep_conflict_state);
if (thd->wsrep_conflict_state == NO_CONFLICT)
{
thd->wsrep_conflict_state = CERT_FAILURE;
WSREP_LOG_CONFLICT(NULL, thd, FALSE);
}
}
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
DBUG_RETURN(WSREP_TRX_ROLLBACK);
case WSREP_CONN_FAIL:
WSREP_ERROR("connection failure");
DBUG_RETURN(WSREP_TRX_ERROR);
default:
WSREP_ERROR("unknown connection failure");
DBUG_RETURN(WSREP_TRX_ERROR);
}
thd->wsrep_query_state= QUERY_EXEC;
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
DBUG_RETURN(WSREP_TRX_OK);
}
static int wsrep_hton_init(void *p)
{
wsrep_hton= (handlerton *)p;
//wsrep_hton->state=opt_bin_log ? SHOW_OPTION_YES : SHOW_OPTION_NO;
wsrep_hton->state= SHOW_OPTION_YES;
wsrep_hton->db_type=DB_TYPE_WSREP;
wsrep_hton->savepoint_offset= sizeof(my_off_t);
wsrep_hton->close_connection= wsrep_close_connection;
wsrep_hton->savepoint_set= wsrep_savepoint_set;
wsrep_hton->savepoint_rollback= wsrep_savepoint_rollback;
wsrep_hton->commit= wsrep_commit;
wsrep_hton->rollback= wsrep_rollback;
wsrep_hton->prepare= wsrep_prepare;
wsrep_hton->flags= HTON_NOT_USER_SELECTABLE | HTON_HIDDEN; // todo: fix flags
wsrep_hton->slot= 0;
return 0;
}
struct st_mysql_storage_engine wsrep_storage_engine=
{ MYSQL_HANDLERTON_INTERFACE_VERSION };
mysql_declare_plugin(wsrep)
{
MYSQL_STORAGE_ENGINE_PLUGIN,
&wsrep_storage_engine,
"wsrep",
"Codership Oy",
"A pseudo storage engine to represent transactions in multi-master synchornous replication",
PLUGIN_LICENSE_GPL,
wsrep_hton_init, /* Plugin Init */
NULL, /* Plugin Deinit */
0x0100 /* 1.0 */,
NULL, /* status variables */
NULL, /* system variables */
NULL, /* config options */
0, /* flags */
}
mysql_declare_plugin_end;

1311
sql/wsrep_mysqld.cc
View File

File diff suppressed because it is too large Load Diff

383
sql/wsrep_mysqld.h
View File

@ -1,383 +0,0 @@
/* Copyright 2008-2012 Codership Oy <http://www.codership.com>
This program 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; version 2 of the License.
This program 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 this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#ifndef WSREP_MYSQLD_H
#define WSREP_MYSQLD_H
#include "mysqld.h"
typedef struct st_mysql_show_var SHOW_VAR;
#include <sql_priv.h>
#include "../wsrep/wsrep_api.h"
class set_var;
class THD;
#ifdef WITH_WSREP
#include "../wsrep/wsrep_api.h"
//#include "wsrep_mysqld.h"
enum wsrep_exec_mode {
LOCAL_STATE,
REPL_RECV,
TOTAL_ORDER,
LOCAL_COMMIT,
};
enum wsrep_query_state {
QUERY_IDLE,
QUERY_EXEC,
QUERY_COMMITTING,
QUERY_EXITING,
QUERY_ROLLINGBACK,
};
enum wsrep_conflict_state {
NO_CONFLICT,
MUST_ABORT,
ABORTING,
ABORTED,
MUST_REPLAY,
REPLAYING,
RETRY_AUTOCOMMIT,
CERT_FAILURE,
};
enum wsrep_consistency_check_mode {
NO_CONSISTENCY_CHECK,
CONSISTENCY_CHECK_DECLARED,
CONSISTENCY_CHECK_RUNNING,
};
#endif
// Global wsrep parameters
extern wsrep_t* wsrep;
// MySQL wsrep options
extern const char* wsrep_provider;
extern const char* wsrep_provider_options;
extern const char* wsrep_cluster_name;
extern const char* wsrep_cluster_address;
extern const char* wsrep_node_name;
extern const char* wsrep_node_address;
extern const char* wsrep_node_incoming_address;
extern const char* wsrep_data_home_dir;
extern const char* wsrep_dbug_option;
extern long wsrep_slave_threads;
extern my_bool wsrep_debug;
extern my_bool wsrep_convert_LOCK_to_trx;
extern ulong wsrep_retry_autocommit;
extern my_bool wsrep_auto_increment_control;
extern my_bool wsrep_drupal_282555_workaround;
extern my_bool wsrep_incremental_data_collection;
extern const char* wsrep_sst_method;
extern const char* wsrep_sst_receive_address;
extern char* wsrep_sst_auth;
extern const char* wsrep_sst_donor;
extern my_bool wsrep_sst_donor_rejects_queries;
extern const char* wsrep_start_position;
extern long long wsrep_max_ws_size;
extern long wsrep_max_ws_rows;
extern const char* wsrep_notify_cmd;
extern my_bool wsrep_certify_nonPK;
extern long wsrep_max_protocol_version;
extern long wsrep_protocol_version;
extern ulong wsrep_forced_binlog_format;
extern ulong wsrep_OSU_method_options;
extern my_bool wsrep_recovery;
extern my_bool wsrep_replicate_myisam;
extern my_bool wsrep_log_conflicts;
extern ulong wsrep_mysql_replication_bundle;
enum enum_wsrep_OSU_method { WSREP_OSU_TOI, WSREP_OSU_RSU };
// MySQL status variables
extern my_bool wsrep_connected;
extern my_bool wsrep_ready;
extern const char* wsrep_cluster_state_uuid;
extern long long wsrep_cluster_conf_id;
extern const char* wsrep_cluster_status;
extern long wsrep_cluster_size;
extern long wsrep_local_index;
extern const char* wsrep_provider_name;
extern const char* wsrep_provider_version;
extern const char* wsrep_provider_vendor;
extern int wsrep_show_status(THD *thd, SHOW_VAR *var, char *buff);
extern void wsrep_free_status(THD *thd);
#define WSREP_SST_ADDRESS_AUTO "AUTO"
// MySQL variables funcs
#define CHECK_ARGS (sys_var *self, THD* thd, set_var *var)
#define UPDATE_ARGS (sys_var *self, THD* thd, enum_var_type type)
#define DEFAULT_ARGS (THD* thd, enum_var_type var_type)
#define INIT_ARGS (const char* opt)
extern int wsrep_init_vars();
extern bool wsrep_on_update UPDATE_ARGS;
extern void wsrep_causal_reads_update UPDATE_ARGS;
extern bool wsrep_start_position_check CHECK_ARGS;
extern bool wsrep_start_position_update UPDATE_ARGS;
extern void wsrep_start_position_init INIT_ARGS;
extern bool wsrep_provider_check CHECK_ARGS;
extern bool wsrep_provider_update UPDATE_ARGS;
extern void wsrep_provider_init INIT_ARGS;
extern bool wsrep_provider_options_check CHECK_ARGS;
extern bool wsrep_provider_options_update UPDATE_ARGS;
extern void wsrep_provider_options_init INIT_ARGS;
extern bool wsrep_cluster_address_check CHECK_ARGS;
extern bool wsrep_cluster_address_update UPDATE_ARGS;
extern void wsrep_cluster_address_init INIT_ARGS;
extern bool wsrep_cluster_name_check CHECK_ARGS;
extern bool wsrep_cluster_name_update UPDATE_ARGS;
extern bool wsrep_node_name_check CHECK_ARGS;
extern bool wsrep_node_name_update UPDATE_ARGS;
extern bool wsrep_node_address_check CHECK_ARGS;
extern bool wsrep_node_address_update UPDATE_ARGS;
extern void wsrep_node_address_init INIT_ARGS;
extern bool wsrep_sst_method_check CHECK_ARGS;
extern bool wsrep_sst_method_update UPDATE_ARGS;
extern void wsrep_sst_method_init INIT_ARGS;
extern bool wsrep_sst_receive_address_check CHECK_ARGS;
extern bool wsrep_sst_receive_address_update UPDATE_ARGS;
extern bool wsrep_sst_auth_check CHECK_ARGS;
extern bool wsrep_sst_auth_update UPDATE_ARGS;
extern void wsrep_sst_auth_init INIT_ARGS;
extern bool wsrep_sst_donor_check CHECK_ARGS;
extern bool wsrep_sst_donor_update UPDATE_ARGS;
extern bool wsrep_slave_threads_check CHECK_ARGS;
extern bool wsrep_slave_threads_update UPDATE_ARGS;
extern bool wsrep_before_SE(); // initialize wsrep before storage
// engines (true) or after (false)
extern int wsrep_init();
extern void wsrep_deinit();
extern void wsrep_recover();
extern "C" enum wsrep_exec_mode wsrep_thd_exec_mode(THD *thd);
extern "C" enum wsrep_conflict_state wsrep_thd_conflict_state(THD *thd);
extern "C" enum wsrep_query_state wsrep_thd_query_state(THD *thd);
extern "C" const char * wsrep_thd_exec_mode_str(THD *thd);
extern "C" const char * wsrep_thd_conflict_state_str(THD *thd);
extern "C" const char * wsrep_thd_query_state_str(THD *thd);
extern "C" wsrep_trx_handle_t* wsrep_thd_trx_handle(THD *thd);
extern "C" void wsrep_thd_set_exec_mode(THD *thd, enum wsrep_exec_mode mode);
extern "C" void wsrep_thd_set_query_state(
THD *thd, enum wsrep_query_state state);
extern "C" void wsrep_thd_set_conflict_state(
THD *thd, enum wsrep_conflict_state state);
extern "C" void wsrep_thd_set_trx_to_replay(THD *thd, uint64 trx_id);
extern "C"void wsrep_thd_LOCK(THD *thd);
extern "C"void wsrep_thd_UNLOCK(THD *thd);
extern "C" uint32 wsrep_thd_wsrep_rand(THD *thd);
extern "C" time_t wsrep_thd_query_start(THD *thd);
extern "C" my_thread_id wsrep_thd_thread_id(THD *thd);
extern "C" int64_t wsrep_thd_trx_seqno(THD *thd);
extern "C" query_id_t wsrep_thd_query_id(THD *thd);
extern "C" char * wsrep_thd_query(THD *thd);
extern "C" query_id_t wsrep_thd_wsrep_last_query_id(THD *thd);
extern "C" void wsrep_thd_set_wsrep_last_query_id(THD *thd, query_id_t id);
extern "C" void wsrep_thd_awake(THD *thd, my_bool signal);
/* wsrep initialization sequence at startup
* @param first wsrep_before_SE() value */
extern void wsrep_init_startup(bool before);
extern void wsrep_close_client_connections(my_bool wait_to_end);
extern int wsrep_wait_committing_connections_close(int wait_time);
extern void wsrep_close_applier(THD *thd);
extern void wsrep_wait_appliers_close(THD *thd);
extern void wsrep_close_applier_threads(int count);
extern void wsrep_create_appliers(long threads = wsrep_slave_threads);
extern void wsrep_create_rollbacker();
extern void wsrep_kill_mysql(THD *thd);
/* new defines */
extern void wsrep_stop_replication(THD *thd);
extern bool wsrep_start_replication();
extern bool wsrep_causal_wait(THD* thd);
extern int wsrep_check_opts (int argc, char* const* argv);
extern void wsrep_prepend_PATH (const char* path);
/* Other global variables */
extern wsrep_seqno_t wsrep_locked_seqno;
#define WSREP_ON \
(global_system_variables.wsrep_on)
#define WSREP(thd) \
(WSREP_ON && (thd && thd->variables.wsrep_on))
#define WSREP_CLIENT(thd) \
(WSREP(thd) && thd->wsrep_client_thread)
#define WSREP_EMULATE_BINLOG(thd) \
(WSREP(thd) && wsrep_emulate_bin_log)
// MySQL logging functions don't seem to understand long long length modifer.
// This is a workaround. It also prefixes all messages with "WSREP"
#define WSREP_LOG(fun, ...) \
{ \
char msg[1024] = {'\0'}; \
snprintf(msg, sizeof(msg) - 1, ## __VA_ARGS__); \
fun("WSREP: %s", msg); \
}
#define WSREP_DEBUG(...) \
if (wsrep_debug) WSREP_LOG(sql_print_information, ##__VA_ARGS__)
#define WSREP_INFO(...) WSREP_LOG(sql_print_information, ##__VA_ARGS__)
#define WSREP_WARN(...) WSREP_LOG(sql_print_warning, ##__VA_ARGS__)
#define WSREP_ERROR(...) WSREP_LOG(sql_print_error, ##__VA_ARGS__)
#define WSREP_LOG_CONFLICT_THD(thd, role) \
WSREP_LOG(sql_print_information, \
"%s: \n " \
" THD: %lu, mode: %s, state: %s, conflict: %s, seqno: %lld\n " \
" SQL: %s", \
role, wsrep_thd_thread_id(thd), wsrep_thd_exec_mode_str(thd), \
wsrep_thd_query_state_str(thd), \
wsrep_thd_conflict_state_str(thd), (long long)wsrep_thd_trx_seqno(thd), \
wsrep_thd_query(thd) \
);
#define WSREP_LOG_CONFLICT(bf_thd, victim_thd, bf_abort) \
if (wsrep_debug || wsrep_log_conflicts) \
{ \
WSREP_LOG(sql_print_information, "cluster conflict due to %s for threads:",\
(bf_abort) ? "high priority abort" : "certification failure" \
); \
if (bf_thd) WSREP_LOG_CONFLICT_THD(bf_thd, "Winning thread"); \
if (victim_thd) WSREP_LOG_CONFLICT_THD(victim_thd, "Victim thread"); \
}
/*! Synchronizes applier thread start with init thread */
extern void wsrep_sst_grab();
/*! Init thread waits for SST completion */
extern bool wsrep_sst_wait();
/*! Signals wsrep that initialization is complete, writesets can be applied */
extern void wsrep_sst_continue();
extern void wsrep_SE_init_grab(); /*! grab init critical section */
extern void wsrep_SE_init_wait(); /*! wait for SE init to complete */
extern void wsrep_SE_init_done(); /*! signal that SE init is complte */
extern void wsrep_SE_initialized(); /*! mark SE initialization complete */
extern void wsrep_ready_wait();
enum wsrep_trx_status {
WSREP_TRX_OK,
WSREP_TRX_ROLLBACK,
WSREP_TRX_ERROR,
};
extern enum wsrep_trx_status
wsrep_run_wsrep_commit(THD *thd, handlerton *hton, bool all);
class Ha_trx_info;
struct THD_TRANS;
void wsrep_register_hton(THD* thd, bool all);
void wsrep_replication_process(THD *thd);
void wsrep_rollback_process(THD *thd);
void wsrep_brute_force_killer(THD *thd);
int wsrep_hire_brute_force_killer(THD *thd, uint64_t trx_id);
extern "C" bool wsrep_consistency_check(void *thd_ptr);
extern "C" int wsrep_thd_is_brute_force(void *thd_ptr);
extern "C" int wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr,
my_bool signal);
extern "C" int wsrep_thd_in_locking_session(void *thd_ptr);
void *wsrep_prepare_bf_thd(THD *thd);
void wsrep_return_from_bf_mode(void *shadow, THD *thd);
/* this is visible for client build so that innodb plugin gets this */
typedef struct wsrep_aborting_thd {
struct wsrep_aborting_thd *next;
THD *aborting_thd;
} *wsrep_aborting_thd_t;
extern mysql_mutex_t LOCK_wsrep_ready;
extern mysql_cond_t COND_wsrep_ready;
extern mysql_mutex_t LOCK_wsrep_sst;
extern mysql_cond_t COND_wsrep_sst;
extern mysql_mutex_t LOCK_wsrep_sst_init;
extern mysql_cond_t COND_wsrep_sst_init;
extern mysql_mutex_t LOCK_wsrep_rollback;
extern mysql_cond_t COND_wsrep_rollback;
extern int wsrep_replaying;
extern mysql_mutex_t LOCK_wsrep_replaying;
extern mysql_cond_t COND_wsrep_replaying;
extern wsrep_aborting_thd_t wsrep_aborting_thd;
extern MYSQL_PLUGIN_IMPORT my_bool wsrep_debug;
extern my_bool wsrep_convert_LOCK_to_trx;
extern ulong wsrep_retry_autocommit;
extern my_bool wsrep_emulate_bin_log;
extern my_bool wsrep_auto_increment_control;
extern my_bool wsrep_drupal_282555_workaround;
extern long long wsrep_max_ws_size;
extern long wsrep_max_ws_rows;
extern int wsrep_to_isolation;
extern my_bool wsrep_certify_nonPK;
extern mysql_mutex_t LOCK_wsrep_slave_threads;
extern PSI_mutex_key key_LOCK_wsrep_ready;
extern PSI_mutex_key key_COND_wsrep_ready;
extern PSI_mutex_key key_LOCK_wsrep_sst;
extern PSI_cond_key key_COND_wsrep_sst;
extern PSI_mutex_key key_LOCK_wsrep_sst_init;
extern PSI_cond_key key_COND_wsrep_sst_init;
extern PSI_mutex_key key_LOCK_wsrep_sst_thread;
extern PSI_cond_key key_COND_wsrep_sst_thread;
extern PSI_mutex_key key_LOCK_wsrep_rollback;
extern PSI_cond_key key_COND_wsrep_rollback;
extern PSI_mutex_key key_LOCK_wsrep_replaying;
extern PSI_cond_key key_COND_wsrep_replaying;
extern PSI_mutex_key key_LOCK_wsrep_slave_threads;
struct TABLE_LIST;
int wsrep_to_isolation_begin(THD *thd, char *db_, char *table_,
const TABLE_LIST* table_list);
void wsrep_to_isolation_end(THD *thd);
void wsrep_prepare_bf_thd(THD *thd, struct wsrep_thd_shadow*);
void wsrep_return_from_bf_mode(THD *thd, struct wsrep_thd_shadow*);
int wsrep_to_buf_helper(
THD* thd, const char *query, uint query_len, uchar** buf, uint* buf_len);
int wsrep_create_sp(THD *thd, uchar** buf, uint* buf_len);
int wsrep_create_trigger_query(THD *thd, uchar** buf, uint* buf_len);
int wsrep_create_event_query(THD *thd, uchar** buf, uint* buf_len);
const wsrep_uuid_t* wsrep_cluster_uuid();
struct xid_t;
void wsrep_set_SE_checkpoint(xid_t*);
void wsrep_xid_init(xid_t*, const wsrep_uuid_t*, wsrep_seqno_t);
const wsrep_uuid_t* wsrep_xid_uuid(const xid_t*);
wsrep_seqno_t wsrep_xid_seqno(const xid_t*);
extern "C" int wsrep_is_wsrep_xid(const void* xid);
#endif /* WSREP_MYSQLD_H */

107
sql/wsrep_notify.cc
View File

@ -1,107 +0,0 @@
/* Copyright 2010 Codership Oy <http://www.codership.com>
This program 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; version 2 of the License.
This program 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 this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include <mysqld.h>
#include "wsrep_priv.h"
const char* wsrep_notify_cmd="";
static const char* _status_str(wsrep_member_status_t status)
{
switch (status)
{
case WSREP_MEMBER_UNDEFINED: return "Undefined";
case WSREP_MEMBER_JOINER: return "Joiner";
case WSREP_MEMBER_DONOR: return "Donor";
case WSREP_MEMBER_JOINED: return "Joined";
case WSREP_MEMBER_SYNCED: return "Synced";
default: return "Error(?)";
}
}
void wsrep_notify_status (wsrep_member_status_t status,
const wsrep_view_info_t* view)
{
if (!wsrep_notify_cmd || 0 == strlen(wsrep_notify_cmd))
{
WSREP_INFO("wsrep_notify_cmd is not defined, skipping notification.");
return;
}
char cmd_buf[1 << 16]; // this can be long
long cmd_len = sizeof(cmd_buf) - 1;
char* cmd_ptr = cmd_buf;
long cmd_off = 0;
cmd_off += snprintf (cmd_ptr + cmd_off, cmd_len - cmd_off, "%s",
wsrep_notify_cmd);
if (status >= WSREP_MEMBER_UNDEFINED && status < WSREP_MEMBER_ERROR)
{
cmd_off += snprintf (cmd_ptr + cmd_off, cmd_len - cmd_off, " --status %s",
_status_str(status));
}
else
{
/* here we preserve provider error codes */
cmd_off += snprintf (cmd_ptr + cmd_off, cmd_len - cmd_off,
" --status 'Error(%d)'", status);
}
if (0 != view)
{
char uuid_str[40];
wsrep_uuid_print (&view->uuid, uuid_str, sizeof(uuid_str));
cmd_off += snprintf (cmd_ptr + cmd_off, cmd_len - cmd_off,
" --uuid %s", uuid_str);
cmd_off += snprintf (cmd_ptr + cmd_off, cmd_len - cmd_off,
" --primary %s", view->view >= 0 ? "yes" : "no");
cmd_off += snprintf (cmd_ptr + cmd_off, cmd_len - cmd_off,
" --index %d", view->my_idx);
cmd_off += snprintf (cmd_ptr + cmd_off, cmd_len - cmd_off, " --members");
for (int i = 0; i < view->memb_num; i++)
{
wsrep_uuid_print (&view->members[i].id, uuid_str, sizeof(uuid_str));
cmd_off += snprintf (cmd_ptr + cmd_off, cmd_len - cmd_off,
"%c%s/%s/%s", i > 0 ? ',' : ' ',
uuid_str, view->members[i].name,
view->members[i].incoming);
}
}
if (cmd_off == cmd_len)
{
WSREP_ERROR("Notification buffer too short (%ld). Aborting notification.",
cmd_len);
return;
}
wsp::process p(cmd_ptr, "r");
p.wait();
int err = p.error();
if (err)
{
WSREP_ERROR("Notification command failed: %d (%s): \"%s\"",
err, strerror(err), cmd_ptr);
}
}

233
sql/wsrep_priv.h
View File

@ -1,233 +0,0 @@
/* Copyright 2010 Codership Oy <http://www.codership.com>
This program 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; version 2 of the License.
This program 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 this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*/
//! @file declares symbols private to wsrep integration layer
#ifndef WSREP_PRIV_H
#define WSREP_PRIV_H
#include "wsrep_mysqld.h"
#include "../wsrep/wsrep_api.h"
#include <log.h>
#include <pthread.h>
#include <cstdio>
extern void wsrep_ready_set (my_bool x);
extern ssize_t wsrep_sst_prepare (void** msg);
extern int wsrep_sst_donate_cb (void* app_ctx,
void* recv_ctx,
const void* msg, size_t msg_len,
const wsrep_uuid_t* current_uuid,
wsrep_seqno_t current_seqno,
const char* state, size_t state_len,
bool bypass);
extern size_t guess_ip (char* buf, size_t buf_len);
extern size_t guess_address(char* buf, size_t buf_len);
extern wsrep_uuid_t local_uuid;
extern wsrep_seqno_t local_seqno;
/*! SST thread signals init thread about sst completion */
extern void wsrep_sst_complete(const wsrep_uuid_t* uuid, wsrep_seqno_t, bool);
extern void wsrep_notify_status (wsrep_member_status_t new_status,
const wsrep_view_info_t* view = 0);
namespace wsp {
class node_status
{
public:
node_status() : status(WSREP_MEMBER_UNDEFINED) {}
void set(wsrep_member_status_t new_status,
const wsrep_view_info_t* view = 0)
{
if (status != new_status || 0 != view)
{
wsrep_notify_status(new_status, view);
status = new_status;
}
}
wsrep_member_status_t get() const { return status; }
private:
wsrep_member_status_t status;
};
} /* namespace wsp */
extern wsp::node_status local_status;
namespace wsp {
/* A small class to run external programs. */
class process
{
private:
const char* const str_;
FILE* io_;
int err_;
pid_t pid_;
public:
/*! @arg type is a pointer to a null-terminated string which must contain
either the letter 'r' for reading or the letter 'w' for writing.
*/
process (const char* cmd, const char* type);
~process ();
FILE* pipe () { return io_; }
int error() { return err_; }
int wait ();
const char* cmd() { return str_; }
};
#ifdef REMOVED
class lock
{
pthread_mutex_t* const mtx_;
public:
lock (pthread_mutex_t* mtx) : mtx_(mtx)
{
int err = pthread_mutex_lock (mtx_);
if (err)
{
WSREP_ERROR("Mutex lock failed: %s", strerror(err));
abort();
}
}
virtual ~lock ()
{
int err = pthread_mutex_unlock (mtx_);
if (err)
{
WSREP_ERROR("Mutex unlock failed: %s", strerror(err));
abort();
}
}
inline void wait (pthread_cond_t* cond)
{
pthread_cond_wait (cond, mtx_);
}
private:
lock (const lock&);
lock& operator=(const lock&);
};
class monitor
{
int mutable refcnt;
pthread_mutex_t mutable mtx;
pthread_cond_t mutable cond;
public:
monitor() : refcnt(0)
{
pthread_mutex_init (&mtx, NULL);
pthread_cond_init (&cond, NULL);
}
~monitor()
{
pthread_mutex_destroy (&mtx);
pthread_cond_destroy (&cond);
}
void enter() const
{
lock l(&mtx);
while (refcnt)
{
l.wait(&cond);
}
refcnt++;
}
void leave() const
{
lock l(&mtx);
refcnt--;
if (refcnt == 0)
{
pthread_cond_signal (&cond);
}
}
private:
monitor (const monitor&);
monitor& operator= (const monitor&);
};
class critical
{
const monitor& mon;
public:
critical(const monitor& m) : mon(m) { mon.enter(); }
~critical() { mon.leave(); }
private:
critical (const critical&);
critical& operator= (const critical&);
};
#endif
class thd
{
class thd_init
{
public:
thd_init() { my_thread_init(); }
~thd_init() { my_thread_end(); }
}
init;
thd (const thd&);
thd& operator= (const thd&);
public:
thd(my_bool wsrep_on);
~thd();
THD* const ptr;
};
class string
{
public:
string() : string_(0) {}
void set(char* str) { if (string_) free (string_); string_ = str; }
~string() { set (0); }
private:
char* string_;
};
} // namespace wsrep
#endif /* WSREP_PRIV_H */

1001
sql/wsrep_sst.cc
View File

File diff suppressed because it is too large Load Diff

468
sql/wsrep_utils.cc
View File

@ -1,468 +0,0 @@
/* Copyright 2010 Codership Oy <http://www.codership.com>
This program 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; version 2 of the License.
This program 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 this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*/
//! @file declares symbols private to wsrep integration layer
#ifndef _GNU_SOURCE
#define _GNU_SOURCE // POSIX_SPAWN_USEVFORK flag
#endif
#include <spawn.h> // posix_spawn()
#include <unistd.h> // pipe()
#include <errno.h> // errno
#include <string.h> // strerror()
#include <sys/wait.h> // waitpid()
#include <sql_class.h>
#include "wsrep_priv.h"
extern char** environ; // environment variables
static wsp::string wsrep_PATH;
void
wsrep_prepend_PATH (const char* path)
{
int count = 0;
while (environ[count])
{
if (strncmp (environ[count], "PATH=", 5))
{
count++;
continue;
}
char* const old_path (environ[count]);
if (strstr (old_path, path)) return; // path already there
size_t const new_path_len(strlen(old_path) + strlen(":") +
strlen(path) + 1);
char* const new_path (reinterpret_cast<char*>(malloc(new_path_len)));
if (new_path)
{
snprintf (new_path, new_path_len, "PATH=%s:%s", path,
old_path + strlen("PATH="));
wsrep_PATH.set (new_path);
environ[count] = new_path;
}
else
{
WSREP_ERROR ("Failed to allocate 'PATH' environment variable "
"buffer of size %zu.", new_path_len);
}
return;
}
WSREP_ERROR ("Failed to find 'PATH' environment variable. "
"State snapshot transfer may not be working.");
}
namespace wsp
{
#define PIPE_READ 0
#define PIPE_WRITE 1
#define STDIN_FD 0
#define STDOUT_FD 1
#ifndef POSIX_SPAWN_USEVFORK
# define POSIX_SPAWN_USEVFORK 0
#endif
process::process (const char* cmd, const char* type)
: str_(cmd ? strdup(cmd) : strdup("")), io_(NULL), err_(EINVAL), pid_(0)
{
if (0 == str_)
{
WSREP_ERROR ("Can't allocate command line of size: %zu", strlen(cmd));
err_ = ENOMEM;
return;
}
if (0 == strlen(str_))
{
WSREP_ERROR ("Can't start a process: null or empty command line.");
return;
}
if (NULL == type || (strcmp (type, "w") && strcmp(type, "r")))
{
WSREP_ERROR ("type argument should be either \"r\" or \"w\".");
return;
}
int pipe_fds[2] = { -1, };
if (::pipe(pipe_fds))
{
err_ = errno;
WSREP_ERROR ("pipe() failed: %d (%s)", err_, strerror(err_));
return;
}
// which end of pipe will be returned to parent
int const parent_end (strcmp(type,"w") ? PIPE_READ : PIPE_WRITE);
int const child_end (parent_end == PIPE_READ ? PIPE_WRITE : PIPE_READ);
int const close_fd (parent_end == PIPE_READ ? STDOUT_FD : STDIN_FD);
char* const pargv[4] = { strdup("sh"), strdup("-c"), strdup(str_), NULL };
if (!(pargv[0] && pargv[1] && pargv[2]))
{
err_ = ENOMEM;
WSREP_ERROR ("Failed to allocate pargv[] array.");
goto cleanup_pipe;
}
posix_spawnattr_t attr;
err_ = posix_spawnattr_init (&attr);
if (err_)
{
WSREP_ERROR ("posix_spawnattr_init() failed: %d (%s)",
err_, strerror(err_));
goto cleanup_pipe;
}
err_ = posix_spawnattr_setflags (&attr, POSIX_SPAWN_SETSIGDEF |
POSIX_SPAWN_USEVFORK);
if (err_)
{
WSREP_ERROR ("posix_spawnattr_setflags() failed: %d (%s)",
err_, strerror(err_));
goto cleanup_attr;
}
posix_spawn_file_actions_t fact;
err_ = posix_spawn_file_actions_init (&fact);
if (err_)
{
WSREP_ERROR ("posix_spawn_file_actions_init() failed: %d (%s)",
err_, strerror(err_));
goto cleanup_attr;
}
// close child's stdout|stdin depending on what we returning
err_ = posix_spawn_file_actions_addclose (&fact, close_fd);
if (err_)
{
WSREP_ERROR ("posix_spawn_file_actions_addclose() failed: %d (%s)",
err_, strerror(err_));
goto cleanup_fact;
}
// substitute our pipe descriptor in place of the closed one
err_ = posix_spawn_file_actions_adddup2 (&fact,
pipe_fds[child_end], close_fd);
if (err_)
{
WSREP_ERROR ("posix_spawn_file_actions_addup2() failed: %d (%s)",
err_, strerror(err_));
goto cleanup_fact;
}
err_ = posix_spawnp (&pid_, pargv[0], &fact, &attr, pargv, environ);
if (err_)
{
WSREP_ERROR ("posix_spawnp(%s) failed: %d (%s)",
pargv[2], err_, strerror(err_));
pid_ = 0; // just to make sure it was not messed up in the call
goto cleanup_fact;
}
io_ = fdopen (pipe_fds[parent_end], type);
if (io_)
{
pipe_fds[parent_end] = -1; // skip close on cleanup
}
else
{
err_ = errno;
WSREP_ERROR ("fdopen() failed: %d (%s)", err_, strerror(err_));
}
cleanup_fact:
int err; // to preserve err_ code
err = posix_spawn_file_actions_destroy (&fact);
if (err)
{
WSREP_ERROR ("posix_spawn_file_actions_destroy() failed: %d (%s)\n",
err, strerror(err));
}
cleanup_attr:
err = posix_spawnattr_destroy (&attr);
if (err)
{
WSREP_ERROR ("posix_spawnattr_destroy() failed: %d (%s)",
err, strerror(err));
}
cleanup_pipe:
if (pipe_fds[0] >= 0) close (pipe_fds[0]);
if (pipe_fds[1] >= 0) close (pipe_fds[1]);
free (pargv[0]);
free (pargv[1]);
free (pargv[2]);
}
process::~process ()
{
if (io_)
{
assert (pid_);
assert (str_);
WSREP_WARN("Closing pipe to child process: %s, PID(%ld) "
"which might still be running.", str_, (long)pid_);
if (fclose (io_) == -1)
{
err_ = errno;
WSREP_ERROR("fclose() failed: %d (%s)", err_, strerror(err_));
}
}
if (str_) free (const_cast<char*>(str_));
}
int
process::wait ()
{
if (pid_)
{
int status;
if (-1 == waitpid(pid_, &status, 0))
{
err_ = errno; assert (err_);
WSREP_ERROR("Waiting for process failed: %s, PID(%ld): %d (%s)",
str_, (long)pid_, err_, strerror (err_));
}
else
{ // command completed, check exit status
if (WIFEXITED (status)) {
err_ = WEXITSTATUS (status);
}
else { // command didn't complete with exit()
WSREP_ERROR("Process was aborted.");
err_ = errno ? errno : ECHILD;
}
if (err_) {
switch (err_) /* Translate error codes to more meaningful */
{
case 126: err_ = EACCES; break; /* Permission denied */
case 127: err_ = ENOENT; break; /* No such file or directory */
}
WSREP_ERROR("Process completed with error: %s: %d (%s)",
str_, err_, strerror(err_));
}
pid_ = 0;
if (io_) fclose (io_);
io_ = NULL;
}
}
else {
assert (NULL == io_);
WSREP_ERROR("Command did not run: %s", str_);
}
return err_;
}
thd::thd (my_bool won) : init(), ptr(new THD)
{
if (ptr)
{
ptr->thread_stack= (char*) &ptr;
ptr->store_globals();
ptr->variables.option_bits&= ~OPTION_BIN_LOG; // disable binlog
ptr->variables.wsrep_on = won;
ptr->security_ctx->master_access= ~(ulong)0;
lex_start(ptr);
}
}
thd::~thd ()
{
if (ptr)
{
delete ptr;
my_pthread_setspecific_ptr (THR_THD, 0);
}
}
} // namespace wsp
extern ulong my_bind_addr;
extern uint mysqld_port;
size_t guess_ip (char* buf, size_t buf_len)
{
size_t ip_len = 0;
if (htonl(INADDR_NONE) == my_bind_addr) {
WSREP_ERROR("Networking not configured, cannot receive state transfer.");
return 0;
}
if (htonl(INADDR_ANY) != my_bind_addr) {
uint8_t* b = (uint8_t*)&my_bind_addr;
ip_len = snprintf (buf, buf_len,
"%hhu.%hhu.%hhu.%hhu", b[0],b[1],b[2],b[3]);
return ip_len;
}
// mysqld binds to all interfaces - try IP from wsrep_node_address
if (wsrep_node_address && wsrep_node_address[0] != '\0') {
const char* const colon_ptr = strchr(wsrep_node_address, ':');
if (colon_ptr)
ip_len = colon_ptr - wsrep_node_address;
else
ip_len = strlen(wsrep_node_address);
if (ip_len >= buf_len) {
WSREP_WARN("default_ip(): buffer too short: %zu <= %zd", buf_len, ip_len);
return 0;
}
memcpy (buf, wsrep_node_address, ip_len);
buf[ip_len] = '\0';
return ip_len;
}
// try to find the address of the first one
#if (TARGET_OS_LINUX == 1)
const char cmd[] = "/sbin/ifconfig | "
// "grep -m1 -1 -E '^[a-z]?eth[0-9]' | tail -n 1 | "
"grep -E '^[[:space:]]+inet addr:' | grep -m1 -v 'inet addr:127' | "
"sed 's/:/ /' | awk '{ print $3 }'";
#elif defined(__sun__)
const char cmd[] = "/sbin/ifconfig -a | "
"/usr/gnu/bin/grep -m1 -1 -E 'net[0-9]:' | tail -n 1 | awk '{ print $2 }'";
#else
char *cmd;
#error "OS not supported"
#endif
wsp::process proc (cmd, "r");
if (NULL != proc.pipe()) {
char* ret;
ret = fgets (buf, buf_len, proc.pipe());
if (proc.wait()) return 0;
if (NULL == ret) {
WSREP_ERROR("Failed to read output of: '%s'", cmd);
return 0;
}
}
else {
WSREP_ERROR("Failed to execute: '%s'", cmd);
return 0;
}
// clear possible \n at the end of ip string left by fgets()
ip_len = strlen (buf);
if (ip_len > 0 && '\n' == buf[ip_len - 1]) {
ip_len--;
buf[ip_len] = '\0';
}
if (INADDR_NONE == inet_addr(buf)) {
if (strlen(buf) != 0) {
WSREP_WARN("Shell command returned invalid address: '%s'", buf);
}
return 0;
}
return ip_len;
}
size_t guess_address(char* buf, size_t buf_len)
{
size_t addr_len = guess_ip (buf, buf_len);
if (addr_len && addr_len < buf_len) {
addr_len += snprintf (buf + addr_len, buf_len - addr_len,
":%u", mysqld_port);
}
return addr_len;
}
/*
* WSREPXid
*/
#define WSREP_XID_PREFIX "WSREPXid"
#define WSREP_XID_PREFIX_LEN MYSQL_XID_PREFIX_LEN
#define WSREP_XID_UUID_OFFSET 8
#define WSREP_XID_SEQNO_OFFSET (WSREP_XID_UUID_OFFSET + sizeof(wsrep_uuid_t))
#define WSREP_XID_GTRID_LEN (WSREP_XID_SEQNO_OFFSET + sizeof(wsrep_seqno_t))
void wsrep_xid_init(XID* xid, const wsrep_uuid_t* uuid, wsrep_seqno_t seqno)
{
xid->formatID= 1;
xid->gtrid_length= WSREP_XID_GTRID_LEN;
xid->bqual_length= 0;
memset(xid->data, 0, sizeof(xid->data));
memcpy(xid->data, WSREP_XID_PREFIX, WSREP_XID_PREFIX_LEN);
memcpy(xid->data + WSREP_XID_UUID_OFFSET, uuid, sizeof(wsrep_uuid_t));
memcpy(xid->data + WSREP_XID_SEQNO_OFFSET, &seqno, sizeof(wsrep_seqno_t));
}
const wsrep_uuid_t* wsrep_xid_uuid(const XID* xid)
{
if (wsrep_is_wsrep_xid(xid))
return reinterpret_cast<const wsrep_uuid_t*>(xid->data
+ WSREP_XID_UUID_OFFSET);
else
return &WSREP_UUID_UNDEFINED;
}
wsrep_seqno_t wsrep_xid_seqno(const XID* xid)
{
if (wsrep_is_wsrep_xid(xid))
{
wsrep_seqno_t seqno;
memcpy(&seqno, xid->data + WSREP_XID_SEQNO_OFFSET, sizeof(wsrep_seqno_t));
return seqno;
}
else
{
return WSREP_SEQNO_UNDEFINED;
}
}
extern "C"
int wsrep_is_wsrep_xid(const void* xid_ptr)
{
const XID* xid= reinterpret_cast<const XID*>(xid_ptr);
return (xid->formatID == 1 &&
xid->gtrid_length == WSREP_XID_GTRID_LEN &&
xid->bqual_length == 0 &&
!memcmp(xid->data, WSREP_XID_PREFIX, WSREP_XID_PREFIX_LEN));
}

548
sql/wsrep_var.cc
View File

@ -1,548 +0,0 @@
/* Copyright 2008 Codership Oy <http://www.codership.com>
This program 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; version 2 of the License.
This program 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 this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include <mysqld.h>
#include <sql_class.h>
#include <sql_plugin.h>
#include <set_var.h>
#include <sql_acl.h>
#include "wsrep_priv.h"
#include <my_dir.h>
#include <cstdio>
#include <cstdlib>
#define WSREP_START_POSITION_ZERO "00000000-0000-0000-0000-000000000000:-1"
#define WSREP_CLUSTER_NAME "my_wsrep_cluster"
const char* wsrep_provider = 0;
const char* wsrep_provider_options = 0;
const char* wsrep_cluster_address = 0;
const char* wsrep_cluster_name = 0;
const char* wsrep_node_name = 0;
static char node_address[256] = { 0, };
const char* wsrep_node_address = node_address; // ???
const char* wsrep_start_position = 0;
ulong wsrep_OSU_method_options;
static int wsrep_thread_change = 0;
int wsrep_init_vars()
{
wsrep_provider = my_strdup(WSREP_NONE, MYF(MY_WME));
wsrep_provider_options= my_strdup("", MYF(MY_WME));
wsrep_cluster_address = my_strdup("", MYF(MY_WME));
wsrep_cluster_name = my_strdup(WSREP_CLUSTER_NAME, MYF(MY_WME));
wsrep_node_name = my_strdup("", MYF(MY_WME));
wsrep_start_position = my_strdup(WSREP_START_POSITION_ZERO, MYF(MY_WME));
global_system_variables.binlog_format=BINLOG_FORMAT_ROW;
return 0;
}
bool wsrep_on_update (sys_var *self, THD* thd, enum_var_type var_type)
{
if (var_type == OPT_GLOBAL) {
// FIXME: this variable probably should be changed only per session
thd->variables.wsrep_on = global_system_variables.wsrep_on;
}
else {
}
#ifdef REMOVED
if (thd->variables.wsrep_on)
thd->variables.option_bits |= (OPTION_BIN_LOG);
else
thd->variables.option_bits &= ~(OPTION_BIN_LOG);
#endif
return false;
}
void wsrep_causal_reads_update (sys_var *self, THD* thd, enum_var_type var_type)
{
if (var_type == OPT_GLOBAL) {
thd->variables.wsrep_causal_reads = global_system_variables.wsrep_causal_reads;
}
else {
}
}
static int wsrep_start_position_verify (const char* start_str)
{
size_t start_len;
wsrep_uuid_t uuid;
ssize_t uuid_len;
start_len = strlen (start_str);
if (start_len < 34)
return 1;
uuid_len = wsrep_uuid_scan (start_str, start_len, &uuid);
if (uuid_len < 0 || (start_len - uuid_len) < 2)
return 1;
if (start_str[uuid_len] != ':') // separator should follow UUID
return 1;
char* endptr;
wsrep_seqno_t const seqno __attribute__((unused)) // to avoid GCC warnings
(strtoll(&start_str[uuid_len + 1], &endptr, 10));
if (*endptr == '\0') return 0; // remaining string was seqno
return 1;
}
bool wsrep_start_position_check (sys_var *self, THD* thd, set_var* var)
{
char buff[FN_REFLEN];
String str(buff, sizeof(buff), system_charset_info), *res;
const char* start_str = NULL;
if (!(res = var->value->val_str(&str))) goto err;
start_str = res->c_ptr();
if (!start_str) goto err;
if (!wsrep_start_position_verify(start_str)) return 0;
err:
my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), var->var->name.str,
start_str ? start_str : "NULL");
return 1;
}
void wsrep_set_local_position (const char* value)
{
size_t value_len = strlen (value);
size_t uuid_len = wsrep_uuid_scan (value, value_len, &local_uuid);
local_seqno = strtoll (value + uuid_len + 1, NULL, 10);
XID xid;
wsrep_xid_init(&xid, &local_uuid, local_seqno);
wsrep_set_SE_checkpoint(&xid);
WSREP_INFO ("wsrep_start_position var submitted: '%s'", wsrep_start_position);
}
bool wsrep_start_position_update (sys_var *self, THD* thd, enum_var_type type)
{
// since this value passed wsrep_start_position_check, don't check anything
// here
wsrep_set_local_position (wsrep_start_position);
if (wsrep) {
wsrep->sst_received (wsrep, &local_uuid, local_seqno, NULL, 0);
}
return 0;
}
void wsrep_start_position_init (const char* val)
{
if (NULL == val || wsrep_start_position_verify (val))
{
WSREP_ERROR("Bad initial value for wsrep_start_position: %s",
(val ? val : ""));
return;
}
wsrep_set_local_position (val);
}
static bool refresh_provider_options()
{
WSREP_DEBUG("refresh_provider_options: %s",
(wsrep_provider_options) ? wsrep_provider_options : "null");
char* opts= wsrep->options_get(wsrep);
if (opts)
{
if (wsrep_provider_options) my_free((void *)wsrep_provider_options);
wsrep_provider_options = (char*)my_memdup(opts, strlen(opts) + 1,
MYF(MY_WME));
}
else
{
WSREP_ERROR("Failed to get provider options");
return true;
}
return false;
}
static int wsrep_provider_verify (const char* provider_str)
{
MY_STAT f_stat;
char path[FN_REFLEN];
if (!provider_str || strlen(provider_str)== 0)
return 1;
if (!strcmp(provider_str, WSREP_NONE))
return 0;
if (!unpack_filename(path, provider_str))
return 1;
/* check that provider file exists */
bzero(&f_stat, sizeof(MY_STAT));
if (!my_stat(path, &f_stat, MYF(0)))
{
return 1;
}
return 0;
}
bool wsrep_provider_check (sys_var *self, THD* thd, set_var* var)
{
char buff[FN_REFLEN];
String str(buff, sizeof(buff), system_charset_info), *res;
const char* provider_str = NULL;
if (!(res = var->value->val_str(&str))) goto err;
provider_str = res->c_ptr();
if (!provider_str) goto err;
if (!wsrep_provider_verify(provider_str)) return 0;
err:
my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), var->var->name.str,
provider_str ? provider_str : "NULL");
return 1;
}
bool wsrep_provider_update (sys_var *self, THD* thd, enum_var_type type)
{
bool rcode= false;
bool wsrep_on_saved= thd->variables.wsrep_on;
thd->variables.wsrep_on= false;
WSREP_DEBUG("wsrep_provider_update: %s", wsrep_provider);
wsrep_stop_replication(thd);
wsrep_deinit();
char* tmp= strdup(wsrep_provider); // wsrep_init() rewrites provider
//when fails
if (wsrep_init())
{
my_error(ER_CANT_OPEN_LIBRARY, MYF(0), tmp);
rcode = true;
}
free(tmp);
// we sure don't want to use old address with new provider
wsrep_cluster_address_init(NULL);
wsrep_provider_options_init(NULL);
thd->variables.wsrep_on= wsrep_on_saved;
refresh_provider_options();
return rcode;
}
void wsrep_provider_init (const char* value)
{
WSREP_DEBUG("wsrep_provider_init: %s -> %s",
(wsrep_provider) ? wsrep_provider : "null",
(value) ? value : "null");
if (NULL == value || wsrep_provider_verify (value))
{
WSREP_ERROR("Bad initial value for wsrep_provider: %s",
(value ? value : ""));
return;
}
if (wsrep_provider) my_free((void *)wsrep_provider);
wsrep_provider = my_strdup(value, MYF(0));
}
bool wsrep_provider_options_check(sys_var *self, THD* thd, set_var* var)
{
return 0;
}
bool wsrep_provider_options_update(sys_var *self, THD* thd, enum_var_type type)
{
wsrep_status_t ret= wsrep->options_set(wsrep, wsrep_provider_options);
if (ret != WSREP_OK)
{
WSREP_ERROR("Set options returned %d", ret);
return true;
}
return refresh_provider_options();
}
void wsrep_provider_options_init(const char* value)
{
if (wsrep_provider_options && wsrep_provider_options != value)
my_free((void *)wsrep_provider_options);
wsrep_provider_options = (value) ? my_strdup(value, MYF(0)) : NULL;
}
static int wsrep_cluster_address_verify (const char* cluster_address_str)
{
/* There is no predefined address format, it depends on provider. */
return 0;
}
bool wsrep_cluster_address_check (sys_var *self, THD* thd, set_var* var)
{
char buff[FN_REFLEN];
String str(buff, sizeof(buff), system_charset_info), *res;
const char* cluster_address_str = NULL;
if (!(res = var->value->val_str(&str))) goto err;
cluster_address_str = res->c_ptr();
if (!wsrep_cluster_address_verify(cluster_address_str)) return 0;
err:
my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), var->var->name.str,
cluster_address_str ? cluster_address_str : "NULL");
return 1 ;
}
bool wsrep_cluster_address_update (sys_var *self, THD* thd, enum_var_type type)
{
bool wsrep_on_saved= thd->variables.wsrep_on;
thd->variables.wsrep_on= false;
wsrep_stop_replication(thd);
if (wsrep_start_replication())
{
wsrep_create_rollbacker();
wsrep_create_appliers(wsrep_slave_threads);
}
thd->variables.wsrep_on= wsrep_on_saved;
return false;
}
void wsrep_cluster_address_init (const char* value)
{
WSREP_DEBUG("wsrep_cluster_address_init: %s -> %s",
(wsrep_cluster_address) ? wsrep_cluster_address : "null",
(value) ? value : "null");
if (wsrep_cluster_address) my_free ((void*)wsrep_cluster_address);
wsrep_cluster_address = (value) ? my_strdup(value, MYF(0)) : NULL;
}
bool wsrep_cluster_name_check (sys_var *self, THD* thd, set_var* var)
{
char buff[FN_REFLEN];
String str(buff, sizeof(buff), system_charset_info), *res;
const char* cluster_name_str = NULL;
if (!(res = var->value->val_str(&str))) goto err;
cluster_name_str = res->c_ptr();
if (!cluster_name_str || strlen(cluster_name_str) == 0) goto err;
return 0;
err:
my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), var->var->name.str,
cluster_name_str ? cluster_name_str : "NULL");
return 1;
}
bool wsrep_cluster_name_update (sys_var *self, THD* thd, enum_var_type type)
{
return 0;
}
bool wsrep_node_name_check (sys_var *self, THD* thd, set_var* var)
{
char buff[FN_REFLEN];
String str(buff, sizeof(buff), system_charset_info), *res;
const char* node_name_str = NULL;
if (!(res = var->value->val_str(&str))) goto err;
node_name_str = res->c_ptr();
if (!node_name_str || strlen(node_name_str) == 0) goto err;
return 0;
err:
my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), var->var->name.str,
node_name_str ? node_name_str : "NULL");
return 1;
}
bool wsrep_node_name_update (sys_var *self, THD* thd, enum_var_type type)
{
return 0;
}
// TODO: do something more elaborate, like checking connectivity
bool wsrep_node_address_check (sys_var *self, THD* thd, set_var* var)
{
char buff[FN_REFLEN];
String str(buff, sizeof(buff), system_charset_info), *res;
const char* node_address_str = NULL;
if (!(res = var->value->val_str(&str))) goto err;
node_address_str = res->c_ptr();
if (!node_address_str || strlen(node_address_str) == 0) goto err;
return 0;
err:
my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), var->var->name.str,
node_address_str ? node_address_str : "NULL");
return 1;
}
bool wsrep_node_address_update (sys_var *self, THD* thd, enum_var_type type)
{
return 0;
}
void wsrep_node_address_init (const char* value)
{
if (wsrep_node_address && strcmp(wsrep_node_address, value))
my_free ((void*)wsrep_node_address);
wsrep_node_address = (value) ? my_strdup(value, MYF(0)) : NULL;
}
bool wsrep_slave_threads_check (sys_var *self, THD* thd, set_var* var)
{
mysql_mutex_lock(&LOCK_wsrep_slave_threads);
wsrep_thread_change = var->value->val_int() - wsrep_slave_threads;
mysql_mutex_unlock(&LOCK_wsrep_slave_threads);
return 0;
}
bool wsrep_slave_threads_update (sys_var *self, THD* thd, enum_var_type type)
{
if (wsrep_thread_change > 0)
{
wsrep_create_appliers(wsrep_thread_change);
}
else if (wsrep_thread_change < 0)
{
wsrep_close_applier_threads(-wsrep_thread_change);
}
return false;
}
/*
* Status variables stuff below
*/
static inline void
wsrep_assign_to_mysql (SHOW_VAR* mysql, wsrep_stats_var* wsrep)
{
mysql->name = wsrep->name;
switch (wsrep->type) {
case WSREP_VAR_INT64:
mysql->value = (char*) &wsrep->value._int64;
mysql->type = SHOW_LONGLONG;
break;
case WSREP_VAR_STRING:
mysql->value = (char*) &wsrep->value._string;
mysql->type = SHOW_CHAR_PTR;
break;
case WSREP_VAR_DOUBLE:
mysql->value = (char*) &wsrep->value._double;
mysql->type = SHOW_DOUBLE;
break;
}
}
#if DYNAMIC
// somehow this mysql status thing works only with statically allocated arrays.
static SHOW_VAR* mysql_status_vars = NULL;
static int mysql_status_len = -1;
#else
static SHOW_VAR mysql_status_vars[512 + 1];
static const int mysql_status_len = 512;
#endif
static void export_wsrep_status_to_mysql(THD* thd)
{
int wsrep_status_len, i;
thd->wsrep_status_vars = wsrep->stats_get(wsrep);
if (!thd->wsrep_status_vars) {
return;
}
for (wsrep_status_len = 0;
thd->wsrep_status_vars[wsrep_status_len].name != NULL;
wsrep_status_len++);
#if DYNAMIC
if (wsrep_status_len != mysql_status_len) {
void* tmp = realloc (mysql_status_vars,
(wsrep_status_len + 1) * sizeof(SHOW_VAR));
if (!tmp) {
sql_print_error ("Out of memory for wsrep status variables."
"Number of variables: %d", wsrep_status_len);
return;
}
mysql_status_len = wsrep_status_len;
mysql_status_vars = (SHOW_VAR*)tmp;
}
/* @TODO: fix this: */
#else
if (mysql_status_len < wsrep_status_len) wsrep_status_len= mysql_status_len;
#endif
for (i = 0; i < wsrep_status_len; i++)
wsrep_assign_to_mysql (mysql_status_vars + i, thd->wsrep_status_vars + i);
mysql_status_vars[wsrep_status_len].name = NullS;
mysql_status_vars[wsrep_status_len].value = NullS;
mysql_status_vars[wsrep_status_len].type = SHOW_LONG;
}
int wsrep_show_status (THD *thd, SHOW_VAR *var, char *buff)
{
export_wsrep_status_to_mysql(thd);
var->type= SHOW_ARRAY;
var->value= (char *) &mysql_status_vars;
return 0;
}
void wsrep_free_status (THD* thd)
{
if (thd->wsrep_status_vars)
{
wsrep->stats_free (wsrep, thd->wsrep_status_vars);
thd->wsrep_status_vars = 0;
}
}

129
support-files/wsrep.cnf.sh
View File

@ -1,129 +0,0 @@
# This file contains wsrep-related mysqld options. It should be included
# in the main MySQL configuration file.
#
# Options that need to be customized:
# - wsrep_provider
# - wsrep_cluster_address
# - wsrep_sst_auth
# The rest of defaults should work out of the box.
##
## mysqld options _MANDATORY_ for correct opration of the cluster
##
[mysqld]
# (This must be substituted by wsrep_format)
binlog_format=ROW
# Currently only InnoDB storage engine is supported
default-storage-engine=innodb
# to avoid issues with 'bulk mode inserts' using autoinc
innodb_autoinc_lock_mode=2
# This is a must for paralell applying
innodb_locks_unsafe_for_binlog=1
# Query Cache is not supported with wsrep
query_cache_size=0
query_cache_type=0
# Override bind-address
# In some systems bind-address defaults to 127.0.0.1, and with mysqldump SST
# it will have (most likely) disastrous consequences on donor node
bind-address=0.0.0.0
##
## WSREP options
##
# Full path to wsrep provider library or 'none'
wsrep_provider=none
# Provider specific configuration options
#wsrep_provider_options=
# Logical cluster name. Should be the same for all nodes.
wsrep_cluster_name="my_wsrep_cluster"
# Group communication system handle
#wsrep_cluster_address="dummy://"
# Human-readable node name (non-unique). Hostname by default.
#wsrep_node_name=
# Base replication <address|hostname>[:port] of the node.
# The values supplied will be used as defaults for state transfer receiving,
# listening ports and so on. Default: address of the first network interface.
#wsrep_node_address=
# Address for incoming client connections. Autodetect by default.
#wsrep_node_incoming_address=
# How many threads will process writesets from other nodes
wsrep_slave_threads=1
# DBUG options for wsrep provider
#wsrep_dbug_option
# Generate fake primary keys for non-PK tables (required for multi-master
# and parallel applying operation)
wsrep_certify_nonPK=1
# Maximum number of rows in write set
wsrep_max_ws_rows=131072
# Maximum size of write set
wsrep_max_ws_size=1073741824
# to enable debug level logging, set this to 1
wsrep_debug=0
# convert locking sessions into transactions
wsrep_convert_LOCK_to_trx=0
# how many times to retry deadlocked autocommits
wsrep_retry_autocommit=1
# change auto_increment_increment and auto_increment_offset automatically
wsrep_auto_increment_control=1
# retry autoinc insert, which failed for duplicate key error
wsrep_drupal_282555_workaround=0
# enable "strictly synchronous" semantics for read operations
wsrep_causal_reads=0
# Command to call when node status or cluster membership changes.
# Will be passed all or some of the following options:
# --status - new status of this node
# --uuid - UUID of the cluster
# --primary - whether the component is primary or not ("yes"/"no")
# --members - comma-separated list of members
# --index - index of this node in the list
wsrep_notify_cmd=
##
## WSREP State Transfer options
##
# State Snapshot Transfer method
wsrep_sst_method=mysqldump
# Address which donor should send State Snapshot to.
# Should be the address of THIS node. DON'T SET IT TO DONOR ADDRESS!!!
# (SST method dependent. Defaults to the first IP of the first interface)
#wsrep_sst_receive_address=
# SST authentication string. This will be used to send SST to joining nodes.
# Depends on SST method. For mysqldump method it is root:<root password>
wsrep_sst_auth=root:
# Desired SST donor name.
#wsrep_sst_donor=
# Reject client queries when donating SST (false)
#wsrep_sst_donor_rejects_queries=0
# Protocol version to use
# wsrep_protocol_version=

102
support-files/wsrep_notify.sh
View File

@ -1,102 +0,0 @@
#!/bin/sh -eu
# This is a simple example of wsrep notification script (wsrep_notify_cmd).
# It will create 'wsrep' schema and two tables in it: 'membeship' and 'status'
# and fill them on every membership or node status change.
#
# Edit parameters below to specify the address and login to server.
USER=root
PSWD=rootpass
HOST=127.0.0.1
PORT=3306
SCHEMA="wsrep"
MEMB_TABLE="$SCHEMA.membership"
STATUS_TABLE="$SCHEMA.status"
BEGIN="
SET wsrep_on=0;
DROP SCHEMA IF EXISTS $SCHEMA; CREATE SCHEMA $SCHEMA;
CREATE TABLE $MEMB_TABLE (
idx INT UNIQUE PRIMARY KEY,
uuid CHAR(40) UNIQUE, /* node UUID */
name VARCHAR(32), /* node name */
addr VARCHAR(256) /* node address */
) ENGINE=MEMORY;
CREATE TABLE $STATUS_TABLE (
size INT, /* component size */
idx INT, /* this node index */
status CHAR(16), /* this node status */
uuid CHAR(40), /* cluster UUID */
prim BOOLEAN /* if component is primary */
) ENGINE=MEMORY;
BEGIN;
DELETE FROM $MEMB_TABLE;
DELETE FROM $STATUS_TABLE;
"
END="COMMIT;"
configuration_change()
{
echo "$BEGIN;"
local idx=0
for NODE in $(echo $MEMBERS | sed s/,/\ /g)
do
echo "INSERT INTO $MEMB_TABLE VALUES ( $idx, "
# Don't forget to properly quote string values
echo "'$NODE'" | sed s/\\//\',\'/g
echo ");"
idx=$(( $idx + 1 ))
done
echo "INSERT INTO $STATUS_TABLE VALUES($idx, $INDEX, '$STATUS', '$CLUSTER_UUID', $PRIMARY);"
echo "$END"
}
status_update()
{
echo "SET wsrep_on=0; BEGIN; UPDATE $STATUS_TABLE SET status='$STATUS'; COMMIT;"
}
COM=status_update # not a configuration change by default
while [ $# -gt 0 ]
do
case $1 in
--status)
STATUS=$2
shift
;;
--uuid)
CLUSTER_UUID=$2
shift
;;
--primary)
[ "$2" = "yes" ] && PRIMARY="1" || PRIMARY="0"
COM=configuration_change
shift
;;
--index)
INDEX=$2
shift
;;
--members)
MEMBERS=$2
shift
;;
esac
shift
done
# Undefined means node is shutting down
if [ "$STATUS" != "Undefined" ]
then
$COM | mysql -B -u$USER -p$PSWD -h$HOST -P$PORT
fi
exit 0
#