內(nèi)核中哪些地方會(huì)用到自旋鎖?看圖:

自旋鎖顧名思義,是一把自動(dòng)旋轉(zhuǎn)的鎖,這很像廁所里的鎖,進(jìn)入前標(biāo)記是綠色可用的,進(jìn)入格子間后,手一帶,里面的鎖轉(zhuǎn)個(gè)圈,外面標(biāo)記變成了紅色表示在使用,外面的只能等待.這是形象的比喻,但實(shí)際也是如此.

在多CPU核環(huán)境中，由于使用相同的內(nèi)存空間，存在對(duì)同一資源進(jìn)行訪問(wèn)的情況，所以需要互斥訪問(wèn)機(jī)制來(lái)保證同一時(shí)刻只有一個(gè)核進(jìn)行操作,自旋鎖就是這樣的一種機(jī)制。

自旋鎖是指當(dāng)一個(gè)線程在獲取鎖時(shí)，如果鎖已經(jīng)被其它CPU中的線程獲取，那么該線程將循環(huán)等待，并不斷判斷是否能夠成功獲取鎖，直到其它CPU釋放鎖后,等鎖CPU才會(huì)退出循環(huán)。

自旋鎖的設(shè)計(jì)理念是它僅會(huì)被持有非常短的時(shí)間,鎖只能被一個(gè)任務(wù)持有,而且持有自旋鎖的CPU是不可以進(jìn)入睡眠模式的,因?yàn)槠渌腃PU在等待鎖,為了防止死鎖上下文交換也是不允許的,是禁止發(fā)生調(diào)度的.

自旋鎖與互斥鎖比較類似，它們都是為了解決對(duì)共享資源的互斥使用問(wèn)題。無(wú)論是互斥鎖，還是自旋鎖，在任何時(shí)刻，最多只能有一個(gè)持有者。但是兩者在調(diào)度機(jī)制上略有不同，對(duì)于互斥鎖，如果鎖已經(jīng)被占用，鎖申請(qǐng)者會(huì)被阻塞；但是自旋鎖不會(huì)引起調(diào)用者阻塞，會(huì)一直循環(huán)檢測(cè)自旋鎖是否已經(jīng)被釋放。

雖然都是共享資源競(jìng)爭(zhēng),但自旋鎖強(qiáng)調(diào)的是CPU核間的競(jìng)爭(zhēng),而互斥量強(qiáng)調(diào)的是任務(wù)(包括同一CPU核)之間的競(jìng)爭(zhēng).

自旋鎖長(zhǎng)什么樣?

typedef struct Spinlock {//自旋鎖結(jié)構(gòu)體

        size_t      rawLock;//原始鎖
    #if (LOSCFG_KERNEL_SMP_LOCKDEP == YES) // 死鎖檢測(cè)模塊開(kāi)關(guān)
        UINT32      cpuid; //持有鎖的CPU
        VOID        *owner; //持有鎖任務(wù)
        const CHAR  *name; //鎖名稱
    #endif
    } SPIN_LOCK_S;

結(jié)構(gòu)體很簡(jiǎn)單,里面有個(gè)宏,用于死鎖檢測(cè),默認(rèn)情況下是關(guān)閉的.所以真正的被使用的變量只有rawLock一個(gè).但C語(yǔ)言代碼中找不到變量的變化過(guò)程,而是通過(guò)一段匯編代碼來(lái)實(shí)現(xiàn).看完本篇會(huì)明白也只能通過(guò)匯編代碼來(lái)實(shí)現(xiàn)自旋鎖.

自旋鎖使用流程

自旋鎖用于多CPU核的情況，解決的是CPU之間競(jìng)爭(zhēng)資源的問(wèn)題.使用流程很簡(jiǎn)單,三步走。

創(chuàng)建自旋鎖：使用LOS_SpinInit初始化自旋鎖，或者使用SPIN_LOCK_INIT初始化靜態(tài)內(nèi)存的自旋鎖。

