rw_semaphore 原理与代码分析

代码基于内核5.10版本

信号量的进阶形式，对读者写者进行区分，写者时互斥的，写者持锁时其余的写者和读者都只能等待。读者是允许并发的，读者持锁时允许其他的读者持锁，但是写者必须等待。

后续的分析不涉及死锁检测，也就是lockdep部分。

常用的API如下：

DECLARE_RWSEM(name)	声明名为name的读写信号量，并初始化它。
void init_rwsem(struct rw_semaphore *sem);	对读写信号量sem进行初始化。
void down_read(struct rw_semaphore *sem);	读者用来获取sem，若没获得时，则调用者睡眠等待。
void up_read(struct rw_semaphore *sem);	读者释放sem。
int down_read_trylock(struct rw_semaphore *sem);	读者尝试获取sem，如果获得返回1，如果没有获得返回0。可在中断上下文使用。
void down_write(struct rw_semaphore *sem);	写者用来获取sem，若没获得时，则调用者睡眠等待。
int down_write_trylock(struct rw_semaphore *sem);	写者尝试获取sem，如果获得返回1，如果没有获得返回0。可在中断上下文使用
void up_write(struct rw_semaphore *sem);	写者释放sem。
void downgrade_write(struct rw_semaphore *sem);	把写者降级为读者。

急用请看这

说实话，这部分机制还是比较难的，并且如果不是需要去深入研究其机制，完全没必要了解这么多。只需要简单debug，我这个给出一个快速的总结

struct rw_semaphore {atomic_long_t count; //用于记录锁的状态atomic_long_t owner; //持锁的写者或者其中一个读者struct optimistic_spin_queue osq; //用于乐观自旋raw_spinlock_t wait_lock; //保护wait_list结构的spinlockstruct list_head wait_list; //等锁者链表void *magic; //魔术数，记录锁的地址，当其被改变时可以认为锁的数据结构被破坏struct lockdep_map dep_map; //死锁检测，暂不关注
}
count bit* Bit  0    - writer locked bit* Bit  1    - waiters present bit* Bit  2    - lock handoff bit* Bits 3-7  - reserved* Bits 8-62 - 55-bit reader count* Bit  63   - read fail bit

遇到读写锁相关问题，那大概率是发生在进程卡在rw_sem上了，这时候需要快速找到持锁者，可以从栈中拿到sem的地址以及数据，然后寻找真正的持锁者。注意owner最后几个bit需要清零才是task结构。

magic不是锁的首地址，那不用看了，结构体被破坏，属于内存踩踏问题
writer locked bit被置位，那肯定是写者持锁，找owner就可以找到持锁者。不要理会reader count是否存在，读者会先斩后奏，这部分不能代表读者持锁。
writer locked bit为0，reader count等于1，这时候是读者持锁，可能没有完全持锁成功，即只写入了count还没来得及走完后面的流程。此时owner不一定准确，可能为NULL，也可能是读者中的一个。如果读者计数为1，并且owner是一个task地址，那么大概率就只有这个读者持锁。
writer locked bit为0，读者计数大于1，此时owner为NULL是正常的，为读者中的一个也是正常的。这种情况是没有办法通过锁结构找完所有读者的。可以尝试开启相关debug功能。如果你正在使用使用crash tool分析一个内核尸体的话， search -t 锁地址来寻找所有线程栈上可能残留的蛛丝马迹。没有收获，也没有bug功能，那就只能自己使用kprobe之类的工具来debug了。
一些奇奇怪怪的中间态：读者计数为0，写者没有持锁，存在等待者。这种情况要么锁当前没有问题，只是panic的时候恰巧碰到了这种状态，要么等待队列的队首进程状态异常，可能是自身非法调度引起，也可能是其所在cpu异常。

handoff机制

handoff这里应该翻译为“接力”，它的出现是为了弥补乐观自旋带来的漏洞，既并非先到先得。锁大多被设计为先到先得，是为了实现公平，防止某些进程一直得不到资源被饿死。而乐观自旋是为了减少进程切换的开销，降低锁带来的延迟，但是乐观自旋会打破先到先得的规则。

semaphore的handoff是为了防止偷锁的情况出现。等待队列的进程都是在睡眠的，尝试持锁的进程可以进入乐观自旋阶段，这个阶段无视等待队列一旦有机会就会获取锁，造成本该最先拿到锁的等待队列队首被跳过。队首被唤醒持锁，发现锁已经被偷取，这时候就会设置handoff bit，后面所有的持锁者看到这个bit就不能再偷锁了。因此队首进程可以顺利获取到锁。

设置时机:
对于读者，当读者在队首并且被唤醒尝试持锁失败之后，如果超时设置handoff。
对于写者，同样在队首，当自身是RT进程或者超时时设置handoff。

生效时机：
所有试图持锁者都需要检测是否有handoff标志位，一旦存在均会持锁失败。设置者会在持锁成功之后才清除这个标志位。

乐观自旋

乐观自旋是性能和速度的均衡考虑，让进程在满足一定的条件下循环等待某些数据，即期待持有锁的进程尽快释放锁所以占用cpu一段时间来死等不退出，避免自身陷入睡眠然后很快被唤醒。一段时间是指调度器没有标记当前进程需要被调度走。伪代码如下：

while(1) {检测数据是否发生变变化是 退出，返回成功当前cpu时间片是否用尽？是 退出，返回失败当前需要等待的进程是否还在cpu上运行否 退出，返回失败
}

数据结构

实际生效的定义如下：
struct rw_semaphore {atomic_long_t count; //用于记录锁的状态atomic_long_t owner; //持锁的写者或者其中一个读者struct optimistic_spin_queue osq; //用于乐观自旋raw_spinlock_t wait_lock; //保护wait_list结构的spinlockstruct list_head wait_list; //等锁者链表void *magic; //魔术数，记录锁的地址，当其被改变时可以认为锁的数据结构被破坏struct lockdep_map dep_map; //死锁检测，暂不关注
}

count

count用来表示当前锁的状态较为复杂，其定义如下：

 * Bit  0    - writer locked bit* Bit  1    - waiters present bit* Bit  2    - lock handoff bit* Bits 3-7  - reserved* Bits 8-62 - 55-bit reader count* Bit  63   - read fail bit

bit 0用来表示是否有写者持锁，写者持锁成功的标志就是这个位被设置

