diff options
Diffstat (limited to 'sql/wsrep_binlog.cc')
| -rw-r--r-- | sql/wsrep_binlog.cc | 400 |
1 files changed, 400 insertions, 0 deletions
diff --git a/sql/wsrep_binlog.cc b/sql/wsrep_binlog.cc new file mode 100644 index 00000000..84392fac --- /dev/null +++ b/sql/wsrep_binlog.cc @@ -0,0 +1,400 @@ +/* Copyright (C) 2013 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 Street, Fifth Floor, Boston, MA 02110-1335 USA. */ + +#include "mariadb.h" +#include "mysql/service_wsrep.h" +#include "wsrep_binlog.h" +#include "log.h" +#include "slave.h" +#include "log_event.h" +#include "wsrep_applier.h" +#include "wsrep_mysqld.h" + +#include "transaction.h" + +extern handlerton *binlog_hton; +/* + Write the contents of a cache to a memory buffer. + + This function quite the same as MYSQL_BIN_LOG::write_cache(), + with the exception that here we write in buffer instead of log file. + */ +int wsrep_write_cache_buf(IO_CACHE *cache, uchar **buf, size_t *buf_len) +{ + *buf= NULL; + *buf_len= 0; + my_off_t const saved_pos(my_b_tell(cache)); + DBUG_ENTER("wsrep_write_cache_buf"); + + if (reinit_io_cache(cache, READ_CACHE, 0, 0, 0)) + { + WSREP_ERROR("failed to initialize io-cache"); + DBUG_RETURN(ER_ERROR_ON_WRITE); + } + + uint length= my_b_bytes_in_cache(cache); + if (unlikely(0 == length)) length= my_b_fill(cache); + + size_t total_length= 0; + + if (likely(length > 0)) do + { + total_length += length; + /* + Bail out if buffer grows too large. + A temporary fix to avoid allocating indefinitely large buffer, + not a real limit on a writeset size which includes other things + like header and keys. + */ + if (total_length > wsrep_max_ws_size) + { + WSREP_WARN("transaction size limit (%lu) exceeded: %zu", + wsrep_max_ws_size, total_length); + goto error; + } + uchar* tmp= (uchar *)my_realloc(PSI_INSTRUMENT_ME, *buf, total_length, + MYF(MY_ALLOW_ZERO_PTR)); + if (!tmp) + { + WSREP_ERROR("could not (re)allocate buffer: %zu + %u", + *buf_len, length); + goto error; + } + *buf= tmp; + + memcpy(*buf + *buf_len, cache->read_pos, length); + *buf_len= total_length; + + if (cache->file < 0) + { + cache->read_pos= cache->read_end; + break; + } + } while ((length= my_b_fill(cache))); + + if (reinit_io_cache(cache, WRITE_CACHE, saved_pos, 0, 0)) + { + WSREP_WARN("failed to initialize io-cache"); + goto cleanup; + } + + DBUG_RETURN(0); + +error: + if (reinit_io_cache(cache, WRITE_CACHE, saved_pos, 0, 0)) + { + WSREP_WARN("failed to initialize io-cache"); + } +cleanup: + my_free(*buf); + *buf= NULL; + *buf_len= 0; + DBUG_RETURN(ER_ERROR_ON_WRITE); +} + +#define STACK_SIZE 4096 /* 4K - for buffer preallocated on the stack: + * many transactions would fit in there + * so there is no need to reach for the heap */ + +/* + Write the contents of a cache to wsrep provider. + + This function quite the same as MYSQL_BIN_LOG::write_cache(), + with the exception that here we write in buffer instead of log file. + + This version uses incremental data appending as it reads it from cache. + */ +static int wsrep_write_cache_inc(THD* const thd, + IO_CACHE* const cache, + size_t* const len) +{ + DBUG_ENTER("wsrep_write_cache_inc"); + my_off_t const saved_pos(my_b_tell(cache)); + + if (reinit_io_cache(cache, READ_CACHE, thd->wsrep_sr().log_position(), 0, 0)) + { + WSREP_ERROR("failed to initialize io-cache"); + DBUG_RETURN(1);; + } + + int ret= 0; + size_t total_length(0); + + uint length(my_b_bytes_in_cache(cache)); + if (unlikely(0 == length)) length= my_b_fill(cache); + + if (likely(length > 0)) + { + do + { + total_length += length; + /* bail out if buffer grows too large + not a real limit on a writeset size which includes other things + like header and keys. + */ + if (unlikely(total_length > wsrep_max_ws_size)) + { + WSREP_WARN("transaction size limit (%lu) exceeded: %zu", + wsrep_max_ws_size, total_length); + ret= 1; + goto cleanup; + } + if (thd->wsrep_cs().append_data(wsrep::const_buffer(cache->read_pos, length))) + goto cleanup; + cache->read_pos= cache->read_end; + } while ((cache->file >= 0) && (length= my_b_fill(cache))); + if (ret == 0) + { + assert(total_length + thd->wsrep_sr().log_position() == saved_pos); + } + } + +cleanup: + *len= total_length; + if (reinit_io_cache(cache, WRITE_CACHE, saved_pos, 0, 0)) + { + WSREP_ERROR("failed to reinitialize io-cache"); + } + DBUG_RETURN(ret); +} + +/* + Write the contents of a cache to wsrep provider. + + This function quite the same as MYSQL_BIN_LOG::write_cache(), + with the exception that here we write in buffer instead of log file. + */ +int wsrep_write_cache(THD* const thd, + IO_CACHE* const cache, + size_t* const len) +{ + return wsrep_write_cache_inc(thd, cache, len); +} + +void wsrep_dump_rbr_buf(THD *thd, const void* rbr_buf, size_t buf_len) +{ + int len= snprintf(NULL, 0, "%s/GRA_%lld_%lld.log", + wsrep_data_home_dir, (longlong) thd->thread_id, + (longlong) wsrep_thd_trx_seqno(thd)); + if (len < 0) + { + WSREP_ERROR("snprintf error: %d, skipping dump.", len); + return; + } + /* + len doesn't count the \0 end-of-string. Use len+1 below + to alloc and pass as an argument to snprintf. + */ + + char *filename= (char *) my_malloc(key_memory_WSREP, len+1, 0); + int len1= snprintf(filename, len+1, "%s/GRA_%lld_%lld.log", + wsrep_data_home_dir, (longlong) thd->thread_id, + (long long)wsrep_thd_trx_seqno(thd)); + + if (len > len1) + { + WSREP_ERROR("RBR dump path truncated: %d, skipping dump.", len); + my_free(filename); + return; + } + + FILE *of= fopen(filename, "wb"); + + if (of) + { + if (fwrite(rbr_buf, buf_len, 1, of) == 0) + WSREP_ERROR("Failed to write buffer of length %llu to '%s'", + (unsigned long long)buf_len, filename); + + fclose(of); + } + else + { + WSREP_ERROR("Failed to open file '%s': %d (%s)", + filename, errno, strerror(errno)); + } + my_free(filename); +} + +/* Dump replication buffer along with header to a file. */ +void wsrep_dump_rbr_buf_with_header(THD *thd, const void *rbr_buf, + size_t buf_len) +{ + DBUG_ENTER("wsrep_dump_rbr_buf_with_header"); + + File file; + IO_CACHE cache; + Log_event_writer writer(&cache, 0); + Format_description_log_event *ev= 0; + + longlong thd_trx_seqno= (long long)wsrep_thd_trx_seqno(thd); + int len= snprintf(NULL, 0, "%s/GRA_%lld_%lld_v2.log", + wsrep_data_home_dir, (longlong)thd->thread_id, + thd_trx_seqno); + /* + len doesn't count the \0 end-of-string. Use len+1 below + to alloc and pass as an argument to snprintf. + */ + char *filename; + if (len < 0 || !(filename= (char*) my_malloc(key_memory_WSREP, len+1, 0))) + { + WSREP_ERROR("snprintf error: %d, skipping dump.", len); + DBUG_VOID_RETURN; + } + + int len1= snprintf(filename, len+1, "%s/GRA_%lld_%lld_v2.log", + wsrep_data_home_dir, (longlong) thd->thread_id, + thd_trx_seqno); + + if (len > len1) + { + WSREP_ERROR("RBR dump path truncated: %d, skipping dump.", len); + my_free(filename); + DBUG_VOID_RETURN; + } + + if ((file= mysql_file_open(key_file_wsrep_gra_log, filename, + O_RDWR | O_CREAT | O_BINARY, MYF(MY_WME))) < 0) + { + WSREP_ERROR("Failed to open file '%s' : %d (%s)", + filename, errno, strerror(errno)); + goto cleanup1; + } + + if (init_io_cache(&cache, file, 0, WRITE_CACHE, 0, 0, MYF(MY_WME | MY_NABP))) + { + goto cleanup2; + } + + if (my_b_safe_write(&cache, BINLOG_MAGIC, BIN_LOG_HEADER_SIZE)) + { + goto cleanup2; + } + + /* + Instantiate an FDLE object for non-wsrep threads (to be written + to the dump file). + */ + ev= (thd->wsrep_applier) ? wsrep_get_apply_format(thd) : + (new Format_description_log_event(4)); + + if (writer.write(ev) || my_b_write(&cache, (uchar*)rbr_buf, buf_len) || + flush_io_cache(&cache)) + { + WSREP_ERROR("Failed to write to '%s'.", filename); + goto cleanup2; + } + +cleanup2: + end_io_cache(&cache); + +cleanup1: + my_free(filename); + mysql_file_close(file, MYF(MY_WME)); + + if (!thd->wsrep_applier) delete ev; + + DBUG_VOID_RETURN; +} + +int wsrep_write_skip_event(THD* thd) +{ + DBUG_ENTER("wsrep_write_skip_event"); + Ignorable_log_event skip_event(thd); + int ret= mysql_bin_log.write_event(&skip_event); + if (ret) + { + WSREP_WARN("wsrep_write_skip_event: write to binlog failed: %d", ret); + } + if (!ret && (ret= trans_commit_stmt(thd))) + { + WSREP_WARN("wsrep_write_skip_event: statt commit failed"); + } + DBUG_RETURN(ret); +} + +int wsrep_write_dummy_event_low(THD *thd, const char *msg) +{ + ::abort(); + return 0; +} + +int wsrep_write_dummy_event(THD *orig_thd, const char *msg) +{ + return 0; +} + +bool wsrep_commit_will_write_binlog(THD *thd) +{ + return (!wsrep_emulate_bin_log && /* binlog enabled*/ + (wsrep_thd_is_local(thd) || /* local thd*/ + (thd->wsrep_applier_service && /* applier and log-slave-updates */ + opt_log_slave_updates))); +} + +/* + The last THD/commit_for_wait registered for group commit. +*/ +static wait_for_commit *commit_order_tail= NULL; + +void wsrep_register_for_group_commit(THD *thd) +{ + DBUG_ENTER("wsrep_register_for_group_commit"); + if (wsrep_emulate_bin_log) + { + /* Binlog is off, no need to maintain group commit queue */ + DBUG_VOID_RETURN; + } + + DBUG_ASSERT(thd->wsrep_trx().ordered()); + + wait_for_commit *wfc= thd->wait_for_commit_ptr= &thd->wsrep_wfc; + + mysql_mutex_lock(&LOCK_wsrep_group_commit); + if (commit_order_tail) + { + wfc->register_wait_for_prior_commit(commit_order_tail); + } + commit_order_tail= thd->wait_for_commit_ptr; + mysql_mutex_unlock(&LOCK_wsrep_group_commit); + + /* + Now we have queued for group commit. If the commit will go + through TC log_and_order(), the commit ordering is done + by TC group commit. Otherwise the wait for prior + commits to complete is done in ha_commit_one_phase(). + */ + DBUG_VOID_RETURN; +} + +void wsrep_unregister_from_group_commit(THD *thd) +{ + DBUG_ASSERT(thd->wsrep_trx().ordered()); + wait_for_commit *wfc= thd->wait_for_commit_ptr; + + if (wfc) + { + mysql_mutex_lock(&LOCK_wsrep_group_commit); + wfc->unregister_wait_for_prior_commit(); + thd->wakeup_subsequent_commits(0); + + /* The last one queued for group commit has completed commit, it is + safe to set tail to NULL. */ + if (wfc == commit_order_tail) + commit_order_tail= NULL; + mysql_mutex_unlock(&LOCK_wsrep_group_commit); + thd->wait_for_commit_ptr= NULL; + } +} |