因為現(xiàn)代操作系統(tǒng)是多處理器計算的架構(gòu)，必然更容易遇到多個進(jìn)程，多個線程訪問共享數(shù)據(jù)的情況，如下圖所示：

圖中每一種顏色代表一種競態(tài)情況，主要歸結(jié)為三類：

進(jìn)程與進(jìn)程之間：單核上的搶占，多核上的SMP；

進(jìn)程與中斷之間：中斷又包含了上半部與下半部，中斷總是能打斷進(jìn)程的執(zhí)行流；

中斷與中斷之間：外設(shè)的中斷可以路由到不同的CPU上，它們之間也可能帶來競態(tài)；

這時候就需要一種同步機(jī)制來保護(hù)并發(fā)訪問的內(nèi)存數(shù)據(jù)。本系列文章分為兩部分，這一章主要討論原子操作，自旋鎖，信號量和互斥鎖

原子操作

原子操作是在執(zhí)行結(jié)束前不可打斷的操作，也是最小的執(zhí)行單位。以 arm 平臺為例，原子操作的 API 包括如下:

原子操作通常是內(nèi)聯(lián)函數(shù)，往往是通過內(nèi)嵌匯編指令來實現(xiàn)的，如果某個函數(shù)本身就是原子的，它往往被定義成一個宏，以下為例。

#define ATOMIC_OP(op, c_op, asm_op)? ? \static inline void atomic_##op(int i, atomic_t *v)? \{? ? ? ? \ unsigned long tmp;? ? ? \ int result;? ? ? \? ? ? ? \ prefetchw(&v->counter);? ? ? \ __asm__ __volatile__("@ atomic_" #op "\n"? \"1: ldrex %0, [%3]\n"? ? ? \" " #asm_op " %0, %0, %4\n"? ? \" strex %1, %0, [%3]\n"? ? ? \" teq %1, #0\n"? ? ? \" bne 1b"? ? ? \ : "=&r" (result), "=&r" (tmp), "+Qo" (v->counter)? \ : "r" (&v->counter), "Ir" (i)? ? \ : "cc");? ? ? \}

可見原子操作的原子性依賴于 ldrex 與 strex 實現(xiàn)，ldrex 讀取數(shù)據(jù)時會進(jìn)行獨占標(biāo)記，防止其他內(nèi)核路徑訪問，直至調(diào)用 strex 完成寫入后清除標(biāo)記。

ldrex 和 strex 指令，是將單純的更新內(nèi)存的原子操作分成了兩個獨立的步驟:

ldrex 用來讀取內(nèi)存中的值，并標(biāo)記對該段內(nèi)存的獨占訪問：

ldrexRx,[Ry]

讀取寄存器 Ry 指向的4字節(jié)內(nèi)存值，將其保存到 Rx 寄存器中，同時標(biāo)記對 Ry 指向內(nèi)存區(qū)域的獨占訪問。如果執(zhí)行 ldrex 指令的時候發(fā)現(xiàn)已經(jīng)被標(biāo)記為獨占訪問了，并不會對指令的執(zhí)行產(chǎn)生影響。

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

strex Rx, Ry, [Rz]

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

ARM 內(nèi)部的實現(xiàn)如下所示，這里不再贅述。

自旋鎖 spin_lock

Linux內(nèi)核中最常見的鎖是自旋鎖，自旋鎖最多只能被一個可執(zhí)行線程持有。如果一個線程試圖獲取一個已被持有的自旋鎖，這個線程會進(jìn)行忙循環(huán)——旋轉(zhuǎn)等待(會浪費處理器時間)鎖重新可用。自旋鎖持有期間不可被搶占。

另一種處理鎖爭用的方式：讓等待線程睡眠，直到鎖重新可用時再喚醒它，這樣處理器不必循環(huán)等待，可以去執(zhí)行其他代碼，但是這會有兩次明顯的上下文切換的開銷，信號量便提供了這種鎖機(jī)制。

自旋鎖的使用接口如下：

以 spin_lock 為例看下它的用法：

DEFINE_SPINLOCK(mr_lock);spin_lock(&mr_lock);/* 臨界區(qū) */spin_unlock(&mr_lock);

staticinlinevoidarch_spin_lock(arch_spinlock_t*lock){unsignedinttmp;arch_spinlock_tlockval,newval;asmvolatile(/* Atomically increment the next ticket. */ARM64_LSE_ATOMIC_INSN(/* LL/SC */" prfm pstl1strm, %3\n""1: ldaxr %w0, %3\n"" add %w1, %w0, %w5\n"" stxr %w2, %w1, %3\n"" cbnz %w2, 1b\n",/* LSE atomics */" mov %w2, %w5\n"" ldadda %w2, %w0, %3\n"__nops(3))/* 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");}

staticinlinevoidarch_spin_unlock(arch_spinlock_t*lock){unsignedlongtmp;asmvolatile(ARM64_LSE_ATOMIC_INSN(/* LL/SC */" ldrh %w1, %0\n"" add %w1, %w1, #1\n"" stlrh %w1, %0",/* LSE atomics */" mov %w1, #1\n"" staddlh %w1, %0\n"__nops(1)):"=Q"(lock->owner),"=&r"(tmp)::"memory");}