bit 1表示等待队列中是否存在等待者，有等待者时读者持锁成功需要唤醒等待队列的其他读者，读写者释放锁时需要唤醒等待队列的队首进程

bit 2表示hand off bit，这个是为了防止饿死设计的。handoff bit会阻止所有的偷锁/持锁行为保证锁的交接。

bit 8-62 表示读者的数目，这个数目并不能真正的反应持锁成功的读者个数。某些情况下读者会采取先斩后奏的方式，先加上这个计数，然后再判断自己能否持锁，不行的话就减去恢复原来的计数。一般而言，当读者持锁时这个计数是真实的，但是写者持锁时也可能有读者计数，这部分就是先斩后奏的读者。

最高位63表示读者个数溢出位，几乎不会被设置，但是在某些路径还是可能会被检查

下面是用于判断count的一些掩码。

#define RWSEM_WRITER_LOCKED (1UL << 0)                            1
#define RWSEM_FLAG_WAITERS  (1UL << 1)                            2
#define RWSEM_FLAG_HANDOFF  (1UL << 2)                            4
#define RWSEM_FLAG_READFAIL (1UL << (BITS_PER_LONG - 1))          1<<63#define RWSEM_READER_SHIFT  8
#define RWSEM_READER_BIAS   (1UL << RWSEM_READER_SHIFT)            1<<8
#define RWSEM_READER_MASK   (~(RWSEM_READER_BIAS - 1))             0xffff ffff ffff ff00
#define RWSEM_WRITER_MASK   RWSEM_WRITER_LOCKED                    1
#define RWSEM_LOCK_MASK     (RWSEM_WRITER_MASK|RWSEM_READER_MASK)  0xffff ffff ffff ff01
#define RWSEM_READ_FAILED_MASK  (RWSEM_WRITER_MASK|RWSEM_FLAG_WAITERS|\RWSEM_FLAG_HANDOFF|RWSEM_FLAG_READFAIL) 0x8000 0000 0000 0007

osq

struct optimistic_spin_queue osq 成员用于乐观自旋，暂不做详细分析，其本质为mcslock的一种变体，当持锁者还在cpu上运行时，等锁者可以期待持锁者很快会释放这个锁，所以不会陷入睡眠而是自旋等待。反之持锁者已经放弃了cpu，那么预计很长一段时间内持锁者都不会释放锁，此时等锁者陷入睡眠不再忙等。

owner

owner用来记录当前持锁的task，由于task必定是cacheline对齐的目前最小的cacheline也有16bit，所以owner的低位可以用来表示一些flag。同时可以有多个reader持锁，因此绣着持锁时owner就是持锁者，但是读者持锁时owner有特殊表示。

#define RWSEM_READER_OWNED  (1UL << 0)                 bit 0 用来表示当前是reader持锁
#define RWSEM_RD_NONSPINNABLE   (1UL << 1)             bit 1 表示是否允许读者自旋等待
#define RWSEM_WR_NONSPINNABLE   (1UL << 2)             bit 2 表示是否允许写者等待
#define RWSEM_NONSPINNABLE  (RWSEM_RD_NONSPINNABLE | RWSEM_WR_NONSPINNABLE)
#define RWSEM_OWNER_FLAGS_MASK  (RWSEM_READER_OWNED | RWSEM_NONSPINNABLE)

读者的owner一般由rwsem_set_reader_owned设置，为 current | RWSEM_READER_OWNED | (之前owner & RWSEM_RD_NONSPINNABLE)

写者的owner由rwsem_set_owner设置，一般为 current

wait_list

等待队列用来放置等待锁的继承，但是由于包含乐观自旋逻辑，等待队列不是全部的等待者，部分等待着可能在乐观自旋等待owner。

作为链表头，连接所有的struct rwsem_waiter.list

初始化

存在两种方式，一种是临时定义一个栈上的变量并且初始化。

count初始化为0
owner初始化为0
wait_lock初始化为未加锁状态
wait_list初始化为空链表头
magic初始化为锁的起始地址
osq lock初始化为未加锁

#define DECLARE_RWSEM(name) \                                                                 struct rw_semaphore name = __RWSEM_INITIALIZER(name)#define __RWSEM_INITIALIZER(name)               \{ __RWSEM_COUNT_INIT(name),             \.owner = ATOMIC_LONG_INIT(0),             \__RWSEM_OPT_INIT(name)                \.wait_lock = __RAW_SPIN_LOCK_UNLOCKED(name.wait_lock),\.wait_list = LIST_HEAD_INIT((name).wait_list),    \__RWSEM_DEBUG_INIT(name)              \__RWSEM_DEP_MAP_INIT(name) }count原子变量初始化为0
#define RWSEM_UNLOCKED_VALUE        0L
#define __RWSEM_COUNT_INIT(name)    .count = ATOMIC_LONG_INIT(RWSEM_UNLOCKED_VALUE)osq lock初始化为未加锁
#define __RWSEM_OPT_INIT(lockname) .osq = OSQ_LOCK_UNLOCKED,magic初始化为锁地址
# define __RWSEM_DEBUG_INIT(lockname) .magic = &lockname,

还可以动态初始化，动态初始化的结果和静态一致。