申請(qǐng)自旋鎖：使用接口LOS_SpinLockLOS_SpinTrylockLOS_SpinLockSave申請(qǐng)指定的自旋鎖，申請(qǐng)成功就繼續(xù)往后執(zhí)行鎖保護(hù)的代碼；申請(qǐng)失敗在自旋鎖申請(qǐng)中忙等，直到申請(qǐng)到自旋鎖為止。

釋放自旋鎖：使用LOS_SpinUnlockLOS_SpinUnlockRestore接口釋放自旋鎖。鎖保護(hù)代碼執(zhí)行完畢后，釋放對(duì)應(yīng)的自旋鎖，以便其他核申請(qǐng)自旋鎖。

幾個(gè)關(guān)鍵函數(shù)

自旋鎖模塊由內(nèi)聯(lián)函數(shù)實(shí)現(xiàn),見(jiàn)于los_spinlock.h代碼不多,主要是三個(gè)函數(shù).

ArchSpinLock(&lock->rawLock);
ArchSpinTrylock(&lock->rawLock)
ArchSpinUnlock(&lock->rawLock);

可以說(shuō)掌握了它們就掌握了自旋鎖,但這三個(gè)函數(shù)全由匯編實(shí)現(xiàn).見(jiàn)于los_dispatch.S文件 因?yàn)橄盗衅延袃善v過(guò)匯編代碼,所以很容易理解這三段代碼.函數(shù)的參數(shù)由r0記錄,即r0保存了lock->rawLock的地址,拿鎖/釋放鎖是讓lock->rawLock在0,1切換 下面逐一說(shuō)明自旋鎖的匯編代碼.

ArchSpinLock 匯編代碼

    FUNCTION(ArchSpinLock)  @死守,非要拿到鎖
        mov     r1, #1      @r1=1
    1:                      @循環(huán)的作用,因SEV是廣播事件.不一定lock->rawLock的值已經(jīng)改變了
        ldrex   r2, [r0]    @r0 = &lock->rawLock, 即 r2 = lock->rawLock
        cmp     r2, #0      @r2和0比較
        wfene               @不相等時(shí),說(shuō)明資源被占用,CPU核進(jìn)入睡眠狀態(tài)
        strexeq r2, r1, [r0]@此時(shí)CPU被重新喚醒,嘗試令lock->rawLock=1,成功寫入則r2=0
        cmpeq   r2, #0      @再來(lái)比較r2是否等于0,如果相等則獲取到了鎖
        bne     1b          @如果不相等,繼續(xù)進(jìn)入循環(huán)
        dmb                 @用DMB指令來(lái)隔離，以保證緩沖中的數(shù)據(jù)已經(jīng)落實(shí)到RAM中
        bx      lr          @此時(shí)是一定拿到鎖了,跳回調(diào)用ArchSpinLock函數(shù)

看懂了這段匯編代碼就理解了自旋鎖實(shí)現(xiàn)的真正機(jī)制,為什么一定要用匯編來(lái)實(shí)現(xiàn). 因?yàn)镃PU寧愿睡眠也非拿要到鎖不可的, 注意這里可不是讓線程睡眠,而是讓CPU進(jìn)入睡眠狀態(tài),能讓CPU進(jìn)入睡眠的只能通過(guò)匯編實(shí)現(xiàn).C語(yǔ)言根本就寫不出讓CPU真正睡眠的代碼.

ArchSpinTrylock 匯編代碼

如果不看下面這段匯編代碼,你根本不可能知道 ArchSpinTrylock 和 ArchSpinLock的真正區(qū)別是什么.

    FUNCTION(ArchSpinTrylock)   @嘗試拿鎖,拿不到就撤
        mov     r1, #1          @r1=1
        mov     r2, r0          @r2 = r0       
        ldrex   r0, [r2]        @r2 = &lock->rawLock, 即 r0 = lock->rawLock
        cmp     r0, #0          @r0和0比較
        strexeq r0, r1, [r2]    @嘗試令lock->rawLock=1,成功寫入則r0=0,否則 r0 =1
        dmb                     @數(shù)據(jù)存儲(chǔ)隔離，以保證緩沖中的數(shù)據(jù)已經(jīng)落實(shí)到RAM中
        bx      lr              @跳回調(diào)用ArchSpinLock函數(shù)

