【精通内核】Linux 内核写锁实现原理与源码解析

本文导读 Linux 内核读锁实现原理,描述自旋锁时,已经顺带描述了读写自旋锁,所以本节将不再描述自旋锁的读写锁实现。读者是否能想到,既然自旋锁有相关的读写锁实现,信号量也应该有呢?答案是一定的。所以可以到,读写锁实际上是在原有锁上进行优化读写的操作。下面讨论源码实现。

一、Linux 内核读写锁核心结构解读 定义一个结构体 rw_semaphore 代表读写信号量,然后义一宏定义表明读写信号量的偏移值。具体源码如下。

struct rw_semaphore{
// 符号长整型,看到long类型,读者就知道,这又是将一个long类型长度大小切割成不同部分来使用的
// 由于使用i38632位来作为例子,因此这里long为32位,同样我们分割为高16位和低16位来使用
signed long count;

#define RWSEM UNLOCKED VALUE 0x0000 0000 // 无锁状态值为0

#define RWSEM_ACTIVE_BIAS 0x0000 0001 // 锁活动偏移值1

// 锁活动位数为4(4个16进制)*4(一个16进制等于4个二进制)=16,即2^16次方个锁位
#define RWSEM ACTIVE MASK Ox0000 ffff

#define RWSEM_WAITING BIAS (-0x00010000) // 锁等待偏移量,即 0xffff 0000

#define RWSEM ACTIVE READ_BIAS RWSEM_ACTIVE_BIAS // 读锁偏移量

// 写锁偏移量0xffff0001 为负数
#define RWSEM ACTIVE WRITE BIAS (RWSEM_WAITING_BIAS+RWSEM_ACTIVE_BIAS)

spinlock t wait_lock; // 保护等待链表的自旋锁
struct list_head wait_list; // 等待链表
};

//等待读写信号量的任务结构体
struct rwsem_waiter{
struct list_head list;

struct task_struct *task;
unsigned int flags; // 标志位声明为等待读锁还是写锁

#define RWSEM_WAITING_FOR_READ 0x00000001
#define RWSEM_WAITING_FOR_WRITE 0x00000002
};

二、Linux 内核获取写锁源码解读 首先原子性减 0xffff0001,然后判断原来的状态是否为 0,如果是,则表明获取写锁成功;

否则需要调用 rwsem_down_write_failed 函数进行阻塞排队操作。

static inline void_down_write(struct rw_semaphore *sem) {
int tmp=RWSEM_ACTIVE_WRITE BIAS;
_asm__volatile_(
//原子性减0xffff001即写锁偏移量,返回旧值被放到edx寄存器中
LOCK_PREFIX" xadd %%edx,(%%eax)"

//查看之前的count值是否为0,因为只有为0,才是无锁状态
" testl %%edx,%%edx"

//如果不为0,则获取锁失败跳到标号2处执行
" jnz 2f""1:"
LOCK_SECTION_START("")
//保存ecx,然后调用rwsem_down_write_failed进行阻塞排队操作
"2:"
" pushl %%есx"
" call rwsem_down_write failed"
" popl %%eсx"
" jmp 1b"
LOCK_SECTION_END
: "=m"(sem->count), "=d"(tmp)
: "a"(sem),"1"(tmp), "m"(sem->count)
: "memory", "cc");
}

// 处理写锁上锁失败逻辑
struct rw_semaphore *rwsem_down_write_failed(struct rw_semaphore *sem) {
// 创建等待节点
struct rwsem waiter waiter;
waiter.flags=RWSEM WAITING FOR WRITE;
// 调用公共处理逻辑执行等待操作。-RWSEM ACTIVE BIAS =Oxffff fff
rwsem_down_failed_common(sem,&waiter,-RWSEM_ACTIVE BIAS);
return sem;
}

三、Linux 内核释放写锁源码解读 首先将锁状态变为无锁状态,如果发现有任务正在等待唤醒,那么调用 rwsem_wake 唤醒等待的任务

static inline void_up_write(struct rw_semaphore *sem) {
_asm__volatile_(
" movl %2,%%edx" // 将写锁偏移量取负数后的值,即0x0000 ffff 放入edx中
// 尝试从Oxffff0001(持有写锁且无等待任务的状态,因为写写、读写互斥)变为 0x00000000
LOCK PREFIX" xaddl %%edx,(%%eax)"
" jnz 2f" //如果之前count值不为0,则有任务正在等待,跳到标号2处执行
" 1:"
LOCK_SECTION_START("")
"2:"
// 对dx也就是释前的lock值低16位自减,看看是否为0,即看看是否有活动的任务
" decw %%dx"
// 如果不为0,则表示写锁被释放后有任务获得了锁,退出;
// 否则,调用rwsem_wake唤醒等待任务
" jnz 1b"
" pushl %%ecx"
" call rwsem_wake"
" popl %%eсx"
" jmp 1b"
LOCK SECTION END
: "=m"(sem->count)
: "a"(sem), "i"(-RWSEM_ACTIVE_WRITE_BIAS),"m"(sem->count)
: "memory", "cc", "edx");
}

