上一篇文章说明了bug出现的原因和原理分析，要修复bug似乎已经水到渠成了，但远没有这么简单，只因为“并发”。要修复问题，首先要做的第一件事情是稳定的复现问题。由于数据库系统是一个并发系统，并且这个bug只有一定的概率出现，更说明了多个线程在一定的执行序列情况下才会出现这个bug。在没有用户请求的情况下，mysql自身的线程就很多，比如主线程，IO线程，监控线程，监听线程等；有用户请求的情况下，还会分配工作线程为用户服务。我们只找和bug相关的线程：purge线程，IO线程，工作线程。Purge线程主要工作是执行purge操作，产生IBUF_OP_DELETE操作；工作线程则主要执行DELETE，INSERT语句，产生IBUF_OP_INSERT和IBUF_OP_DELETE_MARK操作；IO线程则是IBUF的merge操作。为达到重现bug的目的，我们需要产生IBUF_OP_DELETE_MARK，IBUF_OP_INSERT和IBUF_OP_DELETE序列。由于这里面涉及到purge线程和用户线程，因此涉及一个协同问题。如何协同？

DEBUG_SYNC

DEBUG_SYNC是mysql源码中自带的宏，通过信号的方式实现多个线程间的同步。假设存在两个线程thread-1(A,B)，thread-2(a,b)，A,B和a,b分别是thread-1和thread-2的执行顺序。为了得到AaBb执行序列，可以通过DEBUG_SYNC实现。

1) 源代码中，设置同步点

Session-1(thread-1)

Session-2(thread-2)

A

a

DEBUG_SYNC(“after_A”)

DEBUG_SYNC(“after_a”)

B

b

DEBUG_SYNC(“after_B”)

2) Session设置

Session-1(thread-1)

Session-2(thread-2)

set debug_sync="after_Asignal stepA wait_for stepa";

set debug_sync=”after_Bsingal stepB”;

set debug_sync = "nowwait_for stepA ";

set debug_sync = "after_asignal stepa wait_for stepB";

A

a

B

b

通过在代码中设置同步点，以及在会话中设置同步关系即可以实现在并发环境下稳定地得到AaBb序列。注意到2)中的蓝色字体表示同步点，红色字体signal和 wait_for表示动作。比如“after_A signal stepA wait_for  stepa”表示执行到同步点after_A后，发出信号after_A，并且等待stepa信号的到来。另外，其中now是一个特殊的同步点，表示执行到指定的位置后等待。debug_sync似乎已经完美地解决了同步问题，但问题是，通过debug_sync同步，需要会话设置等待或发出信号动作，如果需要和后台线程同步怎么办？而这个bug恰好又需要和后台线程purge，以及IO线程同步。别着急，有需求，就一定有解，mysql还有另外一种协同方法，DBUG_EXECUTE_IF。

DBUG_EXECUTE_IF

DBUG_EXECUTE_IF原理就很简单了，就是在某一点执行一定的操作，至于是什么操作则没有限制。显然，我们可以在某一个点执行sleep函数，虽然无法做到严格的同步，但也基本能满足需求。DEBUG_EXECUTE_IF的基本格式为：DEBUG_EXECUTE_IF(key, code)，比如为了得到AaBb序列，可以通过DBUG_EXECUTE_IF实现

1)  源代码中，设置操作点

Session-1(thread-1)

Session-2(thread-2)

DEBUG_EXECUTE_IF (“before_a”,{my_sleep(1)})

A

a

DEBUG_EXECUTE_IF (“after_A”,{my_sleep(3)})

DEBUG_EXECUTE_IF(“after_b”, {my_sleep(5)})

B

b

DEBUG_EXECUTE_IF (“after_B”, {my_sleep(7)})

2) Session设置

这里为了让后台线程也起作用，可以设置全局debug变量，若只想对本session起作用，可以设置会话debug变量。设置后，执行到对应的操作点时，则会额外执行DEBUG_EXECUTE_IF里面的代码。通过sleep方式来同步是一种取巧的方式，在实际情况中，需要不断调整sleep的时间，尽可能保证能按照预设的顺序执行。设置如下：红色部分是加的代码。

