Linux中读写自旋锁rwlock的实现

一、判断读写锁是否锁定rwlock_is_locked

#define RW_LOCK_BIAS             0x01000000
#define rwlock_is_locked(x) ((x)->lock != RW_LOCK_BIAS)

RW_LOCK_BIAS

• 解锁状态的标志位

rwlock_is_locked

• 如果lock和解锁标志位一致则返回0，表示未锁定
• 不一致返回1，表示已锁定

二、读写锁初始化rwlock_init

#define RW_LOCK_BIAS             0x01000000
typedef struct {volatile unsigned int lock;
} rwlock_t;
#define RW_LOCK_UNLOCKED (rwlock_t) { RW_LOCK_BIAS }#define rwlock_init(x)  do { *(x) = RW_LOCK_UNLOCKED; } while(0)

RW_LOCK_UNLOCKED

• 将读写锁初始化为未锁定状态

rwlock_init

• 动态初始化，x是一个读写锁结构体的指针

三、尝试加写锁的函数_raw_write_trylock

#define RW_LOCK_BIAS             0x01000000
static inline int _raw_write_trylock(rwlock_t *lock)
{atomic_t *count = (atomic_t *)lock;if (atomic_sub_and_test(RW_LOCK_BIAS, count))return 1;atomic_add(RW_LOCK_BIAS, count);return 0;
}

该函数是不阻塞加写锁函数

(atomic_t *)lock

• 对lock进行强转，因为rwlock_t结构体和atomic_t结构体的成员类型是一致的

atomic_sub_and_test

• 将count减去RW_LOCK_BIAS
• 如果结果为0则返回1
• 否则返回0

atomic_add(RW_LOCK_BIAS, count);

• 将减去的值加回去，恢复原始状态

四、写锁获取失败后的自旋等待__write_lock_failed

asm(
".section .sched.text\n"
".align 4\n"
".globl __write_lock_failed\n"
"__write_lock_failed:\n\t"LOCK "addl      $" RW_LOCK_BIAS_STR ",(%eax)\n"
"1:     rep; nop\n\t""cmpl   $" RW_LOCK_BIAS_STR ",(%eax)\n\t""jne    1b\n\t"LOCK "subl      $" RW_LOCK_BIAS_STR ",(%eax)\n\t""jnz    __write_lock_failed\n\t""ret"
);

1.函数整体功能

__write_lock_failed:

• 场景: 当尝试获取写锁失败时调用
• 目的: 自旋等待直到获取写锁成功
• 调用约定: 锁的地址通过eax寄存器传递

2.代码逐行分析

2.1. 恢复锁状态

LOCK "addl      $" RW_LOCK_BIAS_STR ",(%eax)"

作用: 撤销之前失败的写锁获取尝试

2.2. 自旋等待循环

"1:     rep; nop\n\t"           // 短暂的暂停，节约功耗"cmpl   $" RW_LOCK_BIAS_STR ",(%eax)\n\t"  // 检查锁是否空闲"jne    1b\n\t"         // 如果不空闲，继续循环

等待条件: lock == RW_LOCK_BIAS

• 这表示锁处于完全空闲状态
• 没有读者，也没有写者

2.3. 再次尝试获取写锁

LOCK "subl      $" RW_LOCK_BIAS_STR ",(%eax)\n\t"  // 尝试获取写锁"jnz    __write_lock_failed\n\t"           // 如果失败，重试"ret"                                      // 成功，返回

关键操作:

• lock -= RW_LOCK_BIAS
• 如果结果为0：获取成功
• 如果结果非0：获取失败，重新开始

五、写锁获取的内联汇编实现__build_write_lock

#define RW_LOCK_BIAS_STR        "0x01000000"
#define __build_write_lock_ptr(rw, helper) \asm volatile(LOCK "subl $" RW_LOCK_BIAS_STR ",(%0)\n\t" \"jz 1f\n" \"call " helper "\n\t" \"1:\n" \::"a" (rw) : "memory")
#define __build_write_lock(rw, helper)  do { \if (__builtin_constant_p(rw)) \__build_write_lock_const(rw, helper); \else \__build_write_lock_ptr(rw, helper); \} while (0)

1.宏定义分析

1.1. 基础常量

#define RW_LOCK_BIAS_STR "0x01000000"

这是表示读写锁空闲标志的字符串形式，用于内联汇编

1.2. 指针版本的写锁获取

#define __build_write_lock_ptr(rw, helper) \asm volatile(LOCK "subl $" RW_LOCK_BIAS_STR ",(%0)\n\t" \"jz 1f\n" \"call " helper "\n\t" \"1:\n" \::"a" (rw) : "memory")

