MySQL中怎么处理读写锁

52次阅读

共计 4840 个字符，预计需要花费 13 分钟才能阅读完成。

自动写代码机器人，免费开通

这期内容当中丸趣 TV 小编将会给大家带来有关 MySQL 中怎么处理读写锁，文章内容丰富且以专业的角度为大家分析和叙述，阅读完这篇文章希望大家可以有所收获。

1. 创建锁
锁的创建实际上就是初始化一个 RW 结构体（rw_lock_t），实际调用函数如下:

# define rw_lock_create(K, L, level) \
rw_lock_create_func((L),#L)

在 rw_lock_create 上有三个参数，在实际场景锁时只用到第 2 个参数
其中 K 表示 mysql_pfs_key_t，level 显示当前的操作类型（起码看起来是的，在文件 sync0sync.h 中定义），看起来 k 是为 performance schema 准备的，而 k 代表了当前操作所在的层次。
例如：purge 线程的读写锁创建：

rw_lock_create(trx_purge_latch_key,
purge_sys- latch,SYNC_PURGE_LATCH);

我们进去 rw_lock_create_func 看看到底是怎么创建的。
可以看到这个函数的逻辑其实很简单：
lock- lock_word =X_LOCK_DECR; // 关键字段
用于限制读写锁的最大并发数，代码里的注释如下：

/* We decrement lock_word by this amountfor each x_lock. It is also the
start value for the lock_word, meaning thatit limits the maximum number
of concurrent read locks before the rw_lockbreaks. The current value of
0x00100000 allows 1,048,575 concurrentreaders and 2047 recursive writers.*/

在尝试加锁时会调用 rw_lock_lock_word_decr 减少 lock_word
在初始化一系列变量后，执行：

lock- event = os_event_create(NULL);
lock- wait_ex_event = os_event_create(NULL);
os_event_create 用于创建一个系统信号，实际上最终创建的还是互斥量（os_fast_mutex_init((event- os_mutex)); 以及条件变量（os_cond_init((event- cond_var));）
最后将 lock 加入到全局链表 rw_lock_list 中

2. 加锁
加锁函数由宏定义，实际调用函数为：
1)写锁

# define rw_lock_x_lock(M) \
rw_lock_x_lock_func((M),0, __FILE__, __LINE__)

当申请写锁时，执行如下步骤：
(1). 调用 rw_lock_x_lock_low 函数去获取锁，如果得到锁，则 rw_x_spin_round_count += i 后直接返回，如果得不到锁，继续执行
(2).loop 过程中只执行一次 rw_x_spin_wait_count++
(3). 在毫秒级别的 loop 多次等待

while (i SYNC_SPIN_ROUNDS
lock- lock_word = 0) {
if(srv_spin_wait_delay) {
ut_delay(ut_rnd_interval(0,
srv_spin_wait_delay));
}
i++;
}

这里涉及到两个系统变量：
innodb_sync_spin_loops(SYNC_SPIN_ROUNDS)
innodb_spin_wait_delay(srv_spin_wait_delay)

在 SYNC_SPIN_ROUNDS 循环里调用函数 ut_delay，这个函数很简单，就是做了 delay*50 次空循环

Ut_delay(uint delay)：
for(i = 0; i delay * 50; i++) {
j+= i;
UT_RELAX_CPU();
}
其中，UT_RELAX_CPU()会调用汇编指令来独占 CPU，以防止线程切换
(4). 如果 loop 的次数等于 SYNC_SPIN_ROUNDS，调用 os_thread_yield（实际调用 pthread_yield，导致调用线程放弃 CPU 的占用）将线程挂起；否则挑到 1 继续 loop
(5). 在 sync_primary_wait_array 里获取一个 cell（占个坑？）。调用 sync_array_reserve_cell，看起来有 1000 个坑位（sync_primary_wait_array- n_cells）
(6). 再次调用 rw_lock_x_lock_low 函数尝试获取锁，若成功获得，则返回
(7). 调用 sync_array_wait_event 等待条件变量，然后返回 1 继续 loop
具体的加锁函数（rw_lock_x_lock_low）稍后分析

2) 读锁

# define rw_lock_s_lock(M) \
rw_lock_s_lock_func((M),0, __FILE__, __LINE__)

这个函数定义在 sync0rw.ic 里，函数也很简单，如下：

if (rw_lock_s_lock_low(lock, pass, file_name, line)) {
return; /* Success */
}else {
/* Did not succeed, try spin wait */
rw_lock_s_lock_spin(lock, pass, file_name, line);
return;
}