set global debug=”+d,before_a, after_A, after_B, after_b”;

实践

对于我们这个案例而言，为了得到指定的IBUF序列(IBUF_OP_DELETE_MARK，IBUF_OP_INSERT和IBUF_OP_DELETE)，首先保证操作页面不出现在buffer_pool中；其次，执行INSERT操作产生的IBUF_OP_INSERT必需先于PURGE执行，最后，在整个过程中页面不能被MERGE，否则也将功亏一篑。对于第一点，可以通过将buffer_pool设置地比较小，然后扫描一个大表达到清理buffer_pool的目的；对于第二点，由于执行IBUF缓存的函数是IBUF_INSERT,针对类型为IBUF_OP_DELETE，强制其等待；对于第三点，在执行merge ibuf的地方，也强制其等待，并且比PURGE等待的时间要更长。具体设置点如下：

/*storage/innobase/ibuf/ibuf0ibuf.cc: ibuf_insert*/

ibuf_insert

{

DBUG_EXECUTE_IF("sleep_ibuf_insert",{

if (op == IBUF_OP_DELETE)

my_sleep(3000000);

});

......

ibuf_insert_low();

}

/* storage/innobase/ibuf/ibuf0ibuf.cc:ibuf_merge_pages*/

ibuf_merge_pages

{

......

DBUG_EXECUTE_IF("sleep_ibuf_merge",{my_sleep(5000000);});

buf_read_ibuf_merge_pages(sync, space_ids, space_versions, page_nos, *n_pages);//merge操作

}

总结

在整个重现bug的过程中，其实还有很多细节，因为代码中总是会有很多if else的判断，以及特定标记位导致的特殊操作，因此打日志和debug很重要，通过打日志，了解大体执行流程；通过debug来了解细节，从而能掌握关键路径，在关键路径上设置合适的同步点，则能在并发环境下得到想要的执行序列。

参考文档

inlog.cc (sql): DEBUG_SYNC(current_thd,"purge_logs_after_lock_index_before_thread_count");

Binlog.cc (sql): DEBUG_SYNC(thd, "after_show_binlog_event_found_file");

Binlog.cc (sql): DEBUG_SYNC(thd, "wait_in_show_binlog_events_loop");

Binlog.cc (sql): DEBUG_SYNC(thd, "after_show_binlog_events");

Binlog.cc (sql): DEBUG_SYNC(current_thd, "before_rotate_binlog"););

Binlog.cc (sql): DEBUG_SYNC(current_thd, "at_purge_logs_before_date");

Binlog.cc (sql): DEBUG_SYNC(thd, "before_sync_binlog_file");

Binlog.cc (sql): DEBUG_SYNC(thd, "before_process_commit_stage_queue"););

Binlog.cc (sql): DEBUG_SYNC(thd, "ready_to_do_rotation");

CMakeLists.txt (sql): debug_sync.cc

CMakeLists.txt (unittest\gunit): debug_sync

Debug_sync-t.cc (unittest\gunit): DEBUG_SYNC(m_thd,"sync_point");

Debug_sync.cc (sql): the server codebyusing the DEBUG_SYNC macro:

Debug_sync.cc (sql): DEBUG_SYNC(thd, "after_open_tables");

Debug_sync.cc (sql):SET DEBUG_SYNC= 'after_open_tables SIGNAL opened WAIT_FOR flushed';

Debug_sync.cc (sql):SET DEBUG_SYNC= 'after_open_tables SIGNAL opened WAIT_FOR flushed';

Debug_sync.cc (sql):SET DEBUG_SYNC= 'now WAIT_FOR opened';

Debug_sync.cc (sql):SET DEBUG_SYNC= 'after_abort_locks SIGNAL flushed';

Debug_sync.cc (sql):SET DEBUG_SYNC= 'WAIT_FOR flushed NO_CLEAR_EVENT';

