Linux读写锁逻辑解析—Linux为何会引入读写锁？

一、Linux为何会引入读写锁？

除了mutex，在linux内核中，还有一个经常用到的睡眠锁就是rw semaphore（后文简称为rwsem），它到底和mutex有什么不同呢？为何会有rw semaphore？无他，仅仅是为了增加内核的并发，从而增加性能而已。Mutex严格的限制只有一个thread可以进入临界区，但是实际应用中，有些场景对共享资源的访问可以严格区分读和写的，并且是读多写少，这时候，其实多个读的thread同时进入临界区是OK的，使用mutex则限制一个线程进入临界区，从而导致性能的下降。

本文会描述linux5.15.81中读写锁的数据结构和逻辑过程。

二、如何抽象读写锁的数据结构？

下图可以抽象rwsem相关的数据结构：

一个rwsem对象需要记录两种数据：

1、读写锁的状态信息

2、和该读写锁相关的任务信息

我们先看看读写锁的状态。读写锁状态字需要分别记录读锁和写锁的状态：由于多个reader可以同时处于临界区，所以对于reader-owned的场景，读锁状态变成了一个counter，来记录临界区内reader的数量，counter等于0表示读锁为空锁状态。对于writer，其行为和互斥锁一致，因此其写锁状态和mutex一样，仍然使用一个bit表示。

和读写相关的任务有两类，一类是已经持锁的线程（即在临界区的线程），另外一类是无法持锁而需要等待的任务。对于writer持锁情况，由于排他性，我们很清楚的知道是哪个task持锁，那么一个task struct指针就足够了记录owner了。然而对于读侧可以多个reader进入临界区，那么owner们需要组成一个队列才可以记录每一个临界区的reader。

不过在实际的rwsem实现中，由于跟踪owner们开销比较大，因此也是用一个task struct指针指向其一。具体linux代码是这样处理的：reader进入的时候会设置owner task，但是离开读临界区并不会清除task指针。这样，实际上对于读，owner task应该表示该任务曾经拥有该锁，并不表示是目前持锁的owner task，也有可能已经离开临界区，甚至该任务已经销毁。

如果持锁失败，无法进入临界区，我们有两种选择：

1、乐观自旋

2、挂入等待队列

两种选择各有优点和缺点，总结如下：

在5.15的内核中，只有在write持锁路径上有乐观自旋的操作，reader路径没有，只有偷锁的操作。当乐观自旋失败后就会挂入等待队列，阻塞当前线程。（乐观自旋功能有一个很有意思的发展过程，从开始支持writer的乐观自旋，到支持全场景的乐观自旋，然后又回到最初，有兴趣可以查阅内核的patch了解详情）

在了解了rwsem的基本概念之后，我们一起来看看struct rw_semaphore数据结构，其成员描述如下：

由于是sleep lock，我们需要把等待的任务挂入队列。在内核中，struct rwsem_waiter用来抽象等待rwsem的任务，其成员描述如下：

三、Rwsem外部接口API为何？

Rwsem模块的外部接口API如下：

四、尝试获取读锁

和down_read不一样，down_read_trylock只是尝试获取读锁，如果成功，那么自然是好的，直接返回1，如果失败，也不会阻塞，只是返回0就可以了。代码主逻辑在__down_read_trylock函数中，如下：

A、tmp的初始值设定为RWSEM_UNLOCKED_VALUE（0值），因此第一次循环是为当前是空锁而做的优化：如果当前的sem->count等于0，那么给sem->count赋值RWSEM_READER_BIAS，标记持锁成功，然后设定owner返回1即可。

B、如果快速获取空锁不成功，这时候tmp已经赋值（等于sem->count），不再是0值了。通过对当前sem->count的值可以判断是否是可以进入临界区。持读锁失败的情况包括：

如果判断可以进入读临界区（临界区仅有reader并且没有writer等待的场景），那么重新进入循环，如果sem->count保持不变，那么可以持锁成功，给进入临界区的reader数目加一，并设置owner task和reader持锁标记（non-spinnable比特保持不变）。如果这期间有其他线程插入修改了count值，那么需要再次判断是否能持读锁，重复上面的循环。如果判断不可以进入临界区，退出循环，持锁失败。

五、获取读锁

Reader获取读锁的代码主要在__down_read_common函数中，如下：

1、快速路径

rwsem_read_trylock是快速路径，代码如下：

A、reader直接会给sem->count加RWSEM_READER_BIAS来增加读临界区的线程个数，当然这有可能失败，那么就进入慢速路径（需要回退错误增加读临界区线程数量）。如果恰好能够进入临界区，那么就直接设定owner返回即可。注意：这里*cntp保存了atomic add之后的新值。rwsem_down_read_slowpath会使用这个新值作为参数。

