linux 内核同步互斥技术之自旋锁

自旋锁

自旋锁用于处理器之间的互斥，适合保护很短的临界区，并且不允许在临界区睡眠。申请自旋锁的时候，如果自旋锁被其他处理器占有，本处理器自旋等待（也称为忙等待）。
进程、软中断和硬中断都可以使用自旋锁。
目前内核的自旋锁是基于排队的自旋锁（ queued spinlock，也称为“FIFO ticket spinlock”），算法类似于银行柜台的排队叫号。
（1）锁拥有排队号和服务号，服务号是当前占有锁的进程的排队号。
（2）每个进程申请锁的时候，首先申请一个排队号，然后轮询锁的服务号是否等于自己的排队号，如果等于，表示自己占有锁，可以进入临界区，否则继续轮询。
（3）当进程释放锁时，把服务号加 1，下一个进程看到服务号等于自己的排队号，退出自旋，进入临界区。

自旋锁的定义如下：
include/linux/spinlock_types.h
typedef struct spinlock {
union {
struct raw_spinlock rlock;

#ifdef CONFIG_DEBUG_LOCK_ALLOC
# define LOCK_PADSIZE (offsetof(struct raw_spinlock, dep_map))
struct {
u8 __padding[LOCK_PADSIZE];
struct lockdep_map dep_map;
};
#endif
};
} spinlock_t;

typedef struct raw_spinlock {
arch_spinlock_t raw_lock;
#ifdef CONFIG_GENERIC_LOCKBREAK
unsigned int break_lock;
#endif
#ifdef CONFIG_DEBUG_SPINLOCK
unsigned int magic, owner_cpu;
void *owner;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
struct lockdep_map dep_map;
#endif
} raw_spinlock_t;
可以看到，数据类型 spinlock 对数据类型 raw_spinlock 做了封装， spinlock 和 raw_spinlock（原始自旋锁）有什么关系？
Linux 内核有一个实时内核分支（开启配置宏 CONFIG_PREEMPT_RT）来支持硬实时特性，内核主线只支持软实时。
对于没有打上实时内核补丁的内核， spinlock 只是封装 raw_spinlock，它们完全一样。如果打上实时内核补丁，那么 spinlock 使用实时互斥锁保护临界区，在临界区内可以被抢占和睡眠，但 raw_spinlock 还是自旋锁。
目前主线版本还没有合并实时内核补丁，说不定哪天就会合并进来，为了使代码可以兼容实时内核，最好坚持 3 个原则。
（1）尽可能使用 spinlock。
（2）绝对不允许被抢占和睡眠的地方，使用 raw_spinlock，否则使用 spinlock。
（3）如果临界区足够小，使用 raw_spinlock。

各种处理器架构需要自定义数据类型 arch_spinlock_t， ARM64 架构的定义如下：
arch/arm64/include/asm/spinlock_types.h
typedef struct {
#ifdef __AARCH64EB__
u16 next;
u16 owner;
#else
u16 owner;
u16 next;
#endif
} __aligned(4) arch_spinlock_t;
成员 next 是排队号，成员 owner 是服务号。

定义并且初始化静态自旋锁的方法如下：DEFINE_SPINLOCK(x);
在运行时动态初始化自旋锁的方法如下：spin_lock_init(x);

申请自旋锁的函数如下。
（1） void spin_lock(spinlock_t *lock);
申请自旋锁，如果锁被其他处理器占有，当前处理器自旋等待。
（2） void spin_lock_bh(spinlock_t *lock);
申请自旋锁，并且禁止当前处理器的软中断。
（3） void spin_lock_irq(spinlock_t *lock);
申请自旋锁，并且禁止当前处理器的硬中断。
（4） spin_lock_irqsave(lock, flags);
申请自旋锁，保存当前处理器的硬中断状态，并且禁止当前处理器的硬中断。
（5） int spin_trylock(spinlock_t *lock);
申请自旋锁，如果申请成功，返回 1；如果锁被其他处理器占有，当前处理器不等待，立即返回 0。
（6）int spin_is_locked(spinlock_t *lock)
判断自旋锁是否被占用（不自旋等待，占用返回非0、没占用则返回0）。