Debug_sync.cc (sql):ofDEBUG_SYNC accordingly.

Debug_sync.cc (sql):SET DEBUG_SYNC= 'name SIGNAL sig EXECUTE 3';

Debug_sync.cc (sql):SET DEBUG_SYNC= 'name WAIT_FOR sig TIMEOUT 10 EXECUTE 2';

Debug_sync.cc (sql):SET DEBUG_SYNC= 'name HIT_LIMIT 3';

Debug_sync.cc (sql):SET DEBUG_SYNC= 'name SIGNAL sig EXECUTE 2 HIT_LIMIT 3';

Debug_sync.cc (sql):SET DEBUG_SYNC= 'name CLEAR';

Debug_sync.cc (sql):SET DEBUG_SYNC= 'RESET';

Debug_sync.cc (sql):SET DEBUG_SYNC= 'name TEST';

Debug_sync.cc (sql): The stringto "assign" tothe DEBUG_SYNC variable can contain:

Debug_sync.cc (sql): #define DEBUG_SYNC(thd, sync_point_name)

Debug_sync.cc (sql): TheSET DEBUG_SYNC statement adds a requested action to the array orDebug_sync.cc (sql): DEBUG_SYNC(thd, "sync_point_name");

Debug_sync.cc (sql): point emits a signal. DEBUG_SYNC sets its own proc_info, but restores

Debug_sync.cc (sql):[DEBUG_SYNC(thd, "sync_point_name");]Debug_sync.cc (sql): macro DEBUG_SYNC(thd, sync_point_name). The macro needs a'thd'argument.

Debug_sync.cc (sql): DEBUG_SYNC(thd, "thread_end");

Debug_sync.cc (sql): DEBUG_SYNC(thd, "now");

Debug_sync.cc (sql):and shall not be reported as a result of SETDEBUG_SYNC.

Debug_sync.cc (sql): Thisis called when the DEBUG_SYNC system variable is set.

Debug_sync.cc (sql): mutexand cond. This would prohibit the use ofDEBUG_SYNC

Debug_sync.cc (sql): it possibleto use DEBUG_SYNC framework in scheduler whenthis

Debug_sync.cc (sql): mutexand cond. This would prohibit the use ofDEBUG_SYNC

Debug_sync.cc (sql):to be triggered by setting a value toDEBUG_SYNC system variable.

Debug_sync.h (sql):#define DEBUG_SYNC(_thd_, _sync_point_name_) \

Debug_sync.h (sql):#define DEBUG_SYNC(_thd_, _sync_point_name_)/*disabled DEBUG_SYNC*/Filesort.cc (sql): DEBUG_SYNC(thd, "filesort_start");

Filesort.cc (sql): DEBUG_SYNC(thd, "after_index_merge_phase1");

Handler.cc (sql):#include "debug_sync.h"//DEBUG_SYNC

Handler.cc (sql): DEBUG_SYNC(thd, "ha_commit_trans_after_acquire_commit_lock");

Handler0alter.cc (storage\innobase\handler): DEBUG_SYNC(user_thd, "innodb_inplace_alter_table_enter");

Handler0alter.cc (storage\innobase\handler): DEBUG_SYNC(user_thd, "innodb_after_inplace_alter_table");

Handler0alter.cc (storage\innobase\handler): DEBUG_SYNC(user_thd, "innodb_alter_commit_after_lock_table");

Ha_innodb.cc (storage\innobase\handler):#include //DEBUG_SYNC

Ha_innodb.cc (storage\innobase\handler): DEBUG_SYNC(user_thd, "ib_after_row_insert");

Ha_innodb.cc (storage\innobase\handler): DEBUG_SYNC(thd, "after_innobase_rename_table");

Ha_myisammrg.cc (storage\myisammrg): DEBUG_SYNC(current_thd, "before_myisammrg_attach");

Ha_partition.cc (sql): DEBUG_SYNC(ha_thd(), "before_rename_partitions");

