深入決議Linux并發同步

并發

是指在某一時間段內能夠處理多個任務的能力，而 并行 是指同一時間能夠處理多個任務的能力，并發和并行看起來很像，但實際上是有區別的，如下圖：

上圖的意思是，有兩條在排隊買咖啡的佇列，并發只有一架咖啡機在處理，而并行就有兩架的咖啡機在處理，咖啡機的數量越多，并行能力就越強，

linux內核的相關視頻：
深度詳解Linux內核網路結構及分布
epoll的具體實作與epoll執行緒安全|互斥鎖|自旋鎖|原子操作|CAS
手把手帶你實作一個Linux內核檔案系統

可以把上面的兩條佇列看成兩個行程，并發就是指只有單個CPU在處理，而并行就有兩個CPU在處理，為了讓兩個行程在單核CPU中也能得到執行，一般的做法就是讓每個行程交替執行一段時間，比如讓每個行程固定執行 100毫秒，執行時間使用完后切換到其他行程執行，而并行就沒有這種問題，因為有兩個CPU，所以兩個行程可以同時執行，如下圖：

【文章福利】小編推薦自己的C/C++Linux群：812855908！整理了一些個人覺得比較好的學習書籍、視頻資料共享在里面，有需要的可以自行添加哦！~

原子操作

上面介紹過，并發有可能會打斷當前執行的行程，然后替切換成其他行程執行，如果有兩個行程同時對一個共享變數 count 進行加一操作，由于C語言的 count++ 操作會被翻譯成如下指令：

mov eax, [count]
inc eax
mov [count], eax

那么在并發的情況下，有可能出現如下問題：

假設count變數初始值為0：

行程1執行完 mov eax, [count] 后，暫存器eax內保存了count的值0，
行程2被調度執行，行程2執行 count++ 的所有指令，將累加后的count值1寫回到記憶體，
行程1再次被調度執行，計算count的累加值仍為1，寫回到記憶體，
雖然行程1和行程2執行了兩次 count++ 操作，但是count最后的值為1，而不是2，

要解決這個問題就需要使用 原子操作，原子操作是指不能被打斷的操作，在單核CPU中，一條指令就是原子操作，比如上面的問題可以把 count++ 陳述句翻譯成指令 inc [count] 即可，Linux也提供了這樣的原子操作，如對整數加一操作的 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中需要進行如下程序：

從記憶體將count的資料讀取到cpu，
累加讀取的值，
將修改的值寫回count記憶體，
Intel x86 CPU 提供了 lock 前綴來鎖住總線，可以讓指令保證不被其他CPU中斷，如下：

lock
inc [count]

鎖

原子操作 能夠保證操作不被其他行程干擾，但有時候一個復雜的操作需要由多條指令來實作，那么就不能使用原子操作了，這時候可以使用 鎖 來實作，

計算機科學中的 鎖 與日常生活的 鎖 有點類似，舉個例子：比如要上公廁，首先找到一個沒有人的廁所，然后把廁所門鎖上，其他人要使用的話，必須等待當前這人使用完畢，并且把門鎖打開才能使用，在計算機中，要對某個公共資源進行操作時，必須對公共資源進行上鎖，然后才能使用，如果不上鎖，那么就可能導致資料混亂的情況，

在Linux內核中，比較常用的鎖有：自旋鎖、信號量、讀寫鎖 等，下面介紹一下自旋鎖和信號量的實作，

自旋鎖

自旋鎖 只能在多核CPU系統中，其核心原理是 原子操作，原理如下圖：

使用 自旋鎖 時，必須先對自旋鎖進行初始化（設定為1），上鎖程序如下：

對自旋鎖 lock 進行減一操作，判斷結果是否等于0，如果是表示上鎖成功并回傳，
如果不等于0，表示其他行程已經上鎖，此時必須不斷比較自旋鎖 lock 的值是否等于1（表示已經解鎖），
如果自旋鎖 lock 等于1，跳轉到第一步繼續進行上鎖操作，
由于Linux的自旋鎖使用匯編實作，所以比較苦澀難懂，這里使用C語言來模擬一下：

void spin_lock(amtoic_t *lock)
{
again:
    result = --(*lock);
    if (result == 0) {
        return;
    }
    
    while (true) {
        if (*lock == 1) {
	    goto again;
	}
    }
}

上面代碼將 result = --(*lock); 當成原子操作，解鎖程序只需要把 lock 設定為1即可，由于自旋鎖會不斷嘗試上鎖操作，并不會對行程進行調度，所以在單核CPU中可能會導致 100% 的CPU占用率，另外，自旋鎖只適合粒度比較小的操作，如果操作粒度比較大，就需要使用信號量這種可調度行程的鎖，

信號量

與 自旋鎖 不一樣，當當前行程對 信號量 進行上鎖時，如果其他行程已經對其進行上鎖，那么當前行程會進入睡眠狀態，等待其他行程對信號量進行解鎖，程序如下圖：

在Linux內核中，信號量使用 struct semaphore 表示，定義如下：

struct semaphore {
    raw_spinlock_t      lock;
    unsigned int        count;
    struct list_head    wait_list;
};

各個欄位的作用如下：

lock：自旋鎖，用于對多核CPU平臺進行同步，
count：信號量的計數器，上鎖時對其進行減一操作(count–)，如果得到的結果為大于等于0，表示成功上鎖，如果小于0表示已經被其他行程上鎖，
wait_list：正在等待信號量解鎖的行程佇列，
信號量 上鎖通過 down() 函式實作，代碼如下：

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() 函式首先對信號量進行自旋鎖操作（為了避免多核CPU競爭），然后比較計數器是否大于0，如果是對計數器進行減一操作，并且回傳，否則呼叫 __down() 函式進行下一步操作，__down() 函式實作如下：

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;

    // 把當前行程添加到等待佇列中
    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) // 當前行程是否獲得信號量鎖?
            return 0;
    }
    ...
}

__down() 函式最終呼叫 __down_common() 函式，而 __down_common() 函式的操作程序如下：

把當前行程添加到信號量的等待佇列中，
切換到其他行程運行，直到被其他行程喚醒，
如果當前行程獲得信號量鎖（由解鎖行程傳遞），那么函式回傳，
接下來看看解鎖程序，解鎖程序主要通過 up() 函式實作，代碼如下：

void up(struct semaphore *sem)
{
    unsigned long flags;
 
    raw_spin_lock_irqsave(&sem->lock, flags);
    if (likely(list_empty(&sem->wait_list))) // 如果沒有等待的行程, 直接對計數器加一操作
        sem->count++;
    else
        __up(sem); // 如果有等待行程, 那么呼叫 __up() 函式進行喚醒
    raw_spin_unlock_irqrestore(&sem->lock, flags);
}

static noinline void __sched __up(struct semaphore *sem)
{
    // 獲取到等待佇列的第一個行程
    struct semaphore_waiter *waiter = list_first_entry(
        &sem->wait_list, struct semaphore_waiter, list);

    list_del(&waiter->list);       // 把行程從等待佇列中洗掉
    waiter->up = 1;                // 告訴行程已經獲得信號量鎖
    wake_up_process(waiter->task); // 喚醒行程
}

解鎖程序如下：

判斷當前信號量是否有等待的行程，如果沒有等待的行程, 直接對計數器加一操作
如果有等待的行程，那么獲取到等待佇列的第一個行程，
把行程從等待佇列中洗掉，
告訴行程已經獲得信號量鎖，
喚醒行程，