比較兩段匯編代碼可知,ArchSpinTrylock即沒(méi)有循環(huán)也不會(huì)讓CPU進(jìn)入睡眠,直接返回了,而ArchSpinLock會(huì)睡了醒, 醒了睡,一直守到丈夫(lock->rawLock = 0的廣播事件發(fā)生)回來(lái)才肯罷休. 筆者代碼注釋到這里那真是心潮澎湃,心碎了老一地, 真想給ArchSpinLock立一個(gè)貞節(jié)牌坊!

ArchSpinUnlock 匯編代碼

    FUNCTION(ArchSpinUnlock)    @釋放鎖
        mov     r1, #0          @r1=0               
        dmb                     @數(shù)據(jù)存儲(chǔ)隔離，以保證緩沖中的數(shù)據(jù)已經(jīng)落實(shí)到RAM中
        str     r1, [r0]        @令lock->rawLock = 0
        dsb                     @數(shù)據(jù)同步隔離
        sev                     @給各CPU廣播事件,喚醒沉睡的CPU們
        bx      lr              @跳回調(diào)用ArchSpinLock函數(shù)

代碼中涉及到幾個(gè)不常用的匯編指令,一一說(shuō)明:

匯編指令之 WFI / WFE / SEV

WFI(Wait for interrupt):等待中斷到來(lái)指令.WFI一般用于cpuidle,WFI 指令是在處理器發(fā)生中斷或類似異常之前不需要做任何事情。

在鴻蒙源碼分析系列篇(總目錄)線程篇中已說(shuō)過(guò),每個(gè)CPU都有自己的idle任務(wù),CPU沒(méi)事干的時(shí)候就待在里面,就一個(gè)死循環(huán)守著WFI指令,有中斷來(lái)了就觸發(fā)CPU起床干活. 中斷分硬中斷和軟中斷,系統(tǒng)調(diào)用就是通過(guò)軟中斷實(shí)現(xiàn)的,而設(shè)備類的就屬于硬中斷,都能觸發(fā)CPU干活. 具體看下CPU空閑的時(shí)候在干嘛,代碼超級(jí)簡(jiǎn)單:

LITE_OS_SEC_TEXT WEAK VOID OsIdleTask(VOID) //CPU沒(méi)事干的時(shí)候待在這里
{
    while (1) {//只有一個(gè)死循環(huán)
        Wfi();//WFI指令:arm core 立即進(jìn)入low-power standby state，等待中斷，進(jìn)入休眠模式。
    }
}

WFE(Wait for event):等待事件的到來(lái)指令WFE指令是在SEV指令生成事件之前不需要執(zhí)行任何操作,所以用WFE的地方,后續(xù)一定會(huì)對(duì)應(yīng)一個(gè)SEV的指令去喚醒它. WFE的一個(gè)典型使用場(chǎng)景，是用在自旋鎖中,spinlock的功能，是在不同CPU core之間，保護(hù)共享資源。使用WFE的流程是：

開(kāi)始之初資源空閑

CPU核1 訪問(wèn)資源，持有鎖，獲得資源

CPU核2 訪問(wèn)資源，此時(shí)資源不空閑，執(zhí)行WFE指令，讓core進(jìn)入low-power state(睡眠)

CPU核1 釋放資源，釋放鎖，釋放資源，同時(shí)執(zhí)行SEV指令，喚醒CPU核2

CPU核2 獲得資源

另外說(shuō)一下 以往的自旋鎖，在獲得不到資源時(shí)，讓CPU核進(jìn)入死循環(huán)，而通過(guò)插入WFE指令，則大大節(jié)省功耗.

