1berry/MariaDB-server
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

MDEV-4506, parallel replication.

Browse Source 
Some after-review fixes.
This commit is contained in:
unknown 2013-09-13 15:09:57 +02:00
parent 13fddb32de
commit d107bdaa01
16 changed files with 311 additions and 191 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

276
sql/log.cc
View File

@ -6542,26 +6542,87 @@ MYSQL_BIN_LOG::write_transaction_to_binlog(THD *thd,
} }
} }
/*
Put a transaction that is ready to commit in the group commit queue.
The transaction is identified by the ENTRY object passed into this function.
To facilitate group commit for the binlog, we first queue up ourselves in
this function. Then later the first thread to enter the queue waits for
the LOCK_log mutex, and commits for everyone in the queue once it gets the
lock. Any other threads in the queue just wait for the first one to finish
the commit and wake them up. This way, all transactions in the queue get
committed in a single disk operation.
The return value of this function is TRUE if queued as the first entry in
the queue (meaning this is the leader), FALSE otherwise.
The main work in this function is when the commit in one transaction has
been marked to wait for the commit of another transaction to happen
first. This is used to support in-order parallel replication, where
transactions can execute out-of-order but need to be committed in-order with
how they happened on the master. The waiting of one commit on another needs
to be integrated with the group commit queue, to ensure that the waiting
transaction can participate in the same group commit as the waited-for
transaction.
So when we put a transaction in the queue, we check if there were other
transactions already prepared to commit but just waiting for the first one
to commit. If so, we add those to the queue as well, transitively for all
waiters.
*/
bool bool
MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *entry, MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *entry)
wait_for_commit *wfc)
{ {
group_commit_entry *orig_queue; group_commit_entry *orig_queue;
wait_for_commit *list, *cur, *last; wait_for_commit *list, *cur, *last;
wait_for_commit *wfc;
/* /*
To facilitate group commit for the binlog, we first queue up ourselves in Check if we need to wait for another transaction to commit before us.
the group commit queue. Then the first thread to enter the queue waits for
the LOCK_log mutex, and commits for everyone in the queue once it gets the
lock. Any other threads in the queue just wait for the first one to finish
the commit and wake them up.
To support in-order parallel replication with group commit, after we add It is safe to do a quick check without lock first in the case where we do
some transaction to the queue, we check if there were other transactions not have to wait. But if the quick check shows we need to wait, we must do
already prepared to commit but just waiting for the first one to commit. another safe check under lock, to avoid the race where the other
If so, we add those to the queue as well, transitively for all waiters. transaction wakes us up between the check and the wait.
*/ */
wfc= entry->thd->wait_for_commit_ptr;
entry->queued_by_other= false;
if (wfc && wfc->waiting_for_commit)
{
mysql_mutex_lock(&wfc->LOCK_wait_commit);
/* Do an extra check here, this time safely under lock. */
if (wfc->waiting_for_commit)
{
/*
By setting wfc->opaque_pointer to our own entry, we mark that we are
ready to commit, but waiting for another transaction to commit before
us.
This other transaction may then take over the commit process for us to
get us included in its own group commit. If this happens, the
queued_by_other flag is set.
*/
wfc->opaque_pointer= entry;
do
{
mysql_cond_wait(&wfc->COND_wait_commit, &wfc->LOCK_wait_commit);
} while (wfc->waiting_for_commit);
wfc->opaque_pointer= NULL;
}
mysql_mutex_unlock(&wfc->LOCK_wait_commit);
}
/*
If the transaction we were waiting for has already put us into the group
commit queue (and possibly already done the entire binlog commit for us),
then there is nothing else to do.
*/
if (entry->queued_by_other)
return false;
/* Now enqueue ourselves in the group commit queue. */
entry->thd->clear_wakeup_ready(); entry->thd->clear_wakeup_ready();
mysql_mutex_lock(&LOCK_prepare_ordered); mysql_mutex_lock(&LOCK_prepare_ordered);
orig_queue= group_commit_queue; orig_queue= group_commit_queue;
@ -6574,6 +6635,23 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *entry,
This would be natural to do with recursion, but we want to avoid This would be natural to do with recursion, but we want to avoid
potentially unbounded recursion blowing the C stack, so we use the list potentially unbounded recursion blowing the C stack, so we use the list
approach instead. approach instead.
We keep a list of all the waiters that need to be processed in `list',
linked through the next_subsequent_commit pointer. Initially this list
contains only the entry passed into this function.
We process entries in the list one by one. The element currently being
processed is pointed to by `cur`, and the element at the end of the list
is pointed to by `last` (we do not use NULL to terminate the list).
As we process an element, it is first added to the group_commit_queue.
Then any waiters for that element are added at the end of the list, to
be processed in subsequent iterations. This continues until the list
is exhausted, with all elements ever added eventually processed.
The end result is a breath-first traversal of the tree of waiters,
re-using the next_subsequent_commit pointers in place of extra stack
space in a recursive traversal.
*/ */
list= wfc; list= wfc;
cur= list; cur= list;
@ -6594,6 +6672,12 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *entry,
if (!cur) if (!cur)
break; // Can happen if initial entry has no wait_for_commit break; // Can happen if initial entry has no wait_for_commit
/*
Check if this transaction has other transaction waiting for it to commit.
If so, process the waiting transactions, and their waiters and so on,
transitively.
*/
if (cur->subsequent_commits_list) if (cur->subsequent_commits_list)
{ {
bool have_lock; bool have_lock;
@ -6601,63 +6685,66 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *entry,
mysql_mutex_lock(&cur->LOCK_wait_commit); mysql_mutex_lock(&cur->LOCK_wait_commit);
have_lock= true; have_lock= true;
/*
Grab the list, now safely under lock, and process it if still
non-empty.
*/
waiter= cur->subsequent_commits_list; waiter= cur->subsequent_commits_list;
/* Check again, now safely under lock. */ cur->subsequent_commits_list= NULL;
if (waiter) while (waiter)
{ {
/* Grab the list of waiters and process it. */ wait_for_commit *next= waiter->next_subsequent_commit;
cur->subsequent_commits_list= NULL; group_commit_entry *entry2=
do (group_commit_entry *)waiter->opaque_pointer;
if (entry2)
{ {
wait_for_commit *next= waiter->next_subsequent_commit; /*
group_commit_entry *entry2= This is another transaction ready to be written to the binary
(group_commit_entry *)waiter->opaque_pointer; log. We can put it into the queue directly, without needing a
if (entry2) separate context switch to the other thread. We just set a flag
{ so that the other thread will know when it wakes up that it was
/* already processed.
This is another transaction ready to be written to the binary
log. We can put it into the queue directly, without needing a
separate context switch to the other thread. We just set a flag
so that the other thread will know when it wakes up that it was
already processed.
So put it at the end of the list to be processed in a subsequent So put it at the end of the list to be processed in a subsequent
iteration of the outer loop. iteration of the outer loop.
*/ */
entry2->queued_by_other= true; entry2->queued_by_other= true;
last->next_subsequent_commit= waiter; last->next_subsequent_commit= waiter;
last= waiter; last= waiter;
/* /*
As a small optimisation, we do not actually need to set As a small optimisation, we do not actually need to set
waiter->next_subsequent_commit to NULL, as we can use the waiter->next_subsequent_commit to NULL, as we can use the
pointer `last' to check for end-of-list. pointer `last' to check for end-of-list.
*/ */
} }
else else
{ {
/* /*
Wake up the waiting transaction. Wake up the waiting transaction.
For this, we need to set the "wakeup running" flag and release For this, we need to set the "wakeup running" flag and release
the waitee lock to avoid a deadlock, see comments on the waitee lock to avoid a deadlock, see comments on
THD::wakeup_subsequent_commits2() for details. THD::wakeup_subsequent_commits2() for details.
*/ */
if (have_lock) if (have_lock)
{ {
cur->wakeup_subsequent_commits_running= true; have_lock= false;
mysql_mutex_unlock(&cur->LOCK_wait_commit); cur->wakeup_subsequent_commits_running= true;
have_lock= false; mysql_mutex_unlock(&cur->LOCK_wait_commit);
}
waiter->wakeup();
} }
waiter= next; waiter->wakeup();
} while (waiter); }
waiter= next;
} }
if (have_lock) if (have_lock)
mysql_mutex_unlock(&cur->LOCK_wait_commit); mysql_mutex_unlock(&cur->LOCK_wait_commit);
} }
if (cur == last) if (cur == last)
break; break;
/*
Move to the next entry in the flattened list of waiting transactions
that still need to be processed transitively.
*/
cur= cur->next_subsequent_commit; cur= cur->next_subsequent_commit;
entry= (group_commit_entry *)cur->opaque_pointer; entry= (group_commit_entry *)cur->opaque_pointer;
DBUG_ASSERT(entry != NULL); DBUG_ASSERT(entry != NULL);
@ -6691,31 +6778,7 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *entry,
bool bool
MYSQL_BIN_LOG::write_transaction_to_binlog_events(group_commit_entry *entry) MYSQL_BIN_LOG::write_transaction_to_binlog_events(group_commit_entry *entry)
{ {
wait_for_commit *wfc; bool is_leader= queue_for_group_commit(entry);
bool is_leader;
wfc= entry->thd->wait_for_commit_ptr;
entry->queued_by_other= false;
if (wfc && wfc->waiting_for_commit)
{
mysql_mutex_lock(&wfc->LOCK_wait_commit);
/* Do an extra check here, this time safely under lock. */
if (wfc->waiting_for_commit)
{
wfc->opaque_pointer= entry;
do
{
mysql_cond_wait(&wfc->COND_wait_commit, &wfc->LOCK_wait_commit);
} while (wfc->waiting_for_commit);
wfc->opaque_pointer= NULL;
}
mysql_mutex_unlock(&wfc->LOCK_wait_commit);
}
if (entry->queued_by_other)
is_leader= false;
else
is_leader= queue_for_group_commit(entry, wfc);
/* /*
The first in the queue handles group commit for all; the others just wait The first in the queue handles group commit for all; the others just wait
@ -6756,6 +6819,16 @@ MYSQL_BIN_LOG::write_transaction_to_binlog_events(group_commit_entry *entry)
if (next) if (next)
{ {
/*
Wake up the next thread in the group commit.
The next thread can be waiting in two different ways, depending on
whether it put itself in the queue, or if it was put in queue by us
because it had to wait for us to commit first.
So execute the appropriate wakeup, identified by the queued_by_other
field.
*/
if (next->queued_by_other) if (next->queued_by_other)
next->thd->wait_for_commit_ptr->wakeup(); next->thd->wait_for_commit_ptr->wakeup();
else else
@ -6840,14 +6913,18 @@ MYSQL_BIN_LOG::trx_group_commit_leader(group_commit_entry *leader)
*/ */
mysql_mutex_lock(&LOCK_log); mysql_mutex_lock(&LOCK_log);
DEBUG_SYNC(leader->thd, "commit_after_get_LOCK_log"); DEBUG_SYNC(leader->thd, "commit_after_get_LOCK_log");
binlog_id= current_binlog_id;
mysql_mutex_lock(&LOCK_prepare_ordered); mysql_mutex_lock(&LOCK_prepare_ordered);
if (opt_binlog_commit_wait_count) if (opt_binlog_commit_wait_count)
wait_for_sufficient_commits(); wait_for_sufficient_commits();
/*
Note that wait_for_sufficient_commits() may have released and
re-acquired the LOCK_log and LOCK_prepare_ordered if it needed to wait.
*/
current= group_commit_queue; current= group_commit_queue;
group_commit_queue= NULL; group_commit_queue= NULL;
mysql_mutex_unlock(&LOCK_prepare_ordered); mysql_mutex_unlock(&LOCK_prepare_ordered);
binlog_id= current_binlog_id;
/* As the queue is in reverse order of entering, reverse it. */ /* As the queue is in reverse order of entering, reverse it. */
last_in_queue= current; last_in_queue= current;
@ -7141,6 +7218,13 @@ MYSQL_BIN_LOG::write_transaction_or_stmt(group_commit_entry *entry,
} }
/*
Wait for sufficient commits to queue up for group commit, according to the
values of binlog_commit_wait_count and binlog_commit_wait_usec.
Note that this function may release and re-acquire LOCK_log and
LOCK_prepare_ordered if it needs to wait.
*/
void void
MYSQL_BIN_LOG::wait_for_sufficient_commits() MYSQL_BIN_LOG::wait_for_sufficient_commits()
{ {
@ -7152,11 +7236,9 @@ MYSQL_BIN_LOG::wait_for_sufficient_commits()
mysql_mutex_assert_owner(&LOCK_log); mysql_mutex_assert_owner(&LOCK_log);
mysql_mutex_assert_owner(&LOCK_prepare_ordered); mysql_mutex_assert_owner(&LOCK_prepare_ordered);
count= 0; for (e= last_head= group_commit_queue, count= 0; e; e= e->next)
for (e= last_head= group_commit_queue; e; e= e->next) if (++count >= opt_binlog_commit_wait_count)
++count; return;
if (count >= opt_binlog_commit_wait_count)
return;
mysql_mutex_unlock(&LOCK_log); mysql_mutex_unlock(&LOCK_log);
set_timespec_nsec(wait_until, (ulonglong)1000*opt_binlog_commit_wait_usec); set_timespec_nsec(wait_until, (ulonglong)1000*opt_binlog_commit_wait_usec);
@ -7178,7 +7260,25 @@ MYSQL_BIN_LOG::wait_for_sufficient_commits()
last_head= head; last_head= head;
} }
mysql_mutex_lock(&LOCK_log); /*
We must not wait for LOCK_log while holding LOCK_prepare_ordered.
LOCK_log can be held for long periods (eg. we do I/O under it), while
LOCK_prepare_ordered must only be held for short periods.
In addition, waiting for LOCK_log while holding LOCK_prepare_ordered would
violate locking order of LOCK_log-before-LOCK_prepare_ordered. This could
cause SAFEMUTEX warnings (even if it cannot actually deadlock with current
code, as there can be at most one group commit leader thread at a time).
So release and re-acquire LOCK_prepare_ordered if we need to wait for the
LOCK_log.
*/
if (mysql_mutex_trylock(&LOCK_log))
{
mysql_mutex_unlock(&LOCK_prepare_ordered);
mysql_mutex_lock(&LOCK_log);
mysql_mutex_lock(&LOCK_prepare_ordered);
}
} }

