下载掌阅APP，畅读海量书库
立即打开
rwsem是一个复杂版本的rwlock，rwlock相当于spinlock的读写优化版本，但不能睡眠，而rwsem是一个相对高层的锁，增加了可以睡眠等待的特性，复杂度非常高。除非明确要求路径不能睡眠，否则读写锁都应该使用rwsem，而避免使用rwlock。rwsem相当于在rwlock的基础上增加了睡眠等待的语义，允许比较激烈的竞态情况存在。在当前SMP普及的时代，复杂逻辑的激烈竞态几乎是无法避免的。读写信号量的定义如下。
rwsem与普通读写锁都应用在读多于写的场景，并且读锁与写锁都是单独的加锁API。其中rwsem的读锁获得API是down_read，释放读锁是up_read，写锁获得API是down_write，释放写锁是up_write。信号量的语义是有上限个数的信号数量，持有锁的API都是down的语义。但是这只是API层面的语义，在实现上并不是简单地减少和增加信号数量。
rwsem允许获得的写锁直接降级为读锁（downgrade_write），但是不允许获得了读锁后再将其升级为写锁。因此读写锁在API能力上的最大短板是不允许一开始就获得可以并发的读锁，然后在需要写锁的时候原子升级为写锁。必须先释放读锁，然后尝试获得写锁。但是这个释放和获得的过程不是原子的，所以很可能出现释放了读锁就比较难获得写锁的情况，并且在持有读锁的时候已经访问的数据的内容在重新获得了写锁之后是无效的，这是因为无法保证数据的竞态一致性。而从写锁降级到读锁就不存在这些问题，降级是一个很轻量级的操作。拥有锁升级需求场景的最佳做法是一开始就直接获得写锁，或者直接使用互斥锁。Win32 API提供的读写锁也不具备升级的功能。
读写信号量是读写锁和信号量的组合，我们以一个典型的获得读锁的流程down_read引入本结构体各个域的定义和作用。
获得读锁函数定义如下。
might_sleep代表获得读锁函数可能会睡眠，因此在这个函数的入口会允许调度系统运行一次。rwsem_acquire_read是dep_map的锁调试接口。LOCK_CONTENDED是一个封装的获得锁的宏，会尝试使用__down_read_trylock一次性地获得锁，如果获得失败，则会使用__down_read来阻塞获得锁。
__down_read_trylock函数中使用了atomic_long_try_cmpxchg_acquire来尝试进行比较/交换，该函数的三个参数是ptr、pold、new，其中ptr对应＆sem->count，pold对应＆tmp，new对应tmp + RWSEM_READER_BIAS。这个操作是将tmp与sem->count进行值对比，如果值相等，就将sem->count设置为tmp + RWSEM_READER_BIAS，如果值不相等，tmp中就包含sem->count中当前的值。循环的退出条件是tmp中的内容出现了失败条件，也就是count的值被设置了几个特殊意义的值，否则do while循环就会一直尝试获得读锁，读锁的获得方式是在当前sem->count值的基础上增加256，在没有加锁时的默认值是RWSEM_UNLOCKED_VALUE，也就是0。sem->count值有一些特殊的位被用作特殊用途，每次增加256可以并发获得读锁的方式，由于在要求获得读锁时锁不能处于特殊状态，只能处于存在其他读锁获得的状态，所以需要cmpxchg指令来参与保证当前获得的读锁不存在特殊状态。这里典型的特殊状态是有写锁持有。成功获得读锁需要设置owner，表明自己拥有了当前的rwsem锁，设置owner的代码如下。
这里的设置只表示曾经拥有，因为读锁可以并发进入，每个线程都设置了同样的锁的owner域，并且都是持有锁的状态。owner的值以获得锁的线程的task_struct为模板，同时设置RWSEM_READER_OWNED，并且继承之前读锁的RWSEM_RD_NONSPINNABLE标志。owner中task_struct指针的其他特殊位代表当前持有者的状态，对后续的持有者会产生影响。
如果尝试获得读锁成功，就会设置owner为自己。如果不成功，就会进入慢速路径__down_read函数，该函数的定义如下。
这里使用RWSEM_READER_BIAS作为累加值，相当于无条件地给sem->count增加了一个读锁的值，即使sem->count中已经设置了与读锁互斥的位。这个逻辑使得sem->count中出现了读锁与写锁语义互斥的情况。如果增加结果中出现了读互斥的位，rwsem_read_trylock就会返回失败，但是读锁的修改已经在sem->count中生效。这时count在语义上已经获得读锁，但是并没有进入设置owner的逻辑。对于整个rwsem来说，读锁还没有完成获得，读锁的获得必须同时修改count和owner。
之所以要这么做，是因为后面的rwsem_down_read_slowpath慢速路径中包含了多种获得锁的可能路径。如果当前线程需要阻塞等待获得读锁，就需要将自己加入wait_list等待队列中，这个过程需要持有wait_lock锁。获得wait_lock是一个顺序过程，写锁在释放wait_lock之后，又重新进入可获得读锁的状态。所以整个流程相当于先无条件地获得读锁，然后如果可以阻塞等待，就去持有wait_lock，在持有了wait_lock之后，就可以检查当前count是否处于RWSEM_WRITER_MASK状态或RWSEM_FLAG_HANDOFF状态，成功获得锁然后直接返回。这是一个获得读锁的快速路径，相当于在读锁获得失败时，写锁有比较大的概率正处于释放的过程。
获得锁的逻辑都涉及count和owner的定义问题，从上文可以看到owner的定义是task_struct指针和一系列特殊位的标记，代表持有者和持有状态；count则是针对整个原子数进行了不同值范围的定义，代表锁状态。
count和owner都是atomic_long_t类型的原子变量，在64位下，count的定义如下。
● 第0位：写锁位。
● 第1位：等待者存在标志位。
● 第2位：锁交接位。
● 第3～7位：保留。
● 第8～62位：55位的读锁位。
● 第63位：读失败位。
在32位平台下，count的定义如下。
● 第0位：写锁位。
● 第1位：等待者存在标志位。
● 第2位：锁交接位。
● 第3～7位：保留。
● 第8～30位：23位的读锁位。
● 第31位：读失败位。
从count的定义可以看出，无论是32位还是64位的定义，前8位的定义都是一样的，最后一位的读失败的定义也是一样的，从第8位开始到倒数第二位都是读锁位。也就是说，读失败可以用count＜0来表示，读锁加锁可以用count+256来表示，这种API行为在32位和64位都是统一的。
最后一位的读失败位只会发生在读锁持有者溢出的场景，也就是32位下有2 23 个读锁并发持有者，64位下有2 55 个读锁并发持有者，显然这在一个行为正确的读写锁的场景下是不可能发生的。也就是说可以认为count小于0的情况从来不会存在，但是为了语义正确，仍然预留了最后一位来代表读锁用光的语义。在内核的代码实现中，巧妙地利用了整数操作的累加溢出的计算规则来自动获得这种读锁个数用光的效果。利用整数计算溢出结果为负数的规则，也是读锁位被安排在从倒数第二位往前的位置的原因。读锁在判断设置返回值的时候使用RWSEM_READ_FAILED_MASK宏，这个宏代表发生了除读锁之外的持有状态，定义如下。
这个读锁不可获得的count状态是4种情况的组合，4种情况分别是写锁被持有、等待者存在、正在锁交接和读失败。上面的定义就表明了写锁被持有使用了第0位，等待者存在使用了第1位，正在锁交接使用了第2位。
当读写锁被写锁持有的时候，count的RWSEM_WRITER_LOCKED位就会被设置，由于RWSEM_WRITER_LOCKED属于读锁判定的加锁失败的特殊情况RWSEM_READ_FAILED_MASK，也就是说，读锁与写锁是互斥的。当出现写锁持有者后，就一定不能存在其他的读锁持有者，只要还有一个读锁被持有，写锁就不能被持有。这也就是读写锁的多读单写、读/写不并发的语义。
同样属于读锁失败条件的还有RWSEM_FLAG_WAITERS和RWSEM_FLAG_HANDOFF。RWSEM_FLAG_WAITERS代表当前等待队列中有阻塞等待者，阻塞等待者链表中可能有读锁的等待者，也可能有写锁的等待者。当down_read的快速路径失败时，会进入rwsem_down_read_slowpath的慢速路径，在慢速路径中阻塞等待时，就可以进入wait_list阻塞等待，这种情况一般发生在写锁被持有的时候。由于写锁与写锁和读锁都是互斥的，因此写锁更容易进入wait_list等待列表。
锁交接是Linux内核对锁饥饿的一种常见的处理方法。并行的多个CPU出现连续大量的写锁请求，假设一个CPU的写锁释放，但是其紧接着又来写锁请求，那么大量的写锁请求就会更大概率落到刚释放资源的CPU上。这是因为写锁需求是会阻塞的，其他CPU上的阻塞线程很大概率正处于被交换出去的状态，也就是在当前持有锁的CPU释放锁的时候，其他CPU上运行的并不是争抢锁的线程。如果写锁请求大量出现，就会导致一些线程饥饿。在MCS自旋锁的实现中也有类似的交接流程，在争抢和有序交接两种思路下，争抢虽然可以获得最大理论吞吐，但是有序交接却具有公平性，操作系统内核对公平性的需求是非常大的。
rwsem为了解决写锁饥饿的问题也采用了交接的思路，当写锁等待超过一定的时间，就强制触发交接，让锁的持有者将锁固定地让渡给等待队列中的第一个线程。这个机制更多地起到补充作用，因为读写锁本身不太可能出现写锁高并发的情况。但是随着并行程度的增加和主频的提高，竞态概率在逐渐增大，竞态程度在逐渐加深，原来没有问题的逻辑很可能在CPU性能提高之后就出现了竞态问题。
但是这种饥饿带来的不公平的严重性在高层锁上仍然可以接受。spinlock也有专门的排队队列用于按顺序获得锁，只要使用了spinlock，就可以保证获得自旋锁是相对公平的，这也是osq域的价值。
rwsem中的wait_lock是用来保护wait_list链表的，当有写锁在阻塞等待的时候，会加入wait_list进行睡眠等待，操作wait_list的时候需要持有wait_lock锁，而判断强制交接的流程只可能在持有锁的内部发生，也就是说RWSEM_FLAG_HANDOFF标志完整地在wait_lock的保护下，不存在并发的情况。
获得读锁的慢速路径主要包括自旋等待和链表睡眠等待两种情况。因为读写锁的应用条件必然是读锁的概率大于写锁，所以即使当前获得读锁失败，很快也会成功。因为读锁可以并发，就算有很多个读锁在自旋等待，在写锁释放的瞬间，几乎所有的读线程也都可以同时获得读锁。这种假设是通过结构体定义的CONFIG_RWSEM_SPIN_ON_OWNER来控制的。如果这个特性没有打开，获得锁的慢速路径就会直接进入wait_list等待队列的排队逻辑，也就是会直接尝试获得链表锁。
CONFIG_RWSEM_SPIN_ON_OWNER这种自旋等待在读锁和写锁的获得锁流程同时生效，也就是说，在实现中，写锁也有同样的很快获得写锁的假设，这个自旋获得锁相当于在wait_list之前添加了一条优先路径，会插队打破wait_list中获得锁的顺序，带来严重的不公平性，所以对应的也就有了RWSEM_FLAG_HANDOFF的强制锁交接的设计。
自旋等待可以认为是一种一定程度地牺牲CPU计算能力以获得更快速的响应能力的行为，这种行为一般用在spinlock这种绝对保证细粒度的低层锁上。但是rwsem是一个带有睡眠副作用的比较高层的复杂的锁，自旋等待显然需要严格控制自旋条件。
rwsem对自旋逻辑进行了非常严格的进入条件限制，通过在owner域引入的第一位RWSEM_RD_NONSPINNABLE和第二位的RWSEM_WR_NONSPINNABLE，来标识当前rwsem不允许进行读部分的自旋等待或写部分的自旋等待，两部分可以同时生效，同时禁止读自旋和写自旋。
owner域代表的是rwsem的持有状态，与count域一样都是atomic_long_t类型的。因为owner域中存放的是获得锁的线程task_struct的指针，所以指针的长度就自动适配了atomic_long_t的长度。无论是32位还是64位，位定义都是一样的，特殊的位有0、1、2三个位，之所以使用这三个位是因为task_struct结构体指针的这三个位的值一定是0，所以这三个位可以用作其他的用途，且不影响task_struct指针的值。
● 第0位：RWSEM_READER_OWNED-rwsem被读者持有。
● 第1位：RWSEM_RD_NONSPINNABLE读者不能在锁上自旋。
● 第2位：RWSEM_WR_NONSPINNABLE写者不能在锁上自旋。
在读者获得锁时，__rwsem_set_reader_owned中会设置RWSEM_READER_OWNED位，而在写者获得锁时，调用的是如下函数。
直接将当前线程的task_struct指针设置进sem->owner中，也就是说读者与写者在获得锁的时候，只有读者需要设置第0位。
rwsem在阻塞等待时会进入等待链表，等待链表的操作需要使用wait_lock自旋锁进行上锁。Linux下的自旋锁具有有序性，可以保证链表的并发有序进行，链表本身会再次记录这个顺序，给后面正确进行链表操作提供了顺序信息。相当于将并发和有序两种可能混杂的形态变成一个绝对顺序的链表形态。
当一个线程需要加入等待队列时，首先会在栈上生成一个链表节点数据结构，然后将该结构加入rwsem->wait_list等待队列的开头位置。这个链表结构是在当前线程的栈上生成的，因为随着数据结构被唤醒，该数据必然被从rwsem链表中移除，数据的定义也会同步消失。这种链表节点的内存申请的加速与futex和信号量中的等待节点的实现思路一样，该节点数据结构如下。
同一个等待定义被读和写复用，区分读/写节点使用type域的两个不同的定义。list代表加入rwsem->wait_list中的链表节点；task代表当前等待线程的task_struct结构体指针；timeout代表在代表链表中停留的最大时间节点，使用目标节拍来表示；last_rowner代表当前等待节点看到的上一个owner的值。这里的timeout并不是一个定时器，而是一个交付时间，当申请释放锁时，发现wait_list中有节点的timeout超时，就会启动handoff交付机制。由于wait_list中是顺序的，所以只需要检查第一个等待者即可，检查的逻辑位于rwsem_mark_wake函数中。
链表只能起到顺序组织的作用，想要唤醒目标，仍然需要调度系统的结构参与，也就是struct wake_q_head结构，首先通过wake_q_add将要唤醒的task_struct加入唤醒队列，然后通过wake_up_q唤醒队列中的线程。
虽然链表中的读等待和写等待整体的逻辑都是相似的，但是有一个重要的区别是写等待只需要唤醒第一个等待者，而读等待要唤醒很多个等待者，这仍然是因为读写锁的读操作具有并发性和写操作具有互斥性。rwsem_mark_wake函数的调用者可以指定三种唤醒类型，代码如下。
等待队列中可能存在读等待和写等待，所以wait_list的第一个元素就可能是读等待也可能是写等待。调用rwsem_mark_wake函数是为了从wait_list中唤醒线程来获得锁，因此如果发现当前没有任何线程持有锁，就可以使用RWSEM_WAKE_ANY来唤醒任意类型的等待者，这时如果第一个是写锁等待者，就只唤醒第一个即可，如果第一个是读锁等待者，就同步地唤醒最大可能数量的读锁。这个最大数量由MAX_READERS_WAKEUP控制，定义是0x100。
RWSEM_WAKE_READERS和RWSEM_WAKE_READ_OWNED都表示只唤醒读锁等待者，区别是后者必须要在已经确定有线程持有了读锁的情况下使用，通常是本线程在外部刚成功获得了读锁，然后使用RWSEM_WAKE_READ_OWNED来唤醒其他位于等待队列的读锁等待者。因为读操作具有并发性，等待队列中的读队列可以与已经明确存在的读锁持有并发；而RWSEM_WAKE_READERS的意义则是并没有确定外部有线程获得了读锁，只是为了唤醒读线程，在实现上RWSEM_WAKE_READERS也同样是先让本线程获得读锁，然后使用与RWSEM_WAKE_READ_OWNED一样的逻辑唤醒其他的读锁等待者。尽可能快地先获得一个读锁是因为批量唤醒读等待者很耗时，在这个过程中如果有写锁抢占获得了锁，就会出现大量无意义的唤醒。所以先让当前线程抢占一个读锁，以阻止后续写锁的获得，然后再批量唤醒写锁，这就是性能最优的做法。
从上述三种唤醒规则中可以看到，只有RWSEM_WAKE_ANY可以唤醒写锁等待者，在链表的第一个等待者是写锁等待者并且没有指定RWSEM_WAKE_ANY的情况下，rwsem_mark_wake不唤醒任何一个线程。
rwsem_mark_wake代码如下。
整个等待队列的唤醒都要加wait_lock锁，所以wait_lock锁的粒度较大，尤其是在等待队列中元素数量比较多的时候。rwsem_mark_wake唤醒操作输入一个唤醒队列让函数填充，并且指定唤醒的类型，如果是唤醒读锁，还会同时上锁。
从这个逻辑中就能看出为什么rwsem无法支持锁升级，如果要实现从读锁到写锁的原子升级，就要等待除自己以外的其他所有线程都释放了读锁，并且没有新的读锁被获得，因为整个过程会不断产生新的读锁需求。这个等待所需要的条件与普通写锁的获得无异，但是提出了更高的要求，即需要在读锁持有的情况下等待。
rwsem的批量唤醒读等待者语义依赖在读锁持有的情况下写锁不能被获得的事实，而从读锁到写锁的升级却直接打破这个约束。在允许锁升级的情况下，多个读锁同时提出升级要求该如何？也就是说在读锁的持有者数量不固定的情况下，要求只有一个读锁能成功升级为写锁，相当于要求写锁和读锁同时发生，因为并不能区分当前持有的读锁是阻塞升级的读锁还是普通的读锁，毕竟阻塞升级的读锁也由普通读锁转换而来。可以做到这种升级的办法就是定义单独的upgrade状态，并且在一个读锁升级时阻止其他并发读锁的获得。设计合理的链表排队使得等待upgrade的线程也组成有序队列，其复杂度急剧提高。folly实现的RWSpinLock是一个具备升级能力的读写锁，在应用程序层面采用folly的实现是合理的，但是在内核层面就无法实现了。
完整的读锁慢速路径如下，state参数是阻塞等待进程设置的阻塞状态，这种接口在大部分内核锁API上都是一样的。
从down_read到rwsem_down_read_slowpath慢速路径，中间会多次尝试获得锁。rwsem读锁和写锁的获得方式不一样，这是因为rwsem用在读频率远大于写频率的情况下，是一个读优先级更高的流程。写锁在释放之前，如果有读锁请求加入，后面的写锁就不能继续获得，而要持续等待。可以同时进行大量的读锁并发，相当于每一轮读锁加锁都是一个清空等待队列的行为，所以队列中写锁的处理速度远远不如读锁请求的速度。因此rwsem还设计了锁交接机制，允许写锁在等待链表中若等待时间太久就进行插队。在锁交接生效时，也就是第一个写等待者等待太久的时候，后面写等待者的尝试获得锁就会自动失败，进入链表的等待队列，让第一个写等待者更优先获得锁。这个逻辑位于down_write的写锁获得函数中。
osq自旋抢占是一个降低延迟的插队行为，仍然基于读远多于写的使用情况而设计。rwsem通过禁止读自旋或者写自旋来做到锁持有的偏向性，但是这个偏向性的实现过于复杂，整个osq自旋抢占的实现处于比较初级的状态。
慢速路径整体就是对osq自旋和链表等待完全处理的逻辑，写锁获得的慢速路径与之类似。rwsem通过对count、owner、wait_lis这三个元素和保护wait_list的wait_lock锁的综合使用来实现，如果没有osq自旋，那么rwsem可以只使用count和wait_list两个元素来实现。rwsem在数据结构上也与内核的信号量非常相似，只是信号量没有owner域。信号量相当于在rwsem的概念上去掉写锁，也就是只有可以并发的读锁，而将最大并发数进行了限制，控制允许最大的并发数量。rwsem从语义上更像读写锁，但是从数据组织上就更像信号量。