Linux 自旋锁的实现

1.概述

2.Linux的原子变量

Linux 提供了一个原子类型变量 atomic_t

typedef struct {int counter;
} atomic_t;#define ATOMIC_INIT(i) { (i) }#ifdef CONFIG_64BIT
typedef struct {s64 counter;
} atomic64_t;
#endif

上述代码自然不能用普通的代码去读写加减，而是要用 Linux 专门提供的接口函数去操作，否则就不能保证原子性了，代码如下。

//原子读取变量中的值
static __always_inline int arch_atomic_read(const atomic_t *v)
{return __READ_ONCE((v)->counter);
}
//原子写入一个具体的值
static __always_inline void arch_atomic_set(atomic_t *v, int i)
{__WRITE_ONCE(v->counter, i);
}
//原子加上一个具体的值
static __always_inline void arch_atomic_add(int i, atomic_t *v)
{asm volatile(LOCK_PREFIX "addl %1,%0": "+m" (v->counter): "ir" (i) : "memory");
}
//原子减去一个具体的值
static __always_inline void arch_atomic_sub(int i, atomic_t *v)
{asm volatile(LOCK_PREFIX "subl %1,%0": "+m" (v->counter): "ir" (i) : "memory");
}
//原子加1
static __always_inline void arch_atomic_inc(atomic_t *v)
{asm volatile(LOCK_PREFIX "incl %0": "+m" (v->counter) :: "memory");
}
//原子减1
static __always_inline void arch_atomic_dec(atomic_t *v)
{asm volatile(LOCK_PREFIX "decl %0": "+m" (v->counter) :: "memory");
}

__READ_ONCE，__WRITE_ONCE 两个宏，我们来看看它们分别做了什么，如下所示。

#define __READ_ONCE(x)  \
(*(const volatile __unqual_scalar_typeof(x) *)&(x))
#define __WRITE_ONCE(x, val) \
do {*(volatile typeof(x) *)&(x) = (val);} while (0)
//__unqual_scalar_typeof表示声明一个非限定的标量类型，非标量类型保持不变。说人话就是返回x变量的类型，这是GCC的功能，typeof只是纯粹返回x的类型。
//如果 x 是int类型则返回“int” 
#define __READ_ONCE(x)  \
(*(const volatile int *)&(x))
#define __WRITE_ONCE(x, val) \
do {*(volatile int *)&(x) = (val);} while (0) 

结合刚才的代码，我给你做个解读。Linux 定义了 __READ_ONCE，__WRITE_ONCE 这两个宏，是对代码封装并利用 GCC 的特性对代码进行检查，把让错误显现在编译阶段。其中的“volatile int *”是为了提醒编译器：这是对内存地址读写，不要有优化动作，每次都必须强制写入内存或从内存读取。

3.Linux 控制中断

Linux 控制 CPU 响应中断的函数如下

