【精通内核】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
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

二、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;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

三、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");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

四、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;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

五、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;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

总结
实际上，针对读写信号量，如果我们用 C 语言代码高级语言来描述的话，则十分简单，即一个公平的读写锁。也就是说，当有读锁持有时，如果有读任务，则可以直接获得读锁;但如果此时有写仕务在等待的情况下，那么将会导致读锁获取失败，转而进入等待状态。当读锁释放后返回看看有没写者在等待，如果有写者在等待且传入了唤醒写者的标识 1，那么看看等待列表的下一个等待任务是 1 是写节点，如果不是，那么遍历等待列表，唤醒所有读者，直到遇到一个写节点。然而，如果在持有写锁的情况下，那么读锁肯定获取失败，然后进入等待队列中，写锁被释放后，如果有锁等待，那会唤醒等待任务。