2.汇编代码详细解析

LOCK "subl $" RW_LOCK_BIAS_STR ",(%0)\n\t"  # 原子减BIAS
"jz 1f\n"                                    # 如果结果为0，跳转到标签1
"call " helper "\n\t"                        # 否则调用helper函数
"1:\n"                                       # 标签1：继续执行

2.1执行流程

情况1: 快速路径（获取成功）

初始: lock = RW_LOCK_BIAS (0x01000000) - 锁空闲
执行: lock -= RW_LOCK_BIAS → lock = 0
条件: 结果=0 → jz跳转 → 跳过call指令
结果: 直接成功，无需函数调用

情况2: 慢速路径（获取失败）

初始: lock < RW_LOCK_BIAS - 锁被占用
执行: lock -= RW_LOCK_BIAS → lock ≠ 0  
条件: 结果≠0 → 不跳转 → 执行call指令
结果: 调用helper函数处理竞争

3.输入约束和破坏描述符

::"a" (rw) : "memory"

• "a" (rw): 输入操作数，要求rw（锁指针）放入eax寄存器
• "memory": 内存破坏描述符，确保内存访问顺序

4.编译时常数优化

if (__builtin_constant_p(rw))__build_write_lock_const(rw, helper);
else__build_write_lock_ptr(rw, helper);

__builtin_constant_p() 是GCC内置函数：

• 如果rw是编译时常数，使用优化版本
• 如果rw是变量，使用指针版本

六、非抢占式加写锁_raw_write_lock

static inline void _raw_write_lock(rwlock_t *rw)
{__build_write_lock(rw, "__write_lock_failed");
}

调用__build_write_lock函数，指定获取写锁失败时调用__write_lock_failed

七、抢占式加写锁__preempt_write_lock

static inline void __preempt_write_lock(rwlock_t *lock)
{if (preempt_count() > 1) {_raw_write_lock(lock);return;}do {preempt_enable();while (rwlock_is_locked(lock))cpu_relax();preempt_disable();} while (!_raw_write_trylock(lock));
}

1.函数功能

static inline void __preempt_write_lock(rwlock_t *lock)

• 目的: 在支持内核抢占的环境中安全获取写锁
• 特点: 在等待锁时允许被抢占

2.代码逻辑分析

2.1.快速路径检查

if (preempt_count() > 1) {_raw_write_lock(lock);return;
}

preempt_count() 含义：

• preempt_count() = 0: 可抢占状态
• preempt_count() > 0: 不可抢占状态（在中断、软中断、持有自旋锁等）

条件解释：

• 如果preempt_count() > 1，说明已经在原子上下文中
• 此时不能被抢占，直接使用非抢占版本的锁获取

2.2. 慢速路径循环

do {preempt_enable();                    // 允许抢占while (rwlock_is_locked(lock))      // 检查锁是否被占用cpu_relax();                 // 等待时让CPU进入低功耗preempt_disable();                   // 禁止抢占
} while (!_raw_write_trylock(lock));        // 尝试获取锁，不阻塞

2.3.场景1: 在原子上下文中

进程A: preempt_count() = 2 (持有自旋锁)
调用 __preempt_write_lock():- 条件成立，直接调用 _raw_write_lock()- 可能自旋等待，但不会被抢占

2.4.场景2: 在进程上下文中，锁空闲

进程A: preempt_count() = 0
调用 __preempt_write_lock():第一次循环:preempt_enable()    → 允许抢占rwlock_is_locked() → false (锁空闲)preempt_disable()   → 禁止抢占  _raw_write_trylock() → 成功获取，退出循环

2.5.场景3: 在进程上下文中，锁被占用

进程A: preempt_count() = 0  
调用 __preempt_write_lock():第一次循环:preempt_enable()    → 允许抢占while循环: 发现锁被占用 → cpu_relax() 等待// 在等待期间可能被更高优先级进程抢占！被唤醒后继续等待...锁释放后:preempt_disable()   → 禁止抢占_raw_write_trylock() → 可能失败（竞态条件）第二次循环: 重试...

八、加写锁包装函数_write_lock

void __lockfunc _write_lock(rwlock_t *lock)
{preempt_disable();if (unlikely(!_raw_write_trylock(lock)))__preempt_write_lock(lock);
}

preempt_disable();

• 先禁用内核抢占，表示这个函数加写锁不允许抢占

_raw_write_trylock(lock)

• 尝试获取写锁
• 如果失败则调用加写锁函数__preempt_write_lock
• 因为当前处于原子上下文中，所以后面必然调用_raw_write_lock函数

九、解除写锁_write_unlock

