最近简单看了一下Innodb中mutex和rw lock的部分，因为这部分经常在Innodb遇到signal 6的时候会出现，并且是Innodb中并发控制的基础，是分析问题和理解问题根本。这里简单将锁分为Mutex和RW lock。这里的Mutex和RW lock当前来看并不是简单的使用操作系统提供的方式，而是在其基础上进行了大量的封装加入了lock word和spin lock共同实现。这部分比较复杂仅供参考。。。
一、Innodb的event
这里说的event并不是我们常常说道的MySQL的event(定时任务)，它对应的数据结构为os_event实际上是一个mutex 和一个条件变量，还包含2个自带的表示状态的元素，是一个典型的mutex和条件变量的封装，而Innodb中不管mutex和rw lock都依赖于它。其包含如下一些元素，
- m_set:是否已经处于唤醒状态下，如果不处于才可以唤醒，初始化情况下为false
- signal_count: 每次处于唤醒状态就会增加，初始化情况下为1
- mutex:EventMutex类型对OSmutex的封装，根据不同的操作系统使用不同的方法，对于Linux而言就是pthread_mutex，定义在sync0types.h中，这个文件中也描述了加锁顺序相关的内容
- cond_var:条件变量，类型os_cond_t也是根据操作系统自适应的
方法包含：
- destroy 封装 os_event_destroy
析构event - set 封装 os_event_set
进入event mutex
->if (!m_set)
如果m_set设置为false 那么就进行唤醒
如果m_set为true则不唤醒
如下，调用broadcast
broadcast()
释放event mutex
- broadcast
m_set = true
首先设置唤醒为true
++signal_count
并且signal_count+1
ret = pthread_cond_broadcast(&cond_var)
进行唤醒操作
- try_set 封装 os_event_try_set
如果mutex.try_lock() 也就是 mutex封装的try lock方法也就是调用pthread_mutex_trylock
返回为true，则进行唤醒，
否则直接返回false
- reset 封装 os_event_reset
获取mutex，如果m_set 为true，则设置m_set = false，并且返回signal_count
设置m_set 后就代表唤醒完成了，这个变量应该被broadcast唤醒者更改
- wait_low (reset_sig_count) 封装 os_event_wait_low
获取event mutex
->if (!reset_sig_count) {reset_sig_count = signal_count}
这里需要注意如果传入的reset_sig_count为0，则直接把signal_count赋予给reset_sig_count ，也就是判断的时候
排除reset_sig_count的因素
->while (!m_set && signal_count == reset_sig_count)
根据条件!m_set && signal_count == reset_sig_count 进入
->wait();
而wait()就是简单的pthread_cond_t 的等待pthread_cond_wait(&cond_var, mutex)
然后释放event mutex
这里加入reset_sig_count进行判断的意思如注释所解释到的如下，
thread A calls os_event_reset()
thread B calls os_event_set() [event->m_set == true] //唤醒回++signal_count 这样 signal_count == reset_sig_count 条件
thread C calls os_event_reset() [event->m_set == false] //判定失败，这种情况就解除了，因此必须加锁本条件
thread A calls os_event_wait() [infinite wait!]
thread C calls os_event_wait() [infinite wait!]
- wait_time_low(timeout ,reset_sig_count)
这个函数传入了一个时间timeout ，这个timeout参数会转换为LINUX时间后进行等待
获取event mutex
循环
if (m_set || signal_count != reset_sig_count) TODO:具体含义?
break
如果m_set 设置也就是处于唤醒状态
并且signal count不等于reset sig count
这部分和wait_low 完全相反
os_event::timed_wait
调入pthread_cond_timedwait
带入等待时间进行等待
退出event mutex
- signal
注意signal唤醒不设置m_set 条件
ret = pthread_cond_signal(&cond_var)
这里需要注意除了Mutex和RW Lock会用到event以外，很多其他地方也会用到，比如purge线程，协调线程和worker线程之间的唤醒和等待操作，也是用的event。
二、Innodb Mutex
对于Mutex而言，不同线程需要持有Mutex进行临界区更改是需要互相堵塞的，这部分总的来说，当前(8023)使用的是PolicyMutex，而PolicyMutex又分为2种，
- PolicyMutex<TTASEventMutex<GenericPolicy>> SyncArrayMutex
这种Mutex主要用于对于一般目的Mutex，因为存在的个数比较少，因此GenericPolicy在对Mutex spin次数和等待次数计数的时候使用了一个vector 数组单独计数。 - PolicyMutex<TTASEventMutex<BlockMutexPolicy>> BlockSyncArrayMutex
这种Mutex代码中基本用到了BPmutex上，也就是page上，因为堵塞在page上的锁可能很多，因此无法单独计数，作为整体计数放到vector数组中作为一个成员，否则vector数组就可能很大。
其主要的数据结构包含，
- PolicyMutex:比如实例化为PolicyMutex<TTASEventMutex<BlockMutexPolicy>>
- TTASEventMutex:比如实例化为TTASEventMutex<BlockMutexPolicy>
- GenericPolicy和BlockMutexPolicy
同时包含一个计数器的数据结构，主要用于存储各个Mutex spin和wait的情况，整体如下，
LatchMetaData 数据结构
latch_meta_t[0] LatchMeta<LatchCounter>类型
latch_id_t m_id
const char *m_name
latch_level_t m_level
const char *m_level_name
pfs_key_t m_pfs_key
Counter m_counter
mutable Mutex m_mutex
bool m_active
std::vector<Count *> m_counters
Count[0] Count类型
uint32_t m_spins
uint32_t m_waits
uint32_t m_calls
bool m_enabled
Count[1] Count类型
uint32_t m_spins
uint32_t m_waits
uint32_t m_calls
bool m_enabled
...
Count[n] Count类型
uint32_t m_spins
uint32_t m_waits
uint32_t m_calls
bool m_enabled
latch_meta_t[1] LatchMeta<LatchCounter>类型
...
latch_meta_t[n] LatchMeta<LatchCounter>类型
从Mutex 加锁的方式来看，其主要分为3个步骤，
- 根据原子变量lock word判断是否能够直接加锁成功
- 如果不能加锁成功则进行spin
- 如果spin后依旧不能加锁成功，则调用os_event的方法进行等待
其中lock word只有3种可能的状态如下，
- MUTEX_STATE_UNLOCKED:0 锁空闲
- MUTEX_STATE_LOCKED:1 加锁状态
- MUTEX_STATE_WAITERS:2 等待状态
而其包含的方法主要是通过宏封装后的，比如我们以block->mutex为例，
- 初始化：mutex_create(LATCH_ID_BUF_BLOCK_MUTEX, &block->mutex)
宏定义:mutex_create(I, M) mutex_init((M), (I), FILE, LINE)
实际上使用的是PolicyMutex<TTASEventMutex<BlockMutexPolicy>>::init - 加锁：mutex_enter(&block->mutex)
宏定义:mutex_enter(M) (M)->enter(srv_n_spin_wait_rounds, srv_spin_wait_delay, FILE, LINE)
实际上使用的是PolicyMutex<TTASEventMutex<BlockMutexPolicy>>::enter
这里可以看到spin参数传入 - 解锁：mutex_exit(&block->mutex)
宏定义:mutex_exit(M) (M)->exit()
实际调用的是PolicyMutex<TTASEventMutex<BlockMutexPolicy>>::exit
具体的代码细节不展开，放到附件种供参考。
三、RW Lock
对于RW Lock来讲，其和Mutex不同之处就在于对于S锁是兼容的，并且S锁和SX锁之间也是兼容的，这就不能用简单的Mutex来实现了，其主要的奥秘就在于其中的lock word有很多种表示方式，其通过这个原子变量能够判断出能够加锁成功，兼容性如下，
LOCK COMPATIBILITY MATRIX
S SX X
S + + -
SX + - -
X - - -
下面我将lock word种源码的描述放出来如下，
IMPLEMENTATION OF THE RW_LOCK
=============================
The status of a rw_lock is held in lock_word. The initial value of lock_word is
X_LOCK_DECR. lock_word is decremented by 1 for each s-lock and by X_LOCK_DECR
or 1 for each x-lock. This describes the lock state for each value of lock_word:
lock_word == X_LOCK_DECR: Unlocked.
X_LOCK_HALF_DECR < lock_word < X_LOCK_DECR:
S locked, no waiting writers.
(X_LOCK_DECR - lock_word) is the number
of S locks.
lock_word == X_LOCK_HALF_DECR: SX locked, no waiting writers.
0 < lock_word < X_LOCK_HALF_DECR:
SX locked AND S locked, no waiting writers.
(X_LOCK_HALF_DECR - lock_word) is the number
of S locks.
lock_word == 0: X locked, no waiting writers.
-X_LOCK_HALF_DECR < lock_word < 0:
S locked, with a waiting writer.
(-lock_word) is the number of S locks.
lock_word == -X_LOCK_HALF_DECR: X locked and SX locked, no waiting writers.
-X_LOCK_DECR < lock_word < -X_LOCK_HALF_DECR:
S locked, with a waiting writer
which has SX lock.
-(lock_word + X_LOCK_HALF_DECR) is the number
of S locks.
lock_word == -X_LOCK_DECR: X locked with recursive X lock (2 X locks).
-(X_LOCK_DECR + X_LOCK_HALF_DECR) < lock_word < -X_LOCK_DECR:
X locked. The number of the X locks is:
2 - (lock_word + X_LOCK_DECR)
lock_word == -(X_LOCK_DECR + X_LOCK_HALF_DECR):
X locked with recursive X lock (2 X locks)
and SX locked.
lock_word < -(X_LOCK_DECR + X_LOCK_HALF_DECR):
X locked and SX locked.
The number of the X locks is:
2 - (lock_word + X_LOCK_DECR + X_LOCK_HALF_DECR)
当在判断S锁和X锁，都在一定情况下做的-1操作，因此可以简单通过lock word得到加锁线程的个数，通常对于lock word而言初始化为X_LOCK_DECR，对于不同的加锁请求如下，
- X锁：需要减去X_LOCK_DECR，如果同一个线程有递归的本锁的2个以上X锁存在，则来到-X_LOCK_DECR，随后的加递归X锁 lock word -1
- S锁：需要减去lock word-1即可
- SX锁：需要减去X_LOCK_HALF_DECR
因此通过lock word就可以判断出当前加锁的情况，是否不满足需求，不满足则进行spin，如果还不满足则进入os_event等待。
其次再RW Lock种存在一种特殊的情况，就是X锁在等待前置S锁的时候，这种情况下需要X锁的线程会先对RW lock的lock word减去X_LOCK_DECR，然后通过RW Lock种的另外一个os_event wait_ex_event 去等待，S锁释放后也会使用这个event去唤醒等待种的需要持有X锁的线程，因此RW Lock种存在2个os_event，如下
- event：用于其他情况
- wait_ex_event：用于前置S锁唤醒X锁
在最后RW Lock也存在一个计数器，这个计数器主要记录整个RW Lock的spin和wait情况，叫做rw_lock_stats，这是一个全局的变量但是每个指标的类型为ib_counter_t，ib_counter_t是一个类似int64类型的数组，比如
- uint64_counter_t rw_s_spin_wait_count
但是其不能完全保证指标的正确性，因为没有额外的保护机制，同时也不是原子变量，只是一个近似值，不过这已经足够了，如下，
not 100% accurate, since it is not atomic
RW Lock常用的方法大概包含，
- rw_lock_create
- rw_lock_sx_lock
- rw_lock_sx_unlock
- rw_lock_s_lock
- rw_lock_s_unlock
- rw_lock_x_lock
- rw_lock_x_unlock
其中rw_lock_x_lock最为复杂，而rw_lock_s_lock最为简单。
四、关于spin
我们这里可以看到不管Mutex还是RW Lock实际上都是通过下面的步骤进行，
- 首先判断原子变量lock word，如果判断成功直接加锁
- 如果不成功则spin
- 如果spin失败则调用OS的mutex机制等待
可以看到spin实际上关键的一个步骤，而对于spin来讲通常和我们的参数innodb_spin_wait_delay,innodb_spin_wait_pause_multiplier,innodb_sync_spin_loops有关
先进行spin,每次等待时长srv_spin_wait_delay*innodb_spin_wait_pause_multiplier(默认为6*50)次 pause指令(ut_delay函数)
醒来后然后尝试加锁，加锁成功则返回
如果尝试加锁失败，则再次循环直到srv_n_spin_wait_rounds的次数(默认30次)
如果等待满srv_n_spin_wait_rounds则spin结束，进入mutex等待唤醒
等待在os_event上，等待唤醒
而pause指令也是一个让出CPU的指令，且和服务器CPU类型相关，不同的CPU可能等待的时间不一样。如果pause指令周期上升则可能导致自旋时间升高，而CPU利用率下降，导致获取锁时间提升，这个已经有很多文章描述过了。
五、附件和参考
- https://cn.pingcap.com/article/post/2651.html MySQL引擎特性：InnoDB同步机制
- https://zhuanlan.zhihu.com/p/18177571946 innodb rw_lock实现
--- 这个结构还会在锁超时的时候用到
struct srv_slot_t
srv_thread_type type 线程的类型 SRV_MASTER、SRV_PURGE、SRV_WORKER、SRV_NONE ，用户线程的type是什么
bool in_use 是否正在使用，一般都是在使用
bool suspended 是否处于等待状态
std::chrono::steady_clock::time_point suspend_time 每次等待的时间
std::chrono::steady_clock::duration wait_timeout 对于innodb_time_out时间
os_event_t event 如果处于没事可做的状态下，等待在这下面 ！！！！
que_thr_t *thr 对于用户线程来讲，也就是session这个指针指向了用户线程
其中包含了一个os_event_t结构
--- 这个结构主要是master线程/purge线程/purge线程的worker线程用到
struct srv_sys_t
ib_mutex_t tasks_mutex 保护下面的链表，任务链表
UT_LIST_BASE_NODE_T(que_thr_t, queue) 任务链表，任务链表主要用于purge和worker线程之间任务的分发
ib_mutex_t mutex 保护下面的属性
ulint n_sys_threads sys_threads中元素的个数
srv_slot_t *sys_threads 存储srv_slot_t的内存空间 ！！！！
ulint n_threads_active[SRV_MASTER + 1] master、purge、worker线程的数量
srv_stats_t::ulint_ctr_1_t activity_count
(gdb) p *srv_sys->sys_threads
$5 = {type = SRV_MASTER, in_use = true, suspended = false, suspend_time = {__d = {__r = 0}}, reservation_no = 0, wait_timeout = {__r = 0}, event = 0x7fffe0020560, thr = 0x0}
(gdb) p srv_sys->sys_threads[0]
$6 = {type = SRV_MASTER, in_use = true, suspended = false, suspend_time = {__d = {__r = 0}}, reservation_no = 0, wait_timeout = {__r = 0}, event = 0x7fffe0020560, thr = 0x0}
(gdb) p srv_sys->sys_threads[1]
$7 = {type = SRV_PURGE, in_use = true, suspended = true, suspend_time = {__d = {__r = 0}}, reservation_no = 0, wait_timeout = {__r = 0}, event = 0x7fffe0020600, thr = 0x0}
(gdb) p srv_sys->sys_threads[2]
$8 = {type = SRV_WORKER, in_use = true, suspended = true, suspend_time = {__d = {__r = 0}}, reservation_no = 0, wait_timeout = {__r = 0}, event = 0x7fffe00206a0, thr = 0x0}
(gdb) p srv_sys->sys_threads[3]
$9 = {type = SRV_WORKER, in_use = true, suspended = true, suspend_time = {__d = {__r = 0}}, reservation_no = 0, wait_timeout = {__r = 0}, event = 0x7fffe0020740, thr = 0x0}
(gdb) p srv_sys->sys_threads[4]
$10 = {type = SRV_WORKER, in_use = true, suspended = true, suspend_time = {__d = {__r = 0}}, reservation_no = 0, wait_timeout = {__r = 0}, event = 0x7fffe00207e0, thr = 0x0}
(gdb) p srv_sys->sys_threads[5]
$11 = {type = 165, in_use = false, suspended = false, suspend_time = {__d = {__r = 140737212723904}}, reservation_no = 144, wait_timeout = {__r = 301}, event = 0x2000000000, thr = 0x1}
---- event包含一个mutex 一个条件变量，还包含2个自带的表示状态的元素，是一个典型的mutex和条件变量的封装
很多地方都会用到
struct os_event
bool m_set; /*!< this is true when the 是否已经处于唤醒状态下，如果不处于才可以唤醒，初始化情况下为false
event is in the signaled
state, i.e., a thread does
not stop if it tries to wait
for this event */
int64_t signal_count; /*!< this is incremented 每次处于唤醒状态就会增加，初始化情况下为1
each time the event becomes
signaled */
EventMutex mutex; /*!< this mutex protects
the next fields */
os_cond_t cond_var; /*!< condition variable is
used in waiting for the event */
包含方法
-- destroy 封装 os_event_destroy
析构event
-- set 封装 os_event_set
进入event mutex
->if (!m_set)
如果m_set设置为false 那么就进行唤醒
如果m_set为true则不唤醒
如下，调用broadcast
broadcast()
释放event mutex
-- broadcast
m_set = true
首先设置唤醒为true
++signal_count
并且signal_count+1
ret = pthread_cond_broadcast(&cond_var)
进行唤醒操作
-- try_set 封装 os_event_try_set
如果mutex.try_lock() 也就是 mutex封装的try lock方法也就是调用pthread_mutex_trylock
返回为true，则进行唤醒，
否则直接返回false
-- reset 封装 os_event_reset
获取mutex，如果m_set 为true，则设置m_set = false，并且返回signal_count
设置m_set 后就代表唤醒完成了，这个变量应该被broadcast唤醒者更改
-- wait_low (reset_sig_count) 封装 os_event_wait_low
获取event mutex
->if (!reset_sig_count) {reset_sig_count = signal_count}
这里需要注意如果传入的reset_sig_count为0，则直接把signal_count赋予给reset_sig_count ，也就是判断的时候
排除reset_sig_count的因素
->while (!m_set && signal_count == reset_sig_count)
根据条件!m_set && signal_count == reset_sig_count 进入
->wait();
而wait()就是简单的pthread_cond_t 的等待pthread_cond_wait(&cond_var, mutex)
然后释放event mutex
这里加入reset_sig_count进行判断的意思如注释所解释到的如下，
thread A calls os_event_reset()
thread B calls os_event_set() [event->m_set == true] //唤醒回++signal_count 这样 signal_count == reset_sig_count 条件
thread C calls os_event_reset() [event->m_set == false] //判定失败，这种情况就解除了，因此必须加锁本条件
thread A calls os_event_wait() [infinite wait!]
thread C calls os_event_wait() [infinite wait!]
-- wait_time_low(timeout ,reset_sig_count)
这个函数传入了一个时间timeout ，这个timeout参数会转换为LINUX时间后进行等待
获取event mutex
循环
if (m_set || signal_count != reset_sig_count) TODO:具体含义?
break
如果m_set 设置也就是处于唤醒状态
并且signal count不等于reset sig count
这部分和wait_low 完全相反
os_event::timed_wait
调入pthread_cond_timedwait
带入等待时间进行等待
退出event mutex
-- signal
注意signal唤醒不设置m_set 条件
ret = pthread_cond_signal(&cond_var)
这里感觉os_event实现了2个没有意义的条件
为m_set和signal_count
可能是broadcast 线程设置m_set=1 并且 signal_count++
然后被唤醒者调用reset 设置 m_set=0
---------------------------------------------------------
PolicyMutex
---enter
---trylock
---
private:
/** The mutex implementation */
MutexImpl m_impl;
#ifdef UNIV_PFS_MUTEX
/** The performance schema instrumentation hook. */
PSI_mutex *m_ptr;
BlockMutexPolicy
typedef latch_meta_t::CounterType Counter;
/** The user visible counters, registered with the meta-data. */
Counter::Count *m_count;
/** Latch meta data ID */
latch_id_t m_id;
GenericPolicy
typedef latch_meta_t::CounterType Counter;
/** The user visible counters, registered with the meta-data. */
Counter::Count m_count;
/** Latch meta data ID */
latch_id_t m_id;
template <typename Mutex>
MutexDebug
- Context
/** Mutex to check for lock order violation */
const Mutex *m_mutex;
/** Filename from where enter was called */
const char *m_filename;
/** Line mumber in filename */
ulint m_line;
/** Thread ID of the thread that own(ed) the mutex */
os_thread_id_t m_thread_id;
- latch
/** Latch ID */
latch_id_t m_id;
/** true if it is a rw-lock. In debug mode, rw_lock_t derives from
this class and sets this variable. */
bool m_rw_lock;
/** true if it is an temporary space latch */
bool m_temp_fsp;
/** Magic number to check for memory corruption. */
ulint m_magic_n;
/** Latch state of the mutex owner */
Context m_context;
TTASEventMutex<Policy>::wait
TTASEventMutex<Policy>::signal
PolicyMutex<TTASEventMutex<GenericPolicy> >::enter
#ifndef UNIV_HOTBACKUP
#ifdef MUTEX_FUTEX
/** The default mutex type. */
typedef FutexMutex ib_mutex_t;
typedef BlockFutexMutex ib_bpmutex_t;
#define MUTEX_TYPE "Uses futexes"
#elif defined(MUTEX_SYS)
typedef SysMutex ib_mutex_t;
typedef BlockSysMutex ib_bpmutex_t;
#define MUTEX_TYPE "Uses system mutexes"
#elif defined(MUTEX_EVENT)
typedef SyncArrayMutex ib_mutex_t;
typedef BlockSyncArrayMutex ib_bpmutex_t;
#define MUTEX_TYPE "Uses event mutexes"
#else
#error "ib_mutex_t type is unknown"
#endif /* MUTEX_FUTEX */
typedef SyncArrayMutex ib_mutex_t;
typedef BlockSyncArrayMutex ib_bpmutex_t;
UT_MUTEX_TYPE(TTASEventMutex, GenericPolicy, SyncArrayMutex)
UT_MUTEX_TYPE(TTASEventMutex, BlockMutexPolicy, BlockSyncArrayMutex)
#define UT_MUTEX_TYPE(M, P, T) typedef PolicyMutex<M<P>> T;
定义:
PolicyMutex<TTASEventMutex<GenericPolicy>> SyncArrayMutex
PolicyMutex<TTASEventMutex<BlockMutexPolicy>> BlockSyncArrayMutex
#ifdef HAVE_IB_LINUX_FUTEX
UT_MUTEX_TYPE(TTASFutexMutex, GenericPolicy, FutexMutex)
UT_MUTEX_TYPE(TTASFutexMutex, BlockMutexPolicy, BlockFutexMutex)
#endif /* HAVE_IB_LINUX_FUTEX */
UT_MUTEX_TYPE(OSTrackMutex, GenericPolicy, SysMutex)
UT_MUTEX_TYPE(OSTrackMutex, BlockMutexPolicy, BlockSysMutex)
主要使用下面的定义:
UT_MUTEX_TYPE(TTASEventMutex, GenericPolicy, SyncArrayMutex)
UT_MUTEX_TYPE(TTASEventMutex, BlockMutexPolicy, BlockSyncArrayMutex)
#ifndef UNIV_HOTBACKUP
#ifdef MUTEX_FUTEX
/** The default mutex type. */
typedef FutexMutex ib_mutex_t;
typedef BlockFutexMutex ib_bpmutex_t;
#define MUTEX_TYPE "Uses futexes"
#elif defined(MUTEX_SYS)
typedef SysMutex ib_mutex_t;
typedef BlockSysMutex ib_bpmutex_t;
#define MUTEX_TYPE "Uses system mutexes"
#elif defined(MUTEX_EVENT)
主要使用下面的定义:
typedef SyncArrayMutex ib_mutex_t;
typedef BlockSyncArrayMutex ib_bpmutex_t;
#define MUTEX_TYPE "Uses event mutexes"
#else
#error "ib_mutex_t type is unknown"
#endif /* MUTEX_FUTEX */
extern ulong srv_spin_wait_delay;
extern ulong srv_n_spin_wait_rounds;
#define mutex_create(I, M) mutex_init((M), (I), __FILE__, __LINE__)
#define mutex_enter(M) \
(M)->enter(srv_n_spin_wait_rounds, srv_spin_wait_delay, __FILE__, __LINE__)
#define mutex_enter_nospin(M) (M)->enter(0, 0, __FILE__, __LINE__)
#define mutex_enter_nowait(M) (M)->trylock(__FILE__, __LINE__)
#define mutex_exit(M) (M)->exit()
#define mutex_free(M) mutex_destroy(M)
typedef ib_bpmutex_t BPageMutex;
typedef ib_mutex_t BufListMutex;
typedef ib_mutex_t FlushListMutex;
typedef BPageMutex BufPoolZipMutex;
#ifndef UNIV_HOTBACKUP
typedef rw_lock_t BPageLock;
-----------------------------------------------------------------------------------------------------
实际主要使用2种策略锁
UT_MUTEX_TYPE(TTASEventMutex, GenericPolicy, SyncArrayMutex)
UT_MUTEX_TYPE(TTASEventMutex, BlockMutexPolicy, BlockSyncArrayMutex)
展开宏后如下，
PolicyMutex<TTASEventMutex<GenericPolicy>> SyncArrayMutex
PolicyMutex<TTASEventMutex<BlockMutexPolicy>> BlockSyncArrayMutex
定义2个新的类型
typedef SyncArrayMutex ib_mutex_t;
typedef BlockSyncArrayMutex ib_bpmutex_t;
主要使用的方式包含
尝试加锁部分:
mutex_enter(M) 展开后 (M)->enter(srv_n_spin_wait_rounds, srv_spin_wait_delay, __FILE__, __LINE__)
调用的是PolicyMutex的enter方法，这里带入了srv_n_spin_wait_rounds, srv_spin_wait_delay参数
mutex_enter_nospin(M) 展开后 (M)->enter(0, 0, __FILE__, __LINE__)
调用的是PolicyMutex的enter方法,这里带入的srv_n_spin_wait_rounds, srv_spin_wait_delay为0
mutex_enter_nowait(M) 展开后 (M)->trylock(__FILE__, __LINE__)
这里使用了PolicyMutex的trylock方法
解锁部分:
mutex_exit(M) (M)->exit()
调用的是调用的是PolicyMutex 的 exit方法
析构
mutex_free(M) mutex_destroy(M)
本处调用是PolicyMutex的destroy方法
初始化
mutex_create(I, M) mutex_init((M), (I), __FILE__, __LINE__)
本处调用的是mutex_init,也就是PolicyMutex的init 方法
new (mutex) Mutex();
mutex->init(id, file_name, line);
--- PolicyMutex ---
template <typename MutexImpl>
struct PolicyMutex
实例化为:
PolicyMutex<TTASEventMutex<GenericPolicy>> SyncArrayMutex
PolicyMutex<TTASEventMutex<BlockMutexPolicy>> BlockSyncArrayMutex
typedef MutexImpl MutexType;// TTASEventMutex<GenericPolicy> / TTASEventMutex<BlockMutexPolicy> or
typedef typename MutexImpl::MutexPolicy Policy; //TTASEventMutex<BlockMutexPolicy>::MutexPolicy or
成员
/** The mutex implementation */
MutexImpl m_impl; // TTASEventMutex<BlockMutexPolicy> or TTASEventMutex<GenericPolicy>
#ifdef UNIV_PFS_MUTEX
/** The performance schema instrumentation hook. */
PSI_mutex *m_ptr;
#endif /* UNIV_PFS_MUTEX */
--- TTASEventMutex ---
template <template <typename> class Policy = NoPolicy>
struct TTASEventMutex
实例化为:
TTASEventMutex<GenericPolicy>
TTASEventMutex<BlockMutexPolicy>
typedef Policy<TTASEventMutex> MutexPolicy; //GenericPolicy or BlockMutexPolicy
成员
/** lock_word is the target of the atomic test-and-set instruction
when atomic operations are enabled. */
std::atomic_bool m_lock_word;
/** true if there are (or may be) threads waiting
in the global wait array for this mutex to be released. */
std::atomic_bool m_waiters;
/** Used by sync0arr.cc for the wait queue */
os_event_t m_event;
/** Policy data */
MutexPolicy m_policy;
--- GenericPolicy ---
template <typename Mutex>
struct GenericPolicy
--- BlockMutexPolicy ---
template <typename Mutex>
class BlockMutexPolicy
在TTASEventMutex实例化为Policy<TTASEventMutex> MutexPolicy
typedef Mutex MutexType; //TTASEventMutex
typedef typename latch_meta_t::CounterType::Count Count;
typedef latch_meta_t::CounterType Counter
Counter::Count *m_count;
latch_id_t m_id;
Counter部分为计数器如下
typedef LatchMeta<LatchCounter> latch_meta_t;
typedef std::vector<latch_meta_t *, ut_allocator<latch_meta_t *>> LatchMetaData;
定义一个新的类型LatchMetaData，其是一个latch_meta_t的容器，latch_meta_t是一个
定义的LatchMeta<LatchCounter> 的新类型
template <typename Counter = LatchCounter>
-- LatchMeta
/** Latch id */
latch_id_t m_id;
/** Latch name */
const char *m_name;
/** Latch level in the ordering */
latch_level_t m_level;
/** Latch level text representation */
const char *m_level_name;
#ifdef UNIV_PFS_MUTEX
/** PFS key */
pfs_key_t m_pfs_key;
#endif /* UNIV_PFS_MUTEX */
/** For gathering latch statistics */
Counter m_counter;
这里m_counter为一个计数器，是一个LatchCounter的类型
-- LatchCounter
一个计数器用于计算spin次数等
typedef OSMutex Mutex;
typedef std::vector<Count *> Counters;
/** Mutex protecting m_counters */
mutable Mutex m_mutex; 这把锁主要用于保护Counters vector容器
/** Counters for the latches */
Counters m_counters;
/** if true then we collect the data */
bool m_active;
-- struct Count {
/** Constructor */
Count() UNIV_NOTHROW : m_spins(), m_waits(), m_calls(), m_enabled() {
/* No op */
}
/** Rest the values to zero */
void reset() UNIV_NOTHROW {
m_spins = 0;
m_waits = 0;
m_calls = 0;
}
uint32_t m_spins; ///** Number of spins trying to acquire the latch. */ spin的次数
uint32_t m_waits; ///** Number of waits trying to acquire the latch */ 等待的次数
uint32_t m_calls; //** Number of times it was called */ 调用次数
/** true if enabled */
bool m_enabled;
}
这里m_counters为一个vector数组，其中可能包含多个counter，每次注册
LatchMetaData 数据结构
latch_meta_t[0] LatchMeta<LatchCounter>类型
latch_id_t m_id
const char *m_name
latch_level_t m_level
const char *m_level_name
pfs_key_t m_pfs_key
Counter m_counter
mutable Mutex m_mutex
bool m_active
std::vector<Count *> m_counters
Count[0] Count类型
uint32_t m_spins
uint32_t m_waits
uint32_t m_calls
bool m_enabled
Count[1] Count类型
uint32_t m_spins
uint32_t m_waits
uint32_t m_calls
bool m_enabled
...
Count[n] Count类型
uint32_t m_spins
uint32_t m_waits
uint32_t m_calls
bool m_enabled
latch_meta_t[1] LatchMeta<LatchCounter>类型
...
latch_meta_t[n] LatchMeta<LatchCounter>类型
会初始化一个LatchMetaData latch_meta;
#define mutex_create(I, M) mutex_init((M), (I), __FILE__, __LINE__)
#define mutex_enter(M) \
(M)->enter(srv_n_spin_wait_rounds, srv_spin_wait_delay, __FILE__, __LINE__)
#define mutex_enter_nospin(M) (M)->enter(0, 0, __FILE__, __LINE__)
#define mutex_enter_nowait(M) (M)->trylock(__FILE__, __LINE__)
#define mutex_exit(M) (M)->exit()
#define mutex_free(M) mutex_destroy(M)
--初始化:
mutex_create(LATCH_ID_BUF_BLOCK_MUTEX, &block->mutex)
宏定义:mutex_create(I, M) mutex_init((M), (I), __FILE__, __LINE__)
而mutex_init就是
new (mutex) Mutex();
mutex->init(id, file_name, line);
使用指针 &block->mutex为新建mutex的地址
然后调用的block->mutex->init 初始化带入LATCH_ID_BUF_BLOCK_MUTEX
而block->mutex 就是一个ib_bpmutex_t，实际上这里的mutex->init
调用的就是PolicyMutex<TTASEventMutex<BlockMutexPolicy>>::init
-> pfs_add(sync_latch_get_pfs_key(id));
-> m_impl.init(id, filename, line);
这里模板实力化为TTASEventMutex<BlockMutexPolicy>，而m_impl是一个TTASEventMutex<BlockMutexPolicy>定义的类型
TTASEventMutex<BlockMutexPolicy>::init
这里 init为一个初始化，包含如下
->m_event = os_event_create();
初始化一个event
->m_policy.init(*this, id, filename, line)
初始化相关策略GenericPolicy<TTASEventMutex>或者BlockMutexPolicy<TTASEventMutex>中的一个
例子中是BlockMutexPolicy<TTASEventMutex>::init
->m_id = id
这里m_id是latch_id_t，具体如何使用后面分析
->latch_meta_t &meta = sync_latch_get_meta(m_id)
->return (*latch_meta[id])
返回一个LatchMeta<LatchCounter>的类型元素
->m_count = meta.get_counter()->sum_register()
这里需要注意
A:GenericPolicy 策略每个counter放到了一个vector std::vector<Count *>
中，单独保存其数据叫做single_deregister
比如GenericPolicy的初始化函数调用
meta.get_counter()->single_register(&m_count);向m_counters push本次的counter
调用m_counters.push_back(count);
并且多一个->sync_file_created_register(this, filename, line)步骤
B:BlockMutexPolicy 策略每个counter混合放入一个vector std::vector<Count *>
中，只保留一个元素，因为太多了block，可能导致vector过长过大，叫做
sum_deregister，因此sum_deregister如果带参数的调用是空函数，这里就是
BlockMutexPolicy
->ut_d(MutexDebug<MutexType>::init(m_id))
--加锁
mutex_enter(&block->mutex)
宏定义:mutex_enter(M) (M)->enter(srv_n_spin_wait_rounds, srv_spin_wait_delay, __FILE__, __LINE__)
注意这里传入了需要spin的时间和周期
block->mutex->enter
PolicyMutex<TTASEventMutex<BlockMutexPolicy>>::enter(uint32_t n_spins, uint32_t n_delay, const char *name,uint32_t line)
->locker = pfs_begin_lock(&state, name, line);
ps不考虑
->policy().enter(m_impl, name, line)
GenericPolicy<TTASEventMutex>或者BlockMutexPolicy<TTASEventMutex>中的一个
->BlockMutexPolicy<TTASEventMutex>::enter
非debug版本无特殊意义
->m_impl.enter(n_spins, n_delay, name, line)
TTASEventMutex<BlockMutexPolicy>或者TTASEventMutex<BlockMutexPolicy>的一个
->TTASEventMutex<BlockMutexPolicy>::enter
->if (!try_lock())
->bool expected = false;return (m_lock_word.compare_exchange_strong(expected, true));
A.expected期望值为false，如果m_lock_word为false，则修改m_lock_word为true，且返回true
B.expected期望值为false，如果m_lock_word为true，则和期望值不同， 则返回false，修改expected期望值为true
->如果返回为false，说明尝试加锁不成功，可以看出这里也是用原子变量先来判断是否加锁成功
则个和RW LOCK一样，只是lock word变得简单了，只有0 1 2 三种状态分别对应MUTEX_STATE_UNLOCKED/MUTEX_STATE_LOCKED/MUTEX_STATE_WAITERS
无锁/加锁/等待
->spin_and_try_lock(max_spins, max_delay, filename, line)
TTASEventMutex<BlockMutexPolicy>::spin_and_try_lock
进行spin操作然后等待
->for (;;) {
循环加锁成功退出
->if (is_free(max_spins, max_delay, n_spins))
TTASEventMutex<BlockMutexPolicy>::is_free
进行自旋后查看lock word是否空闲
->is_free
->do {..ut_delay..;++n_spins;} while (n_spins < max_spins)
尝试一个周期的自旋操作
如果自旋后lock word空闲则返回true，否则返回false
->如果is_free返回true
if (try_lock())
再次尝试加锁
->如果自旋不能达到目标则
max_spins = n_spins + step
每次max_spins + 4，启发式算法，扩大spin的周期
->++n_waits;
spin失败一次waits+1
-> if (wait(filename, line, 4))
TTASEventMutex<GenericPolicy>::wait
ut0mutex.ic:173
->sync_arr = sync_array_get_and_reserve_cell(this,(m_policy.get_id() == LATCH_ID_BUF_BLOCK_MUTEX || m_policy.get_id() == LATCH_ID_BUF_POOL_ZIP)? SYNC_BUF_BLOCK : SYNC_MUTEX, filename, line, &cell);
这里m_policy.get_id()就是 BlockMutexPolicy<TTASEventMutex> or 的m_id
SYNC_BUF_BLOCK为bpmutex类型，SYNC_MUTEX为mutex类型
->set_waiters()
设置lock word为等待状态 2 MUTEX_STATE_WAITERS
TTASEventMutex<BlockMutexPolicy>::set_waiters
->return m_lock_word.exchange(MUTEX_STATE_WAITERS) == MUTEX_STATE_UNLOCKED
设置为MUTEX_STATE_WAITERS并且返回m_lock_word的old value和MUTEX_STATE_UNLOCKED比较
->for (uint32_t i = 0; i < spin; ++i)
再次尝试4次加锁，主要还是通过lock word进行判断
->try_lock
TTASEventMutex<BlockMutexPolicy>::try_lock
return (m_lock_word.compare_exchange_strong(expected, true))
A.expected期望值为false，如果m_lock_word为false，则修改m_lock_word为true，且返回true
B.expected期望值为false，如果m_lock_word为true，则和期望值不同， 则返回false，修改expected期望值为true
->如果加锁成功则释放cell
->如果失败则进入等待
sync_array_wait_event(sync_arr, cell)
->m_policy.add(n_spins, n_waits)
spin_and_try_lock获取锁成功则需要增加自旋周期和等待次数，显然 n_waits是一次spin周期
拿不到锁才增加一次
->如果返回为true，则说明加锁成功
--解锁
mutex_exit(&block->mutex)
宏定义:mutex_exit(M) (M)->exit()
block->mutex->exit
PolicyMutex<TTASEventMutex<BlockMutexPolicy>>::exit
->pfs_exit();
ps不考虑
->policy().release(m_impl)
GenericPolicy<TTASEventMutex>::release
本函数无特殊含义
->m_impl.exit()
TTASEventMutex<BlockMutexPolicy>或者TTASEventMutex<GenericPolicy>的一个
->TTASEventMutex<GenericPolicy>::exit
->m_lock_word.store(false)
存储lock word为MUTEX_STATE_UNLOCKED，这里释放了设置为MUTEX_STATE_UNLOCKED
-> if (m_waiters.load())
如果存在等待者
->signal()
TTASEventMutex<GenericPolicy>::signal()
->os_event_set(m_event);
唤醒
->sync_array_object_signalled
->++sg_count
---------------------------------------------------------------------------------------------------------------------------------------
读写锁
srv_master_evict_from_table_cache srv0srv.cc line 1989
row0purge.cc line 882
dict0stats.cc line 2412
row0row.cc line 999
btr0sea.cc line 1037
struct rw_lock_t
volatile lint lock_word;
volatile ulint waiters
volatile bool recursive
volatile ulint sx_recursive
bool writer_is_wait_ex
volatile os_thread_id_t writer_thread
os_event_t event
os_event_t wait_ex_event
const char* cfile_name
const char* last_s_file_name
const char* last_x_file_name
unsigned cline:13
unsigned is_block_lock:1
unsigned last_s_line:14
unsigned last_x_line:14
uint32_t count_os_wait
UT_LIST_NODE_T(rw_lock_t) list
struct PSI_rwlock* pfs_psi
mutable ib_mutex_t mutex
ulint magic_n
UT_LIST_BASE_NODE_T(rw_lock_debug_t) debug_list
latch_level_t level
rw_lock_create
rw_lock_s_lock
rw_lock_s_unlock
rw_lock_sx_lock
rw_lock_sx_unlock
rw_lock_x_lock
rw_lock_x_unlock
特别
rw_lock_x_lock_gen(&block->lock, BUF_IO_READ);
define rw_lock_x_lock_gen(M, P) \
rw_lock_x_lock_func((M), (P), __FILE__, __LINE__)
#define X_LOCK_DECR 0x20000000 //512M
#define X_LOCK_HALF_DECR 0x10000000 //256M
rw_lock_create(btr_search_latch_key,btr_search_latches[i], SYNC_SEARCH_SYS)
# define rw_lock_create(K, L, level) \
pfs_rw_lock_create_func((K), (L), (level), #L, __FILE__, __LINE__)
pfs_rw_lock_create_func((btr_search_latch_key),(btr_search_latches[i]),(SYNC_SEARCH_SYS),"btr_search_latches[i]", __FILE__, __LINE__)
pfs_rw_lock_create_func(mysql_pfs_key_t key ,rw_lock_t *lock, latch_level_t level, const char *cmutex_name, const char *cfile_name,ulint cline)
->new (lock) rw_lock_t
建立一个新的rw_lock_t 结构
->lock->pfs_psi = PSI_RWLOCK_CALL(init_rwlock)(key.m_value, lock);
和ps相关
->rw_lock_create_func(lock, level, cmutex_name,cfile_name, cline)
rw_lock_create_func(btr_search_latches[i],SYNC_SEARCH_SYS,const char *cmutex_name, const char *cfile_name,ulint cline)
这里建立的读写锁就是btr_search_latches[i]
->lock->lock_word = X_LOCK_DECR 初始化为X_LOCK_DECR
->lock->waiters = false 是否处于等待状态
->lock->recursive.store(false, std::memory_order_relaxed) 是否有递归锁，比如X锁后获取SX, X后获取X，但是前者用sx_recursive进行标记
->lock->sx_recursive = 0 是否递归获取了SX锁，也就是同一个线程对同一把锁获取了X锁或者SX锁后再次获取SX锁
->lock->writer_thread = std::thread().native_handle() 当前写入线程包含SX和X的线程
->lock->cfile_name = cfile_name 创建lock的代码位置
->lock->cline = cline 创建lock的代码行数
->lock->count_os_wait = 0 总体等待次数
->lock->last_s_file_name = "not yet reserved";
->lock->last_x_file_name = "not yet reserved";
->lock->last_s_line = 0;
->lock->last_x_line = 0;
->lock->event = os_event_create(); //主要的event的唤醒机制，
->lock->wait_ex_event = os_event_create(); //主要用于S和X锁，X锁作为后续读取，等待S锁释放
->lock->is_block_lock = false;
->UT_LIST_ADD_FIRST(rw_lock_list, lock)
加入到rw_lock_list中/* The global list of rw-locks */
----- 加X锁
rw_lock_x_lock(btr_get_search_latch(index));
pfs_rw_lock_x_lock_func((M), 0, __FILE__, __LINE__)
pfs_rw_lock_x_lock_func(rw_lock_t *lock,ulint pass,const char *file_name, ulint line)
->if (lock->pfs_psi != nullptr)
先忽略和ps有关
->rw_lock_x_lock_func(lock, pass, file_name, line)
调用rw_lock_x_lock_func，这里的pass都是0
-> ulint i = 0;sync_array_t *sync_arr;ulint spin_count = 0;uint64_t count_os_wait = 0;bool spinning = false;
定义变量
->if (rw_lock_x_lock_low(lock, pass, file_name, line))
这里调用rw_lock_x_lock_low
->if (rw_lock_lock_word_decr(lock, X_LOCK_DECR, X_LOCK_HALF_DECR)) //512M和256M
->local_lock_word = lock->lock_word
lock->lock_word初始化为X_LOCK_DECR，X锁为减去X_LOCK_DECR
->while (local_lock_word > threshold)
如果local_lock_word大于了X_LOCK_HALF_DECR，则循环但是要看compare_exchange_strong的结果
if (lock->lock_word.compare_exchange_strong(local_lock_word,local_lock_word - amount))
期望值为local_lock_word，设定值为local_lock_word - amount
A.当前值与期望值(expect)相等时，修改当前值为设定值(desired)，返回true
B.当前值与期望值(expect)不等时，将期望值(expect)修改为当前值，返回false
如果返回为true则会直接return
->return (true)
返回为true
如果返回为false则循环
如果local_lock_word小于等于X_LOCK_HALF_DECR则退出 比如0？ 也就是小于等于256M
->return (false)
->如果rw_lock_lock_word_decr返回为true
->rw_lock_set_writer_id_and_recursion_flag(lock, !pass)
->lock->writer_thread.store(os_thread_get_curr_id(), std::memory_order_relaxed);lock->recursive.store(recursive, std::memory_order_release);
设置写线程为自己,且设置递归但是不是SX递归
->rw_lock_x_lock_wait
rw_lock_x_lock_wait(lock, pass, 0, file_name, line)
这里threshold为0，
rw_lock_x_lock_wait(L, P, T, F, O) rw_lock_x_lock_wait_func(L, T, F, O)
!!! 这个函数的作用，虽然已经减去了X_LOCK_DECR，但是可能有前置的读操作，需要
判断是否需要等待
Function for the next writer to call. Waits for readers to exit.The caller must have already decremented lock_word by X_LOCK_DECR
->rw_lock_x_lock_wait_func
调入的threshold为0
->ut_ad(lock->lock_word <= threshold)
因为前面已经减去了X_LOCK_DECR，而初始化最多lock word为X_LOCK_DECR
因此这里一定小于等于0
->while (lock->lock_word < threshold)
当lock word小于0的时候，说明存在持有s锁的读线程，需要不断循环做spin
spin srv_spin_wait_delay*srv_n_spin_wait_rounds
如果spin 后还是没有拿到锁，则进入等待
->sync_arr = sync_array_get_and_reserve_cell(lock, RW_LOCK_X_WAIT, file_name,line, &cell);
注意这里如果有前置S锁存在，则使用的是RW_LOCK_X_WAIT类型的event，
也就是lock->wait_ex_event(sync_cell_get_event)
->if (lock->lock_word < threshold)
再次确认小于threshold
->++count_os_wait;
增加等待次数
->sync_array_wait_event(sync_arr, cell)
进入条件变量等待，等待唤醒
->唤醒后可能是出现还是lock word小于0的情况,因此继续循环判断
这部分是单独本线程用的RW_LOCK_X_WAIT类型的event
因为其他SX/X线程不可能拿得到RW_LOCK_X_WAIT类型的event
堵塞在 while (local_lock_word > threshold)这个条件 直接返回false
拿不到X锁
TODO 什么时候小于0，可能存在多个S锁的情况，每次S锁unlock
唤醒?
如果再次确认大于threshold,说明已经没有S lock可，可以获取X lock了
->sync_array_free_cell(sync_arr, cell);
可以释放cell了
->break
退出循环
下面增加计数器
->rw_lock_stats.rw_x_spin_round_count.add(n_spins);
-> if (count_os_wait > 0)
->lock->count_os_wait += static_cast<uint32_t>(count_os_wait);
->rw_lock_stats.rw_x_os_wait_count.add(count_os_wait);
->如果rw_lock_lock_word_decr返回为flase
->if (!pass && lock->recursive.load(std::memory_order_acquire) &&os_thread_eq(lock->writer_thread.load(std::memory_order_relaxed))
A:lock->recursive 如果是递归
B:os_thread_eq(lock->writer_thread.load(std::memory_order_relaxed)) 如果写线程就是当前线程
C:!pass 这个默认就是0
如果满足条件走下面
->if (rw_lock_lock_word_decr(lock, X_LOCK_DECR, 0))
如果成功，存在一个SX lock 和可能的S lock，这里带入的threshold为0,如果成功则必然减去了X_LOCK_DECR
->rw_lock_x_lock_wait(lock, pass, -X_LOCK_HALF_DECR, file_name, line);
可能有前置的S lock，等待S lock 释放
如果更改lock word失败，lock word小于等于0，且本线程持有X锁一次?
->if (lock->lock_word == 0 || lock->lock_word == -X_LOCK_HALF_DECR)
lock->lock_word -= X_LOCK_DECR
直接减去 X_LOCK_DECR
->否则, lock_word <= -X_LOCK_DECR
--lock->lock_word
说明有本线程已经持有本资源的X 2次 剩下的次数减去1？
！！！那么看起来如果本线程多次获取X锁，两次后就是-1了。
->如果不满足条件走下面
加锁失败
return false
如果前面成功则设置文件名字和行数
->lock->last_x_file_name = file_name;
->lock->last_x_line = line
->return true
如果rw_lock_x_lock_low返回为true
->if (count_os_wait > 0)
这里初始化为0，但是经过rw_lock_x_lock_low中的rw_lock_x_lock_wait，等待前置S锁，可能已经出现了等待
因此要增加计数器
rw_lock_stats.rw_x_os_wait_count和rw_lock_stats.rw_x_spin_round_count 增加
->return
返回成功
->else
否则说明加锁不成功，且不是前置只有S锁的情况，这个时候就需要spin了
都是一样的直接跳过
->进入等待阶段
首先还是找到空闲的cell
sync_arr = sync_array_get_and_reserve_cell(lock, RW_LOCK_X, file_name, line, &cell);
这里和前面不同用的是RW_LOCK_X类型的类型，这里RW_LOCK_X类型的event和上面的不同 ！！！
主要用于需求的锁类型cell->request_type，sync_cell_get_event函数根据不同的锁类型返回不同的
event
->rw_lock_set_waiter_flag(lock);
设置等待标记
->sync_array_wait_event(sync_arr, cell);
等待在event上
->goto lock_loop
唤醒后完成后继续加锁
rw_lock_x_unlock
rw_lock_x_unlock(L) rw_lock_x_unlock_gen(L, 0)
rw_lock_x_unlock_gen(L, P) pfs_rw_lock_x_unlock_func(L)
pfs_rw_lock_x_unlock_func
->if (lock->pfs_psi != nullptr)
ps 不考虑
->rw_lock_x_unlock_func(pass,lock);
其中pass参量主要用于debug
ut_ad(lock->lock_word == 0 || lock->lock_word == -X_LOCK_HALF_DECR || lock->lock_word <= -X_LOCK_DECR);
从这里的断言可以看出这里的lock word只可能小于等于0，因为初始化lock word为+X_LOCK_DECR，每次加X锁需要
减去-X_LOCK_DECR
->if (lock->lock_word == 0)
如果lock_word为0，则说明只有一个X锁，不能存在递归，因此递归变量设置为false
->lock->recursive.store(false, std::memory_order_relaxed);
->if (lock->lock_word == 0 || lock->lock_word == -X_LOCK_HALF_DECR)
如果只有一个 X锁/SX和X锁
-> if (rw_lock_lock_word_incr(lock, X_LOCK_DECR) <= 0)
增加X_LOCK_DECR 必然大于0
-> if (lock->waiters)
??
->rw_lock_reset_waiter_flag(lock);
->os_event_set(lock->event)
唤醒 等待者，通过是event 而不是event_ex
->sync_array_object_signalled()
->else if (lock->lock_word == -X_LOCK_DECR ||lock->lock_word == -(X_LOCK_DECR + X_LOCK_HALF_DECR))
否则可能持有2个X锁和SX锁等
->lock->lock_word += X_LOCK_DECR;
直接增加 X_LOCK_DECR
->else
这里就是大于2个X锁的情况
lock->lock_word += 1;
小于-X_LOCK_DECR 则说明持有2个以上的x锁，则进行+1操作
----- 加S锁
rw_lock_s_lock(btr_get_search_latch(index))
pfs_rw_lock_s_lock_func((M), 0, __FILE__, __LINE__)
pfs_rw_lock_s_lock_func(rw_lock_t *lock,ulint pass,const char *file_name, ulint line)
->if (lock->pfs_psi != nullptr)
ps不考虑
->rw_lock_s_lock_func(lock, pass, file_name, line)
->if (!rw_lock_s_lock_low(lock, pass, file_name, line))
如果加共享锁失败，则进入spin
->if (!rw_lock_lock_word_decr(lock, 1, 0))
->local_lock_word = lock->lock_word
->while (local_lock_word > threshold)
如果local_lock_word大于threshold(0)则不断循环，如果小于等于0则返回false
->if (lock->lock_word.compare_exchange_strong(local_lock_word,local_lock_word - amount))
local_lock_word - 1
期望值(expect)为local_lock_word，设定值(desired)为local_lock_word - amount
A.当前值与期望值(expect)相等时，修改当前值为设定值(desired)，返回true
设置为local_lock_word - 1
B.当前值与期望值(expect)不等时，将期望值(expect)修改为lock->lock_word当前值，返回false
如果能够交换则返回为true
->如果小于等于0则返回false，也就是加锁不成功返回false
->如果rw_lock_lock_word_decr返回false
说明加锁不成功则返回false
如果加锁成功则返回true如下
->lock->last_s_file_name = file_name
->lock->last_s_line = line
->返回true
如果加锁不成功则进行入spin环节
->rw_lock_s_lock_spin(lock, pass, file_name, line)
.....
rw_lock_s_unlock_func
->lint lock_word = rw_lock_lock_word_incr(lock, 1);
这个非常简单，因为S锁只可能是-1情况因此lock word +1就可以了
->if (lock_word == 0 || lock_word == -X_LOCK_HALF_DECR)
这里如果lock word为0/-X_LOCK_HALF_DECR 则说明最后一个S lock已经
释放，且有X锁等待，因此使用wait ex event唤醒x锁的等待者
->os_event_set(lock->wait_ex_event);
->sync_array_object_signalled();
---- 加SX锁
rw_lock_sx_lock(&block->lock);
mtr_latch_type = MTR_MEMO_PAGE_SX_FIX;
rw_lock_sx_lock
pfs_rw_lock_sx_lock_func((M), 0, __FILE__, __LINE__)
pfs_rw_lock_sx_lock_func(rw_lock_t *lock, ulint pass, const char *file_name,ulint line)
->if (lock->pfs_psi != nullptr)
ps不考虑
->rw_lock_sx_lock_func(lock, pass, file_name, line)
ulint i = 0;sync_array_t *sync_arr;ulint spin_count = 0;uint64_t count_os_wait = 0; ulint spin_wait_count = 0;
定义变量
阶段1:加锁和自旋阶段
->if (rw_lock_sx_lock_low(lock, pass, file_name, line))
->if (rw_lock_lock_word_decr(lock, X_LOCK_HALF_DECR, X_LOCK_HALF_DECR))
256M 256M
如果加锁成功
->rw_lock_set_writer_id_and_recursion_flag(lock, !pass)
设置写线程为自己
->lock->sx_recursive = 1
设置递归
->else
如果加锁不成功，也就是小于等于X_LOCK_HALF_DECR，
那么可能自身获得了更高级的X锁，这个时候可以直接上SX锁
( It already has an X or SX lock by this thread or another thread)
->if (!pass && lock->recursive.load(std::memory_order_acquire) &&os_thread_eq(lock->writer_thread.load(std::memory_order_relaxed),os_thread_get_curr_id()))
A.如果需要等待(pass参量为0)
B.当前线程就是写入线程
C.可以递归
D. 递归方式if (lock->sx_recursive++ == 0) 拥有X锁或者SX 锁
断言 加了X锁 或者 同一个线程有多个X锁 递归持有 并且 一个线程 持有多个X和SX ?
->lock->lock_word -= X_LOCK_HALF_DECR
还是直接减去X_LOCK_HALF_DECR
->else
否则无法加SX锁成功，返回false
return false
如果加锁成功，设置加锁文件和位置
->lock->last_x_file_name = file_name
->lock->last_x_line = line
A：如果加锁成功
->if (count_os_wait > 0)
增加计数器
B：如果加锁失败
进入spin，spin后再次获取
阶段2:
->如果进过srv_n_spin_wait_rounds轮次的spin后还是拿不到锁，则没有必要重试了，进入event 等待
->sync_cell_t *cell;
定义一个cell
->sync_arr = sync_array_get_and_reserve_cell(lock, RW_LOCK_SX, file_name, line, &cell)
从sync_wait_array数组中获取一个cell
->sync_cell_get_event
这里可以看出实际上是获取的rw lock中的event但是对于rw lock 有2个event
A:lock->wait_ex_event
B:lock->event
->rw_lock_set_waiter_flag(lock);
设置等待标记
->if (rw_lock_sx_lock_low(lock, pass, file_name, line)) {
再次尝试加锁
....
->++count_os_wait; //等待次数+1
->sync_array_wait_event(sync_arr, cell);
进入等待 等待条件变量 调用的是 os_event_wait_low 等待唤醒
-> i = 0;goto lock_loop;
重新尝试加锁
rw_lock_sx_unlock
rw_lock_sx_unlock(L) pfs_rw_lock_sx_unlock_func(L)
pfs_rw_lock_sx_unlock_func(ulint pass,rw_lock_t *lock)
->if (lock->pfs_psi != nullptr)
ps不考虑
->rw_lock_sx_unlock_func(pass, lock)
->--lock->sx_recursive;
减去递归
->if (lock->sx_recursive == 0)
如果递归位0则自己不会在加SX锁了，但是可能还有X锁
->if (lock->lock_word > 0)
如果当前lock_word大于0 ，这种情况不可能还有X锁了，因此最多还有S锁，因此必然大于X_LOCK_HALF_DECR
->if (rw_lock_lock_word_incr(lock, X_LOCK_HALF_DECR) <= X_LOCK_HALF_DECR)
此时，增加X_LOCK_HALF_DECR 后必然大于X_LOCK_HALF_DECR
->if (lock->waiters)
当前可以唤醒由于本SX堵塞的线程
rw_lock_reset_waiter_flag(lock);os_event_set(lock->event);sync_array_object_signalled();
->else
如果lock->lock_word<=0，说明本线程还有X lock
->lock->lock_word += X_LOCK_HALF_DECR
则直接+X_LOCK_HALF_DECR
--- rw_lock_lock_word_decr
A: X_LOCK_HALF_DECR 256M X_LOCK_DECR 512M
B: amount X锁为X_LOCK_DECR SX锁为X_LOCK_HALF_DECR S锁为 1
C: threshold X锁为X_LOCK_HALF_DECR SX锁为X_LOCK_HALF_DECR S锁为 0
由于初始化lock word为X_LOCK_DECR
-(X_LOCK_DECR + X_LOCK_HALF_DECR)) -X_LOCK_DECR -X_LOCK_HALF_DECR 0 +X_LOCK_HALF_DECR +X_LOCK_DECR
重点函数设计意图
rw_lock_get_writer
rw_lock_get_reader_count
/*
IMPLEMENTATION OF THE RW_LOCK
=============================
The status of a rw_lock is held in lock_word. The initial value of lock_word is
X_LOCK_DECR. lock_word is decremented by 1 for each s-lock and by X_LOCK_DECR
or 1 for each x-lock. This describes the lock state for each value of lock_word:
lock_word == X_LOCK_DECR: Unlocked.
X_LOCK_HALF_DECR < lock_word < X_LOCK_DECR:
S locked, no waiting writers.
(X_LOCK_DECR - lock_word) is the number
of S locks.
lock_word == X_LOCK_HALF_DECR: SX locked, no waiting writers.
0 < lock_word < X_LOCK_HALF_DECR:
SX locked AND S locked, no waiting writers.
(X_LOCK_HALF_DECR - lock_word) is the number
of S locks.
lock_word == 0: X locked, no waiting writers.
-X_LOCK_HALF_DECR < lock_word < 0:
S locked, with a waiting writer.
(-lock_word) is the number of S locks.
lock_word == -X_LOCK_HALF_DECR: X locked and SX locked, no waiting writers.
-X_LOCK_DECR < lock_word < -X_LOCK_HALF_DECR:
S locked, with a waiting writer
which has SX lock.
-(lock_word + X_LOCK_HALF_DECR) is the number
of S locks.
lock_word == -X_LOCK_DECR: X locked with recursive X lock (2 X locks).
-(X_LOCK_DECR + X_LOCK_HALF_DECR) < lock_word < -X_LOCK_DECR:
X locked. The number of the X locks is:
2 - (lock_word + X_LOCK_DECR)
lock_word == -(X_LOCK_DECR + X_LOCK_HALF_DECR):
X locked with recursive X lock (2 X locks)
and SX locked.
lock_word < -(X_LOCK_DECR + X_LOCK_HALF_DECR):
X locked and SX locked.
The number of the X locks is:
2 - (lock_word + X_LOCK_DECR + X_LOCK_HALF_DECR)
LOCK COMPATIBILITY MATRIX
S SX X
S + + -
SX + - -
X - - -
The lock_word is always read and updated atomically and consistently, so that
it always represents the state of the lock, and the state of the lock changes
with a single atomic operation. This lock_word holds all of the information
that a thread needs in order to determine if it is eligible to gain the lock
or if it must spin or sleep. The one exception to this is that writer_thread
must be verified before recursive write locks: to solve this scenario, we make
writer_thread readable by all threads, but only writeable by the x-lock or
sx-lock holder.
The other members of the lock obey the following rules to remain consistent:
recursive: This flag is true iff the lock->writer_thread is allowed to take
the x or sx lock recursively.
In particular you should follow these rules:
- make sure to change recursive to `false` (or reset the
`writer_thread` to `std::thread().native_handle()`) before
releasing the lock
- make sure to change recursive to `false` (but do not reset the
`writer_thread` as it will remove debug info) before passing
the lock to other thread
- make sure to put your thread's native handle before setting
recursive to `true`
- make sure that when you want to read both `recursive` and
`writer_thread` you do this in this precise order
writer_thread: Is used only in recursive x-locking.
This field is initialized to an impossible thread native handle
value at lock creation time and is updated atomically when sx or
x-lock is being acquired or when move_ownership is called.
A thread is only allowed to set the value of this field to its
thread_id i.e.: a thread cannot set writer_thread to some other
thread's id.
The only reasonable way (except reporting in debug info) to use
this field is to compare it to own thread's native handle
AFTER checking that lock->recursive flag is set, to see if we
are the current writer.
waiters: May be set to 1 anytime, but to avoid unnecessary wake-up
signals, it should only be set to 1 when there are threads
waiting on event. Must be 1 when a writer starts waiting to
ensure the current x-locking thread sends a wake-up signal
during unlock. May only be reset to 0 immediately before a
a wake-up signal is sent to event. On most platforms, a
memory barrier is required after waiters is set, and before
verifying lock_word is still held, to ensure some unlocker
really does see the flags new value.
event: Threads wait on event for read or writer lock when another
thread has an x-lock or an x-lock reservation (wait_ex). A
thread may only wait on event after performing the following
actions in order:
(1) Record the counter value of event (with os_event_reset).
(2) Set waiters to 1.
(3) Verify lock_word <= 0.
(1) must come before (2) to ensure signal is not missed.
(2) must come before (3) to ensure a signal is sent.
These restrictions force the above ordering.
Immediately before sending the wake-up signal, we should:
(1) Verify lock_word == X_LOCK_DECR (unlocked)
(2) Reset waiters to 0.
wait_ex_event: A thread may only wait on the wait_ex_event after it has
performed the following actions in order:
(1) Decrement lock_word by X_LOCK_DECR.
(2) Record counter value of wait_ex_event (os_event_reset,
called from sync_array_reserve_cell).
(3) Verify that lock_word < 0.
(1) must come first to ensures no other threads become reader
or next writer, and notifies unlocker that signal must be sent.
(2) must come before (3) to ensure the signal is not missed.
These restrictions force the above ordering.
Immediately before sending the wake-up signal, we should:
Verify lock_word == 0 (waiting thread holds x_lock)
---------------------------------------
struct sync_cell_t {
sync_object_t latch; /*!< pointer to the object the
thread is waiting for; if NULL
the cell is free for use */
ulint request_type; /*!< lock type requested on the
object */
const char *file; /*!< in debug version file where
requested */
ulint line; /*!< in debug version line where
requested */
os_thread_id_t thread_id; /*!< thread id of this waiting
thread */
bool waiting; /*!< TRUE if the thread has already
called sync_array_event_wait
on this cell */
int64_t signal_count; /*!< We capture the signal_count
of the latch when we
reset the event. This value is
then passed on to os_event_wait
and we wait only if the event
has not been signalled in the
period between the reset and
wait call. */
/** Time when the thread reserved the wait cell. */
ib_time_monotonic_t reservation_time;
};
这里可以看到其中包含3种类型，读写锁/普通mutex/block mutex
/** The latch types that use the sync array. */
union sync_object_t {
/** RW lock instance */
rw_lock_t *lock;
/** Mutex instance */
WaitMutex *mutex;
/** Block mutex instance */
BlockWaitMutex *bpmutex;
};