释放自旋锁的函数如下。
（1） void spin_unlock(spinlock_t *lock);
（2） void spin_unlock_bh(spinlock_t *lock);
释放自旋锁，并且开启当前处理器的软中断。
（3） void spin_unlock_irq(spinlock_t *lock);
释放自旋锁，并且开启当前处理器的硬中断。
（4） void spin_unlock_irqrestore(spinlock_t *lock, unsigned long flags);
释放自旋锁，并且恢复当前处理器的硬中断状态。

定义并且初始化静态原始自旋锁的方法如下：DEFINE_RAW_SPINLOCK(x);
在运行时动态初始化原始自旋锁的方法如下：raw_spin_lock_init (x);

申请原始自旋锁的函数如下。
（1） raw_spin_lock(lock)
申请原始自旋锁，如果锁被其他处理器占有，当前处理器自旋等待。
（2） raw_spin_lock_bh(lock)
申请原始自旋锁，并且禁止当前处理器的软中断。
（3） raw_spin_lock_irq(lock)
申请原始自旋锁，并且禁止当前处理器的硬中断。
（4） raw_spin_lock_irqsave(lock, flags)
申请原始自旋锁，保存当前处理器的硬中断状态，并且禁止当前处理器的硬中断。
（5） raw_spin_trylock(lock)
申请原始自旋锁，如果申请成功，返回 1；如果锁被其他处理器占有，当前处理器不等待，立即返回 0。

释放原始自旋锁的函数如下。
（1） raw_spin_unlock(lock)
（2） raw_spin_unlock_bh(lock)
释放原始自旋锁，并且开启当前处理器的软中断。
（3） raw_spin_unlock_irq(lock)
释放原始自旋锁，并且开启当前处理器的硬中断。
（4） raw_spin_unlock_irqrestore(lock, flags)
释放原始自旋锁，并且恢复当前处理器的硬中断状态。
在多处理器系统中，函数 spin_lock()负责申请自旋锁，其代码如下：
spin_lock() -> raw_spin_lock() -> _raw_spin_lock() -> __raw_spin_lock() -> do_raw_spin_lock() -> arch_spin_lock()
arch/arm64/include/asm/spinlock.h
1 static inline void arch_spin_lock(arch_spinlock_t *lock)
2 {
3 nsigned int tmp;
4 arch_spinlock_t lockval, newval;
5
6 sm volatile(
7 RM64_LSE_ATOMIC_INSN(
8 * LL/SC */
9 prfm pstl1strm, %3\n"
10 "1: ldaxr %w0, %3\n"
11 " add %w1, %w0, %w5\n"
12 " stxr %w2, %w1, %3\n"
13 " cbnz %w2, 1b\n",
14 /* 大系统扩展的原子指令 */
15 " mov %w2, %w5\n"
16 " ldadda %w2, %w0, %3\n"
17 __nops(3)
18 )
19
20 /* 我们得到锁了吗？ */
21 " eor %w1, %w0, %w0, ror #16\n"
22 " cbz %w1, 3f\n"
23 " sevl\n"
24 "2: wfe\n"
25 " ldaxrh %w2, %4\n"
26 " eor %w1, %w2, %w0, lsr #16\n"
27 " cbnz %w1, 2b\n"
28 /* 得到锁，临界区从这里开始*/
29 "3:"
30 : "=&r" (lockval), "=&r" (newval), "=&r" (tmp), "+Q" (*lock)
31 : "Q" (lock->owner), "I" (1 << TICKET_SHIFT)
32 : "memory");
33 }
第 7～18 行代码，申请排队号，然后把自旋锁的排队号加 1，这是一个原子操作，有两种实现方法。
1）第 9～13 行代码，使用指令 ldaxr（带有获取语义的独占加载）和 stxr（独占存储）实现，指令 ldaxr 带有获取语义，后面的加载/存储操作必须在指令 ldaxr之后被观察到。
2）第 15 行和第 16 行代码，如果处理器支持大系统扩展，那么使用带有获取语义的原子加法指令 ldadda 实现，指令 ldadda 带有获取语义，后面的加载/存储操作必须在指令 ldadda之后被观察到。
第 21 行和第 22 行代码，如果服务号等于当前进程的排队号，进入临界区。
第 24～27 行代码，如果服务号不等于当前进程的排队号，那么自旋等待。使用指令ldaxrh（带有获取语义的独占加载， h 表示 halfword，即 2 字节）读取服务号，指令 ldaxrh带有获取语义，后面的加载/存储操作必须在指令 ldaxrh 之后被观察到。
第 23 行代码， sevl（ send event local）指令的功能是发送一个本地事件，避免错过其他处理器释放自旋锁时发送的事件。
第 24 行代码， wfe（wait for event）指令的功能是使处理器进入低功耗状态，等待事件。