这里首先调用 rw_lock_s_lock_low 进行加锁，如果加锁不成功，则调用 rw_lock_s_lock_spin 进行等待，rw_lock_s_lock_spin 的代码逻辑与 rw_lock_x_lock_func 有些相似，这里不再赘述。
在 rw_lock_s_lock_spin 里会递归的调用到 rw_lock_s_lock_low 函数；

看起来实际的加锁和解锁操作是通过对计数器来控制的，
(1)在函数 rw_lock_s_lock_low 中
rw_lock_lock_word_decr (lock, 1)，对 lock- lock_word 减去 1
减数成功返回 true，否则返回 false
这部分的逻辑还是很简单的。

(2)在函数 rw_lock_x_lock_low 中，调用：
rw_lock_lock_word_decr(lock, X_LOCK_DECR)，对 lock- lock_word 减去 X_LOCK_DECR
减数成功后，执行：

rw_lock_set_writer_id_and_recursion_flag(lock,pass ? FALSE : TRUE)来设置：
lock- writer_thread = s_thread_get_curr_id()
lock- recursive = TRUE

然后调用 rw_lock_x_lock_wait 函数等待 lock- lock_word=0，也就是说等待所有的读锁退出。

看到一个比较有意思的现象，在.ic 的代码里看到使用了宏
INNODB_RW_LOCKS_USE_ATOMICS，这是跟 gcc 的版本相关的，通过使用 gcc 的内建函数来实现原子操作。

3. 解锁
解锁操作包括解除读锁（#define rw_lock_s_unlock(L) rw_lock_s_unlock_gen(L, 0)）和解除写锁操作（#definerw_lock_x_unlock(L) rw_lock_x_unlock_gen(L, 0)）
实际调用函数为 rw_lock_s_unlock_func 和 rw_lock_x_unlock_func

1)解除读锁（rw_lock_s_unlock_func）
增加计数 rw_lock_lock_word_incr(lock, 1)

2)解除写锁（rw_lock_x_unlock_func）
执行如下操作
(1) 如果是最后一个递归调用锁的线程，设置 lock- recursive= FALSE; 代码里的注释如下：

/* lock- recursive flag also indicatesif lock- writer_thread is
valid or stale. If we are the last of the recursive callers
then we must unset lock- recursive flag to indicate that the
lock- writer_thread is now stale.
Note that since we still hold the x-lock we can safely read the
lock_word. */

(2)增加计数 rw_lock_lock_word_incr(lock,X_LOCK_DECR) == X_LOCK_DECR，这时候需要向等待锁的线程发送信号：

if (lock- waiters) {
rw_lock_reset_waiter_flag(lock);
os_event_set(lock- event);
sync_array_object_signalled(sync_primary_wait_array);
}

os_event_set 函数会发送一个 pthread_cond_broadcast 给等待的线程

4. 监控读写锁
为了防止 mysqld 被 hang 住导致的长时间等待 rw 锁，error 监控线程会对长时间等待的线程进行监控。这个线程每 1 秒 loop 一次
（os_event_wait_time_low(srv_error_event, 1000000, sig_count);）
函数入口：srv_error_monitor_thread
函数 sync_array_print_long_waits()用于处理长时间等待信号量的线程，流程如下：
1. 查看 sync_primary_wait_array 数组中的所有等待线程。
– 大于 240 秒时，向错误日志中输出警告，设置 noticed = TRUE;
– 大于 600 秒时，设置 fatal =TRUE;
2. 当 noticed 为 true 时，打印出 innodb 监控信息，然后 sleep30 秒
3. 返回 fatal 值

当函数 sync_primary_wait_array 返回 true 时，对于同一个等待线程还会有十次机会，也就是 300 + 1*10（监控线程每次 loop sleep 1s）秒的时间；如果挺不过去，监控线程就会执行一个断言失败：

if (fatal_cnt 10) {
fprintf(stderr,
InnoDB:Error: semaphore wait has lasted
%lu seconds\n
InnoDB:We intentionally crash the server,
because it appears to be hung.\n ,
(ulong) srv_fatal_semaphore_wait_threshold);

ut_error;
}

ut_error 是一个宏：

#define ut_error assert(0)
断言失败导致 mysqld crash
在函数 srv_error_monitor_thread 里发现一个比较有意思的参数 srv_kill_idle_transaction，对应的系统变量为 innodb_kill_idle_transaction，用于清理在一段时间内的空闲事务。这个变量指定了空闲事务的最长时间。