#define _raw_write_unlock(rw)	asm volatile("lock ; addl $" RW_LOCK_BIAS_STR ",%0":"=m" ((rw)->lock) : : "memory")
void __lockfunc _write_unlock(rwlock_t *lock)
{_raw_write_unlock(lock);preempt_enable();
}

_raw_write_unlock

• 将解释标志加回到lock
• 启用抢占

十、读锁获取失败后的自旋等待__read_lock_failed

asm(
".section .sched.text\n"
".align 4\n"
".globl __read_lock_failed\n"
"__read_lock_failed:\n\t"LOCK "incl      (%eax)\n"
"1:     rep; nop\n\t""cmpl   $1,(%eax)\n\t""js     1b\n\t"LOCK "decl      (%eax)\n\t""js     __read_lock_failed\n\t""ret"
);

1.函数整体功能

__read_lock_failed:

• 场景: 当尝试获取读锁失败时调用
• 目的: 自旋等待直到成功获取读锁
• 调用约定: 锁的地址通过eax寄存器传递

2.代码逐行分析

2.1. 恢复锁状态

LOCK "incl      (%eax)\n"

作用: 撤销之前失败的读锁获取尝试

执行前:

• 读锁尝试：lock--
• 但发现锁被写者占用（锁值 <= 0），需要恢复

执行后:

• lock++ 恢复锁计数
• 回到尝试获取之前的状态

2.2. 自旋等待循环

"1:     rep; nop\n\t"        // 短暂的暂停，节约功耗"cmpl   $1,(%eax)\n\t"   // 比较锁值和1"js     1b\n\t"          // 如果锁值 < 0，继续循环

等待条件: lock >= 0

• js（Jump if Sign）在结果为负时跳转
• 所以循环条件是：锁值 == 0（有写者持有锁）
• 退出条件是：锁值 > 0（没有写者）

2.3. 再次尝试获取读锁

LOCK "decl      (%eax)\n\t"   // 尝试获取读锁"js     __read_lock_failed\n\t"  // 如果失败，重试"ret"                 // 成功，返回

关键操作:

• lock-- 尝试减少锁计数（获取读锁）
• 如果结果 < 0：获取失败（写者出现），重新开始
• 如果结果 >= 0：获取成功，返回

十一、__build_read_lock

#define __build_read_lock_ptr(rw, helper)   \asm volatile(LOCK "subl $1,(%0)\n\t" \"jns 1f\n" \"call " helper "\n\t" \"1:\n" \::"a" (rw) : "memory")
#define __build_read_lock(rw, helper)   do { \if (__builtin_constant_p(rw)) \__build_read_lock_const(rw, helper); \else \__build_read_lock_ptr(rw, helper); \} while (0)

1.汇编代码详细解析

LOCK "subl $1,(%0)\n\t"    # 原子减1
"jns 1f\n"                  # 如果结果非负，跳转到标签1
"call " helper "\n\t"       # 否则调用helper函数
"1:\n"                      # 标签1：继续执行

情况1: 快速路径（获取成功）

初始: lock > 0 (没有写者持有锁)
执行: lock -= 1 → lock >= 0
条件: 结果非负 → jns跳转 → 跳过call指令
结果: 直接成功，无需函数调用

情况2: 慢速路径（获取失败）

初始: lock == 0 (写者持有锁)
执行: lock -= 1 → lock < 0 (负)
条件: 结果为负 → jns不跳转 → 执行call指令
结果: 调用helper函数处理竞争

十二、_raw_read_lock

static inline void _raw_read_lock(rwlock_t *rw)
{
#ifdef CONFIG_DEBUG_SPINLOCKBUG_ON(rw->magic != RWLOCK_MAGIC);
#endif__build_read_lock(rw, "__read_lock_failed");
}

• 调用__build_read_lock
• 指定获取失败时调用__read_lock_failed函数

十三、_read_lock

void __lockfunc _read_lock(rwlock_t *lock)
{preempt_disable();_raw_read_lock(lock);
}

• 禁用抢占
• 获取读锁

十四、解除读锁_read_unlock

#define _raw_read_unlock(rw)		asm volatile("lock ; incl %0" :"=m" ((rw)->lock) : : "memory")
void __lockfunc _read_unlock(rwlock_t *lock)
{_raw_read_unlock(lock);preempt_enable();
}

_raw_read_unlock

• 给lock加1，恢复初始
• 启用抢占

十五、总结

通过确定读写锁的初始值RW_LOCK_BIAS为0x01000000，其次加一次写锁就直接减去0x01000000，而加一次读锁只减1，这样就可以写锁只能加一次，再加其他锁lock就变负了，而读锁可以加0x01000000这么多次