SEV(send event):發(fā)送事件指令,SEV是一條廣播指令,它會(huì)將事件發(fā)送到多處理器系統(tǒng)中的所有處理器,以喚醒沉睡的CPU.

SEV和WFE的實(shí)現(xiàn)很像設(shè)計(jì)模式的觀察者模式.

匯編指令之 LDREX / STREX

LDREX用來(lái)讀取內(nèi)存中的值，并標(biāo)記對(duì)該段內(nèi)存的獨(dú)占訪問(wèn)：

LDREX Rx, [Ry]上面的指令意味著，讀取寄存器Ry指向的4字節(jié)內(nèi)存值，將其保存到Rx寄存器中，同時(shí)標(biāo)記對(duì)Ry指向內(nèi)存區(qū)域的獨(dú)占訪問(wèn)。

如果執(zhí)行LDREX指令的時(shí)候發(fā)現(xiàn)已經(jīng)被標(biāo)記為獨(dú)占訪問(wèn)了，并不會(huì)對(duì)指令的執(zhí)行產(chǎn)生影響。

而STREX在更新內(nèi)存數(shù)值時(shí)，會(huì)檢查該段內(nèi)存是否已經(jīng)被標(biāo)記為獨(dú)占訪問(wèn)，并以此來(lái)決定是否更新內(nèi)存中的值：

STREX Rx, Ry, [Rz]如果執(zhí)行這條指令的時(shí)候發(fā)現(xiàn)已經(jīng)被標(biāo)記為獨(dú)占訪問(wèn)了，則將寄存器Ry中的值更新到寄存器Rz指向的內(nèi)存，并將寄存器Rx設(shè)置成0。指令執(zhí)行成功后，會(huì)將獨(dú)占訪問(wèn)標(biāo)記位清除。

而如果執(zhí)行這條指令的時(shí)候發(fā)現(xiàn)沒(méi)有設(shè)置獨(dú)占標(biāo)記，則不會(huì)更新內(nèi)存，且將寄存器Rx的值設(shè)置成1。

一旦某條STREX指令執(zhí)行成功后，以后再對(duì)同一段內(nèi)存嘗試使用STREX指令更新的時(shí)候，會(huì)發(fā)現(xiàn)獨(dú)占標(biāo)記已經(jīng)被清空了，就不能再更新了，從而實(shí)現(xiàn)獨(dú)占訪問(wèn)的機(jī)制。

編程實(shí)例

本實(shí)例實(shí)現(xiàn)如下流程。

任務(wù)Example_TaskEntry初始化自旋鎖，創(chuàng)建兩個(gè)任務(wù)Example_SpinTask1、Example_SpinTask2，分別運(yùn)行于兩個(gè)核。

Example_SpinTask1、Example_SpinTask2中均執(zhí)行申請(qǐng)自旋鎖的操作，同時(shí)為了模擬實(shí)際操作，在持有自旋鎖后進(jìn)行延遲操作，最后釋放自旋鎖。

300Tick后任務(wù)Example_TaskEntry被調(diào)度運(yùn)行，刪除任務(wù)Example_SpinTask1和Example_SpinTask2。

#include "los_spinlock.h"
#include "los_task.h"

/* 自旋鎖句柄id */
SPIN_LOCK_S g_testSpinlock;
/* 任務(wù)ID */
UINT32 g_testTaskId01;
UINT32 g_testTaskId02;

VOID Example_SpinTask1(VOID)
{
    UINT32 i;
    UINTPTR intSave;

    /* 申請(qǐng)自旋鎖 */
    dprintf("task1 try to get spinlock\n");
    LOS_SpinLockSave(&g_testSpinlock, &intSave);
    dprintf("task1 got spinlock\n");
    for(i = 0; i < 5000; i++) {
        asm volatile("nop");
    }

    /* 釋放自旋鎖 */
    dprintf("task1 release spinlock\n");
    LOS_SpinUnlockRestore(&g_testSpinlock, intSave);

    return;
}