函数 spin_unlock()负责释放自旋锁，其代码如下
spin_unlock() -> raw_spin_unlock() -> _raw_spin_unlock() -> __raw_spin_unlock() ->do_raw_spin_unlock() -> arch_spin_unlock()
arch/arm64/include/asm/spinlock.h
1 static inline void arch_spin_unlock(arch_spinlock_t *lock)
2 {
3 nsigned long tmp;
4
5 sm volatile(ARM64_LSE_ATOMIC_INSN(
6 * LL/SC */
7 ldrh %w1, %0\n"
8 add %w1, %w1, #1\n"
9 stlrh %w1, %0",
10 /* 大系统扩展的原子指令 */
11 " mov %w1, #1\n"
12 " staddlh %w1, %0\n"
13 __nops(1))
14 : "=Q" (lock->owner), "=&r" (tmp)
15 :
16 : "memory");
17 }
把自旋锁的服务号加 1，有如下两种实现方法。
第 7～9 行代码，使用指令 ldrh（加载， h 表示 halfword，即 2 字节）和 stlrh（带有释放语义的存储）实现，指令 stlrh 带有释放语义，前面的加载/存储操作必须在指令 stlrh 之前被观察到。因为一次只能有一个进程进入临界区，所以只有一个进程把自旋锁的服务号加 1，不需要是原子操作。
第 11 行和第 12 行代码，如果处理器支持大系统扩展，那么使用带有释放语义的原子加法指令 staddlh 实现，指令 staddlh 带有释放语义，前面的加载/存储操作必须在指令 staddlh 之前被观察到。

在单处理器系统中，自旋锁是空的。
include/linux/spinlock_types_up.h
typedef struct { } arch_spinlock_t;
函数 spin_lock()只是禁止内核抢占。
spin_lock() -> raw_spin_lock() -> _raw_spin_lock()
include/linux/spinlock_api_up.h
#define _raw_spin_lock(lock) __LOCK(lock)
#define __LOCK(lock) \
do { preempt_disable(); ___LOCK(lock); } while (0)
#define ___LOCK(lock) \
do { __acquire(lock); (void)(lock); } while (0)

多CPU与单CPU的spin_lock使用上的区别：
1)   如果只要和其他CPU 互斥，就要用spin_lock/spin_unlock
2)   如果要和irq及其他CPU互斥，就要用 spin_lock_irq/spin_unlock_irq
3)   如果既要和irq及其他CPU互斥，又要保存 EFLAG的状态，就要用spin_lock_irqsave/spin_unlock_irqrestore
4)   如果要和bh及其他CPU互斥，就要用spin_lock_bh/spin_unlock_bh
5)   如果不需要和其他CPU互斥，只要和irq互斥，则用local_irq_disable/local_irq_enable
6)   如果不需要和其他CPU互斥，只要和bh互斥，则用local_bh_disable/local_bh_enable

值得指出的是，对同一个数据的互斥，在不同的内核执行路径中，所用的形式有可能不同(见下面的例子)。
1.有些情况下需要在访问共享资源时必须中断失效，而访问完后必须中断使能，这样的情形使用spin_lock_irq和spin_unlock_irq最好；

2.spin_lock_irqsave保存访问共享资源前的中断标志，然后失效中断；spin_unlock_irqrestore将恢复访问共享资源前的中断标志而不是直接使能中断；

3.如果被保护的共享资源只在进程上下文访问和软中断上下文访问，那么当在进程上下文访问共享资源时，可能被软中断打断，从而可能进入软中断上下文来对被保护的共享资源访问，因此对于这种情况，对共享资源的访问必须使用spin_lock_bh和 spin_unlock_bh来保护。当然使用spin_lock_irq和spin_unlock_irq以及spin_lock_irqsave和 spin_unlock_irqrestore也可以，它们失效了本地硬中断，失效硬中断隐式地也失效了软中断。但是使用spin_lock_bh和 spin_unlock_bh是最恰当的，它比其他两个快。如果被保护的共享资源只在进程上下文和tasklet或timer上下文访问，那么应该使用与上面情况相同的获得和释放锁的宏，因为tasklet和timer是用软中断实现的。