2
sql/log.h
View File

@ -540,7 +540,7 @@ class MYSQL_BIN_LOG: public TC_LOG, private MYSQL_LOG
void do_checkpoint_request(ulong binlog_id); void do_checkpoint_request(ulong binlog_id);
void purge(); void purge();
int write_transaction_or_stmt(group_commit_entry *entry, uint64 commit_id); int write_transaction_or_stmt(group_commit_entry *entry, uint64 commit_id);
bool queue_for_group_commit(group_commit_entry *entry, wait_for_commit *wfc); bool queue_for_group_commit(group_commit_entry *entry);
bool write_transaction_to_binlog_events(group_commit_entry *entry); bool write_transaction_to_binlog_events(group_commit_entry *entry);
void trx_group_commit_leader(group_commit_entry *leader); void trx_group_commit_leader(group_commit_entry *leader);
bool is_xidlist_idle_nolock(); bool is_xidlist_idle_nolock();

72
sql/log_event.cc
View File

@ -937,7 +937,7 @@ Log_event::Log_event(const char* buf,
#ifndef MYSQL_CLIENT #ifndef MYSQL_CLIENT
#ifdef HAVE_REPLICATION #ifdef HAVE_REPLICATION
int Log_event::do_update_pos(struct rpl_group_info *rgi) int Log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
/* /*
@ -3756,7 +3756,7 @@ void Query_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
int Query_log_event::do_apply_event(struct rpl_group_info *rgi) int Query_log_event::do_apply_event(rpl_group_info *rgi)
{ {
return do_apply_event(rgi, query, q_len); return do_apply_event(rgi, query, q_len);
} }
@ -3807,8 +3807,8 @@ bool test_if_equal_repl_errors(int expected_error, int actual_error)
mismatch. This mismatch could be implemented with a new ER_ code, and mismatch. This mismatch could be implemented with a new ER_ code, and
to ignore it you would use --slave-skip-errors... to ignore it you would use --slave-skip-errors...
*/ */
int Query_log_event::do_apply_event(struct rpl_group_info *rgi, int Query_log_event::do_apply_event(rpl_group_info *rgi,
const char *query_arg, uint32 q_len_arg) const char *query_arg, uint32 q_len_arg)
{ {
LEX_STRING new_db; LEX_STRING new_db;
int expected_error,actual_error= 0; int expected_error,actual_error= 0;
@ -4244,7 +4244,7 @@ end:
DBUG_RETURN(thd->is_slave_error); DBUG_RETURN(thd->is_slave_error);
} }
int Query_log_event::do_update_pos(struct rpl_group_info *rgi) int Query_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
/* /*
@ -4461,7 +4461,7 @@ bool Start_log_event_v3::write(IO_CACHE* file)
other words, no deadlock problem. other words, no deadlock problem.
*/ */
int Start_log_event_v3::do_apply_event(struct rpl_group_info *rgi) int Start_log_event_v3::do_apply_event(rpl_group_info *rgi)
{ {
DBUG_ENTER("Start_log_event_v3::do_apply_event"); DBUG_ENTER("Start_log_event_v3::do_apply_event");
int error= 0; int error= 0;
@ -4810,7 +4810,7 @@ bool Format_description_log_event::write(IO_CACHE* file)
#endif #endif
#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
int Format_description_log_event::do_apply_event(struct rpl_group_info *rgi) int Format_description_log_event::do_apply_event(rpl_group_info *rgi)
{ {
int ret= 0; int ret= 0;
Relay_log_info const *rli= rgi->rli; Relay_log_info const *rli= rgi->rli;
@ -4867,7 +4867,7 @@ int Format_description_log_event::do_apply_event(struct rpl_group_info *rgi)
DBUG_RETURN(ret); DBUG_RETURN(ret);
} }
int Format_description_log_event::do_update_pos(struct rpl_group_info *rgi) int Format_description_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
if (server_id == (uint32) global_system_variables.server_id) if (server_id == (uint32) global_system_variables.server_id)
@ -5516,7 +5516,7 @@ void Load_log_event::set_fields(const char* affected_db,
1 Failure 1 Failure
*/ */
int Load_log_event::do_apply_event(NET* net, struct rpl_group_info *rgi, int Load_log_event::do_apply_event(NET* net, rpl_group_info *rgi,
bool use_rli_only_for_errors) bool use_rli_only_for_errors)
{ {
LEX_STRING new_db; LEX_STRING new_db;
@ -5919,7 +5919,7 @@ bool Rotate_log_event::write(IO_CACHE* file)
@retval @retval
0 ok 0 ok
*/ */
int Rotate_log_event::do_update_pos(struct rpl_group_info *rgi) int Rotate_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
DBUG_ENTER("Rotate_log_event::do_update_pos"); DBUG_ENTER("Rotate_log_event::do_update_pos");
@ -6096,7 +6096,7 @@ bool Binlog_checkpoint_log_event::write(IO_CACHE *file)
Gtid_log_event::Gtid_log_event(const char *buf, uint event_len, Gtid_log_event::Gtid_log_event(const char *buf, uint event_len,
const Format_description_log_event *description_event) const Format_description_log_event *description_event)
: Log_event(buf, description_event), seq_no(0) : Log_event(buf, description_event), seq_no(0), commit_id(0)
{ {
uint8 header_size= description_event->common_header_len; uint8 header_size= description_event->common_header_len;
uint8 post_header_len= description_event->post_header_len[GTID_EVENT-1]; uint8 post_header_len= description_event->post_header_len[GTID_EVENT-1];
@ -6120,8 +6120,6 @@ Gtid_log_event::Gtid_log_event(const char *buf, uint event_len,
++buf; ++buf;
commit_id= uint8korr(buf); commit_id= uint8korr(buf);
} }
else
commit_id= 0;
} }
@ -6254,7 +6252,7 @@ Gtid_log_event::pack_info(THD *thd, Protocol *protocol)
static char gtid_begin_string[] = "BEGIN"; static char gtid_begin_string[] = "BEGIN";
int int
Gtid_log_event::do_apply_event(struct rpl_group_info *rgi) Gtid_log_event::do_apply_event(rpl_group_info *rgi)
{ {
thd->variables.server_id= this->server_id; thd->variables.server_id= this->server_id;
thd->variables.gtid_domain_id= this->domain_id; thd->variables.gtid_domain_id= this->domain_id;
@ -6295,7 +6293,7 @@ Gtid_log_event::do_apply_event(struct rpl_group_info *rgi)
int int
Gtid_log_event::do_update_pos(struct rpl_group_info *rgi) Gtid_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
rli->inc_event_relay_log_pos(); rli->inc_event_relay_log_pos();
@ -6477,7 +6475,7 @@ Gtid_list_log_event::write(IO_CACHE *file)
int int
Gtid_list_log_event::do_apply_event(struct rpl_group_info *rgi) Gtid_list_log_event::do_apply_event(rpl_group_info *rgi)
{ {
Relay_log_info const *rli= rgi->rli; Relay_log_info const *rli= rgi->rli;
int ret= Log_event::do_apply_event(rgi); int ret= Log_event::do_apply_event(rgi);
@ -6707,7 +6705,7 @@ void Intvar_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
Intvar_log_event::do_apply_event() Intvar_log_event::do_apply_event()
*/ */
int Intvar_log_event::do_apply_event(struct rpl_group_info *rgi) int Intvar_log_event::do_apply_event(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
/* /*
@ -6731,7 +6729,7 @@ int Intvar_log_event::do_apply_event(struct rpl_group_info *rgi)
return 0; return 0;
} }
int Intvar_log_event::do_update_pos(struct rpl_group_info *rgi) int Intvar_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
rli->inc_event_relay_log_pos(); rli->inc_event_relay_log_pos();
@ -6818,7 +6816,7 @@ void Rand_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
int Rand_log_event::do_apply_event(struct rpl_group_info *rgi) int Rand_log_event::do_apply_event(rpl_group_info *rgi)
{ {
Relay_log_info const *rli= rgi->rli; Relay_log_info const *rli= rgi->rli;
/* /*
@ -6835,7 +6833,7 @@ int Rand_log_event::do_apply_event(struct rpl_group_info *rgi)
return 0; return 0;
} }
int Rand_log_event::do_update_pos(struct rpl_group_info *rgi) int Rand_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
rli->inc_event_relay_log_pos(); rli->inc_event_relay_log_pos();
@ -6950,7 +6948,7 @@ void Xid_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
int Xid_log_event::do_apply_event(struct rpl_group_info *rgi) int Xid_log_event::do_apply_event(rpl_group_info *rgi)
{ {
bool res; bool res;
int err; int err;
@ -7416,7 +7414,7 @@ void User_var_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
*/ */
#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
int User_var_log_event::do_apply_event(struct rpl_group_info *rgi) int User_var_log_event::do_apply_event(rpl_group_info *rgi)
{ {
Item *it= 0; Item *it= 0;
CHARSET_INFO *charset; CHARSET_INFO *charset;
@ -7505,7 +7503,7 @@ int User_var_log_event::do_apply_event(struct rpl_group_info *rgi)
DBUG_RETURN(0); DBUG_RETURN(0);
} }
int User_var_log_event::do_update_pos(struct rpl_group_info *rgi) int User_var_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
rli->inc_event_relay_log_pos(); rli->inc_event_relay_log_pos();
@ -7682,7 +7680,7 @@ Slave_log_event::Slave_log_event(const char* buf,
#ifndef MYSQL_CLIENT #ifndef MYSQL_CLIENT
int Slave_log_event::do_apply_event(struct rpl_group_info *rgi) int Slave_log_event::do_apply_event(rpl_group_info *rgi)
{ {
if (mysql_bin_log.is_open()) if (mysql_bin_log.is_open())
return mysql_bin_log.write(this); return mysql_bin_log.write(this);
@ -7726,7 +7724,7 @@ void Stop_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
Start_log_event_v3::do_apply_event(), not here. Because if we come Start_log_event_v3::do_apply_event(), not here. Because if we come
here, the master was sane. here, the master was sane.
*/ */
int Stop_log_event::do_update_pos(struct rpl_group_info *rgi) int Stop_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
/* /*
@ -7958,7 +7956,7 @@ void Create_file_log_event::pack_info(THD *thd, Protocol *protocol)
*/ */
#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
int Create_file_log_event::do_apply_event(struct rpl_group_info *rgi) int Create_file_log_event::do_apply_event(rpl_group_info *rgi)
{ {
char proc_info[17+FN_REFLEN+10], *fname_buf; char proc_info[17+FN_REFLEN+10], *fname_buf;
char *ext; char *ext;
@ -8140,7 +8138,7 @@ int Append_block_log_event::get_create_or_append() const
Append_block_log_event::do_apply_event() Append_block_log_event::do_apply_event()
*/ */
int Append_block_log_event::do_apply_event(struct rpl_group_info *rgi) int Append_block_log_event::do_apply_event(rpl_group_info *rgi)
{ {
char proc_info[17+FN_REFLEN+10], *fname= proc_info+17; char proc_info[17+FN_REFLEN+10], *fname= proc_info+17;
int fd; int fd;
@ -8291,7 +8289,7 @@ void Delete_file_log_event::pack_info(THD *thd, Protocol *protocol)
*/ */
#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
int Delete_file_log_event::do_apply_event(struct rpl_group_info *rgi) int Delete_file_log_event::do_apply_event(rpl_group_info *rgi)
{ {
char fname[FN_REFLEN+10]; char fname[FN_REFLEN+10];
Relay_log_info const *rli= rgi->rli; Relay_log_info const *rli= rgi->rli;
@ -8391,7 +8389,7 @@ void Execute_load_log_event::pack_info(THD *thd, Protocol *protocol)
Execute_load_log_event::do_apply_event() Execute_load_log_event::do_apply_event()
*/ */
int Execute_load_log_event::do_apply_event(struct rpl_group_info *rgi) int Execute_load_log_event::do_apply_event(rpl_group_info *rgi)
{ {
char fname[FN_REFLEN+10]; char fname[FN_REFLEN+10];
char *ext; char *ext;
@ -8664,7 +8662,7 @@ void Execute_load_query_log_event::pack_info(THD *thd, Protocol *protocol)
int int
Execute_load_query_log_event::do_apply_event(struct rpl_group_info *rgi) Execute_load_query_log_event::do_apply_event(rpl_group_info *rgi)
{ {
char *p; char *p;
char *buf; char *buf;
@ -9072,7 +9070,7 @@ int Rows_log_event::do_add_row_data(uchar *row_data, size_t length)
#endif #endif
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
int Rows_log_event::do_apply_event(struct rpl_group_info *rgi) int Rows_log_event::do_apply_event(rpl_group_info *rgi)
{ {
Relay_log_info const *rli= rgi->rli; Relay_log_info const *rli= rgi->rli;
DBUG_ENTER("Rows_log_event::do_apply_event(Relay_log_info*)"); DBUG_ENTER("Rows_log_event::do_apply_event(Relay_log_info*)");
@ -9538,7 +9536,7 @@ static int rows_event_stmt_cleanup(Relay_log_info const *rli, THD * thd)
@retval non-zero Error in the statement commit @retval non-zero Error in the statement commit
*/ */
int int
Rows_log_event::do_update_pos(struct rpl_group_info *rgi) Rows_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
DBUG_ENTER("Rows_log_event::do_update_pos"); DBUG_ENTER("Rows_log_event::do_update_pos");
@ -9777,7 +9775,7 @@ void Annotate_rows_log_event::print(FILE *file, PRINT_EVENT_INFO *pinfo)
#endif #endif
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
int Annotate_rows_log_event::do_apply_event(struct rpl_group_info *rgi) int Annotate_rows_log_event::do_apply_event(rpl_group_info *rgi)
{ {
m_save_thd_query_txt= thd->query(); m_save_thd_query_txt= thd->query();
m_save_thd_query_len= thd->query_length(); m_save_thd_query_len= thd->query_length();
@ -9787,7 +9785,7 @@ int Annotate_rows_log_event::do_apply_event(struct rpl_group_info *rgi)
#endif #endif
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
int Annotate_rows_log_event::do_update_pos(struct rpl_group_info *rgi) int Annotate_rows_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
rli->inc_event_relay_log_pos(); rli->inc_event_relay_log_pos();
@ -10296,7 +10294,7 @@ check_table_map(Relay_log_info const *rli, RPL_TABLE_LIST *table_list)
DBUG_RETURN(res); DBUG_RETURN(res);
} }
int Table_map_log_event::do_apply_event(struct rpl_group_info *rgi) int Table_map_log_event::do_apply_event(rpl_group_info *rgi)
{ {
RPL_TABLE_LIST *table_list; RPL_TABLE_LIST *table_list;
char *db_mem, *tname_mem; char *db_mem, *tname_mem;
@ -10415,7 +10413,7 @@ Table_map_log_event::do_shall_skip(Relay_log_info *rli)
return continue_group(rli); return continue_group(rli);
} }
int Table_map_log_event::do_update_pos(struct rpl_group_info *rgi) int Table_map_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
rli->inc_event_relay_log_pos(); rli->inc_event_relay_log_pos();
@ -11847,7 +11845,7 @@ Incident_log_event::print(FILE *file,
#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
int int
Incident_log_event::do_apply_event(struct rpl_group_info *rgi) Incident_log_event::do_apply_event(rpl_group_info *rgi)
{ {
Relay_log_info const *rli= rgi->rli; Relay_log_info const *rli= rgi->rli;
DBUG_ENTER("Incident_log_event::do_apply_event"); DBUG_ENTER("Incident_log_event::do_apply_event");

74
sql/log_event.h
View File

@ -1317,7 +1317,7 @@ public:
@see do_apply_event @see do_apply_event
*/ */
int apply_event(struct rpl_group_info *rgi) int apply_event(rpl_group_info *rgi)
{ {
return do_apply_event(rgi); return do_apply_event(rgi);
} }
@ -1331,7 +1331,7 @@ public:
@see do_update_pos @see do_update_pos
*/ */
int update_pos(struct rpl_group_info *rgi) int update_pos(rpl_group_info *rgi)
{ {
return do_update_pos(rgi); return do_update_pos(rgi);
} }
@ -1432,7 +1432,7 @@ protected:
@retval 0 Event applied successfully @retval 0 Event applied successfully
@retval errno Error code if event application failed @retval errno Error code if event application failed
*/ */
virtual int do_apply_event(struct rpl_group_info *rgi) virtual int do_apply_event(rpl_group_info *rgi)
{ {
return 0; /* Default implementation does nothing */ return 0; /* Default implementation does nothing */
} }
@ -1461,7 +1461,7 @@ protected:
1). Observe that handler errors are returned by the 1). Observe that handler errors are returned by the
do_apply_event() function, and not by this one. do_apply_event() function, and not by this one.
*/ */
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
/** /**
@ -1986,10 +1986,10 @@ public:
public: /* !!! Public in this patch to allow old usage */ public: /* !!! Public in this patch to allow old usage */
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli); virtual enum_skip_reason do_shall_skip(Relay_log_info *rli);
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
int do_apply_event(struct rpl_group_info *rgi, int do_apply_event(rpl_group_info *rgi,
const char *query_arg, const char *query_arg,
uint32 q_len_arg); uint32 q_len_arg);
static bool peek_is_commit_rollback(const char *event_start, static bool peek_is_commit_rollback(const char *event_start,
@ -2103,7 +2103,7 @@ public:
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
#endif #endif
}; };
@ -2416,12 +2416,12 @@ public:
public: /* !!! Public in this patch to allow old usage */ public: /* !!! Public in this patch to allow old usage */
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi) virtual int do_apply_event(rpl_group_info *rgi)
{ {
return do_apply_event(thd->slave_net,rgi,0); return do_apply_event(thd->slave_net,rgi,0);
} }
int do_apply_event(NET *net, struct rpl_group_info *rgi, int do_apply_event(NET *net, rpl_group_info *rgi,
bool use_rli_only_for_errors); bool use_rli_only_for_errors);
#endif #endif
}; };
@ -2500,7 +2500,7 @@ public:
protected: protected:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info*) virtual enum_skip_reason do_shall_skip(Relay_log_info*)
{ {
/* /*
@ -2596,8 +2596,8 @@ public:
static bool is_version_before_checksum(const master_version_split *version_split); static bool is_version_before_checksum(const master_version_split *version_split);
protected: protected:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli); virtual enum_skip_reason do_shall_skip(Relay_log_info *rli);
#endif #endif
}; };
@ -2675,8 +2675,8 @@ Intvar_log_event(THD* thd_arg,uchar type_arg, ulonglong val_arg,
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli); virtual enum_skip_reason do_shall_skip(Relay_log_info *rli);
#endif #endif
}; };
@ -2754,8 +2754,8 @@ class Rand_log_event: public Log_event
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli); virtual enum_skip_reason do_shall_skip(Relay_log_info *rli);
#endif #endif
}; };
@ -2803,7 +2803,7 @@ class Xid_log_event: public Log_event
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
enum_skip_reason do_shall_skip(Relay_log_info *rli); enum_skip_reason do_shall_skip(Relay_log_info *rli);
#endif #endif
}; };
@ -2870,8 +2870,8 @@ public:
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli); virtual enum_skip_reason do_shall_skip(Relay_log_info *rli);
#endif #endif
}; };
@ -2905,7 +2905,7 @@ public:
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli) virtual enum_skip_reason do_shall_skip(Relay_log_info *rli)
{ {
/* /*
@ -3007,7 +3007,7 @@ public:
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli); virtual enum_skip_reason do_shall_skip(Relay_log_info *rli);
#endif #endif
}; };
@ -3119,8 +3119,8 @@ public:
uint16 flags, bool is_transactional, uint64 commit_id); uint16 flags, bool is_transactional, uint64 commit_id);
#ifdef HAVE_REPLICATION #ifdef HAVE_REPLICATION
void pack_info(THD *thd, Protocol *protocol); void pack_info(THD *thd, Protocol *protocol);
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli); virtual enum_skip_reason do_shall_skip(Relay_log_info *rli);
#endif #endif
#else #else
@ -3249,7 +3249,7 @@ public:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
bool to_packet(String *packet); bool to_packet(String *packet);
bool write(IO_CACHE *file); bool write(IO_CACHE *file);
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
#endif #endif
static bool peek(const char *event_start, uint32 event_len, static bool peek(const char *event_start, uint32 event_len,
uint8 checksum_alg, uint8 checksum_alg,
@ -3328,7 +3328,7 @@ public:
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
#endif #endif
}; };
@ -3383,7 +3383,7 @@ public:
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
#endif #endif
}; };
@ -3424,7 +3424,7 @@ public:
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
#endif #endif
}; };
@ -3464,7 +3464,7 @@ public:
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
#endif #endif
}; };
@ -3563,7 +3563,7 @@ public:
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
#endif #endif
}; };
@ -3635,8 +3635,8 @@ public:
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
private: private:
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info*); virtual enum_skip_reason do_shall_skip(Relay_log_info*);
#endif #endif
@ -4050,8 +4050,8 @@ public:
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli); virtual enum_skip_reason do_shall_skip(Relay_log_info *rli);
#endif #endif
@ -4278,8 +4278,8 @@ protected:
private: private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli); virtual enum_skip_reason do_shall_skip(Relay_log_info *rli);
/* /*
@ -4612,7 +4612,7 @@ public:
#endif #endif
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
#endif #endif
virtual bool write_data_header(IO_CACHE *file); virtual bool write_data_header(IO_CACHE *file);

6
sql/log_event_old.cc
View File

@ -36,7 +36,7 @@
// Old implementation of do_apply_event() // Old implementation of do_apply_event()
int int
Old_rows_log_event::do_apply_event(Old_rows_log_event *ev, struct rpl_group_info *rgi) Old_rows_log_event::do_apply_event(Old_rows_log_event *ev, rpl_group_info *rgi)
{ {
DBUG_ENTER("Old_rows_log_event::do_apply_event(st_relay_log_info*)"); DBUG_ENTER("Old_rows_log_event::do_apply_event(st_relay_log_info*)");
int error= 0; int error= 0;
@ -1451,7 +1451,7 @@ int Old_rows_log_event::do_add_row_data(uchar *row_data, size_t length)
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
int Old_rows_log_event::do_apply_event(struct rpl_group_info *rgi) int Old_rows_log_event::do_apply_event(rpl_group_info *rgi)
{ {
DBUG_ENTER("Old_rows_log_event::do_apply_event(Relay_log_info*)"); DBUG_ENTER("Old_rows_log_event::do_apply_event(Relay_log_info*)");
int error= 0; int error= 0;
@ -1834,7 +1834,7 @@ Old_rows_log_event::do_shall_skip(Relay_log_info *rli)
} }
int int
Old_rows_log_event::do_update_pos(struct rpl_group_info *rgi) Old_rows_log_event::do_update_pos(rpl_group_info *rgi)
{ {
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
DBUG_ENTER("Old_rows_log_event::do_update_pos"); DBUG_ENTER("Old_rows_log_event::do_update_pos");

12
sql/log_event_old.h
View File

@ -214,8 +214,8 @@ protected:
private: private:
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
virtual int do_apply_event(struct rpl_group_info *rgi); virtual int do_apply_event(rpl_group_info *rgi);
virtual int do_update_pos(struct rpl_group_info *rgi); virtual int do_update_pos(rpl_group_info *rgi);
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli); virtual enum_skip_reason do_shall_skip(Relay_log_info *rli);
/* /*
@ -275,7 +275,7 @@ private:
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
int do_apply_event(Old_rows_log_event*, struct rpl_group_info *rgi); int do_apply_event(Old_rows_log_event*, rpl_group_info *rgi);
/* /*
Primitive to prepare for a sequence of row executions. Primitive to prepare for a sequence of row executions.
@ -403,7 +403,7 @@ private:
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
// use old definition of do_apply_event() // use old definition of do_apply_event()
virtual int do_apply_event(struct rpl_group_info *rgi) virtual int do_apply_event(rpl_group_info *rgi)
{ return Old_rows_log_event::do_apply_event(this, rgi); } { return Old_rows_log_event::do_apply_event(this, rgi); }
// primitives for old version of do_apply_event() // primitives for old version of do_apply_event()
@ -481,7 +481,7 @@ private:
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
// use old definition of do_apply_event() // use old definition of do_apply_event()
virtual int do_apply_event(struct rpl_group_info *rgi) virtual int do_apply_event(rpl_group_info *rgi)
{ return Old_rows_log_event::do_apply_event(this, rgi); } { return Old_rows_log_event::do_apply_event(this, rgi); }
// primitives for old version of do_apply_event() // primitives for old version of do_apply_event()
@ -556,7 +556,7 @@ private:
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
// use old definition of do_apply_event() // use old definition of do_apply_event()
virtual int do_apply_event(struct rpl_group_info *rgi) virtual int do_apply_event(rpl_group_info *rgi)
{ return Old_rows_log_event::do_apply_event(this, rgi); } { return Old_rows_log_event::do_apply_event(this, rgi); }
// primitives for old version of do_apply_event() // primitives for old version of do_apply_event()

2
sql/rpl_gtid.cc
View File

@ -62,7 +62,7 @@ rpl_slave_state::update_state_hash(uint64 sub_id, rpl_gtid *gtid)
int int
rpl_slave_state::record_and_update_gtid(THD *thd, struct rpl_group_info *rgi) rpl_slave_state::record_and_update_gtid(THD *thd, rpl_group_info *rgi)
{ {
uint64 sub_id; uint64 sub_id;

15
sql/rpl_parallel.cc
View File

@ -62,7 +62,7 @@ rpt_handle_event(rpl_parallel_thread::queued_event *qev,
struct rpl_parallel_thread *rpt) struct rpl_parallel_thread *rpt)
{ {
int err; int err;
struct rpl_group_info *rgi= qev->rgi; rpl_group_info *rgi= qev->rgi;
Relay_log_info *rli= rgi->rli; Relay_log_info *rli= rgi->rli;
THD *thd= rgi->thd; THD *thd= rgi->thd;
@ -128,8 +128,9 @@ handle_rpl_parallel_thread(void *arg)
old_msg= thd->proc_info; old_msg= thd->proc_info;
thd->enter_cond(&rpt->COND_rpl_thread, &rpt->LOCK_rpl_thread, thd->enter_cond(&rpt->COND_rpl_thread, &rpt->LOCK_rpl_thread,
"Waiting for work from SQL thread"); "Waiting for work from SQL thread");
while (!rpt->stop && !thd->killed && !(events= rpt->event_queue)) while (!(events= rpt->event_queue) && !rpt->stop && !thd->killed)
mysql_cond_wait(&rpt->COND_rpl_thread, &rpt->LOCK_rpl_thread); mysql_cond_wait(&rpt->COND_rpl_thread, &rpt->LOCK_rpl_thread);
/* Mark that this thread is now executing */
rpt->free= false; rpt->free= false;
rpt->event_queue= rpt->last_in_queue= NULL; rpt->event_queue= rpt->last_in_queue= NULL;
thd->exit_cond(old_msg); thd->exit_cond(old_msg);
@ -145,9 +146,15 @@ handle_rpl_parallel_thread(void *arg)
uint64 wait_start_sub_id; uint64 wait_start_sub_id;
bool end_of_group; bool end_of_group;
/* Handle a new event group, which will be initiated by a GTID event. */
if (event_type == GTID_EVENT) if (event_type == GTID_EVENT)
{ {
in_event_group= true; in_event_group= true;
/*
If the standalone flag is set, then this event group consists of a
single statement (possibly preceeded by some Intvar_log_event and
similar), without any terminating COMMIT/ROLLBACK/XID.
*/
group_standalone= group_standalone=
(0 != (static_cast<Gtid_log_event *>(events->ev)->flags2 & (0 != (static_cast<Gtid_log_event *>(events->ev)->flags2 &
Gtid_log_event::FL_STANDALONE)); Gtid_log_event::FL_STANDALONE));
@ -540,12 +547,12 @@ rpl_parallel::wait_for_done()
bool bool
rpl_parallel::do_event(struct rpl_group_info *serial_rgi, Log_event *ev) rpl_parallel::do_event(rpl_group_info *serial_rgi, Log_event *ev)
{ {
rpl_parallel_entry *e; rpl_parallel_entry *e;
rpl_parallel_thread *cur_thread; rpl_parallel_thread *cur_thread;
rpl_parallel_thread::queued_event *qev; rpl_parallel_thread::queued_event *qev;
struct rpl_group_info *rgi; rpl_group_info *rgi;
Relay_log_info *rli= serial_rgi->rli; Relay_log_info *rli= serial_rgi->rli;
enum Log_event_type typ; enum Log_event_type typ;

6
sql/rpl_parallel.h
View File

@ -23,7 +23,7 @@ struct rpl_parallel_thread {
struct queued_event { struct queued_event {
queued_event *next; queued_event *next;
Log_event *ev; Log_event *ev;
struct rpl_group_info *rgi; rpl_group_info *rgi;
} *event_queue, *last_in_queue; } *event_queue, *last_in_queue;
}; };
@ -59,7 +59,7 @@ struct rpl_parallel_entry {
mysql_mutex_t LOCK_parallel_entry; mysql_mutex_t LOCK_parallel_entry;
mysql_cond_t COND_parallel_entry; mysql_cond_t COND_parallel_entry;
uint64 current_sub_id; uint64 current_sub_id;
struct rpl_group_info *current_group_info; rpl_group_info *current_group_info;
/* /*
The sub_id of the last event group in the previous batch of group-committed The sub_id of the last event group in the previous batch of group-committed
transactions. transactions.
@ -78,7 +78,7 @@ struct rpl_parallel {
~rpl_parallel(); ~rpl_parallel();
rpl_parallel_entry *find(uint32 domain_id); rpl_parallel_entry *find(uint32 domain_id);
void wait_for_done(); void wait_for_done();
bool do_event(struct rpl_group_info *serial_rgi, Log_event *ev); bool do_event(rpl_group_info *serial_rgi, Log_event *ev);
}; };

8
sql/rpl_rli.cc
View File

@ -1193,7 +1193,7 @@ bool Relay_log_info::cached_charset_compare(char *charset) const
void Relay_log_info::stmt_done(my_off_t event_master_log_pos, void Relay_log_info::stmt_done(my_off_t event_master_log_pos,
time_t event_creation_time, THD *thd, time_t event_creation_time, THD *thd,
struct rpl_group_info *rgi) rpl_group_info *rgi)
{ {
#ifndef DBUG_OFF #ifndef DBUG_OFF
extern uint debug_not_change_ts_if_art_event; extern uint debug_not_change_ts_if_art_event;
@ -1265,6 +1265,11 @@ void Relay_log_info::cleanup_context(THD *thd, bool error)
{ {
DBUG_ENTER("Relay_log_info::cleanup_context"); DBUG_ENTER("Relay_log_info::cleanup_context");
/*
In parallel replication, different THDs can be used from different
parallel threads. But in single-threaded mode, only the THD of the main
SQL thread is allowed.
*/
DBUG_ASSERT(opt_slave_parallel_threads > 0 || sql_thd == thd); DBUG_ASSERT(opt_slave_parallel_threads > 0 || sql_thd == thd);
/* /*
1) Instances of Table_map_log_event, if ::do_apply_event() was called on them, 1) Instances of Table_map_log_event, if ::do_apply_event() was called on them,
@ -1552,6 +1557,7 @@ event_group_new_gtid(rpl_group_info *rgi, Gtid_log_event *gev)
uint64 sub_id= rpl_global_gtid_slave_state.next_subid(gev->domain_id); uint64 sub_id= rpl_global_gtid_slave_state.next_subid(gev->domain_id);
if (!sub_id) if (!sub_id)
{ {
/* Out of memory caused hash insertion to fail. */
return 1; return 1;
} }
rgi->gtid_sub_id= sub_id; rgi->gtid_sub_id= sub_id;

14
sql/rpl_rli.h
View File

@ -422,7 +422,7 @@ public:
*/ */
void stmt_done(my_off_t event_log_pos, void stmt_done(my_off_t event_log_pos,
time_t event_creation_time, THD *thd, time_t event_creation_time, THD *thd,
struct rpl_group_info *rgi); rpl_group_info *rgi);
/** /**
@ -521,10 +521,14 @@ private:
/* /*
This is data for various state needed to be kept for the processing of This is data for various state needed to be kept for the processing of
one event group in the SQL thread. one event group (transaction) during replication.
For single-threaded replication it is linked from the RLI, for parallel In single-threaded replication, there will be one global rpl_group_info and
replication it is linked into each event group being executed in parallel. one global Relay_log_info per master connection. They will be linked
together.
In parallel replication, there will be one rpl_group_info object for
each running thd. All rpl_group_info will share the same Relay_log_info.
*/ */
struct rpl_group_info struct rpl_group_info
{ {
@ -555,7 +559,7 @@ struct rpl_group_info
for the wrong commit). for the wrong commit).
*/ */
uint64 wait_commit_sub_id; uint64 wait_commit_sub_id;
struct rpl_group_info *wait_commit_group_info; rpl_group_info *wait_commit_group_info;
/* /*
If non-zero, the event group must wait for this sub_id to be committed If non-zero, the event group must wait for this sub_id to be committed
before the execution of the event group is allowed to start. before the execution of the event group is allowed to start.

2
sql/rpl_utility.cc
View File

@ -1143,7 +1143,7 @@ bool Deferred_log_events::is_empty()
return array.elements == 0; return array.elements == 0;
} }
bool Deferred_log_events::execute(struct rpl_group_info *rgi) bool Deferred_log_events::execute(rpl_group_info *rgi)
{ {
bool res= false; bool res= false;

2
sql/slave.cc
View File

@ -3019,7 +3019,7 @@ static int has_temporary_error(THD *thd)
ev->update_pos(). ev->update_pos().
*/ */
int apply_event_and_update_pos(Log_event* ev, THD* thd, int apply_event_and_update_pos(Log_event* ev, THD* thd,
struct rpl_group_info *rgi, rpl_group_info *rgi,
rpl_parallel_thread *rpt) rpl_parallel_thread *rpt)
{ {
int exec_res= 0; int exec_res= 0;

2
sql/sql_binlog.cc
View File

@ -80,7 +80,7 @@ void mysql_client_binlog_statement(THD* thd)
my_bool have_fd_event= TRUE; my_bool have_fd_event= TRUE;
int err; int err;
Relay_log_info *rli; Relay_log_info *rli;
struct rpl_group_info *rgi; rpl_group_info *rgi;
rli= thd->rli_fake; rli= thd->rli_fake;
if (!rli) if (!rli)

7
sql/sql_class.cc
View File

@ -5666,6 +5666,10 @@ wait_for_commit::register_wait_for_prior_commit(wait_for_commit *waitee)
waiting_for_commit= false; waiting_for_commit= false;
else else
{ {
/*
Put ourself at the head of the waitee's list of transactions that must
wait for it to commit first.
*/
this->next_subsequent_commit= waitee->subsequent_commits_list; this->next_subsequent_commit= waitee->subsequent_commits_list;
waitee->subsequent_commits_list= this; waitee->subsequent_commits_list= this;
} }
@ -5704,7 +5708,7 @@ wait_for_commit::wait_for_prior_commit2()
The waiter needs to lock the waitee to delete itself from the list in The waiter needs to lock the waitee to delete itself from the list in
unregister_wait_for_prior_commit(). Thus wakeup_subsequent_commits() can not unregister_wait_for_prior_commit(). Thus wakeup_subsequent_commits() can not
hold its own lock while locking waiters, lest we deadlock. hold its own lock while locking waiters, as this could lead to deadlock.
So we need to prevent unregister_wait_for_prior_commit() running while wakeup So we need to prevent unregister_wait_for_prior_commit() running while wakeup
is in progress - otherwise the unregister could complete before the wakeup, is in progress - otherwise the unregister could complete before the wakeup,
@ -5727,6 +5731,7 @@ wait_for_commit::wait_for_prior_commit2()
would not be woken up until next wakeup, which could be potentially much would not be woken up until next wakeup, which could be potentially much
later than necessary. later than necessary.
*/ */
void void
wait_for_commit::wakeup_subsequent_commits2() wait_for_commit::wakeup_subsequent_commits2()
{ {

2
sql/sql_class.h
View File

@ -1615,7 +1615,7 @@ struct wait_for_commit
*/ */
bool waiting_for_commit; bool waiting_for_commit;
/* /*
Flag set when wakeup_subsequent_commits_running() is active, see commonts Flag set when wakeup_subsequent_commits_running() is active, see comments
on that function for details. on that function for details.
*/ */
bool wakeup_subsequent_commits_running; bool wakeup_subsequent_commits_running;