# 同步，第 4 部分：臨界區問題 > 原文：<https://github.com/angrave/SystemProgramming/wiki/Synchronization%2C-Part-4%3A-The-Critical-Section-Problem> ## 候選解決方案 ## 什么是臨界區問題？ 正如[同步，第 3 部分：使用互斥鎖和信號量](/angrave/SystemProgramming/wiki/Synchronization%2C-Part-3%3A-Working-with-Mutexes-And-Semaphores)中已經討論的那樣，我們的代碼的關鍵部分一次只能由一個線程執行。我們將此要求描述為“互斥”;只有一個線程（或進程）可以訪問共享資源。 在多線程程序中，我們可以使用互斥鎖和解鎖調用來包裝關鍵部分： ```c pthread_mutex_lock() - one thread allowed at a time! (others will have to wait here) ... Do Critical Section stuff here! pthread_mutex_unlock() - let other waiting threads continue ``` 我們如何實施這些鎖定和解鎖通話？我們可以創建一個確保相互排斥的算法嗎？下面顯示了不正確的實現， ```c pthread_mutex_lock(p_mutex_t *m) { while(m->lock) {}; m->lock = 1;} pthread_mutex_unlock(p_mutex_t *m) { m->lock = 0; } ``` 乍一看，代碼似乎有效;如果一個線程試圖鎖定互斥鎖，則后一個線程必須等到鎖被清除。然而，該實現 _ 不滿足互斥 _。讓我們從兩個同時運行的線程的角度仔細研究這個“實現”。在下表中，時間從上到下 - | 時間 | 線程 1 | 線程 2 | | --- | --- | --- | | 1 | `while(lock) {}` | | | 2 | | `while(lock) {}` | | 3 | 鎖= 1 | lock = 1 | 哎呀！有競爭條件。在不幸的情況下，兩個線程都檢查了鎖并讀取了一個假值，因此能夠繼續。 ## 臨界區問題的候選解決方案。 為了簡化討論，我們只考慮兩個線程。請注意，這些參數適用于線程和進程，而經典的 CS 文獻則根據需要對關鍵部分或共享資源進行獨占訪問（即互斥）的兩個進程來討論這些問題。 提升標志表示線程/進程進入臨界區的意圖。 請記住，下面列出的偽代碼是更大程序的一部分;線程或進程通常需要在進程的生命周期內多次進入臨界區。因此，想象每個示例都包含在一個循環中，在這個循環中，線程或進程正在處理其他內容。 下面描述的候選解決方案有什么問題嗎？ ``` // Candidate #1 wait until your flag is lowered raise my flag // Do Critical Section stuff lower my flag ``` 答案：候選解決方案＃1 也會遇到競爭條件，即它不滿足互斥，因為兩個線程/進程都可以讀取彼此的標志值（=降低）并繼續。 這表明我們應該在檢查另一個線程的標志之前提高標志 _- 這是下面的候選解決方案＃2。_ ``` // Candidate #2 raise my flag wait until your flag is lowered // Do Critical Section stuff lower my flag ``` 候選人＃2 滿足互斥 - 兩個線程不可能同時進入臨界區。但是這段代碼遇到了死鎖！假設兩個線程希望同時進入臨界區： | Time | Thread 1 | Thread 2 | | --- | --- | --- | | 1 | 升旗 | | | 2 | | raise flag | | 3 | 等...... | wait ... | Ooops 兩個線程/進程現在正在等待另一個線程/進程降低其標志。兩個人都不會進入臨界區，因為它們現在都永遠被卡住了！ 這表明我們應該使用回合制變量來嘗試解決誰應該繼續進行。 ## 回合制解決方案 以下候選解決方案＃3 使用基于回合的變量來禮貌地允許一個線程然后另一個繼續 ``` // Candidate #3 wait until my turn is myid // Do Critical Section stuff turn = yourid ``` 候選者＃3 滿足互斥（每個線程或進程獲得對臨界區的獨占訪問權限），但是兩個線程/進程必須采用嚴格的回合制方法來使用臨界區;即，它們被迫進入交替的臨界區段訪問模式。例如，如果線程 1 希望每毫秒讀取一個哈希表，但另一個線程每秒寫入一個哈希表，則讀取線程必須再等待 999ms 才能再次從哈希表中讀取。這個“解決方案”無效，因為如果臨界區中沒有其他線程，我們的線程應該能夠進行并進入臨界區。 ## 解決臨界區問題所需的屬性？ 在解決臨界區問題時，我們希望有三個主要的理想屬性 * 相互排斥 - 線程/進程獲得獨占訪問權限;其他人必須等到它退出臨界區。 * 有界等待 - 如果線程/進程必須等待，那么它只需要等待一段有限的時間（不允許無限等待時間！）。有界等待的確切定義是，在給定進程進入之前，任何其他進程可以進入其臨界區的次數存在上限（非無限）。 * 進度 - 如果臨界區內沒有線程/進程，則線程/進程應該能夠繼續（進行中）而不必等待。 考慮到這些想法，讓我們檢查另一個使用轉向標志的候選解決方案，前提是兩個線程同時需要訪問。 ## Turn 和 Flag 解決方案 以下是 CSP 的正確解決方案嗎？ ``` \\ Candidate #4 raise my flag if your flag is raised, wait until my turn // Do Critical Section stuff turn = yourid lower my flag ``` 一位教練和另一位 CS 教員最初這么認為！但是，分析這些解決方案很棘手。即使是關于這一特定主題的同行評審論文也包含不正確的解乍一看它似乎滿足相互排斥，有界等待和進步：回合制標志僅用于平局事件（因此允許進度和有界等待）并且似乎滿足互斥。但是......也許你能找到一個反例？ 候選者＃4 失敗，因為線程不會等到另一個線程降低其標志。經過一些思考（或靈感）后，可以創建以下場景來演示如何不滿足互斥。 想象一下，第一個線程運行此代碼兩次（因此轉向標志現在指向第二個線程）。當第一個線程仍在 Critical Section 中時，第二個線程到達。第二個線程可以立即繼續進入臨界區！ | Time | 轉 | 線程＃1 | 線程＃2 | | --- | --- | --- | --- | | 1 | 2 | 舉起我的旗幟 | | | 2 | 2 | 如果你的旗幟升起，請等到輪到我 | raise my flag | | 3 | 2 | //做關鍵部分的事情 | 如果你的旗幟升起，請等到輪到我了（TRUE！） | | 4 | 2 | // Do Critical Section stuff | //做關鍵部分 - OOPS | ## 工作方案 ## 彼得森的解決方案是什么？ 彼得森在 1981 年的一篇 2 頁的論文中發表了他的小說和令人驚訝的簡單解決方案。他的算法的一個版本如下所示，使用共享變量`turn`： ``` \\ Candidate #5 raise my flag turn = your_id wait until your flag is lowered and turn is yourid // Do Critical Section stuff lower my flag ``` 該解決方案滿足互斥，有界等待和進步。如果線程＃2 已設置為 2 并且當前位于關鍵部分內。線程＃1 到達，_ 將轉回設置為 1_ ，現在等待直到線程 2 降低標志。 鏈接彼得森的原始文章 pdf： [G. L. Peterson：“關于互斥問題的神話”，信息處理快報 12（3）1981,115-116](http://dl.acm.org/citation.cfm?id=945527) ## 彼得森的解決方案是第一個解決方案嗎？ 不，Dekkers 算法（1962）是第一個可證明正確的解決方案。該算法的一個版本如下。 ``` raise my flag while(your flag is raised) : if it's your turn to win : lower my flag wait while your turn raise my flag // Do Critical Section stuff set your turn to win lower my flag ``` 注意如果循環迭代為零，一次或多次，如何在臨界區期間始終引發進程的標志。此外，該標志可以被解釋為進入臨界區的直接意圖。只有當其他進程也引發了標志時，一個進程才會推遲，降低其意圖標志并等待。 ## 我可以在 C 或匯編程序中實現 Peterson（或 Dekkers）算法嗎？ 是的 - 通過一些搜索，即使在今天也可以在生產中找到特定的簡單移動處理器：Peterson 的算法用于為 Tegra 移動處理器實現低級 Linux 內核鎖（片上系統 ARM 處理和 Nvidia 的 GPU 核心） [https://android.googlesource.com/kernel/tegra.git/+/android-tegra-3.10/arch/arm/mach-tegra/sleep.S#58](https://android.googlesource.com/kernel/tegra.git/+/android-tegra-3.10/arch/arm/mach-tegra/sleep.S#58) 但是，通常，如果另一個核心更新共享變量，CPU 和 C 編譯器可以重新排序 CPU 指令或使用過時的 CPU 核心本地緩存值。因此，對于大多數平臺來說，簡單的 C 代碼偽代碼實在太天真了。你現在可以停止閱讀了。 哦......你決定繼續讀書。好吧，這里是龍！不要說我們沒有警告你。考慮這個高級和粗糙的話題，但（劇透警報）一個快樂的結局。 考慮以下代碼， ```c while(flag2 ) { /* busy loop - go around again */ ``` 一個有效的編譯器會推斷出`flag2`變量永遠不會在循環內部發生變化，因此測試可以優化到`while(true)`使用`volatile`來防止這種編譯器優化。 獨立指令可以由優化編譯器重新排序，或者在運行時由 CPU 進行無序執行優化。如果代碼需要修改和檢查變量以及精確的順序，則進行這些復雜的優化。 相關的挑戰是 CPU 核心包括用于存儲最近讀取或修改的主存儲器值的數據高速緩存。修改后的值可能無法寫回主存儲器或立即從存儲器重新讀取。因此，在兩個 CPU 代碼之間可能不共享數據改變，例如上述示例中的標志和轉彎變量的狀態。 但結局很快樂。幸運的是，現代硬件使用“內存屏障”（也稱為內存屏障）CPU 指令來解決這些問題，以確保主內存和 CPU 的緩存處于合理且連貫的狀態。更高級別的同步原語（例如`pthread_mutex_lock`）將調用這些 CPU 指令作為其實現的一部分。因此，在實踐中，使用互斥鎖定和解鎖調用的周圍臨界區域足以忽略這些較低級別的問題。 進一步閱讀：我們建議以下網絡帖子討論在 x86 進程上實現 Peterson 的算法以及關于內存障礙的 linux 文檔。 [http://bartoszmilewski.com/2008/11/05/who-ordered-memory-fences-on-an-x86/](http://bartoszmilewski.com/2008/11/05/who-ordered-memory-fences-on-an-x86/) [http://lxr.free-electrons.com/source /Documentation/memory-barriers.txt](http://lxr.free-electrons.com/source/Documentation/memory-barriers.txt) ## 硬件方案 ## 我們如何在硬件上實現關鍵部分問題？ 我們可以使用 C11 Atomics 完美地完成這項工作！這里詳細介紹了一個完整的解決方案（這是一個自旋鎖互斥， [futex](https://locklessinc.com/articles/mutex_cv_futex/) 實現可以在網上找到）。 ```c typedef struct mutex_{ atomic_int_least8_t lock; pthread_t owner; } mutex; #define UNLOCKED 0 #define LOCKED 1 #define UNASSIGNED_OWNER 0 int mutex_init(mutex* mtx){ if(!mtx){ return 0; } atomic_init(&mtx->lock, UNLOCKED); // Not thread safe the user has to take care of this mtx->owner = UNASSIGNED_OWNER; return 1; } ``` 這是初始化代碼，這里沒什么特別的。我們將互斥鎖的狀態設置為已解鎖并將所有者設置為已鎖定。 ```c int mutex_lock(mutex* mtx){ int_least8_t zero = UNLOCKED; while(!atomic_compare_exchange_weak_explicit (&mtx->lock, &zero, LOCKED, memory_order_relaxed, memory_order_relaxed)){ zero = UNLOCKED; sched_yield(); //Use system calls for scheduling speed } //We have the lock now!!!! mtx->owner = pthread_self(); return 1; } ``` 哎呀！這段代碼有什么作用？好吧，啟動它初始化一個我們將保持為解鎖狀態的變量。 [原子比較和交換](https://en.wikipedia.org/wiki/Compare-and-swap)是大多數現代架構支持的指令（在 x86 上它是`lock cmpxchg`）。此操作的偽代碼如下所示 ```c int atomic_compare_exchange_pseudo(int* addr1, int* addr2, int val){ if(*addr1 == *addr2){ *addr1 = val; return 1; }else{ *addr2 = *addr1; return 0; } } ``` 除了全部完成 _ 原子 _ 意味著在一個不間斷的操作中。 _ 弱 _ 部分意味著什么？原子指令也容易出現**虛假失敗**意味著這些原子函數有兩個版本 _ 強 _ 和 _ 弱 _ 部分，強有力保證成功或失敗而弱者可能會失敗。我們使用弱，因為弱更快，我們處于循環中！這意味著如果它經常失敗，我們就可以了，因為無論如何我們都會繼續旋轉。 這個記憶訂單業務是什么？我們之前討論的是記憶圍欄，現在就是這樣！我們不會詳細說明，因為它超出了本課程的范圍，而不是本文 [的范圍。](https://gcc.gnu.org/wiki/Atomic/GCCMM/AtomicSync) 在 while 循環中，我們未能抓住鎖！我們將零重置為解鎖并暫停一會兒。當我們醒來時，我們試圖再次抓住鎖。一旦我們成功交換，我們就處于關鍵部分！我們將互斥鎖的所有者設置為解鎖方法的當前線程并返回成功。 這如何保證互斥，在使用原子時我們并不完全確定！但是在這個簡單的例子中我們可以因為能夠成功地期望鎖定為 UNLOCKED（0）并將其交換為 LOCKED（1）狀態的線程被認為是贏家。我們如何實施解鎖？ ```c int mutex_unlock(mutex* mtx){ if(unlikely(pthread_self() != mtx->owner)){ return 0; //You can't unlock a mutex if you aren't the owner } int_least8_t one = 1; //Critical section ends after this atomic mtx->owner = UNASSIGNED_OWNER; if(!atomic_compare_exchange_strong_explicit( &mtx->lock, &one, UNLOCKED, memory_order_relaxed, memory_order_relaxed)){ //The mutex was never locked in the first place return 0; } return 1; } ``` 為了滿足 api，你不能解鎖互斥鎖，除非你是擁有它的人。然后我們取消分配互斥鎖所有者，因為關鍵部分在原子之后結束了。我們想要一個強大的交換，因為我們不想阻止（pthread_mutex_unlock 不會阻塞）。我們希望鎖定互斥鎖，然后將其交換為解鎖。如果交換成功，我們解鎖互斥鎖。如果交換不是，這意味著互斥鎖已解鎖，我們嘗試將其從 UNLOCKED 切換到 UNLOCKED，從而保持解鎖的非阻塞。