4.对tasklet和timer和互斥操作，如果被保护的共享资源只在一个tasklet或timer上下文访问，那么不需要任何自旋锁保护，因为同一个tasklet或timer只能在一个CPU上运行，即使是在SMP环境下也是如此;如果被保护的共享资源只在两个或多个tasklet或timer上下文访问，那么对共享资源的访问仅需要用spin_lock和spin_unlock来保护，不必使用_bh版本，因为当tasklet或timer运行时，不可能有其他tasklet或timer在当前CPU上运行。

5.spin_lock用于阻止在不同CPU上的执行单元对共享资源的同时访问以及不同进程上下文互相抢占导致的对共享资源的非同步访问，而中断失效和软中断失效却是为了阻止在同一CPU上软中断或中断对共享资源的非同步访问

实例

#include <linux/init.h>
#include <linux/module.h>
#include <linux/kthread.h>
#include <linux/spinlock.h>
#include <linux/delay.h>

DEFINE_SPINLOCK(sp_lock);
struct task_struct * task1;
struct task_struct * task2;
struct task_struct * task3;
int i = 100000;

int thread_print_first(void *p)
{
if(kthread_should_stop()){
return 0;
}

printk(KERN_ALERT"Hello World:first writer cpu=%d\n",raw_smp_processor_id());
spin_lock(&sp_lock);
printk(KERN_ALERT"Hello World:first writting cpu=%d\n",raw_smp_processor_id());
while(i--);
spin_unlock(&sp_lock);
printk(KERN_ALERT"Hello World:first written cpu=%d\n",raw_smp_processor_id());
do {
msleep(1000);
}while(!kthread_should_stop());
return 0;
}
int thread_print_second(void *p)
{
if(kthread_should_stop()){
return 0;
}
//msleep(1000);
printk(KERN_ALERT"Hello World:second writer cpu=%d\n",raw_smp_processor_id());
spin_lock(&sp_lock);
printk(KERN_ALERT"Hello World:second writting cpu=%d\n",raw_smp_processor_id());
while(i--);
spin_unlock(&sp_lock);
printk(KERN_ALERT"Hello World:second written cpu=%d\n",raw_smp_processor_id());
do {
msleep(1000);
}while(!kthread_should_stop());
return 0;
}

int thread_print_third(void *p)
{
if(kthread_should_stop()){
return 0;
}
//msleep(2000);
printk(KERN_ALERT"Hello World:third writer cpu=%d\n",raw_smp_processor_id());
spin_lock(&sp_lock);
printk(KERN_ALERT"Hello World:third writting cpu=%d\n",raw_smp_processor_id());
while(i--);
spin_unlock(&sp_lock);
printk(KERN_ALERT"Hello World:third written cpu=%d\n",raw_smp_processor_id());
do {
msleep(1000);
}while(!kthread_should_stop());
return 0;
}

static int hello_init(void)
{
printk(KERN_ALERT"Hello World enter\n");

task1 = kthread_create(thread_print_first,NULL,"first");
if(IS_ERR(task1))
{
printk(KERN_ALERT"kthread_create error!\n");
return -1;
}

task2 = kthread_create(thread_print_second,NULL,"second");
if(IS_ERR(task2))
{
printk(KERN_ALERT"kthread_create error!\n");
kthread_stop(task1);
return -1;
}

task3 = kthread_create(thread_print_third,NULL,"third");
if(IS_ERR(task3))
{
printk(KERN_ALERT"kthread_create error!\n");
kthread_stop(task1);
kthread_stop(task2);
return -1;
}

kthread_bind(task1,1);
kthread_bind(task2,0);
kthread_bind(task3,1);
wake_up_process(task1);
wake_up_process(task2);
wake_up_process(task3);

return 0;
}
static void hello_exit(void)
{
int ret;
if (!IS_ERR(task1)) {
ret = kthread_stop(task1);
printk("<<<<<<<<task1 exit, ret = %d\n", ret);
}

if (!IS_ERR(task2)) {
ret = kthread_stop(task2);
printk("<<<<<<<<task2 exit, ret = %d\n", ret);
}

if (!IS_ERR(task3)) {
ret = kthread_stop(task3);
printk("<<<<<<<<task3 exit, ret = %d\n", ret);
}