B、当reader的数量过多（以至于都溢出了）的时候，需要禁止乐观自旋。

C、这里是持锁成功的路径。RWSEM_READ_FAILED_MASK上一节已经解释，这里不再赘述。这里需要注意的是rwsem_set_reader_owned函数中flag的设定，由于reader进入临界区，因此RWSEM_READER_OWNED也需要设定。RWSEM_RD_NONSPINNABLE标记保持不变。

在快速路径中，有两种常见的情况会持锁成功：一种是空锁，另外一种是没有任何waiter等待的纯reader并发。

2、慢速路径

如果快速路径持锁失败，那么进入慢速路径。慢速路径代码比较长，我们分段解析。首先是防止等待队列中waiter任务饿死的代码：

如果当前的锁被reader持有（至少有一个reader在临界区），那么不再乐观偷锁而是直接进行挂等待队列的操作。为何怎么做呢？因为需要在饿死waiter和reader吞吐量上进行平衡。一方面，连续的reader持续偷锁的话会饿死等待队列上的任务。另外，在唤醒路径上，被唤醒的top reader会顺便将队列中的若干（不大于256个）reader也同时唤醒，以便增加rwsem的吞吐量。所以这里的reader直接挂入队列，累计多个reader以便可以批量唤醒。

Reader偷锁的场景主要发生在唤醒top waiter的过程中，这时候临界区没有线程，被唤醒的reader或者writer也没有持锁（writer需要被调度到CPU上执行之后才会试图持锁，高负载的场景下，锁被偷的概率比较大，reader是唤醒后立刻持锁，被偷的几率小一点）。具体乐观偷锁（optimistic lock stealing）的代码如下：

A、所谓偷锁就是不乐观自旋（要有排队），不管先来后到，直接获取锁。允许偷锁的场景是这样的：临界区没有writer持锁，也没有设置handoff，正在唤醒top waiter的过程中，并且有任务在等待队列的情况。这时候进入慢速路径的reader可以先于top waiter唤醒之前把锁偷走。需要特别说明的是：这时候reader counter已经加一，还是尽量让reader偷锁成功，否则还需要回退。

B、当前线程获得了读锁，需要设置owner，毕竟它是临界区的新客

C、如果偷锁成功并且它是临界区第一个reader，那么它还会把等待队列中的reader都唤醒（前提是top waiter不是writer），带领大家一起往前冲（这里会打破FIFO的顺序，惩罚了队列中的writer）。具体是通过rwsem_mark_wake来标记唤醒的reader，然后通过wake_up_q将reader唤醒并进入读临界区。为了减低对等待中的writer线程的影响，这时候对reader的并发是受限的，最多可以唤醒MAX_READERS_WAKEUP个reader。

如果偷锁不成功，当前的reader还是需要进入阻塞状态：

A、准备好挂入等待队列的rwsem waiter数据，需要特别说明的是这里的timeout时间：目前手机平台的HZ设置的是250，也就是说在触发handoff机制之前waiter需要至少在队列中等待一个tick（4ms）的时间。这里的timeout是指handoff timeout，为了防止偷锁或者自旋导致等待队列中的top waiter有一个长时间的持锁延迟。在timeout时间内，乐观偷锁或者自旋可以顺利进行，但是一旦超时就会设定handoff标记，乐观偷锁或者自旋被禁止，锁的所有权需要递交给等待队列中的top waiter。

B、如果目前等待队列为空，那么要做一些额外的处理。例如入队之前肯定给安排上RWSEM_FLAG_WAITERS这个标记。

C、当然，在入队之前还要垂死挣扎一下（等待队列为空的时候逻辑简单一些，不需要唤醒队列上的wait），看看是不是当前有机可乘，如果是这样，那么就顺势而为，直接持锁成功，而且counter都已经准备好了，前面已经加一了。

D、等待队列非空的时候，逻辑稍微负载一点。调用rwsem_add_waiter函数即可以把当前任务挂入等待队列尾部。这时候也需要把之前武断增加的counter给修正回来了（adjustment初始化为-RWSEM_READER_BIAS）。如果是第一个waiter，也顺便设置了RWSEM_FLAG_WAITERS标记。

在当前线程进入阻塞之前，我们需要进行试图持锁的动作（上面是空队列场景检查，这里的逻辑稍微复杂一点，由于已经入队，这里需要调用rwsem_mark_wake函数来完成阻塞后唤醒的动作），毕竟这时候可能恰好owner离开临界区，变成空锁。

