自旋锁spin_lock
自旋锁spin_lock
自旋锁的引入
原子变量适用在多核之间多单一共享变量进行互斥访问，如果要保护多个变量，并且这些变量之间有逻辑关系时，原子变量就不适用了。例如：常见的双向链表。假设有三个链表节点Ａ、Ｂ、Ｃ。需要将节点Ｂ插入节点Ａ、Ｃ之间。如果ＣＰＵ Ａ刚好将Ａ节点的后向指针指向Ｂ，但是还没有将Ｂ的后向指针指向Ｃ。此时ＣＰＵ Ｂ要遍历链表，这将会一个灾难性的后果。
如果共享数据段在中断上下文或者进程上下文被访问呢？ 如果在进程上下文被访问，完全可以使用信号量semaphore机制来实现互斥。如果在中断上下文被访问呢？ 就不能使用semaphore来实现互斥，因为semaphore会引起睡眠的。这时候就引入了spin_lock
spin_lock的实现思想
先说生活中一个示例，如果机智的你乘坐过火车的话，就一定知道早上6点-7点在火车上厕所的感受了。如果机智你的起来上厕所，发现一大堆人都等着上厕所，男女老少。接设你前面排了三个人，分别为A， B， C。 当A进入厕所之后，关闭了厕所的门，然后就会看见一个红灯亮着“有人“，这时候B，C和机智的你都在等待。当A出来后，B进去不到20s就出来了。然后进去了C，然后你就苦苦的在等待，一直在观察这什么时候红灯熄灭，这让机智的你等待了10min, 然后机智的你进去就10s搞定。好了关于生活的例子说完了，再回到spin_lock中。
可以将厕所当作临界区。A， B， C， 机智的你是四个cpu， 红灯是临界区时候有cpu进入状态。 当A进入临界区（厕所），然后就会将进入状态修改为忙（红灯亮），然后B，C以及机智的你都会判断当前状态，如果是忙，就等待，不忙就让B先进去，B进入之后同样的操作。
spin_lock早期代码分析
因为spin_lock在ARM平台上的实现策略发生过变化，所以先分析以前版本2.6.18的spin_lock。
主要是以SMP系统分析，后面会稍带分析UP系统。
<include/linux/spinlock.h>
----------------------------------------------------------
#define spin_lock(lock) _spin_lock(lock)
<kernel/spinlock.c>
--------------------------------------------------------
void __lockfunc _spin_lock(spinlock_t *lock)
{
preempt_disable();
spin_acquire(&lock->dep_map, 0, 0, _RET_IP_);
_raw_spin_lock(lock);
}其中preempt_disable()是用来关闭掉抢占的。如果系统中打开了CONFIG_PREEMPT该选项的话，就是用来关闭系统的抢占，如果没有开启相当于什么都没干，只是为了统一代码。至于这里为什么需要关闭抢占，在后面会说。
spin_acquire(&lock->dep_map, 0, 0, _RET_IP_);这段代码使用来调试使用的，没有系统没有开启CONFIG_DEBUG_LOCK_ALLOC配置的话，这样代码也相当于什么都没干。继续往下。
define _raw_spin_lock(lock) __raw_spin_lock(&(lock)->raw_lock)
static inline void __raw_spin_lock(raw_spinlock_t *lock)
{
unsigned long tmp;
__asm__ __volatile__(
"1: ldrex %0, [%1]\n"
" teq %0, #0\n"
" strexeq %0, %2, [%1]\n"
" teqeq %0, #0\n"
" bne 1b"
: "=&r" (tmp)
: "r" (&lock->lock), "r" (1)
: "cc");
smp_mb();
}回头看看spinlock_t变量的定义：
typedef struct {
raw_spinlock_t raw_lock;
} spinlock_t;
typedef struct {
volatile unsigned int lock;
} raw_spinlock_t;通过层层的调用，最后spinlock_t就是一个volatile unsigned int型变量。
汇编代码 | C语言 | 解释 |
|---|---|---|
1: ldrex %0, [%1] | tmp=lock->lock | 读取lock的状态赋值给tmp |
teq %0, #0 | if(tmp == 0) | 判断lock的状态是否为0。如果是0说明可以获得锁；如果不为0，说明自旋锁处于上锁状态，不能访问，执行bne 1b指令，跳到标号1处不停执行。 |
strexeq %0, %2, [%1] | lock->lock=1 | 使用常量1来更新锁的状态，并将执行结果放入到tmp中 |
teqeq %0, #0 | if(tmp == 0) | 用来判断tmp是否为0，如果为0，表明更新锁的状态成功；如果不为0表明锁的状态没哟更新成功，执行”bne 1b”，跳转到标号1继续执行。 |
早期spin_lock存在的不公平性
还是回到火车上上厕所的故事中，某天早上去上厕所，发现有一大堆的人都在排队。但是进去厕所的人已经进去了半个小时，后面的人已经开始等待不急了，有的谩骂起来，有人大喊憋不住了，机智你的刚好肚子疼，快憋不住了。刚好排在第一位是你的媳妇，然后你就插队立马上了厕所。你出来后，接着是你儿子，然后你全家。后面的人就一直等待了1个小时终于进入了厕所。
将这个现象转移到程序中就是，在现代多核的cpu中，因为每个cpu都有chach的存在，导致不需要去访问主存获取lock，所以当当前获取lock的cpu，释放锁后，使其他cpu的cache都失效，然后释放的锁在下一次就比较容易进入临界去，导致出现了不公平。
ticket机制原理
先看最新的spin_lock的结构体定义：
typedef struct spinlock {
struct raw_spinlock rlock;
} spinlock_t;
typedef struct raw_spinlock {
arch_spinlock_t raw_lock;
} raw_spinlock_t;
typedef struct {
union {
u32 slock;
struct __raw_tickets {
#ifdef __ARMEB__
u16 next;
u16 owner;
#else
u16 owner;
u16 next;
#endif
} tickets;
};
} arch_spinlock_t;在分析代码之前，还需要解释一下tickets中的owner和next的含义。详细可见提交： 546c2896a42202dbc7d02f7c6ec9948ac1bf511b
因为有cache的作用，导致本次释放lock的cpu在下一次就可以更快的获取锁。所以在ARMv6上引入了”票”算法来保证每个cpu都是像“FIFO“访问临界区。
还是说回到火车上厕所的事件，还是早上排队上厕所。这时候好多人都插队，导致没有熟人的人一直上不了厕所，于是火车管理员（虚拟的，只是为了讲解原理而已）出现了。火车管理员说“从现在开始不准插队，我来监督，所有人排位一队“。管理员站在厕所门口，让大家都按次序排队上厕所，这时候就没有人插队了。
将这个事件转移到程序中的ticket中。刚开始的时候临界区没有cpu进入，状态是空闲的。next和owner的值都是0，当cpu1进入临界区后。将next++, 当cpu1从临界区域执行完后，将owner++。这时候next和owner都为1,说明临界区没有cpu进入。这时候cpu2进入临界区，将next++, 然后cpu2好像干的活比较多，当cpu3进来后，next++，这时候next已经是3了，当cpu2执行完毕后，owner++，owner的值变为2, 表示让cpu2进入临界区，这就保障了各个cpu之间都是先来后到的执行。
ARM32 上spin_lock代码实现
static inline void arch_spin_lock(arch_spinlock_t *lock)
{
unsigned long tmp;
u32 newval;
arch_spinlock_t lockval;
prefetchw(&lock->slock);
__asm__ __volatile__(
"1: ldrex %0, [%3]\n"
" add %1, %0, %4\n"
" strex %2, %1, [%3]\n"
" teq %2, #0\n"
" bne 1b"
: "=&r" (lockval), "=&r" (newval), "=&r" (tmp)
: "r" (&lock->slock), "I" (1 << TICKET_SHIFT)
: "cc");
while (lockval.tickets.next != lockval.tickets.owner) {
wfe();
lockval.tickets.owner = ACCESS_ONCE(lock->tickets.owner);
}
smp_mb();
}汇编 | C语言 | 解释 |
|---|---|---|
1: ldrex %0, [%3] | lockval = lock | 读取锁的值赋值给lockval |
add %1, %0, %4 | newval = lockval + (1 << 16) | 将next++之后的值存在newval中 |
strex %2, %1, [%3] | lock = newval | 将新的值存在lock中，将是否成功结果存入在tmp中 |
teq %2, #0 | if(tmp == 0) | 判断上条指令是否成功，如果不成功执行”bne 1b”跳到标号1执行 |
while (lockval.tickets.next != lockval.tickets.owner) {
wfe();
lockval.tickets.owner = ACCESS_ONCE(lock->tickets.owner);
}当tickets中的next和owner不相等的时候，说明临界区在忙， 需要等待。然后cpu会执行wfe指令。当其他cpu忙完之后，会更新owner的值，如果owner的值如果与next值相同，那到next号的cpu执行。
ARM64 上spin_lock代码实现
static inline void arch_spin_lock(arch_spinlock_t *lock)
{
unsigned int tmp;
arch_spinlock_t lockval, newval;
asm volatile(
/* Atomically increment the next ticket. */
" prfm pstl1strm, %3\n"
"1: ldaxr %w0, %3\n"
" add %w1, %w0, %w5\n"
" stxr %w2, %w1, %3\n"
" cbnz %w2, 1b\n"
/* Did we get the lock? */
" eor %w1, %w0, %w0, ror #16\n"
" cbz %w1, 3f\n"
/*
* No: spin on the owner. Send a local event to avoid missing an
* unlock before the exclusive load.
*/
" sevl\n"
"2: wfe\n"
" ldaxrh %w2, %4\n"
" eor %w1, %w2, %w0, lsr #16\n"
" cbnz %w1, 2b\n"
/* We got the lock. Critical section starts here. */
"3:"
: "=&r" (lockval), "=&r" (newval), "=&r" (tmp), "+Q" (*lock)
: "Q" (lock->owner), "I" (1 << TICKET_SHIFT)
: "memory");
}汇编 | C语言 | 解释 |
|---|---|---|
prfm pstl1strm, %3 | 将lock变量读到cache,增加访问速度 | |
1: ldaxr %w0, %3 | lockval = lock | 将lock的值赋值给lockval |
add %w1, %w0, %w5 | newval=lockval + (1 << 16) | 将lock中的next++， 然后将结果赋值给newval |
stxr %w2, %w1, %3 | lock = newval | 将newval赋值给lock，同时将是否设置成功结果存放到tmp |
cbnz %w2, 1b | if（tmp != 0）goto 1 | 如果tmp不为0，跳到标号1执行 |
eor %w1, %w0, %w0, ror #16 | if(next == owner) | 判断next是否等于owner |
cbz %w1, 3f | if(newval == 0) | 进入临界区 |
2: wfe | 自旋等待 | |
ldaxrh %w2, %4 | tmp = lock->owner | 获取当前的Owner值存放在tmp中 |
eor %w1, %w2, %w0, lsr #16 | if(next == owner) | 判断next是否等于owner |
cbnz %w1, 2b | 如果不等跳到标号2自旋，负责进入临界区域 |
ARM64 上spin_unlock代码实现
static inline void arch_spin_unlock(arch_spinlock_t *lock)
{
asm volatile(
" stlrh %w1, %0\n"
: "=Q" (lock->owner)
: "r" (lock->owner + 1)
: "memory");
}解锁的操作相对简单，就是给owner执行加1的操作。