//实际保存eflags寄存器
extern __always_inline unsigned long native_save_fl(void){unsigned long flags;asm volatile("# __raw_save_flags\n\t""pushf ; pop %0":"=rm"(flags)::"memory");return flags;
}
//实际恢复eflags寄存器
extern inline void native_restore_fl(unsigned long flags){asm volatile("push %0 ; popf"::"g"(flags):"memory","cc");
}
//实际关中断
static __always_inline void native_irq_disable(void){asm volatile("cli":::"memory");
}
//实际开启中断
static __always_inline void native_irq_enable(void){asm volatile("sti":::"memory");
}
//arch层关中断
static __always_inline void arch_local_irq_disable(void){native_irq_disable();
}
//arch层开启中断
static __always_inline void arch_local_irq_enable(void){ native_irq_enable();
}
//arch层保存eflags寄存器
static __always_inline unsigned long           arch_local_save_flags(void){return native_save_fl();
}
//arch层恢复eflags寄存器
static  __always_inline void arch_local_irq_restore(unsigned long flags){native_restore_fl(flags);
}
//实际保存eflags寄存器并关中断
static __always_inline unsigned long arch_local_irq_save(void){unsigned long flags = arch_local_save_flags();arch_local_irq_disable();return flags;
}
//raw层关闭开启中断宏
#define raw_local_irq_disable()     arch_local_irq_disable()
#define raw_local_irq_enable()      arch_local_irq_enable()
//raw层保存恢复eflags寄存器宏
#define raw_local_irq_save(flags)           \do {                        \typecheck(unsigned long, flags);    \flags = arch_local_irq_save();      \} while (0)#define raw_local_irq_restore(flags)            \do {                        \typecheck(unsigned long, flags);    \arch_local_irq_restore(flags);      \} while (0)#define raw_local_save_flags(flags)         \do {                        \typecheck(unsigned long, flags);    \flags = arch_local_save_flags();    \} while (0)
//通用层接口宏 
#define local_irq_enable()              \do { \raw_local_irq_enable();         \} while (0)#define local_irq_disable()             \do {                        \raw_local_irq_disable();        \} while (0)#define local_irq_save(flags)               \do {                        \raw_local_irq_save(flags);      \} while (0)#define local_irq_restore(flags)            \do {                        \raw_local_irq_restore(flags);       \} while (0)

带 native_ 前缀之类的函数则跟我们之前实现的 hal_ 前缀对应，而 Linux 为了支持不同的硬件平台，做了多层封装。

Linux 原始自旋锁

Linux 的原始自旋锁本质上用一个整数来表示，值为 1 代表锁未被占用，为 0 或者负数则表示被占用。你可以结合上节课的这张图，理解后面的内容。当某个 CPU 核心执行进程请求加锁时，如果锁是未加锁状态，则加锁，然后操作共享资源，最后释放锁；如果锁已被加锁，则进程并不会转入睡眠状态，而是循环等待该锁，一旦锁被释放，则第一个感知此信息的进程将获得锁。

Linux 原始自旋锁的数据结构

//最底层的自旋锁数据结构
typedef struct{
volatile unsigned long lock;//真正的锁值变量，用volatile标识
}spinlock_t;

Linux 原始自旋锁数据结构封装了一个 unsigned long 类型的变量。有了数据结构，我们再来看看操作这个数据结构的函数，即自旋锁接口，代码如下。

#define spin_unlock_string \  "movb $1,%0" \ //写入1表示解锁:"=m" (lock->lock) : : "memory"#define spin_lock_string \"\n1:\t" \  "lock ; decb %0\n\t" \ //原子减1"js 2f\n" \    //当结果小于0则跳转到标号2处，表示加锁失败".section .text.lock,\"ax\"\n" \ //重新定义一个代码段，这是优化技术，避免后面的代码填充cache，因为大部分情况会加锁成功，链接器会处理好这个代码段的"2:\t" \  "cmpb $0,%0\n\t" \  //和0比较"rep;nop\n\t" \  //空指令"jle 2b\n\t" \   //小于或等于0跳转到标号2"jmp 1b\n" \   //跳转到标号1  ".previous"
//获取自旋锁
static inline void spin_lock(spinlock_t*lock){__asm__ __volatile__(spin_lock_string:"=m"(lock->lock)::"memory");
}
//释放自旋锁
static inline void spin_unlock(spinlock_t*lock){
__asm__ __volatile__(spin_unlock_string);
}

上述代码中用 spin_lock_string、spin_unlock_string 两个宏，定义了获取、释放自旋锁的汇编指令。spin_unlock_string 只是简单将锁值变量设置成 1，表示释放自旋锁，spin_lock_string 中并没有像我们 Cosmos 一样使用 xchg 指令，而是使用了 decb 指令，这条指令也能原子地执行减 1 操作。开始锁值变量为 1 时，执行 decb 指令就变成了 0，0 就表示加锁成功。如果小于 0，则表示有其它进程已经加锁了，就会导致循环比较。

Linux 排队自旋锁