A、如果这时候发现锁的owner恰好都离开了临界区，那么我们是需要执行唤醒top waiter操作的，唤醒之前需要清除禁止乐观自旋的标记，毕竟目前临界区没有任何线程。

B、除了上面说的场景需要唤醒，在reader持锁并且我们是队列中的第一个waiter的时候，也需要唤醒的动作（唤醒自己）。

阻塞部分的代码逻辑如下：

A、在rwsem_mark_wake函数中我们会唤醒reader并将其等待对象的task成员（waiter.task）设置为NULL。因此，这里如果发现waiter.task等于NULL，那么说明是该线程被正常唤醒，那么从阻塞状态返回，持锁成功。

B、如果在该线程阻塞的时候，有其他任务发送信号给该线程，那么就持锁失败退出。如果已经被唤醒，同时又收到信号，这时候需要首先完成唤醒，持锁成功，然后在其他的合适点再处理该信号。当然，大部分的rwsem都是D状态，也就不需要处理信号了。

C、进入阻塞状态，让调度器选择next task

六、释放读锁

释放读锁的代码逻辑主要在__up_read函数中，如下：

需要强调的是：这里仅仅是减去了读临界区的counter计数，并没有清除owner中的task pointer。此外，当等待队列有waiter并且没有writer或者reader在临界区的时候，我们会调用rwsem_wake来唤醒等待队列的线程。因为临界区已经没有线程，所以需要清除nonspinable标记。唤醒的动作主要是通过rwsem_mark_wake和wake_up_q来完成的，wake_up_q比较简单，我们就不赘述了，主要看看rwsem_mark_wake的逻辑。

我们首先给出wake type的解释：

在RWSEM_WAKE_READERS场景中，多个reader被唤醒，并且当前很可能是空锁状态，为了防止writer抢锁，因此会先让top waiter持有读锁，然后慢慢处理后续。RWSEM_WAKE_READ_OWNED则没有这个顾虑，因为唤醒者已经持有读锁。

在释放读锁的场景中，rwsem_mark_wake使用的是RWSEM_WAKE_ANY参数，具体的代码如下：

这段代码是处理top waiter是writer的逻辑。这时候，如果wake type是RWSEM_WAKE_ANY，即不关心唤醒的是reader还是writer，只要唤醒等待队列头部的waiter就好。如果top waiter是writer，我们只需要将这个writer唤醒即可，不需要修改锁的状态，出队等操作，这些都是在唤醒之后完成。如果wake type是其他两种类型（都是唤醒reader的），那么就直接返回。也就是说在rwsem_mark_wake想要唤醒reader的场景中，如果top waiter是writer，那么将不会唤醒任何reader线程。如果top waiter是reader的话，那么基本上是需要唤醒一组reader了。

A、执行到这里，我们需要唤醒等待队列头部的若干reader线程去持锁。由于writer有可能会在这个阶段偷锁，因此，这里我们会先让top waiter（reader）持锁，然后再慢慢去计算到底需要唤醒多少个reader并将其唤醒。如果当前线程已经持有了读锁（wake type的类型是RWSEM_WAKE_READ_OWNED），则不需要提前持锁，直接越过这部分的逻辑即可。

B、如果的确发生了writer通过乐观自旋偷锁，那么我们需要检查设置handoff的条件。如果reader被writer阻塞太久，那么我们设定handoff标记，要求rwsem的writer停止通过乐观自旋偷锁，将锁的所有权转交给top waiter（reader）

C、上面已经向rwsem的count增加reader计数，这里把owner也设定上（flag也同步安排，这里non-spinnable bit保持不变）。随后top waiter的reader会唤醒若干队列中的non top reader，但是它们都不配拥有名字。

读锁已经安排的妥妥的了，下面就是慢慢唤醒等待队列的reader了。我们通过两步来完成唤醒：

1、将等待队列中的reader摘下放入到一个单独的列表中（wlist），同时对reader进行计数。后续这个计数会写入rwsem 的reader counte域。

2、对于wlist中的每一个waiter对象（reader任务），清除waiter->task并将它们放入wake_q以便稍后被唤醒。

我们先看第一轮计算唤醒reader个数的计数：

A、对于rwsem，其公平性是区分读写的。对于读，如果top waiter是reader，那么所有的reader都可以进入临界区，不管reader在队列中的顺序。对于writer，我们要确保其公平性，我们要按照writer在队列中的顺序依次持锁。根据上面的原则，我们会略过队列中的writer，将尽量多的reader唤醒并进入临界区

B、唤醒数量不能大于256，否则会饿死writer

C、根据唤醒的reader数量计算count调整值