-
-
[原创]明明白白自旋锁
-
发表于: 2008-10-12 13:39
-
自某日看了iceboy和MJ0011关于多处理器同步的一些讨论，才发现原来我对自旋锁的理解也有一定错误，还好现在明白了~~为了加深理解，就深入分析了一下自旋锁的实现，一篇小小的自旋锁分析文章，献给大家。写得比较碎，欢迎各位大牛小牛指出错误~
一、自旋锁是什么？
先进行下简单科普，自旋锁是一种轻量级的多处理器间的同步机制。因此，自旋锁对于单处理器是没有实际意义的。它要求持有锁的处理器所占用的时间尽可能短，因为此时别的处理器正在高速运转并等待锁的释放，所以不能长时间占有。
曾经有个经典的例子来比喻自旋锁：A，B两个人合租一套房子，共用一个厕所，那么这个厕所就是共享资源，且在任一时刻最多只能有一个人在使用。当厕所闲置时，谁来了都可以使用，当A使用时，就会关上厕所门，而B也要使用，但是急啊，就得在门外焦急地等待，急得团团转，是为“自旋”，呵呵。这个比喻还算恰当吧，大家也明白为什么要求锁的持有时间尽量短了吧！尤其b4占着茅坑不拉屎的行为~~
二、操作系统如何实现自旋锁？
在Linux和Windows中都实现了自旋锁，下面我们就来看一看Windows下是如何实现的吧。
自旋锁的结构：
KSPIN_LOCK SpinLock；
KSPIN_LOCK实际是一个操作系统相关的无符号整数，32位系统上是32位的unsigned long，64位系统则定义为unsigned __int64。
在初始化时，其值被设置为0，为空闲状态。
参见WRK：
FORCEINLINE
VOID
NTAPI
KeInitializeSpinLock (
__out PKSPIN_LOCK SpinLock
)
{
*SpinLock = 0;
}
关于自旋锁的两个基本操作：获取和释放
VOID
KeAcquireSpinLock(
IN PKSPIN_LOCK SpinLock,
OUT PKIRQL OldIrql
);
VOID
KeReleaseSpinLock(
IN PKSPIN_LOCK SpinLock,
IN KIRQL NewIrql
);
获取时做了哪些工作呢？
Ntddk.h中是这样定义的：
#define KeAcquireSpinLock(SpinLock, OldIrql) \
*(OldIrql) = KeAcquireSpinLockRaiseToDpc(SpinLock)
很明显，核心的操作对象是SpinLock，同时也与IRQL有关。
再翻翻WRK，找到KeAcquireSpinLockRaiseToDpc的定义：
__forceinline
KIRQL
KeAcquireSpinLockRaiseToDpc (
__inout PKSPIN_LOCK SpinLock
)
{
KIRQL OldIrql;
//
// Raise IRQL to DISPATCH_LEVEL and acquire the specified spin lock.
//
OldIrql = KfRaiseIrql(DISPATCH_LEVEL);
KxAcquireSpinLock(SpinLock);
return OldIrql;
}
首先会提升IRQL到DISPATCH_LEVEL，然后调用KxAcquireSpinLock()。（若当前IRQL就是DISPATCH_LEVEL，那么就调用KeAcquireSpinLockAtDpcLevel，省去提升IRQL一步）。因为线程调度也是发生在DISPATCH_LEVEL，所以提升IRQL之后当前处理器上就不会发生线程切换。单处理器时，当前只能有一个线程被执行，而这个线程提升IRQL至DISPATCH_LEVEL之后又不会因为调度被切换出去，自然也可以实现我们想要的互斥“效果”，其实只操作IRQL即可，无需SpinLock。实际上单核系统的内核文件ntosknl.exe中导出的有关SpinLock的函数都只有一句话，就是return，呵呵。
而多处理器呢？提升IRQL只会影响到当前处理器，保证当前处理器的当前线程不被切换，那还得考虑其它处理器啊，继续看 KxAcquireSpinLock()函数吧。在WRK中找到的KxAcquireSpinLock()函数是Amd64位处理器上的代码（位于(\inc\private\ntos\inc\Amd64.h)中），32位x86的没找到。不过原理相通，一样可以参考
__forceinline
VOID
KxAcquireSpinLock (
__inout PKSPIN_LOCK SpinLock
)
{
if (InterlockedBitTestAndSet64((LONG64 *)SpinLock, 0))//64位函数
{
KxWaitForSpinLockAndAcquire(SpinLock); //只有声明没有定义的函数,应该是做了测试,等待的工作
}
}
InterlockedBitTestAndSet64()函数的32位版本如下：
ps:我汇编功底不太好，见谅~
BOOLEAN
FORCEINLINE
InterlockedBitTestAndSet (
IN LONG *Base,
IN LONG Bit
)
{
__asm {
mov eax, Bit
mov ecx, Base
lock bts [ecx], eax
setc al
};
}
关键就在bts指令，是一个进行位测试并置位的指令。，这里在进行关键的操作时有lock前缀，保证了多处理器安全。InterLockedXXX函数都有这个特点。显然，KxAcquireSpinLock（）函数先测试锁的状态。若锁空闲，则*SpinLock为0，那么InterlockedBitTestAndSet()将返回0，并使*SpinLock置位，不再为0。这样KxAcquireSpinLock()就成功得到了锁，并设置锁为占用状态（*SpinLock不为0），函数返回。若锁已被占用呢？InterlockedBitTestAndSet()将返回1，此时将调用KxWaitForSpinLockAndAcquire()等待并获取这个锁。这表明，SPIN_LOCK为0则锁空闲，非0则已被占有。
由于WRK中仅有KxWaitForSpinLockAndAcquire（）的声明而无定义，我们只能从名字猜测其做了什么。在WRK中看到了这两个函数：
__forceinline
BOOLEAN
KxTryToAcquireSpinLock (
__inout PKSPIN_LOCK SpinLock
)
{
if (*(volatile LONG64 *)SpinLock == 0)
{
return !InterlockedBitTestAndSet64((LONG64 *)SpinLock, 0);
}
else
{
KeYieldProcessor();
return FALSE;
}
}
从名字看应该是试图获取自旋锁，先判断锁是否被占有。若空闲，则设置其为占用状态，这就成功地抢占了。若已被占用，则调用KeYieldProcessor()，这个函数其实只是一个宏：
#define KeYieldProcessor() __asm { rep nop } //空转
都知道nop干啥的，CPU就是在空转进行等待而已。
下面这个函数则是仅测试自旋锁的状态：
__forceinline
BOOLEAN
KeTestSpinLock (
__in PKSPIN_LOCK SpinLock
)
{
KeMemoryBarrierWithoutFence();//这个函数我也不知道干啥的
if (*SpinLock != 0) {
KeYieldProcessor();//若被占用，则空转
return FALSE;
} else {
return TRUE;
}
}
好，看了获取部分，再看看释放锁的时候做了什么。
__forceinline
VOID
KeReleaseSpinLock (
__inout PKSPIN_LOCK SpinLock,
__in KIRQL OldIrql
)
{
KxReleaseSpinLock(SpinLock);//先释放锁
KeLowerIrql(OldIrql);//恢复至原IRQL
return;
}
继续看KxReleaseSpinLock（）
__forceinline
VOID
KxReleaseSpinLock (
__inout PKSPIN_LOCK SpinLock
)
{
InterlockedAnd64((LONG64 *)SpinLock, 0);//释放时进行与操作设置其为0
}
好了，对于自旋锁的初始化、获取、释放，都有了了解。但是只是谈谈原理，看看WRK，似乎有种纸上谈兵的感觉？那就实战一下，看看真实系统中是如何实现的。以双核系统中XP SP2下内核中关于SpinLock的实现细节为例：
用IDA分析双核系统的内核文件ntkrnlpa.exe，关于自旋锁操作的两个基本函数是KiAcquireSpinLock和KiReleaseSpinLock，其它几个类似。
.text:004689C0 KiAcquireSpinLock proc near ; CODE XREF:
sub_416FEE+2D p
.text:004689C0 ; sub_4206C0+5 j ...
.text:004689C0 lock bts dword ptr [ecx], 0
.text:004689C5 jb short loc_4689C8
.text:004689C7 retn
.text:004689C8 ; ---------------------------------------------------------------------------
.text:004689C8
.text:004689C8 loc_4689C8: ; CODE XREF: KiAcquireSpinLock+5 j
.text:004689C8 ; KiAcquireSpinLock+12 j
.text:004689C8 test dword ptr [ecx], 1
.text:004689CE jz short KiAcquireSpinLock
.text:004689D0 pause
.text:004689D2 jmp short loc_4689C8
.text:004689D2 KiAcquireSpinLock endp
代码比较简单，还原成源码是这样子的（偷懒用了F5）：
void __fastcall KiAcquireSpinLock(int _ECX)
{
while ( 1 )
{
__asm { lock bts dword ptr [ecx], 0 }
if ( !_CF )
break;
while ( *(_DWORD *)_ECX & 1 )
__asm { pause }//应是rep nop，IDA将其翻译成pause
}
}
fastcall方式调用，参数KSPIN_LOCK在ECX中，可以看到是一个死循环，先测试其是否置位，若否，则CF将置0，并将ECX置位，即获取锁的操作成功；若是，即锁已被占有，则一直对其进行测试并进入空转状态，这和前面分析的完全一致，只是代码似乎更精炼了一点，毕竟是实用的玩意嘛。
再来看看释放时：
.text:004689E0 public KiReleaseSpinLock
.text:004689E0 KiReleaseSpinLock proc near ; CODE XREF: sub_41702E+E p
.text:004689E0 ; sub_4206D0+5 j ...
.text:004689E0 mov byte ptr [ecx], 0
.text:004689E3 retn
.text:004689E3 KiReleaseSpinLock endp
这个再清楚不过了，直接设置为0就代表了将其释放，此时那些如虎狼般疯狂空转的其它处理器将马上获知这一信息，于是，下一个获取、释放的过程开始了。这就是最基本的自旋锁，其它一些自旋锁形式是对这种基本形式的扩充。比如排队自旋锁，是为了解决多处理器竞争时的无序状态等等，不多说了。
现在对自旋锁可谓真的是明明白白了，之前我犯的错误就是以为用了自旋锁就能保证多核同步，其实不是的，用自旋锁来保证多核同步的前提是大家都要用这个锁。若当前处理器已占有自旋锁，只有别的处理器也来请求这个锁时，才会进入空转，不进行别的操作，这时你的操作将不会受到干扰。但是假如某个需要互斥的操作只有你这个线程才做而别人根本不去做（以iceboy的安全实现Inline Hook为例，请求锁、修改代码、释放锁的过程只有这个线程才会做，别的处理器上的
线程如果要执行这里还是照样执行，人家又不用修改），所以人家不请求锁时还是该干嘛干嘛啊，自己在那儿自旋只是一厢情愿。所以MJ说“别人都不跟你旋，你自己旋个头啊”，经典经典…….
[培训]内核驱动高级班，冲击BAT一流互联网大厂工作，每周日13:00-18:00直播授课
|
|
|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