四、Linux 内核读写锁锁降级源码解读 有时候我们需要在获取到写锁后,进行降级为读锁,这可以通过 downgrade_write 方法进行锁降级有先原子性的降锁状态从写锁状态置为读锁状态,如果结果小于 0,则表明有任务正在等待被唤醒,此时可以调用 rwsem_downgrade_wake 函数唤醒等待读锁的任务,因为此时写锁已经被释放,可以让等待读锁的任务一起并行执行。

// 写锁降级为读锁
static inline void___downgrade_write(struct rw_semaphore*sem) {
_asm__volatile_(
LOCK PREFIX" addl %2,(%%eax)" //将状态从0xZZZZ0001变为0xYYYY0001
// 如果小于0,即锁正在等待被释放,则跳到标号2处执行rwsem_downgrade_wake函数,降级唤醒操作
" js 2f"
"1:"
LOCK_SECTION_START("")
"2:"
" pushl %%ecx"
" pushl %%edx"
" call rwsem_downgrade_wake" // 调用rwsem_downgrade_wake 函数
" popl %%edx"
" popl %%есx"
" jmp 1b"
LOCK_SECTION_END
: "=m"(sem->count)
: "a"(sem), "i"(-RWSEM_WAITING_BIAS), "m"(sem->count):
: "memory", "cc");
}


// 接下来查看rwsem_downgrade_wake 函数实现过程。
struct rw_semaphore*rsem_downgrade_wake(struct rw_semaphore*sem){
// 获取自旋锁
spin_lock(&sem->wait lock);
// 如果等待队列不为空,那么调用_rwsem_do_wake函数唤醒
// 注意,这里传入为0,表明只唤醒读任务 if(!list_empty(&sem->wait list))
sem=___rwsem_do_wake(sem,0); // 释放自旋锁
spin_unlock(&sem->wait lock);
return sem;
}

五、Linux 内核读写锁唤醒线程过程 首先获取保护等待队列的自旋锁,然后检测队列是否为空,如果不为空,那么调用 rwsem_do_wake 函数唤醒等待的任务。

struct rw_semaphore *rwsem wake(struct rw semaphore*sem) {
spin lock(&sem->wait lock); // 获取自旋锁
// 如果等待链表为空,则什么也不做,否则调用rwsemdo wake函数唤醒任务
// 注:这里传入为0,表名只唤醒读任务
if(!listempty(&sem->wait list))
sem =_rwsem_do_wake(sem,1);// 1表明唤醒写任务
spin_unlock(&sem->wait_lock);
return sem;
}

// 真正唤醒流程
static inline struct rw_semaphore*__rwsem_do_wake(struct rw_semaphore *sem,int wakewrite) {
struct rwsem waiter *waiter;
struct list head *next;
signed long oldcount; int woken, loop;
// 如果不唤醒写任务,那么直接跳转到 dont_wake_writers执行
if(!wakewrite)
goto dont_wake_writers;
try again:
oldcount =rwsem_atomic_update(RWSEM_ACTIVE_BIAS,sem)-RWSEM_ACTIVE_BIAS;

// 如果之前count与上RWSEM_ACTIVE_MASK不为0,也就是还有活动的任务,则还原修改之前的值
if (oldcount & RWSEM_ACTIVE_MASK)
goto undo;

// 否则取出下一个等待任务,如果下一个等待的任务不是一个写任务,那么调用readers_only
//函数唤醒读任务
waiter =list_entry(sem->wait_list.next,struct rwsem_waiter,list);
if(!(waiter->flags&RWSEM_WAITING_FOR_WRITE))
goto readers_only;

//否则将写者从队列中移出,修改 flags 为0,调用wake_up_process函数唤醒任务,并且退出
list_del(&waiter->list);
waiter->flags =0;
wake_up_process(waiter->task);
goto out;

不唤醒写者操作流程,取出下一个等待者,如果等待者是写者,那么直接退出
dont wake writers:
waiter =list_entry(sem->wait_list.next,struct rwsem_waiter,list);
if(waiter->flags &RWSEM_WAITING_FOR_WRITE)
goto out;

// 只唤醒读者操作流程,遍历等待链表,直到等待者为写者时停下
readers_only:
woken =0;
do {
woken++;
if (waiter->list.next==&sem->wait_list)
break;
waiter =list_entry(waiter->listnext,struct rwsem_waiter,list);
} while (waiter->flags &RWSEM_WAITING_FOR_READ);
loop=woken;
woken *= RWSEM_ACTIVE_BIAS-RWSEM_WAITING_BIAS; woken-=RWSEM_ACTIVE_BIAS;
rwsem_atomic_add(woken,sem); // 更新counter 值
next = sem->wait_list.next; // 获取循环开始节点
for (; loop>0;loop--){ // 从当前节点一直遍历唤醒所有读等待任务
waiter =list_entry(next,struct rwsem_waiter,list);
next = waiter->list.next;
waiter->flags =0;
wake_up_process(waiter->task);

// 然后将唤醒了的一系列链表断开链接 sem->wait_list.next=next; next->prev = &sem->wait_list;
// 退出流程
out:
return sem;
//还原操作流程
undo:
// 再次判断,如果还有活动任务,则退出
if (rwsem_atomic_update(-RWSEM_ACTIVE_BIAS,sem)!=0)
goto out;
goto try_again;
}
发表评论

相关文章