Linux内核机制—rwsem
作者：互联网
基于linux-5.10.66
一、相关结构
1. rwsem 表示结构体 struct rw_semaphore
/*
* 对于无竞争的 rwsem，count 和 owner 是任务在获取 rwsem 时需要触及的唯一字段。
* 因此，它们彼此相邻放置，以增加它们共享相同cacheline的机会。
*
* 在竞争 rwsem 中，owner 可能是此结构中最常访问的字段，因为持有 osq 锁的乐观
* 等待者将在 owner 上自旋。 对于嵌入的 rwsem，上层结构中的其它热字段应远离rwsem，
* 以减少它们共享相同cacheline导致cacheline弹跳问题的机会。
*/
struct rw_semaphore {
/* 计数和位掩码组成 */
atomic_long_t count;
/* task_struct指针值和位掩码组成 */
atomic_long_t owner;
#ifdef CONFIG_RWSEM_SPIN_ON_OWNER //默认不使能
struct optimistic_spin_queue osq; /* spinner MCS lock */
#endif
/* wait_lock 用于保护wait_list成员 */
raw_spinlock_t wait_lock;
struct list_head wait_list;
#ifdef CONFIG_DEBUG_RWSEMS //默认不使能
void *magic;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC //默认不使能
struct lockdep_map dep_map;
#endif
ANDROID_VENDOR_DATA(1); //u64 android_vendor_data1;
ANDROID_OEM_DATA_ARRAY(1, 2); //u64 android_oem_data1[2];
};
成员介绍：
(1) 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 */ #define RWSEM_WRITER_LOCKED (1UL << 0) //标记有writer在临界区 #define RWSEM_FLAG_WAITERS (1UL << 1) //标记是否有waiter在等待队列上等待 #define RWSEM_FLAG_HANDOFF (1UL << 2) #define RWSEM_FLAG_READFAIL (1UL << (BITS_PER_LONG - 1)) //1<<63 最高bit位为1。表示在临界区的reader太多，计数溢出 #define RWSEM_READER_SHIFT 8 #define RWSEM_READER_BIAS (1UL << RWSEM_READER_SHIFT) //1<<8，在临界区的reader从此bit开始加1计数 #define RWSEM_READER_MASK (~(RWSEM_READER_BIAS - 1)) //低8bit清0，其它bit全为1 #define RWSEM_WRITER_MASK RWSEM_WRITER_LOCKED //1<<0 #define RWSEM_LOCK_MASK (RWSEM_WRITER_MASK|RWSEM_READER_MASK) //(1<<0)|(低8bit清0，其它bit全为1) 表示是否有writer或reader在临界区中 //只要sem->count 和其与不为0，reader就不能走快速路径持锁 #define RWSEM_READ_FAILED_MASK (RWSEM_WRITER_MASK|RWSEM_FLAG_WAITERS|RWSEM_FLAG_HANDOFF|RWSEM_FLAG_READFAIL) //(1<<0)|(1<<1)|(1<<2)|(1<<63)
RWSEM_FLAG_READFAIL: 最高有效位（读取失败位）不太可能被设置。 无论如何，在 down_read() 快速路径中仍会检查此保护位，以防我们将来需要将更多reader位用于其他目的。
RWSEM_READ_FAILED_MASK: reader获取锁时若sem->count中包含其内的任意一个掩码，就无法走快速获取获取读锁。
RWSEM_FLAG_HANDOFF: 可以设置或清除handoff标志位的三个位置。
1) rwsem_mark_wake() 读者使用的此函数。
2) rwsem_try_write_lock() 写者使用的此函数。
3) rwsem_down_write_slowpath() 中的错误路径。
对于上述所有情况，wait_lock 都会被持有。writer还必须是 wait_list 中第一个waiter以便可以设置handoff bit位。 因此无法同时设置/清除handoff bit位。
(2) owner
由于至少是8字节对齐，owner的低3bit可以另做它用，如下。其它bit位保存的是持有此rwsem的task_struct结构。
#define RWSEM_READER_OWNED (1UL << 0) // The rwsem is owned by readers #define RWSEM_RD_NONSPINNABLE (1UL << 1) // Readers cannot spin on this lock #define RWSEM_WR_NONSPINNABLE (1UL << 2) // Writers cannot spin on this locks #define RWSEM_NONSPINNABLE (RWSEM_RD_NONSPINNABLE | RWSEM_WR_NONSPINNABLE) //(1<<1)|(1<<2) #define RWSEM_OWNER_FLAGS_MASK (RWSEM_READER_OWNED | RWSEM_NONSPINNABLE) //(1<<0)|(1<<1)|(1<<2)
只有 reader_owned 宏，没有 writer_owned 宏。有设置这个 reader_owned 的逻辑，设置位置为reader持锁进入临界区的位置，但是没有直接清
理这个 reader_owned 的逻辑，但是不需要做额外的清理动作，因为writer持锁时rwsem_set_owner() 中直接 sem->owner = current 进行赋值，
会清理掉reader_owner标志位。
对于多个reader持锁进入临界区，owner保存的是最后一个进入临界区的reader的task_struct结构指针。
(3) wait_list
等待队列，当reader或writer获取不到rwsem而阻塞时，会挂载此链表上。默认是尾插法，从头部唤醒，当做一个队列来使用了。
2. rwsem 等待者结构 rwsem_waiter
struct rwsem_waiter {
/* 通过此成员挂载 rwsem->wait_list 链表上 */
struct list_head list;
/* waiter的 task_struct 结构 */
struct task_struct *task;
/* 标记是reader还是writer，分别取值为 RWSEM_WAITING_FOR_READ/RWSEM_WAITING_FOR_WRITE */
enum rwsem_waiter_type type;
/* 挂入等待链表之前设置，通常是4ms */
unsigned long timeout;
/* 上一个reader owner */
unsigned long last_rowner;
};
二、相关函数
1. 初始化函数
/* 初始化宏，将 sem->count 设置为 RWSEM_UNLOCKED_VALUE(8), 将 sem->owner 设置为0 */ init_rwsem(sem)
2. 获取读sem
/* 阻塞时是 TASK_UNINTERRUPTIBLE 状态 */ void __sched down_read(struct rw_semaphore *sem); /* 阻塞时是 TASK_INTERRUPTIBLE 状态，writer没有类似接口 */ int __sched down_read_interruptible(struct rw_semaphore *sem) /* 阻塞时是 TASK_KILLABLE 状态 */ int __sched down_read_killable(struct rw_semaphore *sem) /* 尝试获取读锁，不会阻塞，成功获取返回1，失败返回0 */ int down_read_trylock(struct rw_semaphore *sem)
3. 获取写sem
/* 阻塞时是 TASK_UNINTERRUPTIBLE 状态 */ void __sched down_write(struct rw_semaphore *sem) /* 阻塞时是 TASK_KILLABLE 状态 */ int __sched down_write_killable(struct rw_semaphore *sem) /* 尝试获取写锁，不会阻塞，成功获取返回1，失败返回0 */ int down_write_trylock(struct rw_semaphore *sem)
4. 释放读sem
void up_read(struct rw_semaphore *sem)
5. 释放写sem
void up_write(struct rw_semaphore *sem)
可见rwsem只是一个二值信号量，不支持递归获取，获取不到要进入休眠。
三、获取读信号流程
1. down_read
void __sched down_read(struct rw_semaphore *sem)
{
might_sleep();
/* 若没有使能 CONFIG_LOCKDEP 就是空函数，默认不使能的 */
rwsem_acquire_read(&sem->dep_map, 0, 0, _RET_IP_);
/* 若没有使能 CONFIG_LOCK_STAT 就等于 __down_read(sem) */
LOCK_CONTENDED(sem, __down_read_trylock, __down_read);
}
static inline void __down_read(struct rw_semaphore *sem)
{
/*
* 先执行尝试获取信号量，若获取不到再进入慢速路径，若获取
* 到了就将sem->owner设置为当前线程。
*/
if (!rwsem_read_trylock(sem)) {
rwsem_down_read_slowpath(sem, TASK_UNINTERRUPTIBLE);
/*若不使能 CONFIG_DEBUG_RWSEMS 就是空函数，默认不使能 */
DEBUG_RWSEMS_WARN_ON(!is_rwsem_reader_owned(sem), sem);
} else {
rwsem_set_reader_owned(sem);
}
}
走慢速路径之前会调用一次 rwsem_read_trylock() 尝试获取锁，若获取不到就会走慢速路径，获取到后走快速路径。
(1) rwsem_read_trylock(sem) 判断是否能持有读取锁
/* 这个函数只是一个判断，能成功持锁返回1，否则返回0 */
static inline bool rwsem_read_trylock(struct rw_semaphore *sem)
{
/* 作用：sem->count += RWSEM_READER_BIAS */
long cnt = atomic_long_add_return_acquire(RWSEM_READER_BIAS, &sem->count);
/*
* 最高bit位为1才会小于0, RWSEM_FLAG_READFAIL 是bit63,但是没有使用过，
* 这里应该只是表示reader持锁个数已经导致bit位溢出，应该不会发生这种情况。
*/
if (WARN_ON_ONCE(cnt < 0))
rwsem_set_nonspinnable(sem);
/* 只要这些标志中sem->count中有1个，reader就不能在快速路径中获取到锁 */
return !(cnt & RWSEM_READ_FAILED_MASK);
}
若此时已经有 writer 在临界区了，也就是说 cnt 中包含了 RWSEM_READ_FAILED_MASK 中的 RWSEM_WRITER_MASK，reader 就无法走快速路径获取到锁了，而是要挂到等待链表中等待。
可以再看下面 RWSEM_READ_FAILED_MASK 的定义，只要有writer在临界区、有waiter在等待、写者持锁慢速路径中对wait状态的写者标记了handoff、写者标记了handoff、reader等待的太多(bit63==1)，读者都不能走快速路径持锁。
(2) rwsem_set_reader_owned(sem) 持锁快速路径
reader持锁，在sem->owner上或上 RWSEM_READER_OWNED 并保留owner中原来的 RWSEM_RD_NONSPINNABLE 位的值。最后一个持锁的reader才会将其task_struct结构指针设置到sem->owner中。
static inline void rwsem_set_reader_owned(struct rw_semaphore *sem)
{
__rwsem_set_reader_owned(sem, current);
}
/*
* 最后拥有的 reader 的 task_struct 指针将留在 owner 字段中。
*
* 请注意，owner 的值仅表示该任务之前拥有了该 rwsem，当检查该字段时，它可能不
* 再是真正的所有者或真正的所有者之一，因此请谨慎对待。
*
* reader的non-spinnable bit位被保留下来。
*
* 传参：owner = current
* 作用：sem->owner = owner | RWSEM_READER_OWNED | (sem->owner & RWSEM_RD_NONSPINNABLE)
*/
static inline void __rwsem_set_reader_owned(struct rw_semaphore *sem, struct task_struct *owner)
{
unsigned long val = (unsigned long)owner | RWSEM_READER_OWNED | (atomic_long_read(&sem->owner) & RWSEM_RD_NONSPINNABLE);
atomic_long_set(&sem->owner, val);
}
(3) rwsem_down_read_slowpath(sem, TASK_UNINTERRUPTIBLE) 持读取锁的慢速路径
/*
* Wait for the read lock to be granted.
* __down_read()传参：(sem, TASK_UNINTERRUPTIBLE)
*/
static struct rw_semaphore __sched * rwsem_down_read_slowpath(struct rw_semaphore *sem, int state)
{
/* 每增加一个等待的reader就从bit8的reader计算bit位开始增加1 */
long count, adjustment = -RWSEM_READER_BIAS;
struct rwsem_waiter waiter;
DEFINE_WAKE_Q(wake_q);
bool wake = false;
bool already_on_list = false;
/*
* Save the current read-owner of rwsem, if available, and the
* reader nonspinnable bit.
* 先将 sem->owner 保存到 waiter.last_rowner。在reader的持锁快速路径中
* sem->owner 保存的是最后一个持读取锁的reader.
*/
waiter.last_rowner = atomic_long_read(&sem->owner);
/*
* 在reader快速路径持锁时，都会 sem->owner 都会或上 RWSEM_READER_OWNED
* 标志。若没有此标志说明上次reader已经不是走快速路径持锁的了。此时
* waiter.last_rowner 中只保留 RWSEM_RD_NONSPINNABLE bit位的值【为啥？
* 此bit在默认deconfig下是否有使用 ？】。
*/
if (!(waiter.last_rowner & RWSEM_READER_OWNED))
waiter.last_rowner &= RWSEM_RD_NONSPINNABLE;
/* 默认没有使能 CONFIG_RWSEM_SPIN_ON_OWNER，此函数恒返回0 */
if (!rwsem_can_spin_on_owner(sem, RWSEM_RD_NONSPINNABLE))
goto queue;
....
queue:
/* 标记waiter的线程、类型、超时时间【超时时间怎么使用？】 */
waiter.task = current;
waiter.type = RWSEM_WAITING_FOR_READ;
waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT; //jiffies+1, 即是4ms
raw_spin_lock_irq(&sem->wait_lock);
if (list_empty(&sem->wait_list)) {
/*
* 翻译：如果等待队列为空，并且锁没有被writer持有或handoff位未被设置，
* 则此reader可以退出慢速路径并立即返回，因为其 RWSEM_READER_BIAS
* 位已被设置到count成员中【什么时候设置进去的？】。
*
* 默认配置下 adjustment = -RWSEM_READER_BIAS 恒成立。
* 没有 writer 持锁，且没有指定 handoff。
*/
if (adjustment && !(atomic_long_read(&sem->count) &
(RWSEM_WRITER_MASK | RWSEM_FLAG_HANDOFF))) {
/* Provide lock ACQUIRE */
smp_acquire__after_ctrl_dep(); //只是smp_rmb()
raw_spin_unlock_irq(&sem->wait_lock);
/*
* 将current线程当做reader设置进sem->owner中，然后就返回了。
* 这体现了多个reader可以同时获取锁。
*/
rwsem_set_reader_owned(sem);
/* CONFIG_LOCK_EVENT_COUNTS 默认没有使能，是空函数 */
lockevent_inc(rwsem_rlock_fast);
return sem;
}
/* 等待链表为空，但是有writer持锁或指定了handoff bit位，准备好有waiter等待bit位 */
adjustment += RWSEM_FLAG_WAITERS;
}
/*----下面就是等待链表不为空(有writer持锁或指定了handoff标志)的处理情况了----*/
trace_android_vh_alter_rwsem_list_add(&waiter, sem, &already_on_list);
if (!already_on_list)
/* 默认逻辑是添加到等待队列的尾部 */
list_add_tail(&waiter.list, &sem->wait_list);
/* we're now waiting on the lock, but no longer actively locking */
if (adjustment)
/*
* 相当于 sem->count = sem->count - RWSEM_READER_BIAS + RWSEM_FLAG_WAITERS
* 减去 RWSEM_READER_BIAS 应该是因为在前面的trey_lock中加了。多了一个waiter。
*/
count = atomic_long_add_return(adjustment, &sem->count);
else
count = atomic_long_read(&sem->count); //不会执行到
/*
* 翻译：
* 如果没有处于active状态的锁，唤醒排队前面的任务。
* 如果没有writers并且我们排在队列的首位，唤醒我们自己的waiter加入现有的活跃readers！
* active 锁应该表示获取不到sem而挂在等待链表上的reader和writer.
*/
if (!(count & RWSEM_LOCK_MASK)) {
/* 没有使能 CONFIG_RWSEM_SPIN_ON_OWNER, 是个空函数 */
clear_wr_nonspinnable(sem);
wake = true;
}
/*
* 若没有writer持锁，并且sem->wait_list链表为空(adjustment中有RWSEM_FLAG_WAITERS
* bit位表示), 那么就标记要唤醒的任务。
*/
if (wake || (!(count & RWSEM_WRITER_MASK) && (adjustment & RWSEM_FLAG_WAITERS)))
rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q); /*-又重新看了-*/
trace_android_vh_rwsem_wake(sem);
raw_spin_unlock_irq(&sem->wait_lock);
/* 唤醒queue入等待队列中的任务 */
wake_up_q(&wake_q); /* 这不是睡眠吗，，为啥执行起了唤醒逻辑？*/
/* ----下面就是有writer持有信号量，reader要进入休眠的情况了---- */
/* wait to be given the lock */
trace_android_vh_rwsem_read_wait_start(sem);
for (;;) {
/* __down_read()传参：state == TASK_UNINTERRUPTIBLE) */
set_current_state(state);
/* 如果 waiter.task 为空，退出死循环 */
if (!smp_load_acquire(&waiter.task)) {
/* Matches rwsem_mark_wake()'s smp_store_release(). */
break;
}
/* 若传参state==TASK_UNINTERRUPTIBLE，这里不可能为真 */
if (signal_pending_state(state, current)) {
raw_spin_lock_irq(&sem->wait_lock);
if (waiter.task)
goto out_nolock;
raw_spin_unlock_irq(&sem->wait_lock);
/* Ordered by sem->wait_lock against rwsem_mark_wake(). */
break;
}
schedule();
lockevent_inc(rwsem_sleep_reader);
}
__set_current_state(TASK_RUNNING);
trace_android_vh_rwsem_read_wait_finish(sem);
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);
trace_android_vh_rwsem_read_wait_finish(sem);
lockevent_inc(rwsem_rlock_fail);
/* 被信号唤醒时的返回值 */
return ERR_PTR(-EINTR);
}
在获取不到信号量，进入休眠之前，还尝试标记唤醒。
rwsem_mark_wake() 函数：
/*
* 翻译：
* 当被阻塞的线程可以运行时处理释放锁的逻辑：
* - 如果我们从 up_xxxx() 来到这里，那么 RWSEM_FLAG_WAITERS 位必须已被设置。
* - 队列中必须有排队的任务
* - wait_lock 必须被调用者持有
* - 任务标记唤醒，调用者必须稍后调用 wake_up_q() 来唤醒阻塞的任务，并降低引用计数，最好是在释放 wait_lock 时做。
* - 任务归零后被唤醒的任务从链表中删除
* - 只有当降级为假时，writers才会被标记为唤醒
*/
/* rwsem_down_read_slowpath()传参：(sem, RWSEM_WAKE_ANY, &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，以便我们确定要执行的唤醒。
* 是从 sem->wait_list 链表首，取一个任务来唤醒的。
*/
waiter = rwsem_first_waiter(sem);
/*
* 若等待链表头部排队的是一个writer, 而且唤醒类型是ANY, 那么直接
* 将这个writer放到待唤醒链表上。
*/
if (waiter->type == RWSEM_WAITING_FOR_WRITE) {
if (wake_type == RWSEM_WAKE_ANY) {
/*
* 翻译：在队列前面标记 writer 以唤醒。直到调用者稍后真正唤醒了任务，
* 其他writer才能窃取它。另一方面，读者会阻塞，因为他们会注意到排队的writer。
*/
wake_q_add(wake_q, waiter->task);
/* 没使能CONFIG_LOCK_EVENT_COUNTS，是个空函数 */
lockevent_inc(rwsem_wake_writer);
}
return;
}
/*---- 下面就是sem的等待链表首不是writer或唤醒类型不是ANY的情况 ----*/
/*
* 翻译：如果有太多读者等待了，则不会唤醒读者。最高bit为1才是小于0，实际情况下
* 应该不大会出现。
*/
if (unlikely(atomic_long_read(&sem->count) < 0))
return;
/*
* 翻译：在我们将锁授予下一个读者之前，写者可能会偷走它。
* 我们更喜欢在计算读者数之前进行第一读者授权，这样如果写者偷了锁，
* 我们可以提前退出。
*/
if (wake_type != RWSEM_WAKE_READ_OWNED) {
struct task_struct *owner;
adjustment = RWSEM_READER_BIAS;
/* 应该是先将sem->count的值返回，再加上adjustment的值 (1)*/
oldcount = atomic_long_fetch_add(adjustment, &sem->count);
/* 若此刻有写者持有锁 */
if (unlikely(oldcount & RWSEM_WRITER_MASK)) {
/*
* 翻译：当我们已经等待“太”长时间(等待写者释放锁)时，请求 HANDOFF 以强制
* 解决该问题。
*/
if (!(oldcount & RWSEM_FLAG_HANDOFF) && time_after(jiffies, waiter->timeout)) {
/* 这里是减，减减就是加【handoff是专门针对写者吗？】 */
adjustment -= RWSEM_FLAG_HANDOFF;
lockevent_inc(rwsem_rlock_handoff);
}
/* 对 RWSEM_READER_BIAS 加减相互抵消，相当于 sem->count += RWSEM_FLAG_HANDOFF */
atomic_long_add(-adjustment, &sem->count);
return;
}
/*---- 下面就是当前没有writer在等待sem的情况了 ----*/
/*
* 翻译：将其设置为 reader-owned 可以让 spinners 及早感知到现在是读者拥有锁。
* 在读者的慢速路径入口处看到的读者不可spin位被复制。
*/
owner = waiter->task;
/* 若上个被阻塞的reader的nom-spin位被设置，则继承，仍然被设置 */
if (waiter->last_rowner & RWSEM_RD_NONSPINNABLE) {
owner = (void *)((unsigned long)owner | RWSEM_RD_NONSPINNABLE);
lockevent_inc(rwsem_opt_norspin);
}
/* 设置等待链表上的首个reader持有锁 */
__rwsem_set_reader_owned(sem, owner);
}
/*
* 翻译：
* 向队列中的所有读者授予最多 MAX_READERS_WAKEUP 读取锁。 我们知道被唤醒任
* 务至少是1个，正如我们上面所解释的。请注意，在唤醒任何进程之前，我们将计
* 数的“active part”增加读者的数量。
*
* 这是对相位公平 R/W 锁的改编，其中在读者阶段（第一个等待者是读者），所有
* 读者都有资格同时获取锁，而不管他们在队列中的顺序如何。 写者根据他们在队
* 列中的顺序获取锁。
*
* 我们必须在 2 遍中进行唤醒，以防止读者计数可能在增加之前减少。 因为将被唤
* 醒的写者可能还没睡。所以它可能会看到 waiter->task 被清除，执行完它的临界
* 区部分并在读者计数增加之前进行解锁。
*
* 1) 在单独的链表中收集读取等待者，对它们进行计数并在 rwsem 中增加读者计数。
* 2) 对于新链表中的每一个waiter，清空waiter->task 并放入wake_q 以便稍后唤醒。
*/
INIT_LIST_HEAD(&wlist);
list_for_each_entry_safe(waiter, tmp, &sem->wait_list, list) {
/* 跳过等待链表上的writer */
if (waiter->type == RWSEM_WAITING_FOR_WRITE)
continue;
woken++;
/* 将等待链表上的reader摘取下来挂在临时链表wlist的尾部 */
list_move_tail(&waiter->list, &wlist);
/* 翻译：限制每次调用可以唤醒的读者数量，256个, 应该很难达到。*/
if (woken >= MAX_READERS_WAKEUP)
break;
}
/* 减去应该是去除 (1) 位置的影响*/
adjustment = woken * RWSEM_READER_BIAS - adjustment;
lockevent_cond_inc(rwsem_wake_reader, woken);
/*
* 等待链表已经为空了说明等待链表上没有writer(因为没有往下摘writer)，
* 也没有reader了，这里减去有waiter的标志
*/
if (list_empty(&sem->wait_list)) {
/* hit end of list above */
adjustment -= RWSEM_FLAG_WAITERS;
}
/*
* 翻译：当我们唤醒读者后，就不再需要强制写者放弃锁，可以清除 HANDOFF 标志了。【？】
*/
if (woken && (atomic_long_read(&sem->count) & RWSEM_FLAG_HANDOFF))
adjustment -= RWSEM_FLAG_HANDOFF;
/*
* 相当于进来时 sem->count 的状态加上 woken * RWSEM_READER_BIAS，并且没有handoff
* 标志。
* 可见(要)唤醒状态的reader才会出记录在sem->count中，已经被阻塞的不会。
*/
if (adjustment)
atomic_long_add(adjustment, &sem->count);
/* 2nd pass */
/* 临时链表 wlist 上只挂入了reader */
list_for_each_entry_safe(waiter, tmp, &wlist, list) {
struct task_struct *tsk;
tsk = waiter->task;
/* 增加计数，防止使用过程中 task_struct 结构被释放 */
get_task_struct(tsk);
/*
* 翻译：确保在将 reader waiter 设置为 NULL 之前调用 get_task_struct()，
* 这样 rwsem_down_read_slowpath() 在整个唤醒任务的过程中始终保持引用
* 计数而不会和 do_exit() 存在竞争。
*/
/* waiter->task == NULL 【waiter->task 设置为NULL是什么意思？】*/
smp_store_release(&waiter->task, NULL);
/* 翻译：确保在将 reader waiter 设置为 NULL 后开始唤醒（由我们或其他人） */
wake_q_add_safe(wake_q, tsk);
}
}
主要逻辑：若链表首是个writer，且唤醒类型是ANY，只唤醒此writer就返回。若链表首不是writer，就唤醒等待队列中的所有reader。唤醒所有的reader能获得更高的并发效果，因为所有的reader能同时进入临界区。queue在等待队列中的writer是还没有进入临界区的。
sem->count 中对于 writer 只有一个标记，唤醒 writer 没有像唤醒 reader 那样从 sem->count 中减去对 reader 的计数。
四、获取写信号量流程
1. down_write()
/*
* lock for writing
*/
void __sched down_write(struct rw_semaphore *sem)
{
might_sleep();
rwsem_acquire(&sem->dep_map, 0, 0, _RET_IP_);
/* 没有使能CONFIG_LOCK_STAT，等效于__down_write(sem) */
LOCK_CONTENDED(sem, __down_write_trylock, __down_write);
}
EXPORT_SYMBOL(down_write);
/*
* lock for writing
*/
static inline void __down_write(struct rw_semaphore *sem)
{
long tmp = RWSEM_UNLOCKED_VALUE;
/*
* atomic_long_try_cmpxchg_acquire(atomic_long_t *v, long *old, long new)
* 若*v==*old, *v=new, return true
* 若*v!=*old, *old=*v, return false
*
* 作用是原子的执行：
* if (sem->count == RWSEM_UNLOCKED_VALUE) {
* sem->count = RWSEM_WRITER_LOCKED;
* }
*/
if (unlikely(!atomic_long_try_cmpxchg_acquire(&sem->count, &tmp, RWSEM_WRITER_LOCKED))) {
rwsem_down_write_slowpath(sem, TASK_UNINTERRUPTIBLE);
} else {
/* 当前sem还没被持有，writer快速持有路径。直接 sem->owner = current */
rwsem_set_owner(sem);
}
}
(1) 尝试获取写锁
判断 sem->count 是否为0，若是，就将起赋值为 RWSEM_WRITER_LOCKED，表示 writer 持有锁了，返回真。
(2) 持写锁的快速路径
上面尝试获取锁成功后，直接 sem->owner = current，表示持锁的是当前任务。
(3) 持写锁的慢速路径
若前面尝试持写锁失败，将进入持写锁慢速路径。执行 rwsem_down_write_slowpath() 函数
/*
* Wait until we successfully acquire the write lock.
*
* __down_write 传参：(sem, TASK_UNINTERRUPTIBLE)
*/
static 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);
bool already_on_list = false;
/* do optimistic spinning and steal lock if possible */
/* CONFIG_RWSEM_SPIN_ON_OWNER 默认不使能，恒返回false，不执行 */
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 bit位被设置时，在获取写锁后关闭此rwsem的reader的乐观自旋。
* 检索此标志没有被赋值过，disable_rspin 应该恒为假
*/
disable_rspin = atomic_long_read(&sem->owner) & RWSEM_NONSPINNABLE;
/* 翻译：乐观自旋失败，继续慢速路径并阻塞，直到可以获取sem。*/
waiter.task = current;
waiter.type = RWSEM_WAITING_FOR_WRITE;
waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT; //jiffies + 4ms
raw_spin_lock_irq(&sem->wait_lock);
/* account for this before adding a new element to the list */
wstate = list_empty(&sem->wait_list) ? WRITER_FIRST : WRITER_NOT_FIRST;
trace_android_vh_alter_rwsem_list_add(&waiter, sem, &already_on_list);
if (!already_on_list)
/* 默认是插入等待队列的尾部 */
list_add_tail(&waiter.list, &sem->wait_list);
/* we're now waiting on the lock */
/*
* 插入时等待队列不为空就是NOT_FIRST。此时可能是只有一个writer在临界区，等待链表上
* 都是reader，也可能已经有writer在等待链表上了，reader和writer都有。
*/
if (wstate == WRITER_NOT_FIRST) {
count = atomic_long_read(&sem->count);
/*
* 翻译：如果在我们之前已经有排队等待的任务了，并且：
* 1) 没有active状态的锁，唤醒前面排队的进程，因为可能设置了handoff bit位。
* 2) 没有active状态的写者和一些读者，锁必须是读拥有的；所以我们尝试唤醒任
* 何排在我们前面的读锁等待者(由于是尾插法，等待链表上已有的所有读者都在我们前面)。
*/
/*
* 之前持锁的writer是不是已经释放锁了，若是还没有释放，那么当前等待持锁的writer
* 直接去休眠等待。
*/
if (count & RWSEM_WRITER_MASK)
goto wait;
/* 若之前持锁的writer已经释放锁了，那么接下来就把等待的reader都唤醒吧 */
rwsem_mark_wake(sem, (count & RWSEM_READEMASK) ? RWSEM_WAKE_READERS : RWSEM_WAKE_ANY, &wake_q);
if (!wake_q_empty(&wake_q)) {
/* 翻译：我们希望最小化 wait_lock 保持时间，尤其是在要唤醒大量读者时。*/
raw_spin_unlock_irq(&sem->wait_lock);
/* 唤醒等待队列上的所有任务 */
wake_up_q(&wake_q);
/* Used again, reinit */
wake_q_init(&wake_q);
raw_spin_lock_irq(&sem->wait_lock);
}
} else {
/*
* 若插入时writer是等待链表上为空, 也就是此时只有reader持有锁，
* 只或上这个标志位，表示有writer在等待锁。
*/
atomic_long_or(RWSEM_FLAG_WAITERS, &sem->count);
}
wait:
trace_android_vh_rwsem_wake(sem);
/* wait until we successfully acquire the lock */
trace_android_vh_rwsem_write_wait_start(sem);
/* __down_write 传参：state == TASK_UNINTERRUPTIBLE */
set_current_state(state);
/* 第一层循环 */
for (;;) {
/*
* 成功获取sem返回true, 失败返回false。这里若传 wstate = WRITER_HANDOFF,
* 会赋值到 sem->count 中。
*/
if (rwsem_try_write_lock(sem, wstate)) {
/*
* 外层for循环唯一退出位置。唯一退出条件是成功持有sem。退出时
* sem->wait_lock 是锁定状态的。
*/
break;
}
raw_spin_unlock_irq(&sem->wait_lock);
/*
* 翻译：
* 在未能获取锁和设置了handoff bit位后，尝试在sem的owner上自
* 旋以加速锁的传递。如果前一个owner是一个on-cpu(正在运行)的writer
* 并且它刚刚释放了锁，则将返回 OWNER_NULL。在这种情况下，我们尝试
* 在不休眠的情况下再次获取锁。
*
* CONFIG_RWSEM_SPIN_ON_OWNER 默认不使能，此if恒不成立。
*/
if (wstate == WRITER_HANDOFF && rwsem_spin_on_owner(sem, RWSEM_NONSPINNABLE) == OWNER_NULL)
goto trylock_again; //恒不执行
/* Block until there are no active lockers. */
/* active lockers：位于临界区的入任务。睡眠等待，直到自己是 WRITER_FIRST 才能退出 */
/* 第二层循环 */
for (;;) {
/* __down_write 传参 state==TASK_UNINTERRUPTABLE, 恒不成立*/
if (signal_pending_state(state, current))
goto out_nolock;
/* 将当前任务切走 */
schedule();
lockevent_inc(rwsem_sleep_writer);
set_current_state(state);
/* If HANDOFF bit is set, unconditionally do a trylock */
if (wstate == WRITER_HANDOFF)
break;
/*
* 若之前writer不是等待链表中的第一个waiter, 而现在是一个waiter了，更新wstate。
* 这里是一级循环退出的唯一条件：WRITER_FIRST --> WRITER_HANDOFF --> break
*/
if (wstate == WRITER_NOT_FIRST && rwsem_first_waiter(sem) == &waiter)
wstate = WRITER_FIRST;
/* 若此时writer和reader都没有再持有sem */
count = atomic_long_read(&sem->count);
if (!(count & RWSEM_LOCK_MASK))
break;
/*
* 翻译：handoff 位的设置被推迟到 rwsem_try_write_lock() 被调用时。
*
* 此 writer 必须是等待链表上的首个waiter了，并且已经在这里是spin大于4ms了
*/
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);
trace_android_vh_rwsem_write_wait_finish(sem);
/* 将此waiter从等待链表中移除 */
list_del(&waiter.list);
/* 若之前是no-spin，退出时还是设置为no-spin */
rwsem_disable_reader_optspin(sem, disable_rspin);
raw_spin_unlock_irq(&sem->wait_lock);
lockevent_inc(rwsem_wlock);
return ret;
out_nolock: //被信号唤醒，此Case这里恒不会被执行
__set_current_state(TASK_RUNNING);
trace_android_vh_rwsem_write_wait_finish(sem);
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);
else
rwsem_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);
}
a. reader 获取sem锁时，若是没有 writer 持有，只是在 sem->count 上增加读者计数。之后在 writer 持锁时，发现 sem->count 不为0就会
进入慢速持锁路径，放在等待链表上休眠。若之前持锁的reader在休眠，这里什么也没有做，只是静静的等待reader释放锁。只有在 writer
持锁进入慢速路径时，发现等待链表上已经有任务了，并且之前持锁的writer已经释放锁了，才会尝试将之前阻塞的reader一下子都唤醒。
b. 进入慢速路径的writer若发现 sem->wait_list 等待链表不为空，这说明之前已经有一个writer和若干个reader在等待链表上了。
c. 虽然 sem->count 上 writer 只有一个bit的标志，标识同一时间只能有一个writer进入临界区。但是 sem->wait_list 链表上等待的 writer 却可以同时有多个，它们都不在临界区中。
d. 一直等待，直到此writer成功等待链表上首个waiter了。若它是rt线程或等待时间已经超过了4ms, 就会启用 WRITER_HANDOFF 进行锁交接。
(2) rwsem_try_write_lock() 函数
/*
* 翻译：
* 必须在持有 sem->wait_lock 的情况下调用此函数，以防止检查 rwsem 等待链表和
* 相应地设置 sem->count 之间的竞争条件。
*
* 如果 wstate 为 WRITER_HANDOFF，它将确保设置了切换位或在清除切换位的情况下
* 获取锁。
*
* rwsem_down_write_slowpath 传参 wstate 为 WRITER_FIRST 或 WRITER_NOT_FIRST,
* 第二轮也可能传参为 WRITER_HANDOFF.
*
* 成功获取sem返回true, 失败返回false并设置handoff bit位
*/
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);
count = atomic_long_read(&sem->count);
do {
bool has_handoff = !!(count & RWSEM_FLAG_HANDOFF);
/* 看来handoff主要是针对等待队列中首个waiter是writer的情况 */
if (has_handoff && wstate == WRITER_NOT_FIRST)
return false;
new = count;
/* 是否有writer或reader在等待sem */
if (count & RWSEM_LOCK_MASK) {
/*
* 虽然还有持锁，但是传参handoff标志，也不返回false退出了，
* 后面这个handoff标志会设置到sem->count中。
*/
if (has_handoff || wstate != WRITER_HANDOFF)
return false;
new |= RWSEM_FLAG_HANDOFF;
} else {
/*
* 既没有writer又没有reader在等待sem，此writer就可以持锁了，
* 或上writer标志，清除掉handoff标志。
*/
new |= RWSEM_WRITER_LOCKED;
new &= ~RWSEM_FLAG_HANDOFF;
/* 若等待链表中只有一个元素(本writer自己)，标记等待链表上没有waiters */
if (list_is_singular(&sem->wait_list))
new &= ~RWSEM_FLAG_WAITERS;
}
/*
* 若 sem->count==count, sem->count=new, return true;
* 若 sem->count!=count, count=sem->count, return false;
*/
} while (!atomic_long_try_cmpxchg_acquire(&sem->count, &count, new));
/*
* We have either acquired the lock with handoff bit cleared or
* set the handoff bit.
* 翻译：走到这里了还没退出，说明我们要么获得了锁并清除了handoff bit位，就返回true;
* 要么设置了handoff bit位, 返回false.
*/
if (new & RWSEM_FLAG_HANDOFF)
return false;
/* 直接单纯是sem->owner=current */
rwsem_set_owner(sem);
return true;
}
五、释放读信号流程
1. up_read() 函数
/* release a read lock */
void up_read(struct rw_semaphore *sem)
{
rwsem_release(&sem->dep_map, _RET_IP_);
__up_read(sem);
}
EXPORT_SYMBOL(up_read);
/* unlock after reading */
static inline void __up_read(struct rw_semaphore *sem)
{
long tmp;
/* CONFIG_DEBUG_RWSEMS 默认不使能，是空函数 */
DEBUG_RWSEMS_WARN_ON(sem->magic != sem, sem);
DEBUG_RWSEMS_WARN_ON(!is_rwsem_reader_owned(sem), sem);
/* CONFIG_DEBUG_RWSEMS 默认不使能，是个空函数 */
rwsem_clear_reader_owned(sem);
/* sem->count 中的 reader 计数减去1 */
tmp = atomic_long_add_return_release(-RWSEM_READER_BIAS, &sem->count);
/* CONFIG_DEBUG_RWSEMS 默认不使能，是个空函数 */
DEBUG_RWSEMS_WARN_ON(tmp < 0, sem);
/* 若是有waiter, 才执行唤醒流程。[或上 RWSEM_LOCK_MASK 有什么意义呢，作者笔误？] */
if (unlikely((tmp & (RWSEM_LOCK_MASK|RWSEM_FLAG_WAITERS)) == RWSEM_FLAG_WAITERS)) {
clear_wr_nonspinnable(sem);
rwsem_wake(sem, tmp);
}
}
(1) rwsem_wake() 函数
/*
* 翻译：
* 处理唤醒信号量上的waiter
* - 如果我们来到这里，up_read/up_write 会减少 count 的活动部分
*/
static struct rw_semaphore *rwsem_wake(struct rw_semaphore *sem, long count)
{
unsigned long flags;
DEFINE_WAKE_Q(wake_q);
raw_spin_lock_irqsave(&sem->wait_lock, flags);
/* 若等待链表不为空，就执行唤醒动作 */
if (!list_empty(&sem->wait_list))
/* 标记待唤醒的waiter */
rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);
trace_android_vh_rwsem_wake_finish(sem);
raw_spin_unlock_irqrestore(&sem->wait_lock, flags);
/* 执行唤醒操作 */
wake_up_q(&wake_q);
return sem;
}
六、释放写信号量流程
1. up_write() 函数
/* release a write lock */
void up_write(struct rw_semaphore *sem)
{
rwsem_release(&sem->dep_map, _RET_IP_);
trace_android_vh_rwsem_write_finished(sem);
__up_write(sem);
}
EXPORT_SYMBOL(up_write);
/* unlock after writing */
static inline void __up_write(struct rw_semaphore *sem)
{
long tmp;
/* CONFIG_DEBUG_RWSEMS 默认不使能，是个空函数 */
DEBUG_RWSEMS_WARN_ON(sem->magic != sem, sem);
/*
* 翻译：如果通过设置 RWSEM_NONSPINNABLE 位将所有权转移给匿名写入者，则
* sem->owner 可能与当前不同。
*
* CONFIG_DEBUG_RWSEMS 默认不使能，是个空函数
*/
DEBUG_RWSEMS_WARN_ON((rwsem_owner(sem) != current) && !rwsem_test_oflags(sem, RWSEM_NONSPINNABLE), sem);
/* 直接 sem->owner=0 */
rwsem_clear_owner(sem);
/* 原子的读取-修改-写入，等效于 return sem->count; sem->count += -RWSEM_WRITER_LOCKED; */
tmp = atomic_long_fetch_add_release(-RWSEM_WRITER_LOCKED, &sem->count);
/* 有等待的waiter才执行唤醒操作 */
if (unlikely(tmp & RWSEM_FLAG_WAITERS))
rwsem_wake(sem, tmp);
}
七、总结
1. rwsem可以有多个读者同时进入临界区，但是读者和写者，写者和写者之间只能有一个进入临界区。
2. 有写者在临界区，对应成功获取写锁，从 down_write() 到 up_write() 之间，前者会在 sem->count 上或上 RWSEM_WRITER_LOCKED 标志，i后者会清除此标志。RWSEM_WRITER_LOCKED 标志标识的是写者在临界区中，慢速路径中写者被挂在等待链表上时是不会或上这个标志的。在读者持锁时会因为 sem->count 中有这个标志而被迫进入慢速路径在等待链表上进行等待。
标签：count,waiter,RWSEM,rwsem,reader,内核,Linux,sem 来源： https://www.cnblogs.com/hellokitty2/p/16343272.html