Ha_partition.cc (sql): DEBUG_SYNC(ha_thd(), "partition_open_error");

Item_func.cc (sql): DEBUG_SYNC(current_thd, "before_acos_function");

Item_func.cc (sql): DEBUG_SYNC(current_thd, "after_getting_user_level_lock_info");

Item_func.cc (sql): DEBUG_SYNC(current_thd, "after_getting_user_level_lock_info");

Item_subselect.cc (sql): DEBUG_SYNC(thd, "before_index_end_in_subselect");

Lock.cc (sql): DEBUG_SYNC(thd, "after_wait_locked_schema_name");

Lock.cc (sql): DEBUG_SYNC(thd, "before_wait_locked_pname");

Lock.cc (sql): DEBUG_SYNC(thd, "after_wait_locked_pname");

Mdl-t.cc (unittest\gunit): We don't need DEBUG_SYNC functionality in a unit test.

Mdl.cc (sql): DEBUG_SYNC(get_thd(), "mdl_acquire_lock_wait");

Mdl.cc (sql): DEBUG_SYNC(get_thd(), "mdl_upgrade_lock");

Mdl.h (sql): This one is only used for DEBUG_SYNC.

Mysqld.cc (sql): DEBUG_SYNC here, but that's not possible because the THD's debug

Mysqld.cc (sql): DEBUG_SYNC(thd, "dsync_show_heartbeat_period");

Rpl_master.cc (sql): DEBUG_SYNC(thd, "show_binlogs_after_lock_log_before_lock_index");

Rpl_slave.cc (sql): DEBUG_SYNC(thd, "wait_after_lock_active_mi_and_rli_data_lock_is_acquired");

Rpl_slave.cc (sql): DEBUG_SYNC(thd, "after_start_slave");

Semisync_master.cc (plugin\semisync): DEBUG_SYNC(current_thd, "rpl_semisync_master_commit_trx_before_lock");

Sql_admin.cc (sql):#include "debug_sync.h" // DEBUG_SYNC

Sql_admin.cc (sql): DEBUG_SYNC(thd, "after_admin_flush");

Sql_admin.cc (sql): DEBUG_SYNC(thd, "ha_admin_try_alter");

Sql_admin.cc (sql): DEBUG_SYNC(thd, "ha_admin_open_ltable");

Sql_base.cc (sql): DEBUG_SYNC(thd, "after_flush_unlock");

Sql_base.cc (sql): DEBUG_SYNC(thd, "before_close_thread_tables");

Sql_base.cc (sql): DEBUG_SYNC(thd, "before_open_table_wait_refresh");

Sql_base.cc (sql): DEBUG_SYNC(thd, "after_open_table_mdl_shared");

Sql_base.cc (sql): DEBUG_SYNC(thd, "before_upgrading_lock_from_S_to_X_for_create_table");

Sql_base.cc (sql): DEBUG_SYNC(thd, "after_upgrading_lock_from_S_to_X_for_create_table");

Sql_base.cc (sql): DEBUG_SYNC(m_thd, "recover_ot_repair");

Sql_base.cc (sql): DEBUG_SYNC(thd, "after_shared_lock_pname");

Sql_base.cc (sql): DEBUG_SYNC(thd, "open_and_process_table");

Sql_base.cc (sql): DEBUG_SYNC(thd, "open_tables_after_open_and_process_table");

Sql_base.cc (sql): DEBUG_SYNC(thd, "before_lock_tables_takes_lock");

Sql_base.cc (sql): DEBUG_SYNC(thd, "after_lock_tables_takes_lock");

Sql_cache.cc (sql):#include "debug_sync.h" // DEBUG_SYNC

Sql_cache.cc (sql): DEBUG_SYNC(thd, name); \

Sql_cache.cc (sql): DEBUG_SYNC(thd, "wait_after_query_cache_invalidate");

Sql_cache.cc (sql): DEBUG_SYNC(thd, "wait_in_query_cache_invalidate1");

Sql_cache.cc (sql): DEBUG_SYNC(thd, "wait_in_query_cache_invalidate2");

