@[TOC] # 核心中的併發和競爭簡介   在早期的 Linux核心中，併發的來源相對較少。早期核心不支援對稱多處理（ symmetric multi processing,SMP），因此，導致併發執行的唯一原因是對硬體中斷的服務。這種情況處理起來較為簡單，但並不適用於為獲得更好的效能而使用更多處理器且強調快速響應事件的系統。   為了響應現代硬體和應用程式的需求， Linux核心已經發展到同時處理更多事情的時代。Linux系統是個多工作業系統，會存在多個任務同時訪問同一片記憶體區域的情況，這些任務可能會相互覆蓋這段記憶體中的資料，造成記憶體資料混亂。針對這個問題必須要做處理，嚴重的話可能會導致系統崩潰。現在的 Linux系統併發產生的原因很複雜，總結一下有下面幾個主要原因： 1. 多執行緒併發訪問， Linux是多工（執行緒）的系統，所以多執行緒訪問是最基本的原因。 2. 搶佔式併發訪問，核心程式碼是可搶佔的，因此，我們的驅動程式程式碼可能在任何時候丟失對處理器的獨佔 3. 中斷程式併發訪問，裝置中斷是非同步事件，也會導致程式碼的併發執行。 4. SMP（多核）核間併發訪問，現在ARM架構的多核SOC很常見，多核CPU存在核間併發訪問。正在執行的多個使用者空間程序可能以一種令人驚訝的組合方式訪問我們的程式碼，SMP系統甚至可在不同的處理器上同時執行我們的程式碼。   只要我們的程式在運轉當中，就有可能發生併發和競爭。比如，當兩個執行執行緒需要訪問**相同的資料結構**（或硬體資源）時，混合的可能性就永遠存在。因此在設計自己的驅動程式時，就應該避免資源的共享。**如果沒有併發的訪問，也就不會有競態的產生**。因此，仔細編寫的核心程式碼應該具有**最少的共享**。這種思想的最明顯應用就是**避免使用全域性變數**。如果我們將資源放在多個執行執行緒都會找到的地方（臨界區），則必須有足夠的理由。   如何防止我們的資料被併發訪問呢？這個時候就要建立一種保護機制，下面介紹幾種核心提供的幾種併發和競爭的處理方法。 # 原子操作 ## 原子操作簡介   原子，在早接觸到是在化學概念中。原子指化學反應不可再分的基本微粒。同樣的，在核心中所說的原子操作表示這一個訪問是一個步驟，必須一次性執行完，不能被打斷，不能再進行拆分。   例如，在多執行緒訪問中，我們的執行緒一對a進行賦值操作，`a=1`，執行緒二也對a進行賦值操作`a=2`，我們理想的執行順序是執行緒一先執行，執行緒二再執行。但是很有可能線上程一執行的時候被其他操作打斷，使得執行緒一最後的執行結果變為`a=2`。要解決這個問題，必須保證我們的執行緒一在對資料訪問的過程中不能被其他的操作打斷，一次性執行完成。 ## 整型原子操作函式 函式 | 描述 -------- | ----- ATOMIC_INIT(int i) | 定義原子變數的時候對其初始化。 int atomic_read(atomic_t*v)| 讀取 v的值，並且返回 void atomic_set(atomic_t *v, int i) | 向 v寫入 i值。 void atomic_add(int i, atomic_t *v)| 給 v加上 i值。 void atomic_sub(int i, atomic_t *v)| 從 v減去 i值。 void atomic_inc(atomic_t *v)| 給 v加 1，也就是自增。 void atomic_dec(atomic_t *v)| 從 v減 1，也就是自減 。 int atomic_dec_return(atomic_t *v)| 從 v減 1，並且返回v的值 。 int atomic_inc_return(atomic_t *v)| 給 v加 1，並且返回 v的值。 int atomic_sub_and_test(int i, atomic_t *v)| 從 v減 i，如果結果為0就返回真，否則就返回假 int atomic_dec_and_test(atomic_t *v)| 從 v減 1，如果結果為0就返回真，否則就返回假 int atomic_inc_and_test(atomic_t *v)| 給 v加 1，如果結果為0就返回真，否則就返回假 int atomic_add_negative(int i, atomic_t *v) | 給 v加 i，如果結果為負就返回真，否則返回假 注：64位的整型原子操作只是將“atomic_”字首換成“atomic64_”，將int換成long long。 ## 位原子操作函式 函式 | 描述 -------- | ----- void set_bit(int nr, void *p)| 將p地址的nr位置1 void clear_bit(int nr,void *p) | 將p地址的nr位清零 void change_bit(int nr, void *p)| 將p地址的nr位反轉 int test_bit(int nr, void *p)| 獲取p地址的nr位的值 int test_and_set_bit(int nr, void *p)| 將p地址的nr位置1，並且返回nr位原來的值 int test_and_clear_bit(int nr, void *p)| 將p地址的nr位清0，並且返回nr位原來的值 int test_and_change_bit(int nr, void *p)| 將p地址的nr位翻轉，並且返回nr位原來的值 ## 原子操作例程 ```c /* 定義原子變數，初值為1*/ static atomic_t xxx_available = ATOMIC_INIT(1); static int xxx_open(struct inode *inode, struct file *filp) { ... /* 通過判斷原子變數的值來檢查LED有沒有被別的應用使用 */ if (!atomic_dec_and_test(&xxx_available)) { /*小於0的話就加1,使其原子變數等於0*/ atomic_inc(&xxx_available); /* LED被使用，返回忙*/ return - EBUSY; } ... /* 成功 */ return 0; static int xxx_release(struct inode *inode, struct file *filp) { /* 關閉驅動檔案的時候釋放原子變數 */ atomic_inc(&xxx_available); return 0; } ``` # 自旋鎖   上面我們介紹了原子變數，從它的操作函式可以看出，原子操作只能針對**整型變數或者位**。假如我們有一個結構體變數需要被執行緒A所訪問，線上程A訪問期間不能被其他執行緒訪問，這怎麼辦呢？自旋鎖就可以完成對結構體變數的保護。 ## 自旋鎖簡介   自旋鎖，顧名思義，我們可以把他理解成廁所門上的一把鎖。這個廁所門只有一把鑰匙，當我們進去時，把鑰匙取下來，進去後反鎖。那麼當第二個人想進來，必須等我們出去後才可以。當第二個人在外面等待時，可能會一直等待在門口轉圈。   我們的自旋鎖也是這樣，自旋鎖只有鎖定和解鎖兩個狀態。當我們進入拿上鑰匙進入廁所，這就相當於自旋鎖鎖定的狀態，期間誰也不可以進來。當第二個人想要進來，這相當於執行緒B想要訪問這個共享資源，但是目前不能訪問，所以執行緒B就一直在原地等待，一直查詢是否可以訪問這個共享資源。當我們從廁所出來後，這個時候就“解鎖”了，只有再這個時候執行緒B才能訪問。   假如，在廁所的人待的時間太長怎麼辦？外面的人一直等待嗎？如果換做是我們，肯定不會這樣，簡直浪費時間，可能我們會尋找其他方法解決問題。自旋鎖也是這樣的，如果執行緒A持有**自旋鎖時間過長，顯然會浪費處理器的時間，降低了系統性能**。我們知道CPU最偉大的發明就在於多執行緒操作，這個時候讓執行緒B在這裡傻傻的不知道還要等待多久，顯然是不合理的。因此，如果**自旋鎖只適合短期持有**，如果遇到需要長時間持有的情況，我們就要換一種方式了（下文的互斥體）。 ## 自旋鎖操作函式 函式 | 描述 -------- | ----- DEFINE_SPINLOCK(spinlock_t lock) | 定義並初始化一個自旋變數 int spin_lock_init(spinlock_t *lock)| 初始化自旋鎖 void spin_lock(spinlock_t *lock)| 獲取指定的自旋鎖，也叫加鎖 void spin_unlock(spinlock_t *lock)| 釋放指定的自旋鎖。 int spin_trylock(spinlock_t *lock)| 嘗試獲取指定的鎖，如果沒有獲取到，返回0 int spin_is_locked(spinlock_t *lock)| 檢查指定的自旋鎖是否被獲取，如果沒有被獲取返回非0，否則返回0.   自旋鎖是主要為了**多處理器系統**設計的。對於**單處理器且核心不支援搶佔的系統**，一旦進入了自旋狀態，則會永遠自旋下去。因為，沒有任何執行緒可以獲取CPU來釋放這個鎖。因此，在單處理器且核心不支援搶佔的系統中，**自旋鎖會被設定為空操作**。   以上列表中的函式**適用於SMP或支援搶佔的單CPU**下執行緒之間的併發訪問，也就是用於執行緒與執行緒之間，被自旋鎖保護的臨界區**一定不能呼叫任何能夠引起睡眠和阻塞（其實本質仍然是睡眠）的API函式**，否則的話會可能會導致死鎖現象的發生。自旋鎖會**自動禁止搶佔**，也就說當執行緒A得到鎖以後會**暫時禁止核心搶佔**。如果執行緒A在持有鎖期間進入了**休眠狀態**，那麼執行緒A會**自動放棄CPU使用權**。執行緒B開始執行，執行緒B也想要獲取鎖，但是此時鎖被A執行緒持有，而且**核心搶佔還被禁止了**！執行緒B無法被排程岀去，那麼執行緒A就無法執行，鎖也就無法釋放**死鎖**發生了！   當執行緒之間發生併發訪問時，如果此時中斷也要插一腳，中斷也想訪問共享資源，那該怎麼辦呢？首先可以肯定的是，中斷裡面使用自旋鎖，但是在中斷裡面使用自旋鎖的時候，在獲取鎖之前一定要**先禁止本地中斷**（也就是**本CPU中斷**，對於多核SOC來說會有多個CPU核），否則可能導致鎖死現象的發生。看下下面一個例子： ```c //執行緒A spin_lock(&lock); ....... functionA(); ....... spin_unlock(&lock); //中斷髮生，執行執行緒B spin_lock(&lock); ....... functionA(); ....... spin_unlock(&lock); ```   執行緒A先執行，並且獲取到了lock這個鎖，當執行緒A執行 functionA函式的時候中斷髮生了，**中斷搶走了CPU使用權**。下邊的中斷服務函式也要獲取lock這個鎖，但是這個鎖被執行緒A佔有著，中斷就會一直自旋，等待鎖有效。但是在中斷服務函式執行完之前，執行緒A是不可能執行的，執行緒A說“你先放手”，中斷說“你先放手”，場面就這麼僵持著死鎖發生！   使用了自旋鎖之後可以保證臨界區不受別的CPU和本CPU內的搶佔程序的打擾，但是得到鎖的程式碼在執行臨界區的時候，還可能受到中斷和底半部的影響，為了防止這種影響，建議使用以下列表中的函式： 函式 | 描述 -------- | ----- void spin_lock_irq(spinlock_t *lock) |禁止本地中斷，並獲取自旋鎖 void spin_unlock_irq(spinlock_t *lock)| 啟用本地中斷，並釋放自旋鎖 void spin_lock_irqsave(spinlock_t *lock, unsigned long flags)| 儲存中斷狀態，禁止本地中斷，並獲取自旋鎖 void spin_unlock_irqrestore(spinlock_t *lock, unsigned long flags)      | 將中斷狀態恢復到以前的狀態，並且啟用本地中斷，釋放自旋鎖   在多核程式設計的時候，如果程序和中斷可能訪問同一片臨界資源，我們一般需要在程序上下文中呼叫spin_ lock irqsave（） spin_unlock_irqrestore（），在中斷上下文中呼叫 spin_lock（） spin _unlock（）。這樣，在CPU上，無論是程序上下文，還是中斷上下文獲得了自旋鎖，此後，如果CPU1無論是程序上下文，還是中斷上下文，想獲得同一自旋鎖，都必須忙等待，這避免一切核間併發的可能性。同時，由於每個核的程序上下文持有鎖的時候用的是 spin_lock_irgsave（），所以該核上的中斷是不可能進入的，這避免了核內併發的可能性。 ```c DEFINE_SPINLOCK(lock) /* 定義並初始化一個鎖 */ /* 執行緒A */ void functionA (){ unsigned long flags; /* 中斷狀態 */ spin_lock_irqsave(&lock, flags) /* 獲取鎖 */ /* 臨界區 */ spin_unlock_irqrestore(&lock, flags) /* 釋放鎖 */ } /* 中斷服務函式 */ void irq() { spin_lock(&lock) /* 獲取鎖 */ /* 臨界區 */ spin_unlock(&lock) /* 釋放鎖 */ } ``` ## 自旋鎖例程 ```c static int xxx_open(struct inode *inode, struct file *filp) { ... spinlock(&xxx_lock); if (xxx_count) {/* 已經開啟*/ spin_unlock(&xxx_lock); return -EBUSY; } xxx_count++;/* 增加使用計數*/ spin_unlock(&xxx_lock); ... return 0;/* 成功 */ } static int xxx_release(struct inode *inode, struct file *filp) { ... spinlock(&xxx_lock); xxx_count--;/* 減少使用計數*/ spin_unlock(&xxx_lock); return 0; } ``` ## 讀寫自旋鎖   當臨界區的一個檔案可以被**同時讀取**，但是並**不能被同時讀和寫**。如果一個執行緒在讀，另一個執行緒在寫，那麼很可能會讀取到錯誤的**不完整的資料**。讀寫自旋鎖是可以**允許對臨界區的共享資源進行併發讀操作的。但是並不允許多個執行緒併發讀寫操作**。如果想要併發讀寫，就要用到了順序鎖。   讀寫自旋鎖的讀操作函式如下所示： 函式 | 描述 -------- | ----- DEFINE_RWLOCK(rwlock_t lock) | 定義並初始化讀寫鎖 void rwlock_init(rwlock_t *lock) | 初始化讀寫鎖 void read_lock(rwlock_t *lock) | 獲取讀鎖 void read_unlock(rwlock_t *lock | 釋放讀鎖 void read_unlock_irq(rwlock_t *lock) | 開啟本地中斷，並且釋放讀鎖 void read_lock_irqsave(rwlock_t *lock,unsigned long flags) | 儲存中斷狀態，禁止本地中斷，並獲取讀鎖 void read_unlock_irqrestore(rwlock_t *lock,unsigned long flags) | 將中斷狀態恢復到以前的狀態，並且啟用本地中斷，釋放讀鎖 void read_lock_bh(rwlock_t *lock) | 關閉下半部，並獲取讀鎖 void read_unlock_bh(rwlock_t *lock) | 開啟下半部，並釋放讀鎖   讀寫自旋鎖的寫操作函式如下所示： 函式 | 描述 -------- | ----- void write_lock(rwlock_t *lock)| 獲取寫鎖 void write_unlock(rwlock_t *lock) | 釋放寫鎖 void write_lock_irq(rwlock_t *lock)| 禁止本地中斷，並且獲取寫鎖。 void write_unlock_irq(rwlock_t *lock)| 開啟本地中斷，並且釋放寫鎖 void write_lock_irqsave(rwlock_t *lock,unsigned long flags) | 儲存中斷狀態，禁止本地中斷，並獲取寫鎖 void write_unlock_irqrestore(rwlock_t *lock,unsigned long flags) | 將中斷狀態恢復到以前的狀態，並且啟用本地中斷，釋放寫鎖 void write_lock_bh(rwlock_t *lock) | 關閉下半部，並獲取寫鎖 void write_unlock_bh(rwlock_t *lock) | 開啟下半部，並釋放寫鎖 ## 讀寫鎖例程 ```c rwlock_t lock; /* 定義rwlock */ rwlock_init(&lock); /* 初始化rwlock */ /* 讀時獲取鎖*/ read_lock(&lock); ... /* 臨界資源 */ read_unlock(&lock); /* 寫時獲取鎖*/ write_lock_irqsave(&lock, flags); ... /* 臨界資源 */ write_unlock_irqrestore(&lock, flags); ``` ## 順序鎖   順序鎖是讀寫鎖的優化版本，讀寫鎖不允許同時讀寫，而使用**順序鎖可以完成同時進行讀和寫的操作**，**但並不允許同時的寫**。雖然順序鎖可以同時進行讀寫操作，但並不建議這樣，讀取的過程並不能保證資料的完整性。 ## 順序鎖操作函式   順序鎖的讀操作函式如下所示： 函式 | 描述 -------- | ----- DEFINE_SEQLOCK(seqlock_t sl) | 定義並初始化順序鎖 void seqlock_ini seqlock_t *sl) | 初始化順序鎖 void write_seqlock(seqlock_t *sl) | 順序鎖寫操作 void write_sequnlock(seqlock_t *sl) | 獲取寫順序鎖 void write_seqlock_irq(seqlock_t *sl) | 禁止本地中斷，並且獲取寫順序鎖 void write_sequnlock_irq(seqlock_t *sl) | 開啟本地中斷，並且釋放寫順序鎖 void write_seqlock_irqsave(seqlock_t *sl,unsigned long flags)| 儲存中斷狀態，禁止本地中斷，並獲取寫順序 void write_sequnlock_irqrestore(seqlock_t *sl,unsigned long flags)|將中斷狀態恢復到以前的狀態，並且啟用本地中斷，釋放寫順序鎖 void write_seqlock_bh(seqlock_t *sl) | 關閉下半部，並獲取寫讀鎖 void write_sequnlock_bh(seqlock_t *sl) | 開啟下半部，並釋放寫讀鎖   順序鎖的寫操作函式如下所示： 函式 | 描述 -------- | ----- DEFINE_RWLOCK(rwlock_t lock) | 讀單元訪問共享資源的時候呼叫此函式，此函式會返回順序鎖的順序號 unsigned read_seqretry(const seqlock_t *sl,unsigned start) | 讀結束以後呼叫此函式檢查在讀的過程中有沒有對資源進行寫操作，如果有的話就要重讀 ## 自旋鎖使用注意事項 1. 因為在等待自旋鎖的時候處於“自旋”狀態，因此鎖的持有時間不能太長，一定要短，否則的話會降低系統性能。如果臨界區比較大，執行時間比較長的話要選擇其他的併發處理方式，比如稍後要講的訊號量和互斥體。 2. 自旋鎖保護的臨界區內不能呼叫任何可能導致執行緒休眠的API函式，比如copy_from_user（）、copy_to_user（）、kmalloc（）和msleep（）等函式，否則的話可能導致死鎖。 3. 不能遞迴申請自旋鎖，因為一旦通過遞迴的方式申請一個你正在持有的鎖，那麼你就必須“自旋”，等待鎖被釋放，然而你正處於“自旋”狀態，根本沒法釋放鎖。結果就是自己把自己鎖死了 4. 在編寫驅動程式的時候我們必須考慮到驅動的可移植性，因此不管你用的是單核的還是多核的SOC，都將其當做多核SOC來編寫驅動程式。 > copy_from_user的使用是結合程序上下文的，因為他們要訪問“user”的記憶體空間，這個“user”必須是某個特定的程序。如果在驅動中使用這兩個函式，必須是在實現系統呼叫的函式中使用，不可在實現中斷處理的函式中使用。如果在中斷上下文中使用了，那程式碼就很可能操作了根本不相關的程序地址空間。其次由於操作的頁面可能被換出，這兩個函式可能會休眠，所以同樣不可在中斷上下文中使用。 # 訊號量 ## 訊號量簡介   訊號量和自旋鎖有些相似，不同的是訊號量會發出一個訊號告訴你還需要等多久。因此，不會出現傻傻等待的情況。比如，有100個停車位的停車場，門口電子顯示屏上實時更新的停車數量就是一個訊號量。這個停車的數量就是一個訊號量，他告訴我們是否可以停車進去。當有車開進去，訊號量加一，當有車開出來，訊號量減一。   比如，廁所一次只能讓一個人進去，當A在裡面的時候，B想進去，如果是自旋鎖，那麼B就會一直在門口傻傻等待。如果是訊號量，A就會給B一個訊號，你先回去吧，我出來了叫你。這就是一個訊號量的例子，B聽到A發出的訊號後，可以先回去睡覺，等待A出來。   因此，訊號量顯然可以提高系統的執行效率，避免了許多無用功。訊號量具有以下特點： 1. 因為訊號量可以使等待資源執行緒進入休眠狀態，因此適用於那些**佔用資源比較久**的場合。 2. 因此訊號量不能用於中斷中，因為訊號量會引起休眠，**中斷不能休眠**。 3. **如果共享資源的持有時間比較短，那就不適合使用訊號量了，因為頻繁的休眠、切換執行緒引起的開銷要遠大於訊號量帶來的那點優勢**。 ## 訊號量操作函式 函式 | 描述 -------- | ----- DEFINE_SEAMPHORE(name)| 定義一個訊號量，並且設定訊號量的值為1 void sema_init(struct semaphore *sem, int val) | 初始化訊號量sem，設定訊號量值為val void down(struct semaphore *sem) | 獲取訊號量，因為會導致休眠，因此不能在中斷中使用 int down_trylock(struct semaphore *sem); | 嘗試獲取訊號量，如果能獲取到訊號量就獲取，並且返回0.如果不能就返回非0，並且不會進入休眠 int down_interruptible(struct semaphore                       | 獲取訊號量，和down類似，只是使用dow進入休眠狀態的執行緒不能被訊號打斷。而使用此函式進入休眠以後是可以被訊號打斷的 void up(struct semaphore *sem) | 釋放訊號量 ## 訊號量例程 ```c struct semaphore sem; /* 定義訊號量 */ sema_init(&sem, 1)； /* 初始化訊號量 表示只能由一個執行緒同時訪問這塊資源 */ down(&sem); /* 申請訊號量 */ /* 臨界區 */ up(&sem); /* 釋放訊號量 */ ``` # 互斥體 ## 互斥體簡介   互斥體表示一次只有一個執行緒訪問共享資源，**不可以遞迴申請互斥體**。   訊號量也可以用於互斥體，當訊號量用於互斥時（即避免多個程序同時在一個臨界區中執行），訊號量的值應初始化為1.這種訊號量在任何給定時刻只能由單個程序或執行緒擁有。在這種使用模式下，一個訊號量有時也稱為一個“互斥體（ mutex）”，它是互斥（ mutual exclusion）的簡稱。Linux核心中**幾平所有的訊號量均用於互斥**。 ## 互斥體操作函式 函式 | 描述 -------- | ----- DEFINE_MUTEX(name)| 定義並初始化一個 mutex變數 void mutex_init(mutex *lock)| 初始化 mutex void mutex_lock(struct mutex *lock)| 獲取 mutex，也就是給 mutex上鎖。如果獲取不到就進休眠 void mutex_unlock(struct mutex *lock) | 釋放 mutex，也就給 mutex解鎖 int mutex_trylock(struct mutex *lock)                       | 判斷 mutex是否被獲取，如果是的話就返回，否則返回0 int mutex_lock_interruptible(struct mutex *lock) | 使用此函式獲取訊號量失敗進入休眠以後可以被訊號打斷 ## 互斥體例程 ```c struct mutex lock; /* 定義一個互斥體 */ mutex_init(&lock); /* 初始化互斥體 */ mutex_lock(&lock); /* 上鎖 */ /* 臨界區 */ mutex_unlock(&lock); /* 解鎖*/ ``` ## 互斥體與自旋鎖   互斥體和自旋鎖都是解決互斥問題的一種手段。互斥體是程序級別的，互斥體在使用的時候會發生程序間的切換，因此，使用互斥體資源開銷比較大。自旋鎖可以節省上下文切換的時間，如果持有鎖的時間不長，使用自旋鎖是比較好的選擇，如果持有鎖時間比較長，互斥體顯然是更好的選擇。 ## 互斥體使用注意事項 1. 當鎖不能被獲取到時，使用互斥體的開銷是**程序上下文切換時間**，使用自旋鎖的開銷是等待**獲取自旋鎖**（由臨界區執行時間決定）。若臨界區比較小，宜使用自旋鎖，若臨界區很大，應使用互斥體。 2. 互斥體所保護的臨界區可包含可能引起阻塞的程式碼，而自旋鎖則絕對要避免用來保護包含這樣程式碼的臨界區。因為**阻塞意味著要進行程序的切換**，如果程序被切換岀去後，另一個程序企圖獲取本自旋鎖，死鎖就會發生。 3. 互斥體存在於程序上下文。因此，如果被保護的共享資源需要在中斷或軟中斷情況下使用，則在互斥體和自旋鎖之間只能選擇自旋鎖。當然，如果一定要使用互斥體，則只能通過mutex trylock（）方式進行，不能獲取就立即返回以避免阻塞。 **  大家的鼓勵是我繼續創作的動力，如果覺得寫的不錯，歡迎關注，點贊，收藏，轉發，謝謝！** > 有任何問題，均可通過公告中的二維碼