一文讀懂 | 進(jìn)程并發(fā)與同步
并發(fā) 是指在某一時(shí)間段內(nèi)能夠處理多個(gè)任務(wù)的能力，而 并行 是指同一時(shí)間能夠處理多個(gè)任務(wù)的能力。并發(fā)和并行看起來(lái)很像，但實(shí)際上是有區(qū)別的，如下圖（圖片來(lái)源于網(wǎng)絡(luò)）：
上圖的意思是，有兩條在排隊(duì)買(mǎi)咖啡的隊(duì)列，并發(fā)只有一架咖啡機(jī)在處理，而并行就有兩架的咖啡機(jī)在處理?？Х葯C(jī)的數(shù)量越多，并行能力就越強(qiáng)。
可以把上面的兩條隊(duì)列看成兩個(gè)進(jìn)程，并發(fā)就是指只有單個(gè)CPU在處理，而并行就有兩個(gè)CPU在處理。為了讓兩個(gè)進(jìn)程在單核CPU中也能得到執(zhí)行，一般的做法就是讓每個(gè)進(jìn)程交替執(zhí)行一段時(shí)間，比如讓每個(gè)進(jìn)程固定執(zhí)行 100毫秒，執(zhí)行時(shí)間使用完后切換到其他進(jìn)程執(zhí)行。而并行就沒(méi)有這種問(wèn)題，因?yàn)橛袃蓚€(gè)CPU，所以?xún)蓚€(gè)進(jìn)程可以同時(shí)執(zhí)行。如下圖：
原子操作
上面介紹過(guò)，并發(fā)有可能會(huì)打斷當(dāng)前執(zhí)行的進(jìn)程，然后替切換成其他進(jìn)程執(zhí)行。如果有兩個(gè)進(jìn)程同時(shí)對(duì)一個(gè)共享變量 count 進(jìn)行加一操作，由于C語(yǔ)言的 count++ 操作會(huì)被翻譯成如下指令：
mov eax, [count]
inc eax
mov [count], eax
那么在并發(fā)的情況下，有可能出現(xiàn)如下問(wèn)題：
假設(shè)count變量初始值為0：
進(jìn)程1執(zhí)行完 mov eax, [count]后，寄存器eax內(nèi)保存了count的值0。進(jìn)程2被調(diào)度執(zhí)行。進(jìn)程2執(zhí)行 count++的所有指令，將累加后的count值1寫(xiě)回到內(nèi)存。進(jìn)程1再次被調(diào)度執(zhí)行，計(jì)算count的累加值仍為1，寫(xiě)回到內(nèi)存。
雖然進(jìn)程1和進(jìn)程2執(zhí)行了兩次 count++ 操作，但是count最后的值為1，而不是2。
要解決這個(gè)問(wèn)題就需要使用 原子操作，原子操作是指不能被打斷的操作，在單核CPU中，一條指令就是原子操作。比如上面的問(wèn)題可以把 count++ 語(yǔ)句翻譯成指令 inc [count] 即可。Linux也提供了這樣的原子操作，如對(duì)整數(shù)加一操作的 atomic_inc()：
static __inline__ void atomic_inc(atomic_t *v)
{
__asm__ __volatile__(
LOCK "incl %0"
:"=m" (v->counter)
:"m" (v->counter));
}
在多核CPU中，一條指令也不一定是原子操作，比如 inc [count] 指令在多核CPU中需要進(jìn)行如下過(guò)程：
從內(nèi)存將count的數(shù)據(jù)讀取到cpu。 累加讀取的值。 將修改的值寫(xiě)回count內(nèi)存。
Intel x86 CPU 提供了 lock 前綴來(lái)鎖住總線，可以讓指令保證不被其他CPU中斷，如下：
lock
inc [count]
鎖
原子操作 能夠保證操作不被其他進(jìn)程干擾，但有時(shí)候一個(gè)復(fù)雜的操作需要由多條指令來(lái)實(shí)現(xiàn)，那么就不能使用原子操作了，這時(shí)候可以使用 鎖 來(lái)實(shí)現(xiàn)。
計(jì)算機(jī)科學(xué)中的 鎖 與日常生活的 鎖 有點(diǎn)類(lèi)似，舉個(gè)例子：比如要上公廁，首先找到一個(gè)沒(méi)有人的廁所，然后把廁所門(mén)鎖上。其他人要使用的話，必須等待當(dāng)前這人使用完畢，并且把門(mén)鎖打開(kāi)才能使用。在計(jì)算機(jī)中，要對(duì)某個(gè)公共資源進(jìn)行操作時(shí)，必須對(duì)公共資源進(jìn)行上鎖，然后才能使用。如果不上鎖，那么就可能導(dǎo)致數(shù)據(jù)混亂的情況。
在Linux內(nèi)核中，比較常用的鎖有：自旋鎖、信號(hào)量、讀寫(xiě)鎖 等，下面介紹一下自旋鎖和信號(hào)量的實(shí)現(xiàn)。
自旋鎖
自旋鎖 只能在多核CPU系統(tǒng)中，其核心原理是 原子操作，原理如下圖：
使用 自旋鎖 時(shí)，必須先對(duì)自旋鎖進(jìn)行初始化（設(shè)置為1），上鎖過(guò)程如下：
對(duì)自旋鎖 lock進(jìn)行減一操作，判斷結(jié)果是否等于0，如果是表示上鎖成功并返回。如果不等于0，表示其他進(jìn)程已經(jīng)上鎖，此時(shí)必須不斷比較自旋鎖 lock的值是否等于1（表示已經(jīng)解鎖）。如果自旋鎖 lock等于1，跳轉(zhuǎn)到第一步繼續(xù)進(jìn)行上鎖操作。
由于Linux的自旋鎖使用匯編實(shí)現(xiàn)，所以比較苦澀難懂，這里使用C語(yǔ)言來(lái)模擬一下：
void spin_lock(amtoic_t *lock)
{
again:
result = --(*lock);
if (result == 0) {
return;
}
while (true) {
if (*lock == 1) {
goto again;
}
}
}
上面代碼將 result = --(*lock); 當(dāng)成原子操作，解鎖過(guò)程只需要把 lock 設(shè)置為1即可。由于自旋鎖會(huì)不斷嘗試上鎖操作，并不會(huì)對(duì)進(jìn)程進(jìn)行調(diào)度，所以在單核CPU中可能會(huì)導(dǎo)致 100% 的CPU占用率。另外，自旋鎖只適合粒度比較小的操作，如果操作粒度比較大，就需要使用信號(hào)量這種可調(diào)度進(jìn)程的鎖。
信號(hào)量
與 自旋鎖 不一樣，當(dāng)當(dāng)前進(jìn)程對(duì) 信號(hào)量 進(jìn)行上鎖時(shí)，如果其他進(jìn)程已經(jīng)對(duì)其進(jìn)行上鎖，那么當(dāng)前進(jìn)程會(huì)進(jìn)入睡眠狀態(tài)，等待其他進(jìn)程對(duì)信號(hào)量進(jìn)行解鎖。過(guò)程如下圖：
在Linux內(nèi)核中，信號(hào)量使用 struct semaphore 表示，定義如下：
struct semaphore {
raw_spinlock_t lock;
unsigned int count;
struct list_head wait_list;
};
各個(gè)字段的作用如下：
lock：自旋鎖，用于對(duì)多核CPU平臺(tái)進(jìn)行同步。count：信號(hào)量的計(jì)數(shù)器，上鎖時(shí)對(duì)其進(jìn)行減一操作(count--)，如果得到的結(jié)果為大于等于0，表示成功上鎖，如果小于0表示已經(jīng)被其他進(jìn)程上鎖。wait_list：正在等待信號(hào)量解鎖的進(jìn)程隊(duì)列。
信號(hào)量 上鎖通過(guò) down() 函數(shù)實(shí)現(xiàn)，代碼如下：
void down(struct semaphore *sem)
{
unsigned long flags;
spin_lock_irqsave(&sem->lock, flags);
if (likely(sem->count > 0))
sem->count--;
else
__down(sem);
spin_unlock_irqrestore(&sem->lock, flags);
}
上面代碼可以看出，down() 函數(shù)首先對(duì)信號(hào)量進(jìn)行自旋鎖操作（為了避免多核CPU競(jìng)爭(zhēng)），然后比較計(jì)數(shù)器是否大于0，如果是對(duì)計(jì)數(shù)器進(jìn)行減一操作，并且返回，否則調(diào)用 __down() 函數(shù)進(jìn)行下一步操作。__down() 函數(shù)實(shí)現(xiàn)如下：
static noinline void __sched __down(struct semaphore *sem)
{
__down_common(sem, TASK_UNINTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
}
static inline int __down_common(struct semaphore *sem,
long state, long timeout)
{
struct task_struct *task = current;
struct semaphore_waiter waiter;
// 把當(dāng)前進(jìn)程添加到等待隊(duì)列中
list_add_tail(&waiter.list, &sem->wait_list);
waiter.task = task;
waiter.up = 0;
for (;;) {
...
__set_task_state(task, state);
spin_unlock_irq(&sem->lock);
timeout = schedule_timeout(timeout);
spin_lock_irq(&sem->lock);
if (waiter.up) // 當(dāng)前進(jìn)程是否獲得信號(hào)量鎖?
return 0;
}
...
}
__down() 函數(shù)最終調(diào)用 __down_common() 函數(shù)，而 __down_common() 函數(shù)的操作過(guò)程如下：
把當(dāng)前進(jìn)程添加到信號(hào)量的等待隊(duì)列中。 切換到其他進(jìn)程運(yùn)行，直到被其他進(jìn)程喚醒。 如果當(dāng)前進(jìn)程獲得信號(hào)量鎖（由解鎖進(jìn)程傳遞），那么函數(shù)返回。
接下來(lái)看看解鎖過(guò)程，解鎖過(guò)程主要通過(guò) up() 函數(shù)實(shí)現(xiàn)，代碼如下：
void up(struct semaphore *sem)
{
unsigned long flags;
raw_spin_lock_irqsave(&sem->lock, flags);
if (likely(list_empty(&sem->wait_list))) // 如果沒(méi)有等待的進(jìn)程, 直接對(duì)計(jì)數(shù)器加一操作
sem->count++;
else
__up(sem); // 如果有等待進(jìn)程, 那么調(diào)用 __up() 函數(shù)進(jìn)行喚醒
raw_spin_unlock_irqrestore(&sem->lock, flags);
}
static noinline void __sched __up(struct semaphore *sem)
{
// 獲取到等待隊(duì)列的第一個(gè)進(jìn)程
struct semaphore_waiter *waiter = list_first_entry(
&sem->wait_list, struct semaphore_waiter, list);
list_del(&waiter->list); // 把進(jìn)程從等待隊(duì)列中刪除
waiter->up = 1; // 告訴進(jìn)程已經(jīng)獲得信號(hào)量鎖
wake_up_process(waiter->task); // 喚醒進(jìn)程
}
解鎖過(guò)程如下：
判斷當(dāng)前信號(hào)量是否有等待的進(jìn)程，如果沒(méi)有等待的進(jìn)程, 直接對(duì)計(jì)數(shù)器加一操作 如果有等待的進(jìn)程，那么獲取到等待隊(duì)列的第一個(gè)進(jìn)程。 把進(jìn)程從等待隊(duì)列中刪除。 告訴進(jìn)程已經(jīng)獲得信號(hào)量鎖 喚醒進(jìn)程。