Sql_class.cc (sql): DEBUG_SYNC(this, "thd_cleanup_start");

Sql_class.cc (sql): DEBUG_SYNC is used in commit code.

Sql_cursor.cc (sql): DEBUG_SYNC(thd, "after_table_close");

Sql_db.cc (sql): DEBUG_SYNC(thd, "after_freeing_thd_db");

Sql_db.cc (sql): DEBUG_SYNC(thd, "before_db_dir_check");

Sql_insert.cc (sql): DEBUG_SYNC(thd, "before_write_delayed");

Sql_insert.cc (sql): DEBUG_SYNC(thd, "after_write_delayed");

Sql_insert.cc (sql): DEBUG_SYNC(thd, "write_row_replace");

Sql_insert.cc (sql): DEBUG_SYNC(thd, "write_row_noreplace");

Sql_insert.cc (sql): DEBUG_SYNC(thd,"create_table_select_before_create");

Sql_insert.cc (sql): DEBUG_SYNC(thd,"create_table_select_before_open");

Sql_insert.cc (sql): DEBUG_SYNC(thd,"create_table_select_before_check_if_exists");

Sql_insert.cc (sql): DEBUG_SYNC(thd,"create_table_select_before_lock");

Sql_optimizer.cc (sql):#include "debug_sync.h" // DEBUG_SYNC

Sql_optimizer.cc (sql): DEBUG_SYNC(thd, "before_join_optimize");

Sql_parse.cc (sql): DEBUG_SYNC(thd, "before_do_command_net_read");

Sql_parse.cc (sql): DEBUG_SYNC(thd,"before_execute_sql_command");

Sql_parse.cc (sql): DEBUG_SYNC(thd, "execute_command_after_close_tables");

Sql_parse.cc (sql): DEBUG_SYNC(thd, "after_table_open");

Sql_partition.cc (sql): DEBUG_SYNC(current_thd, "swap_partition_first_row_read");

Sql_partition_admin.cc (sql):#include "debug_sync.h" // DEBUG_SYNC

Sql_partition_admin.cc (sql): DEBUG_SYNC(thd, "swap_partition_after_compare_tables");

Sql_partition_admin.cc (sql): DEBUG_SYNC(thd, "swap_partition_after_wait");

Sql_partition_admin.cc (sql): DEBUG_SYNC(thd, "swap_partition_before_rename");

Sql_plugin.cc (sql): DEBUG_SYNC(thd, "acquired_LOCK_plugin");

Sql_reload.cc (sql): DEBUG_SYNC(thd,"flush_tables_with_read_lock_after_acquire_locks");

Sql_show.cc (sql): DEBUG_SYNC(thd,"before_copying_threads");

Sql_show.cc (sql): DEBUG_SYNC(thd,"after_copying_threads");

Sql_show.cc (sql): DEBUG_SYNC(thd, "processlist_after_LOCK_thd_count_before_LOCK_thd_data");

Sql_show.cc (sql): DEBUG_SYNC(thd, "processlist_after_LOCK_thd_count_before_LOCK_thd_data");

Sql_show.cc (sql): DEBUG_SYNC(thd, "after_open_table_ignore_flush");

Sql_show.cc (sql): DEBUG_SYNC(thd, "before_open_in_get_all_tables");

Sql_show.cc (sql): DEBUG_SYNC(thd, "get_schema_column");

Sql_show.cc (sql): DEBUG_SYNC(thd, "acquired_LOCK_system_variables_hash");

Sql_show.cc (sql): DEBUG_SYNC(thd, "before_preparing_global_status_array");

Sql_show.cc (sql): DEBUG_SYNC(thd, "after_preparing_global_status_array");

Sql_table.cc (sql): DEBUG_SYNC(thd, "rm_table_no_locks_before_delete_table");

Sql_table.cc (sql): DEBUG_SYNC(thd, "rm_table_no_locks_before_binlog");

Sql_table.cc (sql): DEBUG_SYNC(thd, "locked_table_name");