上邊的代碼中，核心邏輯在于 asm volatile() 內(nèi)聯(lián)匯編中，有很多獨占的操作指令，只有基于指令的獨占操作，才能保證軟件上的互斥。把核心邏輯翻譯成 C 語言：

可以看出，Linux 中針對每一個 spin_lock 有兩個計數(shù)。分別是 next 和 owner（初始值為0）。進(jìn)程 A 申請鎖時，會判斷 next 和 owner 的值是否相等。如果相等就代表鎖可以申請成功，否則原地自旋。直到 owner 和 next 的值相等才會退出自旋。

信號量 Semaphore

信號量是在多線程環(huán)境下使用的一種措施，它負(fù)責(zé)協(xié)調(diào)各個進(jìn)程，以保證他們能夠正確、合理的使用公共資源。它和 spin_lock 最大的不同之處就是：無法獲取信號量的進(jìn)程可以睡眠，因此會導(dǎo)致系統(tǒng)調(diào)度。

信號量的定義如下：

structsemaphore{raw_spinlock_tlock;//利用自旋鎖同步unsignedintcount;//用于資源計數(shù)structlist_headwait_list;//等待隊列};

信號量在創(chuàng)建時設(shè)置一個初始值 count，用于表示當(dāng)前可用的資源數(shù)。一個任務(wù)要想訪問共享資源，首先必須得到信號量，獲取信號量的操作為 count - 1。若當(dāng)前 count 為負(fù)數(shù)，表明無法獲得信號量，該任務(wù)必須掛起在該信號量的等待隊列等待；若當(dāng)前 count 為非負(fù)數(shù)，表示可獲得信號量，因而可立刻訪問被該信號量保護(hù)的共享資源。

當(dāng)任務(wù)訪問完被信號量保護(hù)的共享資源后，必須釋放信號量，釋放信號量是操作 count + 1，如果加一后的 count 為非正數(shù)，表明有任務(wù)等待，則喚醒所有等待該信號量的任務(wù)。

了解了信號量的結(jié)構(gòu)與定義，接下來我們看下常用的信號量接口：

這里我們看下最核心的兩個實現(xiàn) down 和 up。

down

down 用于調(diào)用者獲得信號量，若 count 大于0，說明資源可用，將其減一即可。

voiddown(structsemaphore*sem){unsignedlongflags;raw_spin_lock_irqsave(&sem->lock,flags);if(likely(sem->count>0))sem->count--;else__down(sem);raw_spin_unlock_irqrestore(&sem->lock,flags);}EXPORT_SYMBOL(down);

若 count < 0，調(diào)用函數(shù) __down()，將 task 加入等待隊列，并進(jìn)入等待隊列，并進(jìn)入調(diào)度循環(huán)等待，直至其被 __up 喚醒，或者因超時以被移除等待隊列。

staticinlineint__sched__down_common(structsemaphore*sem,longstate,longtimeout){structsemaphore_waiterwaiter;list_add_tail(&waiter.list,&sem->wait_list);waiter.task=current;waiter.up=false;for(;;){if(signal_pending_state(state,current))gotointerrupted;if(unlikely(timeout<=0))gototimed_out;__set_current_state(state);raw_spin_unlock_irq(&sem->lock);timeout=schedule_timeout(timeout);raw_spin_lock_irq(&sem->lock);if(waiter.up)return0;}timed_out:list_del(&waiter.list);return-ETIME;interrupted:list_del(&waiter.list);return-EINTR;}

up

up 用于調(diào)用者釋放信號量，若 waitlist 為空，說明無等待任務(wù)，count + 1，該信號量可用。

voidup(structsemaphore*sem){unsignedlongflags;raw_spin_lock_irqsave(&sem->lock,flags);if(likely(list_empty(&sem->wait_list)))sem->count++;else__up(sem);raw_spin_unlock_irqrestore(&sem->lock,flags);}EXPORT_SYMBOL(up);

若 waitlist 非空，將 task 從等待隊列移除，并喚醒該 task，對應(yīng) __down 條件。

staticnoinlinevoid__sched__up(structsemaphore*sem){structsemaphore_waiter*waiter=list_first_entry(&sem->wait_list,structsemaphore_waiter,list);list_del(&waiter->list);waiter->up=true;wake_up_process(waiter->task);}

互斥鎖 mutex

Linux 內(nèi)核中，還有一種類似信號量的同步機(jī)制叫做互斥鎖?；コ怄i類似于 count 等于 1 的信號量。所以說信號量是在多個進(jìn)程/線程訪問某個公共資源的時候，進(jìn)行保護(hù)的一種機(jī)制。而互斥鎖是單個進(jìn)程/線程訪問某個公共資源的一種保護(hù)，于互斥操作。

互斥鎖有一個特殊的地方：只有持鎖者才能解鎖。如下圖所示：

用一句話來講信號量和互斥鎖的區(qū)別，就是信號量用于線程的同步，互斥鎖用于線程的互斥。

互斥鎖的結(jié)構(gòu)體定義：

structmutex{atomic_long_towner;//互斥鎖的持有者spinlock_twait_lock;//利用自旋鎖同步#ifdef CONFIG_MUTEX_SPIN_ON_OWNERstructoptimistic_spin_queueosq;/* Spinner MCS lock */#endifstructlist_headwait_list;//等待隊列......};

其常用的接口如下所示：