#define init_rwsem(sem)                     \
do {                                \static struct lock_class_key __key;         \\__init_rwsem((sem), #sem, &__key);          \
} while (0)void __init_rwsem(struct rw_semaphore *sem, const char *name,struct lock_class_key *key)
{#ifdef CONFIG_DEBUG_RWSEMSsem->magic = sem;
#endifatomic_long_set(&sem->count, RWSEM_UNLOCKED_VALUE);raw_spin_lock_init(&sem->wait_lock);INIT_LIST_HEAD(&sem->wait_list);atomic_long_set(&sem->owner, 0L);
#ifdef CONFIG_RWSEM_SPIN_ON_OWNERosq_lock_init(&sem->osq);
#endiftrace_android_vh_rwsem_init(sem);
}

down_read 读者持锁

读者持锁的判定条件较为复杂，即使有写者持锁时，也可能向count增加RWSEM_READER_BIAS，增加总是能成功的，但是不代表持锁成功，增加之后一般会检测是否有writer以及handoff，失败需要回退，增加只是为了防止写者抢占。读者持锁的条件如下：

没有读者写者以及等待者，即count == 0，此时向count增加RWSEM_READER_BIAS并且设置owner
有其他读者持锁，向count增加RWSEM_READER_BIAS即算成功
有等待者但是没有写者，偷锁时向count增加RWSEM_READER_BIAS并且设置owner
被其他人唤醒时waiter.task被清除，说明别人已经替当前进程持锁了，当前进程只需要返回即可

note:

读者不管持锁是否成功都可能会增加RWSEM_READER_BIAS一段时间，因此不能用此标志位判断是否是读者持锁
持锁顺序不是先来后到，乐观自旋不管是读者还是写者都能通过rwsem_try_xxx_lock_unqueued无视等待队列偷锁
队首进程有特殊处理，如果队首是写者任何不会唤醒写者(! RWSEM_WAKE_ANY)的动作都会失败，如果队首是读者，唤醒读者的时候锁可能会被另一个写者被偷取，此时队首读者如果等待超时会设置handoff不再进行竞争持锁。
读者有两种持锁方式：在无人持锁以及乐观自旋的情况下是读者自己获取锁，陷入睡眠被唤醒的情况下别人已经设置好了count和owner，自身不需要持锁了，醒来检测自己的waiter.task被清空了即可退出。
读者持锁的owner：在无人持锁以及乐观自旋的情况下owner为第一个持锁的读者，如果是处于等待队列中被其他人协助唤醒的读者，是不会设置owner的。所以存在两种情况：1.多个读者，但是只有最先持锁的读者是owner。2.任意数目的读者，最先持锁的task已经释放锁并且清除owner，导致owner中的task为空。

void __sched down_read(struct rw_semaphore *sem)
{//检测是否有其他高优先级任务需要抢占，如果有让出cpumight_sleep();//死锁检测rwsem_acquire_read(&sem->dep_map, 0, 0, _RET_IP_);//持锁流程LOCK_CONTENDED(sem, __down_read_trylock, __down_read);
} //如果穿入了trt_lock的接口，那么先尝试不需要等待的try_lock，成功就直接持锁退出
//否则调用会阻塞的lock接口
#define LOCK_CONTENDED(_lock, try, lock)            \
do {                                \if (!try(_lock)) {                  \lock_contended(&(_lock)->dep_map, _RET_IP_);    \lock(_lock);                    \}                           \lock_acquired(&(_lock)->dep_map, _RET_IP_);         \
} while (0)简化一下，流程为：
down_read
{//尝试非阻塞持锁if(！__down_read_trylock)//尝试失败，进入阻塞流程__down_read
}

__down_read_trylock

读者尝试非阻塞的持锁，仅当count == 0，也就是没有reader也没有writer也没有waiter的情况下才能持锁成功。成功之后设置自己的task | RWSEM_READER_OWNED 到owner。

返回1表示成功，0表示失败。

__down_read_trylock
{if(sem->count == 0)sem->count += RWSEM_READER_BIAS(1 << 8)sem->owner &= current | RWSEM_READER_OWNED | RWSEM_RD_NONSPINNABLEreturn 1return 0
}

static inline int __down_read_trylock(struct rw_semaphore *sem)
{long tmp;//检测magic是否被改写DEBUG_RWSEMS_WARN_ON(sem->magic != sem, sem);//设置tmp为锁没有持有者也没有等待者的状态的值，为0tmp = RWSEM_UNLOCKED_VALUE;do {//判断sem->count == tmp//是，当前无人持锁无人等待，设置sem->count = tmp + RWSEM_READER_BIAS，返回1表示更新成功//否，锁不是空闲状态，更新count失败，返回0if (atomic_long_try_cmpxchg_acquire(&sem->count, &tmp,tmp + RWSEM_READER_BIAS)) {//sem->owner &= current | RWSEM_READER_OWNED | RWSEM_RD_NONSPINNABLErwsem_set_reader_owned(sem);//持锁成功返回1return 1;}} while (!(tmp & RWSEM_READ_FAILED_MASK));//尝试拿锁失败，返回0return 0;
}

__down_read

阻塞性的等锁直到持锁成功。

__down_read
{rwsem_read_trylock{sem->count += RWSEM_READER_BIASreturn !(cnt & (RWSEM_WRITER_MASK|RWSEM_FLAG_WAITERS|RWSEM_FLAG_HANDOFF|RWSEM_FLAG_READFAIL))}if (!rwsem_read_trylock) {rwsem_down_read_slowpath} else {rwsem_set_reader_owned}}

static inline void __down_read(struct rw_semaphore *sem)
{//读者尝试持锁，0表示失败//不管失败成功与否，都会在count中增加读者的计数RWSEM_READER_BIAS。if (!rwsem_read_trylock(sem)) {//读者持锁失败，进入慢速路径rwsem_down_read_slowpath(sem, TASK_UNINTERRUPTIBLE);DEBUG_RWSEMS_WARN_ON(!is_rwsem_reader_owned(sem), sem);} else {//读者持锁成功，设置owner = current | RWSEM_READER_OWNED，并且继承之前设置的RWSEM_RD_NONSPINNABLErwsem_set_reader_owned(sem);}
}

rwsem_read_trylock

读者尝试持锁，返回0表示失败，1表示成功。不管失败成功与否，都会在count中增加读者的计数RWSEM_READER_BIAS。

失败的原因有：

存在写者持锁
等待队列不为空
锁处于handoff交接状态
读者数目溢出

static inline bool rwsem_read_trylock(struct rw_semaphore *sem)
{//将count增加RWSEM_READER_BIAS，返回增加后的值//即读者做一次预抢占，先设置抢占值再检查是否能够抢占成功long cnt = atomic_long_add_return_acquire(RWSEM_READER_BIAS, &sem->count);//数值小于0表示bit 63被设置，即读者数目溢出了，禁止读写者在锁上乐观自旋等待if (WARN_ON_ONCE(cnt < 0))rwsem_set_nonspinnable(sem);//返回此时读者是否能够持有这个锁，如果存在写者或者存在等待者或者锁处于特殊的handoff状态，都会返回0，也就是失败return !(cnt & RWSEM_READ_FAILED_MASK);
}

rwsem_down_read_slowpath

读者直接持锁失败，进入慢速路径。

注意，在此函数之前读者已经将自己的标志位RWSEM_READER_BIAS加入了count。

rwsem_down_read_slowpath
{if(rwsem_can_spin_on_owner) {//判断自身是否需要调度，持锁者是否在运行，以及flag中有没有明确禁止当前身份的自旋减去之前rwsem_read_trylock中加上的RWSEM_READER_BIAS (2)atomic_long_add(-RWSEM_READER_BIAS, &sem->count); //减去之前rwsem_read_trylock中加上的RWSEM_READER_BIASif(rwsem_optimistic_spin){ //乐观自旋，自旋等待成功会设置当前task为owner，设置count对应bit等待队列不为空？rwsem_mark_wake 唤醒等待队列其他读者return 持锁成功，退出} else if (rwsem_reader_phase_trylock()) //乐观自旋失败，再尝试一次持锁return 持锁成功，退出}乐观自旋彻底失败，加入等待队列初始化waiter结构，设置超时时间如果进程没有经历过乐观自旋(1)处设置的count一直在生效，会阻止写者持锁，当前不是写者持锁和handoff状态，那读者持锁成功了持锁成功，退出否则加上RWSEM_FLAG_WAITERS，将自身加入等待队列加入等待队列之后锁被释放，或者另一个读者持锁我们是队首，那么rwsem_mark_wake唤醒所有读者持锁包括我们自身持锁成功，退出while(1)等待waiter.task被清除}

读者将自身加入等待队列时，会创建唤醒队列和等待队列结构
struct rwsem_waiter {struct list_head list; //链表头struct task_struct *task; //等待持锁是task，被唤醒时会清除这个成员enum rwsem_waiter_type type; //表明自身是等待读锁还是写锁unsigned long timeout; //保存超时时间，只对队首进程生效，超时之后并不会退出，而是设置handoff标志unsigned long last_rowner; //保存当前的持锁者
}struct wake_q_head {struct wake_q_node *first; //初始化为WAKE_Q_TAILstruct wake_q_node **lastp; //初始化为自身first成员的地址
}

static struct rw_semaphore __sched *
rwsem_down_read_slowpath(struct rw_semaphore *sem, int state)
{//adjustment用来存储将要count增加或者减少的值//初始化为非0，在乐观自旋逻辑中清零，这样可以通过该值判断进程有没有经历过乐观自旋的逻辑long count, adjustment = -RWSEM_READER_BIAS;struct rwsem_waiter waiter;//初始化唤醒队列DEFINE_WAKE_Q(wake_q);bool wake = false;//保存当前owner信息waiter.last_rowner = atomic_long_read(&sem->owner);//如果当前不是读者持锁if (!(waiter.last_rowner & RWSEM_READER_OWNED))//在保存的owner中加上禁止读者自旋等待的flagwaiter.last_rowner &= RWSEM_RD_NONSPINNABLE;//当前是否允许自旋等待，不允许则加入等待队列if (!rwsem_can_spin_on_owner(sem, RWSEM_RD_NONSPINNABLE))goto queue;//乐观自旋等待的部分
---------------------------------------------------------------------- (1)//减去之前rwsem_read_trylock中加上的RWSEM_READER_BIASatomic_long_add(-RWSEM_READER_BIAS, &sem->count);adjustment = 0;if (rwsem_optimistic_spin(sem, false)) {//自旋等待成功，成功设置owner为当前task，持有count//读者持锁成功，就可以将其他等待的读者唤醒了if ((atomic_long_read(&sem->count) & RWSEM_FLAG_WAITERS)) {raw_spin_lock_irq(&sem->wait_lock);if (!list_empty(&sem->wait_list))//唤醒队列中所有能唤醒的读者rwsem_mark_wake(sem, RWSEM_WAKE_READ_OWNED,                                            &wake_q);raw_spin_unlock_irq(&sem->wait_lock);wake_up_q(&wake_q);}return sem;//乐观自旋失败，进行最后一次尝试} else if (rwsem_reader_phase_trylock(sem, waiter.last_rowner)) {return sem;}//乐观自旋彻底失败，加入等待队列
---------------------------------------------------------------------- (2)
queue:waiter.task = current;waiter.type = RWSEM_WAITING_FOR_READ;waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT;raw_spin_lock_irq(&sem->wait_lock);//等待队列为空，当前进程为队首if (list_empty(&sem->wait_list)) {//adjustment!=0意味着没有经历过代码(1)部分的乐观自旋等待，即之前设置的count中的RWSEM_READER_BIAS没有被减去//当前不是写者持锁和handoff状态，那读者持锁成功了if (adjustment && !(atomic_long_read(&sem->count) &(RWSEM_WRITER_MASK | RWSEM_FLAG_HANDOFF))) {smp_acquire__after_ctrl_dep();raw_spin_unlock_irq(&sem->wait_lock);//count值前面已经增加过RWSEM_READER_BIAS了，现在不需要调整，设置owner之后就退出rwsem_set_reader_owned(sem);lockevent_inc(rwsem_rlock_fast);return sem;}//否则添加RWSEM_FLAG_WAITERS，表明等待队列非空，当前进程不是队首adjustment += RWSEM_FLAG_WAITERS;}//将自身加入等待队列list_add_tail(&waiter.list, &sem->wait_list);if (adjustment)count = atomic_long_add_return(adjustment, &sem->count);else                                                                                               count = atomic_long_read(&sem->count);//加入等待队列完毕
----------------------------------------------------//加入等待队列之后情况如果发生改变，锁被释放了if (!(count & RWSEM_LOCK_MASK)) {clear_wr_nonspinnable(sem);wake = true;}//或者没有写者持锁并且我们是队首//那么唤醒并且协助队列中所有的读者持锁，包括我们自身//rwsem_mark_wake会清除所有被唤醒者的waiter.taskif (wake || (!(count & RWSEM_WRITER_MASK) &&(adjustment & RWSEM_FLAG_WAITERS))) //adjustment被设置了RWSEM_FLAG_WAITERS意味着我们是队首rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);raw_spin_unlock_irq(&sem->wait_lock);wake_up_q(&wake_q);//在队列中等待waiter.task被清除
----------------------------------------------------------for (;;) {set_current_state(state);//当waiter.task被清除时，说明我们持锁成功if (!smp_load_acquire(&waiter.task)) {/* Matches rwsem_mark_wake()'s smp_store_release(). */break;}//处理信号
...schedule();lockevent_inc(rwsem_sleep_reader);                                                             }__set_current_state(TASK_RUNNING);lockevent_inc(rwsem_rlock);return sem;out_nolock:list_del(&waiter.list);if (list_empty(&sem->wait_list)) {atomic_long_andnot(RWSEM_FLAG_WAITERS|RWSEM_FLAG_HANDOFF,&sem->count);}raw_spin_unlock_irq(&sem->wait_lock);__set_current_state(TASK_RUNNING);lockevent_inc(rwsem_rlock_fail);return ERR_PTR(-EINTR);
}

rwsem_can_spin_on_owner

检测是否能够乐观自旋。

static inline bool rwsem_can_spin_on_owner(struct rw_semaphore *sem,unsigned long nonspinnable)
{struct task_struct *owner;unsigned long flags;bool ret = true;//当前任务被调度器标记为需要调度，那么不能自旋等待if (need_resched()) {return false;}
...//获取当前owner以及flagowner = rwsem_owner_flags(sem, &flags);//如果flag中禁止了当前身份的自旋或者当前是写者持锁并且写者放弃了cpu，自旋失败if ((flags & nonspinnable) ||(owner && !(flags & RWSEM_READER_OWNED) && !owner_on_cpu(owner)))ret = false;
...return ret;
}

rwsem_optimistic_spin

在owner上乐观自旋，第一步通过乐观自旋获取sem->osq，只有持有这个锁才能在sem->owner上自旋等待。返回自旋等待是否成功，如果是被迫退出自旋返回0.

自旋等待成功会设置当前task为owner。

rwsem_optimistic_spin
{osq_lock 获取乐观自旋所osq失败，退出while(1) {owner_state = rwsem_spin_on_owner 更新当前owner状态OWNER_SPINNABLE? -> breakrwsem_try_xxx_lock_unqueued 调用不加入队列尝试获取锁的函数,此函数能够抢占等待队列的持锁机会reader只检测是否存在writer持锁和handoffwriter只检测锁是否被人持有和handoff，两者均不关心waiter成功 -> break写者等待读者持锁的情况，读者太多不能确定每个读者是否都在线每16次循环检测超时时间超时 -> breakRT任务不能被阻塞，读者持锁情况向读者太多不能确定每个读者是否都在线两次机会获取锁或者带写者持锁开始spin}}

@sem   指向的指针锁
@wlock 决定是否是写者函数中使用owner_state表示当前owner的状态
enum owner_state {OWNER_NULL = 1          owner为NULLOWNER_WRITER = 2        owner为写者OWNER_READER = 4        owner为读者OWNER_NONSPINNABLE = 8  当前不允许在owner上spin
}
#define OWNER_SPINNABLE     (OWNER_NULL | OWNER_WRITER | OWNER_READER)static bool rwsem_optimistic_spin(struct rw_semaphore *sem, bool wlock)
{bool taken = false;int prev_owner_state = OWNER_NULL;int loop = 0;u64 rspin_threshold = 0;                                                                           unsigned long nonspinnable = wlock ? RWSEM_WR_NONSPINNABLE: RWSEM_RD_NONSPINNABLE;preempt_disable();/* 先获取osqlock，osqlock也是一种乐观自旋机制，在soqlock上自旋失败，则不尝试在owner上自旋等待，直接退出 */if (!osq_lock(&sem->osq))goto done;//由于到达此步之前需要持有osq，因此即使有多个读写者抢锁也只能有一个自旋等待owner//乐观自旋等待owner发生变化时持锁for (;;) {enum owner_state owner_state;//自旋等待owner发生改变，返回改变之后的stateowner_state = rwsem_spin_on_owner(sem, nonspinnable);//除OWNER_NONSPINNABLE之外都是允许继续spin的if (!(owner_state & OWNER_SPINNABLE))break;//根据身份调用不加入队列尝试获取锁的函数，不加入等待队列意味着有可能抢占等待队列进程的持锁机会//对于写者而言，等待count中既没有读者持锁也没有写者持锁，也不是handoff状态//对于读者而言，等待count中既没有没有写者持锁，也不是handoff状态//成功返回1，失败返回0taken = wlock ? rwsem_try_write_lock_unqueued(sem): rwsem_try_read_lock_unqueued(sem);//持锁成功，退出循环if (taken)break;//其他的读者可以直接持锁，写者需要等待所有读者释放锁才能持锁，写者无法弄清楚是否所有读者都在线//所以如果当前是写者等锁，并且读者持有锁，给自旋等待的时间加个限制if (wlock && (owner_state == OWNER_READER)) {//prev_owner_state初始化为OWNER_NULL，第一次此条件必定成立if (prev_owner_state != OWNER_READER) {//不允许写者在owner上spin，那么退出if (rwsem_test_oflags(sem, nonspinnable))break;//计算写者的最大等待时间rspin_threshold = rwsem_rspin_threshold(sem);//循环次数清零loop = 0;并且已经超过了该读者的预定等}//每16次循环计算当前等待是否超时，如果超时设置RWSEM_NONSPINNABLE禁止所有的在owner上的spinelse if (!(++loop & 0xf) && (sched_clock() > rspin_threshold)) {rwsem_set_nonspinnable(sem);lockevent_inc(rwsem_opt_nospin);break;}}//读者可能有很多，RT任务同样不能弄清楚每个读者的状态//上次不是写者持锁，则RT任务不再等待，读者一直持锁RT任务将有两次等待机会if (owner_state != OWNER_WRITER) {//调度器认为我们需要让出cpu，那么退出等待if (need_resched())break;//读者持锁的情况向RT任务有两次机会spinif (rt_task(current) &&(prev_owner_state != OWNER_WRITER))break;}//记录这次循环的持锁者，下次循环时会用到prev_owner_state = owner_state;cpu_relax();}osq_unlock(&sem->osq);
done:preempt_enable();lockevent_cond_inc(rwsem_opt_fail, !taken);return taken;
}

rwsem_spin_on_owner

自旋等待owner或者flag发生改变，返回新的state

static noinline enum owner_state
rwsem_spin_on_owner(struct rw_semaphore *sem, unsigned long nonspinnable)
{struct task_struct *new, *owner;unsigned long flags, new_flags;enum owner_state state;                                                                                                                                                                   //获取ownerowner = rwsem_owner_flags(sem, &flags);/* 获取onwer的状态，这里有4种情况* 1.owner中存在禁止当前身份spin的flag，比如当前是读者owner中存在RWSEM_RD_NONSPINNABLE，返回 OWNER_NONSPINNABLE* 2.当前是读者持锁，返回 OWNER_READER* 3.在等待的过程中所已经被释放了，返回 OWNER_NULL* 4.写者持锁，OWNER_WRITER */state = rwsem_owner_state(owner, flags, nonspinnable);//当前不是写者持锁，停止等待owner//因为读者持锁时间可能很长，OWNER_NONSPINNABLE意味着禁止自旋等待if (state != OWNER_WRITER)return state;rcu_read_lock();for (;;) {//读取新的flag和ownernew = rwsem_owner_flags(sem, &new_flags);//如果owner或者flag发生了改变if ((new != owner) || (new_flags != flags)) {//重新获取stat，终止循环state = rwsem_owner_state(new, new_flags, nonspinnable);break;}
...//如果自选条件不满足if (need_resched() || !owner_on_cpu(owner)) {state = OWNER_NONSPINNABLE;break;}cpu_relax();}rcu_read_unlock();return state;
}

rwsem_try_xxx_lock_unqueued

不加入等待队列的尝试持锁，这此持锁没有判断等待队列是否有等待者，可能会抢占等待队列中进程的持锁机会，所以需要进行handoff的判断，避免等待着被强占次数过多饿死。

static inline bool rwsem_try_write_lock_unqueued(struct rw_semaphore *sem)
{long count = atomic_long_read(&sem->count);//写者等待无人持锁以及非handoff状态，之后尝试持锁//使用循环是为了给写者更多的持锁机会，读者持锁就只有一次判断while (!(count & (RWSEM_LOCK_MASK|RWSEM_FLAG_HANDOFF))) {if (atomic_long_try_cmpxchg_acquire(&sem->count, &count,count | RWSEM_WRITER_LOCKED)) {rwsem_set_owner(sem);lockevent_inc(rwsem_opt_wlock);return true;}}return false;
}

前面读者实际上是有非阻塞持锁的函数的，rwsem_read_trylock，与此函数区别在于是否判断waiter。

static inline bool rwsem_try_read_lock_unqueued(struct rw_semaphore *sem)
{long count = atomic_long_read(&sem->count);//判断是否有写者持锁以及是否在handoff状态，读者持锁不影响另一个读者if (count & (RWSEM_WRITER_MASK | RWSEM_FLAG_HANDOFF))return false;//不在的话尝试持锁，将count + RWSEM_READER_BIAScount = atomic_long_fetch_add_acquire(RWSEM_READER_BIAS, &sem->count);//成功则开始设置ownerif (!(count & (RWSEM_WRITER_MASK | RWSEM_FLAG_HANDOFF))) {rwsem_set_reader_owned(sem);lockevent_inc(rwsem_opt_rlock);return true;}//失败则回退自己设置的RWSEM_READER_BIASatomic_long_add(-RWSEM_READER_BIAS, &sem->count);return false;
}

rwsem_mark_wake

唤醒等待队列的进程。

如果队首是写者但是唤醒类型不是RWSEM_WAKE_ANY，那么唤醒不了任何进程。

队首是读者，当前不是读者持锁唤醒其他读者，那么帮助这个读者提前占据锁。抢锁失败并且超时，就设置handoff位。


@wake_type 需要唤醒对象的类型
enum rwsem_wake_type {RWSEM_WAKE_ANY,     /* 唤醒队列的第一个等待者 */RWSEM_WAKE_READERS, /* 只唤醒读者 */RWSEM_WAKE_READ_OWNED   /* 读者持锁唤醒其他读者 */                                                             };
@wake_q 唤醒队列static void rwsem_mark_wake(struct rw_semaphore *sem,                   enum rwsem_wake_type wake_type,struct wake_q_head *wake_q)
{struct rwsem_waiter *waiter, *tmp;long oldcount, woken = 0, adjustment = 0;struct list_head wlist;lockdep_assert_held(&sem->wait_lock);//如果第一个等待者是写者，并且唤醒对象是任意队首等待者，那么不唤醒所有进程进行竞争，只唤醒这个写者//如果第一个等待者是写者，但是唤醒对象是其他的，比如读者持锁唤醒其他读者或者仅限于唤醒读者的情况//直接返回不进行唤醒动作，因为第一个等待的写者应该最先被唤醒waiter = rwsem_first_waiter(sem);if (waiter->type == RWSEM_WAITING_FOR_WRITE) {if (wake_type == RWSEM_WAKE_ANY) {wake_q_add(wake_q, waiter->task);lockevent_inc(rwsem_wake_writer);}return;}//此处之后处理队首是读者的情况
-------------------------------------------------------------------------//读者计数溢出处理if (unlikely(atomic_long_read(&sem->count) < 0))return;//不是读者持锁唤醒其他读者的情况，说明当前锁是未被人持有的状态,需要抢锁if (wake_type != RWSEM_WAKE_READ_OWNED) {struct task_struct *owner;//先构造一个虚假的读者持锁，帮助这个队首的读者提前获取锁adjustment = RWSEM_READER_BIAS;oldcount = atomic_long_fetch_add(adjustment, &sem->count);//抢锁失败了，写者偷到了锁，队首的读者错过了一次持锁机会，并且已经超过了该读者的预定等待时间//现在设置handoff，屏蔽所有竞争锁以及窃取锁的动作，让下个持锁者直接把锁交给队首的读者，退出if (unlikely(oldcount & RWSEM_WRITER_MASK)) {if (!(oldcount & RWSEM_FLAG_HANDOFF) &&time_after(jiffies, waiter->timeout)) {adjustment -= RWSEM_FLAG_HANDOFF;lockevent_inc(rwsem_rlock_handoff);}atomic_long_add(-adjustment, &sem->count);                                                                                                                                        return;}//帮助队首的读者抢锁成功了，设置队首进程为新的ownerowner = waiter->task;if (waiter->last_rowner & RWSEM_RD_NONSPINNABLE) {owner = (void *)((unsigned long)owner | RWSEM_RD_NONSPINNABLE);lockevent_inc(rwsem_opt_norspin);}__rwsem_set_reader_owned(sem, owner);}/*
这里之后，队首是读者，持锁者也是读者，需要做的是唤醒队列中的其他读者
1.读者持锁唤醒其他读者，
2.帮助队首的读者获取了锁，现在需要继续唤醒剩余的读者*/
-------------------------------------------------------------------//现在，持锁者是读者了，我们可以唤醒队列中至多MAX_READERS_WAKEUP个其他的读者，一般是1600个INIT_LIST_HEAD(&wlist);list_for_each_entry_safe(waiter, tmp, &sem->wait_list, list) {if (waiter->type == RWSEM_WAITING_FOR_WRITE)continue;woken++;list_move_tail(&waiter->list, &wlist);if (woken >= MAX_READERS_WAKEUP)break;}//增加需要唤醒的读者的计数，如果等待队列为空需要移除掉waiter标志adjustment = woken * RWSEM_READER_BIAS - adjustment;lockevent_cond_inc(rwsem_wake_reader, woken);if (list_empty(&sem->wait_list)) {adjustment -= RWSEM_FLAG_WAITERS;}//handoff只作用于队首进程，现在我们已经协助队首拿到锁了，清除它if (woken && (atomic_long_read(&sem->count) & RWSEM_FLAG_HANDOFF))adjustment -= RWSEM_FLAG_HANDOFF;//将增加的读者数目写入countif (adjustment)atomic_long_add(adjustment, &sem->count);/* 将需要唤醒的进程放入wake_q */list_for_each_entry_safe(waiter, tmp, &wlist, list) {struct task_struct *tsk;tsk = waiter->task;get_task_struct(tsk);//唤醒之前清除waiter->tasksmp_store_release(&waiter->task, NULL);wake_q_add_safe(wake_q, tsk);}
}

up_read 读者释放锁

如果owner是当前task，清除掉owner中task_struct部分，保留flag部分。

将count减去RWSEM_READER_BIAS，即减少一名读者计数。

若无人持锁并且等待队列不为空，调用rwsem_wake唤醒等待队列进程，最终盗用的是rwsem_mark_wake，唤醒类型为RWSEM_WAKE_ANY。

void up_read(struct rw_semaphore *sem)
{rwsem_release(&sem->dep_map, _RET_IP_);__up_read(sem);
}static inline void __up_read(struct rw_semaphore *sem)
{long tmp;DEBUG_RWSEMS_WARN_ON(sem->magic != sem, sem);DEBUG_RWSEMS_WARN_ON(!is_rwsem_reader_owned(sem), sem);//onwer是当前进程，将flag中的task_struct清除，保留flagrwsem_clear_reader_owned(sem);//减少读者数目tmp = atomic_long_add_return_release(-RWSEM_READER_BIAS, &sem->count);DEBUG_RWSEMS_WARN_ON(tmp < 0, sem);//当前无人持锁if (unlikely((tmp & (RWSEM_LOCK_MASK|RWSEM_FLAG_WAITERS)) ==RWSEM_FLAG_WAITERS)) {clear_wr_nonspinnable(sem);//唤醒等待队列rwsem_wake(sem, tmp);}
}

down_write 写者持锁

写者和其他写着以及读者互斥，因此需要没有任何人持锁的情况下才能获取锁。

写者持锁的条件比读者简单，只需要设置count的RWSEM_WRITER_LOCKED bit就算持锁成功。

读者会有其他人协助帮忙持锁，自身被唤醒只用检查waiter.task是否被清空，写者无人协助，被唤醒之后会自己去检查锁的状态，然后拿锁。

挟制持锁，owner必定是自身。

down_write 和读者持锁流程一样，都是先尝试__down_write_trylock失败则调用__down_write。

伪代码为：
if !__down_write_trylock__down_writevoid __sched down_write(struct rw_semaphore *sem)
{might_sleep();rwsem_acquire(&sem->dep_map, 0, 0, _RET_IP_);LOCK_CONTENDED(sem, __down_write_trylock, __down_write);
}

__down_write_trylock

写者尝试非阻塞持锁。仅当count == 0时，将其设置为RWSEM_WRITER_LOCKED，并且设置owner为当前task。

static inline int __down_write_trylock(struct rw_semaphore *sem)
{long tmp;DEBUG_RWSEMS_WARN_ON(sem->magic != sem, sem);tmp  = RWSEM_UNLOCKED_VALUE;//如果count == 0，那么设置count = RWSEM_WRITER_LOCKED，即表明写者持锁if (atomic_long_try_cmpxchg_acquire(&sem->count, &tmp,RWSEM_WRITER_LOCKED)) {                                                                                                              //设置owner = current                                 rwsem_set_owner(sem);return true;}return false;
}

__down_write

static inline void __down_write(struct rw_semaphore *sem)
{long tmp = RWSEM_UNLOCKED_VALUE;//再尝试一次__down_write_trylock的流程if (unlikely(!atomic_long_try_cmpxchg_acquire(&sem->count, &tmp, RWSEM_WRITER_LOCKED)))//失败则进入慢速路径rwsem_down_write_slowpath(sem, TASK_UNINTERRUPTIBLE);elserwsem_set_owner(sem);
}

rwsem_down_write_slowpath

乐观自旋持锁以及加入等待队列死等的慢速路径。

rwsem_down_write_slowpath
{和读者类似的乐观自旋尝试加入等待队列不是队首，写者持锁什么都不干，读者持锁唤醒队列其他读者，无人持锁唤醒队列队首while(1) {rwsem_try_write_lock 尝试持锁，当前队首进程等待超时设置handoff当前进程设置了handoff，那么尝试在owner上乐观自旋}
}

wstate 用来表明当前写着的状态
WRITER_FIRST 当前是等待队列的队首
WRITER_NOT_FIRST 当前不是队首
WRITER_HANDOFF 当前进程已经等待超时或者当前时RT进程，允许在count中设置handoffstatic struct rw_semaphore *
rwsem_down_write_slowpath(struct rw_semaphore *sem, int state)
{long count;bool disable_rspin;enum writer_wait_state wstate;struct rwsem_waiter waiter;struct rw_semaphore *ret = sem;DEFINE_WAKE_Q(wake_q);//尝试乐观自旋偷锁//和前面读者流程一致，不过是以写者身份spin，等待count中没有写者持锁，读者持锁，handoff标记if (rwsem_can_spin_on_owner(sem, RWSEM_WR_NONSPINNABLE) &&rwsem_optimistic_spin(sem, true)) {/* rwsem_optimistic_spin() implies ACQUIRE on success */return sem;}//获取禁止spin的flagdisable_rspin = atomic_long_read(&sem->owner) & RWSEM_NONSPINNABLE;//初始化waiter，准备加入等待队列waiter.task = current;waiter.type = RWSEM_WAITING_FOR_WRITE;waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT;raw_spin_lock_irq(&sem->wait_lock);                            //当前进程是等待队列的队首吗？wstate = list_empty(&sem->wait_list) ? WRITER_FIRST : WRITER_NOT_FIRST;list_add_tail(&waiter.list, &sem->wait_list);//如果我们不是队首，需要处理下等待队列if (wstate == WRITER_NOT_FIRST) {count = atomic_long_read(&sem->count);//当前是写者持锁，没什么好做的慢慢等待即可if (count & RWSEM_WRITER_MASK)goto wait;//当前是读者持锁，尝试使用RWSEM_WAKE_READERS唤醒队列内的其他读者持锁//当前无人持锁，RWSEM_WAKE_READERS唤醒队列的队首rwsem_mark_wake(sem, (count & RWSEM_READER_MASK)? RWSEM_WAKE_READERS: RWSEM_WAKE_ANY, &wake_q);//唤醒进程if (!wake_q_empty(&wake_q)) {                                                                                                                            raw_spin_unlock_irq(&sem->wait_lock);wake_up_q(&wake_q);wake_q_init(&wake_q);   /* Used again, reinit */raw_spin_lock_irq(&sem->wait_lock);}} else {//我们是队首，增加RWSEM_FLAG_WAITERS标志atomic_long_or(RWSEM_FLAG_WAITERS, &sem->count);}wait:/* 死循环等锁 */set_current_state(state);for (;;) {//以写者的身份尝试获取锁，成功返回1if (rwsem_try_write_lock(sem, wstate)) {break;}raw_spin_unlock_irq(&sem->wait_lock);//如果当前进程允许设置handoff还是持锁失败，那么尝试在owner上自旋等待owner为NULL，避免睡眠//因为设置了handoff之后应该不会再有竞争者if (wstate == WRITER_HANDOFF &&rwsem_spin_on_owner(sem, RWSEM_NONSPINNABLE) == OWNER_NULL)goto trylock_again;/* Block until there are no active lockers. */for (;;) {...//让出cpu，等待唤醒schedule();//唤醒之后的节点，更新自身以及锁的状态，发生改变就进入外层循环再次尝试持锁，否则回来继续睡眠
------------------------------------------------------//当前进程能够设置handoff状态，马上退出内存循环到外层循环rwsem_try_write_lock给count设置handoffif (wstate == WRITER_HANDOFF)break;//当前不是队首进程，更新下自身状态看下是否已经成为新的队首if ((wstate == WRITER_NOT_FIRST) &&(rwsem_first_waiter(sem) == &waiter))wstate = WRITER_FIRST;//更新锁的状态，如果无人持锁马上去外层循环尝试持锁count = atomic_long_read(&sem->count);if (!(count & RWSEM_LOCK_MASK))break;//当前是队首并且是RT进程或者等待超时，那么允许设置handoff状态if ((wstate == WRITER_FIRST) && (rt_task(current) ||time_after(jiffies, waiter.timeout))) {wstate = WRITER_HANDOFF;lockevent_inc(rwsem_wlock_handoff);break;}}
trylock_again:raw_spin_lock_irq(&sem->wait_lock);}//持锁成功，恢复进程状态，将自身从等待队列删除__set_current_state(TASK_RUNNING);list_del(&waiter.list);rwsem_disable_reader_optspin(sem, disable_rspin);raw_spin_unlock_irq(&sem->wait_lock);lockevent_inc(rwsem_wlock);return ret;out_nolock:__set_current_state(TASK_RUNNING);raw_spin_lock_irq(&sem->wait_lock);                                                                                                                                                       list_del(&waiter.list);if (unlikely(wstate == WRITER_HANDOFF))atomic_long_add(-RWSEM_FLAG_HANDOFF,  &sem->count);if (list_empty(&sem->wait_list))atomic_long_andnot(RWSEM_FLAG_WAITERS, &sem->count);elserwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);raw_spin_unlock_irq(&sem->wait_lock);wake_up_q(&wake_q);lockevent_inc(rwsem_wlock_fail);return ERR_PTR(-EINTR);
}

rwsem_try_write_lock

写者尝试获取锁，主要用来处理写者的handoff的标记。

存在count handoff标记当前进程不是队首的情况下会直接失败。
有人持锁，根据wstate选择是否设置handoff，然后返回
无人持锁，清除handoff，然后持锁

@wstate 用来表明当前写着的状态
WRITER_FIRST 当前是等待队列的队首
WRITER_NOT_FIRST 当前不是队首
WRITER_HANDOFF 当前进程已经等待超时或者当前时RT进程，允许在count中设置handoff
static inline bool rwsem_try_write_lock(struct rw_semaphore *sem,enum writer_wait_state wstate)
{long count, new;lockdep_assert_held(&sem->wait_lock);//读取新的countcount = atomic_long_read(&sem->count);do {//count中是否被设置了handoff bit？bool has_handoff = !!(count & RWSEM_FLAG_HANDOFF);//如果存在handoff那意味着锁会被交接给队首，当前如果不是队首，不可能获取到锁，因此直接返回失败if (has_handoff && wstate == WRITER_NOT_FIRST)return false;new = count;//当前仍然有人持锁if (count & RWSEM_LOCK_MASK) {//存在handoff或者当前进程暂时不能设置handoff，返回失败if (has_handoff || (wstate != WRITER_HANDOFF))return false;//我们可以设置handoff，加上handoff标记new |= RWSEM_FLAG_HANDOFF;} else {//无人持锁了，尝试持锁new |= RWSEM_WRITER_LOCKED;new &= ~RWSEM_FLAG_HANDOFF;//持锁成功如果等待队列为空，需要删除RWSEM_FLAG_WAITERSif (list_is_singular(&sem->wait_list))new &= ~RWSEM_FLAG_WAITERS;}//如果这期间count没有发生变化，那么将sem->count设置为new，返回更新是否成功，count被更新为sem->count} while (!atomic_long_try_cmpxchg_acquire(&sem->count, &count, new));if (new & RWSEM_FLAG_HANDOFF)return false;//设置owner == currentrwsem_set_owner(sem);return true;
}