Sql_table.cc (sql): DEBUG_SYNC(thd, "create_table_like_after_open");

Sql_table.cc (sql): DEBUG_SYNC(thd, "create_table_like_before_binlog");

Sql_table.cc (sql): DEBUG_SYNC(thd, "alter_table_inplace_after_lock_upgrade");

Sql_table.cc (sql): DEBUG_SYNC(thd, "alter_table_inplace_after_lock_downgrade");

Sql_table.cc (sql): DEBUG_SYNC(thd, "alter_table_inplace_before_commit");

Sql_table.cc (sql): DEBUG_SYNC(thd,"alter_table_enable_indexes");

Sql_table.cc (sql): DEBUG_SYNC(thd, "alter_table_before_open_tables");

Sql_table.cc (sql): DEBUG_SYNC(thd, "alter_opened_table");

Sql_table.cc (sql): DEBUG_SYNC(thd, "locked_table_name");

Sql_table.cc (sql): DEBUG_SYNC(thd, "alter_table_before_create_table_no_lock");

Sql_table.cc (sql): DEBUG_SYNC(thd, "alter_table_copy_after_lock_upgrade");

Sql_table.cc (sql): DEBUG_SYNC(thd, "alter_table_manage_keys");

Sql_table.cc (sql): DEBUG_SYNC(thd, "alter_table_before_rename_result_table");

Sql_table.cc (sql): DEBUG_SYNC(thd, "alter_table_before_main_binlog");

Sql_tmp_table.cc (sql): DEBUG_SYNC(thd, "tmp_table_created");

Sql_truncate.cc (sql):#include "debug_sync.h" // DEBUG_SYNC

Sql_truncate.cc (sql): DEBUG_SYNC(thd, "upgrade_lock_for_truncate");

Sys_vars.cc (sql):#include "debug_sync.h" // DEBUG_SYNC

Sys_vars.cc (sql): DEBUG_SYNC(thd, "fix_slave_net_timeout");

Table.cc (sql): DEBUG_SYNC(thd, "flush_complete");

Table.cc (sql): DEBUG_SYNC(thd, "share_destroyed");

Thread_pool_priv.h (include\mysql): A thread pool can also use DEBUG_SYNC and must thus include

Transaction.cc (sql):#include "debug_sync.h" // DEBUG_SYNC

Transaction.cc (sql): DEBUG_SYNC(thd, "trans_xa_commit_after_acquire_commit_lock");

Tztime.cc (sql): DEBUG_SYNC(thd, "my_tz_find");

DEBUG_EXECUTE_IF主要用于当设置了某个关键字key时，执行后面code

可以简单的表示为:

DEBUG_EXECUTE_IF(key, code)

例如在open_and_lock_tables函数中.5527 if (open_tables(thd, &tables, &counter, flags, prelocking_strategy))5528 gotoerr;5529

5530DBUG_EXECUTE_IF("sleep_open_and_lock_after_open", {5531 const char *old_proc_info= thd->proc_info;5532 thd->proc_info="DBUG sleep";5533 my_sleep(6000000);5534 thd->proc_info=old_proc_info;});5535

5536 if(lock_tables(thd, tables, counter, flags))5537 gotoerr;

那么如何让其生效呢，执行如下语句即可:set session debug=”+d, sleep_open_and_lock_after_open”

这时候当进入这个函数时，在执行完open_tables语句后，就会设置thd的状态为DBUG sleep，再sleep 6秒钟，然后再调用lock_tables函数

—————-另外我们在写test case时，也可能用到用户层的锁，使用SQL FUNCTION的方法实现:

GET_LOCK(str, timeout)

—尝试获取一个名为str的锁，等待timeout，返回1表示获取锁成功

IS_FREE_LOCK(str)

—检查名为str的锁是否已经被释放

IS_USED_LOCK(str)

—检查名为str的锁是否已经被占用，如果是，返回占有该锁的线程ID，否则返回NULL

RELEASE_LOCK(str)

—释放名为str的锁