VOID Example_SpinTask2(VOID)
{
    UINT32 i;
    UINTPTR intSave;

    /* 申請(qǐng)自旋鎖 */
    dprintf("task2 try to get spinlock\n");
    LOS_SpinLockSave(&g_testSpinlock, &intSave);
    dprintf("task2 got spinlock\n");
    for(i = 0; i < 5000; i++) {
        asm volatile("nop");
    }

    /* 釋放自旋鎖 */
    dprintf("task2 release spinlock\n");
    LOS_SpinUnlockRestore(&g_testSpinlock, intSave);

    return;
}

UINT32 Example_TaskEntry(VOID)
{
    UINT32 ret;
    TSK_INIT_PARAM_S stTask1;
    TSK_INIT_PARAM_S stTask2;

    /* 初始化自旋鎖 */
    LOS_SpinInit(&g_testSpinlock);

    /* 創(chuàng)建任務(wù)1 */
    memset(&stTask1, 0, sizeof(TSK_INIT_PARAM_S));
    stTask1.pfnTaskEntry  = (TSK_ENTRY_FUNC)Example_SpinTask1;
    stTask1.pcName        = "SpinTsk1";
    stTask1.uwStackSize   = LOSCFG_TASK_MIN_STACK_SIZE;
    stTask1.usTaskPrio    = 5;
#ifdef LOSCFG_KERNEL_SMP
    /* 綁定任務(wù)到CPU0運(yùn)行 */
    stTask1.usCpuAffiMask = CPUID_TO_AFFI_MASK(0);
#endif
    ret = LOS_TaskCreate(&g_testTaskId01, &stTask1);
    if(ret != LOS_OK) {
        dprintf("task1 create failed .\n");
        return LOS_NOK;
    }

    /* 創(chuàng)建任務(wù)2 */
    memset(&stTask2, 0, sizeof(TSK_INIT_PARAM_S));
    stTask2.pfnTaskEntry = (TSK_ENTRY_FUNC)Example_SpinTask2;
    stTask2.pcName       = "SpinTsk2";
    stTask2.uwStackSize  = LOSCFG_TASK_MIN_STACK_SIZE;
    stTask2.usTaskPrio   = 5;
#ifdef LOSCFG_KERNEL_SMP
    /* 綁定任務(wù)到CPU1運(yùn)行 */
    stTask1.usCpuAffiMask = CPUID_TO_AFFI_MASK(1);
#endif
    ret = LOS_TaskCreate(&g_testTaskId02, &stTask2);
    if(ret != LOS_OK) {
        dprintf("task2 create failed .\n");
        return LOS_NOK;
    }

    /* 任務(wù)休眠300Ticks */
    LOS_TaskDelay(300);

    /* 刪除任務(wù)1 */
    ret = LOS_TaskDelete(g_testTaskId01);
    if(ret != LOS_OK) {
        dprintf("task1 delete failed .\n");
        return LOS_NOK;
    }
    /* 刪除任務(wù)2 */
    ret = LOS_TaskDelete(g_testTaskId02);
    if(ret != LOS_OK) {
        dprintf("task2 delete failed .\n");
        return LOS_NOK;
    }

    return LOS_OK;
}
運(yùn)行結(jié)果

task2 try to get spinlock
task2 got spinlock
task1 try to get spinlock
task2 release spinlock
task1 got spinlock
task1 release spinlock

總結(jié)

自旋鎖用于解決CPU核間競(jìng)爭(zhēng)資源的問(wèn)題

因?yàn)樽孕i會(huì)讓CPU陷入睡眠狀態(tài),所以鎖的代碼不能太長(zhǎng),否則容易導(dǎo)致意外出現(xiàn),也影響性能.

必須由匯編代碼實(shí)現(xiàn),因?yàn)镃語(yǔ)言寫不出讓CPU進(jìn)入真正睡眠,核間競(jìng)爭(zhēng)的代碼.

編輯：hfy