现在我们再来看看 100 个进程获取同一个自旋锁的情况，开始 1 个进程获取了自旋锁 L，后面继续来了 99 个进程，它们都要获取自旋锁 L，但是它们必须等待，这时第 1 进程释放了自旋锁 L。请问，这 99 个进程中谁能先获取自旋锁 L 呢？答案是不确定，因为这个次序依赖于哪个 CPU 核心能最先访问内存，而哪个 CPU 核心可以访问内存是由总线仲裁协议决定的。很有可能最后来的进程最先获取自旋锁 L，这对其它等待的进程极其不公平，为了解决获取自旋锁的公平性，Linux 开发出了排队自旋锁。你可以这样理解，想要给进程排好队，就需要确定顺序，也就是进程申请获取锁的先后次序，Linux 的排队自旋锁通过保存这个信息，就能更公平地调度进程了。为了保存顺序信息，排队自旋锁重新定义了数据结构。

//RAW层的自旋锁数据结构
typedef struct raw_spinlock{unsigned int slock;//真正的锁值变量
}raw_spinlock_t;
//最上层的自旋锁数据结构
typedef struct spinlock{struct raw_spinlock rlock;
}spinlock_t;
//Linux没有这样的结构，这只是为了描述方便
typedef struct raw_spinlock{union {unsigned int slock;//真正的锁值变量struct {u16 owner;u16 next;}}
}raw_spinlock_t;

slock 域被分成两部分，分别保存锁持有者和未来锁申请者的序号，如上述代码 10～16 行所示。只有 next 域与 owner 域相等时，才表示自旋锁处于未使用的状态（此时也没有进程申请该锁）。在排队自旋锁初始化时，slock 被置为 0，即 next 和 owner 被置为 0，Linux 进程执行申请自旋锁时，原子地将 next 域加 1，并将原值返回作为自己的序号。

如果返回的序号等于申请时的 owner 值，说明自旋锁处于未使用的状态，则进程直接获得锁；否则，该进程循环检查 owner 域是否等于自己持有的序号，一旦相等，则表明锁轮到自己获取。进程释放自旋锁时，原子地将 owner 域加 1 即可，下一个进程将会发现这一变化，从循环状态中退出。进程将严格地按照申请顺序依次获取排队自旋锁。这样一来，原先进程无序竞争的乱象就迎刃而解了。

static inline void __raw_spin_lock(raw_spinlock_t*lock){
int inc = 0x00010000;
int tmp;
__asm__ __volatile__(
"lock ; xaddl %0, %1\n" //将inc和slock交换，然后 inc=inc+slock//相当于原子读取next和owner并对next+1
"movzwl %w0, %2\n\t"//将inc的低16位做0扩展后送tmp tmp=(u16)inc
"shrl $16, %0\n\t" //将inc右移16位 inc=inc>>16
"1:\t"
"cmpl %0, %2\n\t" //比较inc和tmp，即比较next和owner 
"je 2f\n\t" //相等则跳转到标号2处返回
"rep ; nop\n\t" //空指令
"movzwl %1, %2\n\t" //将slock的低16位做0扩展后送tmp 即tmp=owner
"jmp 1b\n" //跳转到标号1处继续比较
"2:"
:"+Q"(inc),"+m"(lock->slock),"=r"(tmp)
::"memory","cc"
);
}
#define UNLOCK_LOCK_PREFIX LOCK_PREFIX
static inline void __raw_spin_unlock(raw_spinlock_t*lock){
__asm__ __volatile__(
UNLOCK_LOCK_PREFIX"incw %0"//将slock的低16位加1 即owner+1
:"+m"(lock->slock)
::"memory","cc");
}

上述代码中的注释已经描述得很清楚了，每条指令都有注解，供你参考。这里需要注意的是 Linux 为了避免差异性，在 spinlock_t 结构体中包含了 raw_spinlock_t，而在 raw_spinlock_t 结构体中并没使用 next 和 owner 字段，而是在代码中直接操作 slock 的高 16 位和低 16 位来实现的。

参考：LMOS 操作系