可重復(fù)使用的并行數(shù)據(jù)結(jié)構(gòu)和算法

1、9 種可重復(fù)使用的并行數(shù)據(jù)結(jié)構(gòu)和算法目錄 倒計(jì)數(shù)鎖存 (Countdown Latch) 可重用旋轉(zhuǎn)等待 (Spin Wait) 屏障 (Barrier) 阻塞隊(duì)列 受限緩沖區(qū) (Bounded Buffer) Thin 事件 無鎖定 LIFO 堆棧 循環(huán)分塊 并行分拆 總結(jié) 本專欄并未涉及很多公共語言運(yùn)行庫 (CLR) 功能的機(jī)制問題，而是更多介紹了如何有效使用您手頭所具有的工具。身為一名程序員，必須做出很多決策，而選擇正確的數(shù)據(jù)結(jié)構(gòu)和算法無疑是最常見的，也是最重要的決策之一。錯(cuò)誤的選擇可能導(dǎo)致程序無法運(yùn)行，而大多數(shù)情況下，那么決定了性能的好壞。鑒于并行編程通常旨在改良性能，并且要難于串行編

2、程，因此所作的選擇對(duì)您程序的成功就更為重要。在本專欄中，我們將介紹九種可重復(fù)使用的數(shù)據(jù)結(jié)構(gòu)和算法，這些結(jié)構(gòu)和算法是許多并行程序所常用的，您應(yīng)該能夠輕松將它們應(yīng)用到自己的 .NET 軟件中。專欄中每個(gè)例如隨附的代碼都是可用的，但尚未經(jīng)過完全定型、測(cè)試和優(yōu)化。這里列舉的模式雖然并不詳盡，但卻代表了一些較為常見的模式。如您所見，很多例如都是互為補(bǔ)充的。在開始前，我想還是先介紹一些相關(guān)內(nèi)容。Microsoft .NET Framework 提供了幾個(gè)現(xiàn)有的并發(fā)基元。雖然我要為您講解如何構(gòu)建自己的基元，但實(shí)際上現(xiàn)有基元是足以應(yīng)付大多數(shù)情況的。我只是想說某些可選的方案有時(shí)也是有參考價(jià)值的。此外，了解這些技

3、巧如何應(yīng)用于實(shí)際操作也有助于加深您對(duì)并行編程的整體理解。在開始講解前，我假定您對(duì)現(xiàn)有基元已經(jīng)有了一個(gè)根本的了解。您也可以參閱?MSDN 雜志?2005 年 8 月版的文章“關(guān)于多線程應(yīng)用程序：每個(gè)開發(fā)人員都應(yīng)了解的內(nèi)容，以全面了解其概念。一、倒計(jì)數(shù)鎖存 (Countdown Latch)Semaphore 之所以成為并發(fā)編程中一種較為知名的數(shù)據(jù)結(jié)構(gòu)，原因是多方面的，而并不只是因?yàn)樗谟?jì)算機(jī)科學(xué)領(lǐng)域有著悠久的歷史可以追溯到 19 世紀(jì) 60 年代的操作系統(tǒng)設(shè)計(jì)。Semaphore 只是一種帶有一個(gè)計(jì)數(shù)字段的數(shù)據(jù)結(jié)構(gòu)，它只支持兩種操作：放置和取走通常分別稱為 P 和 V。一次放置操作會(huì)增加一個(gè) s

4、emaphore 計(jì)數(shù)，而一次取走操作會(huì)減少一個(gè) semaphore 計(jì)數(shù)。當(dāng) semaphore 計(jì)數(shù)為零時(shí)，除非執(zhí)行一項(xiàng)并發(fā)的放置操作使計(jì)數(shù)變?yōu)榉橇阒担衲敲慈魏魏罄m(xù)的取走嘗試都將阻塞等待。這兩種操作均為不可再分 (atomic) 操作，并發(fā)時(shí)不會(huì)產(chǎn)生錯(cuò)誤，能夠確保并發(fā)的放置和取走操作有序地進(jìn)行。Windows具有根底內(nèi)核和對(duì) semaphore 對(duì)象的 Win32支持請(qǐng)參閱 CreateSemaphore 和相關(guān) API，并且在 .NET Framework 中這些對(duì)象可以通過 System.Threading.Semaphore 類公開到上層。Mutex 和 Monitor 所支持的臨

5、界區(qū)，通常被認(rèn)為是一種特殊的 semaphore，其計(jì)數(shù)會(huì)在 0 和 1 之間來回切換，換句話說，是一個(gè)二進(jìn)制的 semaphore。另外還有一種“反向 semaphore也是非常有用。也就是說，有時(shí)您需要數(shù)據(jù)結(jié)構(gòu)能夠等待數(shù)據(jù)結(jié)構(gòu)計(jì)數(shù)歸零。Fork/join 式并行模式在數(shù)據(jù)并行編程中是極為常見的，其中由單個(gè)“主線程控制執(zhí)行假設(shè)干“輔助線程并等待線程執(zhí)行完畢。在這類情況下，使用反向 semaphore 會(huì)很有幫助。大多數(shù)時(shí)候，您其實(shí)并不想喚醒線程來修改計(jì)數(shù)。因此在這種情況下，我們將結(jié)構(gòu)稱為倒計(jì)數(shù)“鎖存，用來表示計(jì)數(shù)的減少，同時(shí)還說明一旦設(shè)置為“Signaled狀態(tài)，鎖存將保持“signaled

6、這是一個(gè)與鎖存相關(guān)的屬性。遺憾的是，Windows 和 .NET Framework 均不支持這種數(shù)據(jù)結(jié)構(gòu)。但令人欣慰的是，構(gòu)建這種數(shù)據(jù)閉鎖并不難。要構(gòu)建倒計(jì)數(shù)鎖存，只需將其計(jì)數(shù)器初始值設(shè)為 n，并讓每項(xiàng)輔助任務(wù)在完成時(shí)不可再分地將 n 減掉一個(gè)計(jì)數(shù)，這可以通過為減量操作加上“鎖或調(diào)用 Interlocked.Decrement 來實(shí)現(xiàn)。接下來，線程可以不執(zhí)行取走操作，而是減少計(jì)數(shù)并等待計(jì)數(shù)器歸零；而當(dāng)線程被喚醒時(shí)，它就可以得知已經(jīng)有 n 個(gè)信號(hào)向鎖存注冊(cè)。在 while (count != 0) 循環(huán)中，讓等待的線程阻塞通常是不錯(cuò)的選擇這種情況下，您稍后將不得不使用事件，而不是使用旋轉(zhuǎn)。pu

7、blic class CountdownLatch private int m_remain; private EventWaitHandle m_event; public CountdownLatch(int count) m_remain = count; m_event = new ManualResetEvent(false); public void Signal() / The last thread to signal also sets the event. if (Interlocked.Decrement(ref m_remain) = 0) m_event.Set();

8、 public void Wait() m_event.WaitOne(); 這看上去極為簡單，但要正確運(yùn)用還需要技巧。稍后我們將通過一些例如來講解如何使用這種數(shù)據(jù)結(jié)構(gòu)。請(qǐng)注意，此處所示根本實(shí)現(xiàn)還有很多可以改良地方，例如：在事件上調(diào)用 WaitOne 之前添加某種程度的旋轉(zhuǎn)等待、緩慢分配事件而不是在構(gòu)造器中進(jìn)行分配以防足夠的旋轉(zhuǎn)會(huì)防止出現(xiàn)阻塞，如本專欄稍后介紹的 ThinEvent 演示的那樣、添加重置功能以及提供 Dispose 方法以便在不再需要內(nèi)部事件對(duì)象時(shí)將對(duì)象關(guān)閉。二、可重用旋轉(zhuǎn)等待 (Spin Wait)雖然忙碌等待 (busy waiting) 更容易實(shí)現(xiàn)阻塞，但在某些情況下，您

9、也許確實(shí)想在退回到真正的等待狀態(tài)前先旋轉(zhuǎn) (spin) 一段時(shí)間。我們很難理解為何這樣做會(huì)有幫助，而大多數(shù)人之所以一開始就防止旋轉(zhuǎn)等待，是因?yàn)樾D(zhuǎn)看上去像是在做無用功；如果上下文切換每當(dāng)線程等待內(nèi)核事件時(shí)都會(huì)發(fā)生需要幾千個(gè)周期在 Windows 上確實(shí)是這樣，我們稱之為 c，并且線程所等待的條件出現(xiàn)的時(shí)間少于 2c 周期時(shí)間1c 用于等待自身，1c 用于喚醒，那么旋轉(zhuǎn)可以降低等待所造成的系統(tǒng)開銷和滯后時(shí)間，從而提升算法的整體吞吐量和可伸縮性。如果您決定使用旋轉(zhuǎn)等待，就必須謹(jǐn)慎行事。因?yàn)槿绻@樣做，您可能需要注意很多問題，比方：要確保在旋轉(zhuǎn)循環(huán)內(nèi)調(diào)用 Thread.SpinWait，以提高 In

10、tel 超線程技術(shù)的計(jì)算機(jī)上硬件對(duì)其他硬件線程的可用性；偶爾使用參數(shù) 1 而非 0 來調(diào)用 Thread.Sleep，以防止優(yōu)先級(jí)反向問題；通過輕微的回退 (back-off) 來引入隨機(jī)選擇，從而改善訪問的局部性假定調(diào)用方持續(xù)重讀共享狀態(tài)并可能防止活鎖；當(dāng)然，在單 CPU 的計(jì)算機(jī)最好不要采用這種方法因?yàn)樵谶@種環(huán)境下旋轉(zhuǎn)是非常浪費(fèi)資源的。SpinWait 類需要被定義為值類型，以便分配起來更加節(jié)省資源?，F(xiàn)在，我們可以使用此算法來防止前述 CountdownLatch 算法中出現(xiàn)的阻塞。public struct SpinWait private int m_count; private st

11、atic readonly bool s_isSingleProc = (Environment.ProcessorCount = 1); private const int s_yieldFrequency = 4000; private const int s_yieldOneFrequency = 3*s_yieldFrequency; public int Spin() int oldCount = m_count; / On a single-CPU machine, we ensure our counter is always / a multiple of s_yieldFre

12、quency, so we yield every time. / Else, we just increment by one. m_count += (s_isSingleProc ? s_yieldFrequency : 1); / If not a multiple of s_yieldFrequency spin (w/ backoff). int countModFrequency = m_count % s_yieldFrequency; if (countModFrequency 0) Thread.SpinWait(int)(1 + (countModFrequency *

13、0.05f); else Thread.Sleep(m_count = s_yieldOneFrequency ? 0 : 1); return oldCount; private void Yield() Thread.Sleep(m_count 0) if (s.Spin() = s_spinCount) m_event.WaitOne(); 不可否認(rèn)，選擇頻率和旋轉(zhuǎn)計(jì)數(shù)是不確定的。與 Win32 臨界區(qū)旋轉(zhuǎn)計(jì)數(shù)類似，我們應(yīng)該根據(jù)測(cè)試和實(shí)驗(yàn)的結(jié)果來選擇合理的數(shù)值，而且即使合理的數(shù)值在不同系統(tǒng)中也會(huì)發(fā)生變化。例如，根據(jù) Microsoft Media Center 和 Windows ker

14、nel 團(tuán)隊(duì)的經(jīng)驗(yàn)，MSDN 文檔建議臨界區(qū)旋轉(zhuǎn)計(jì)數(shù)為 4,000 ，但您的選擇可以有所不同。理想的計(jì)數(shù)取決于多種因素，包括在給定時(shí)間等待事件的線程數(shù)和事件出現(xiàn)的頻率等。大多數(shù)情況下，您會(huì)希望通過等待事件來消除顯式讓出時(shí)間，如鎖存的例如中所示。您甚至可以選擇動(dòng)態(tài)調(diào)整計(jì)數(shù)：例如，從中等數(shù)量的旋轉(zhuǎn)開始，每次旋轉(zhuǎn)失敗就增加計(jì)數(shù)。一旦計(jì)數(shù)到達(dá)預(yù)定的最大值，就完全停止旋轉(zhuǎn)并立即發(fā)出 WaitOne。邏輯如下所示：您希望立即增加到達(dá)預(yù)定的最大周期數(shù)，但卻無法超過最大周期數(shù)。如果您發(fā)現(xiàn)此最大值缺乏以阻止上下文切換，那么立即執(zhí)行上下文切換總的算來占用的資源更少。慢慢您就會(huì)希望旋轉(zhuǎn)計(jì)數(shù)能夠到達(dá)一個(gè)穩(wěn)定的值。三、

15、屏障 (Barrier)屏障，又稱集合點(diǎn)，是一種并發(fā)性基元，它無需另一“主線程控制即可實(shí)現(xiàn)各線程之間簡單的互相協(xié)調(diào)。每個(gè)線程在到達(dá)屏障時(shí)都會(huì)不可再分地發(fā)出信號(hào)并等待。僅當(dāng)所有 n 都到達(dá)屏障時(shí)，才允許所有線程繼續(xù)。這種方法可用于協(xié)調(diào)算法 (cooperative algorithms)，該算法廣泛應(yīng)用于科學(xué)、數(shù)學(xué)和圖形領(lǐng)域。很多計(jì)算中都適合使用屏障，實(shí)際上，甚至 CLR 的垃圾收集器都在使用它們。屏障只是將較大的計(jì)算分割為假設(shè)干較小的協(xié)作階段 (cooperative phase)，例如：const int P = .;Barrier barrier = new Barrier(P);Data

16、 partitions = new DataP;/ Running on P separate threads in parallel:public void Body(int myIndex) FillMyPartition(partitionsmyIndex); barrier.Await(); ReadOtherPartition(partitionsP myIndex - 1); barrier.Await(); / .您會(huì)很快發(fā)現(xiàn)在這種情況下是可以使用倒計(jì)數(shù)鎖存的。每個(gè)線程都可以在調(diào)用 Signal 后立即調(diào)用 Wait，而不是調(diào)用 Await；在到達(dá)屏障后，所有線程都會(huì)被釋放。但這

17、里存在一個(gè)問題：前面所講的鎖存并不支持屢次重復(fù)使用同一對(duì)象，而這卻是所有屏障都支持的一個(gè)常用屬性。實(shí)際上，上述例如就需要使用此屬性。您可以通過單獨(dú)的屏障對(duì)象來實(shí)現(xiàn)這一點(diǎn)，但這樣做非常浪費(fèi)資源；而由于所有線程每次只出現(xiàn)在一個(gè)階段，因此根本無需多個(gè)屏障對(duì)象。要解決這個(gè)問題，您可以使用相同的根底倒計(jì)數(shù)鎖存算法來處理減少計(jì)數(shù)器計(jì)數(shù)、發(fā)信號(hào)指示事件、等待等方面的問題，從而將其擴(kuò)展為可重復(fù)使用。要實(shí)現(xiàn)這一點(diǎn)，您需要使用所謂的感應(yīng)反向屏障 (sense reversing barrier)，該屏障需要在“偶數(shù)和“奇數(shù)階段之間交替。在交替階段需要使用單獨(dú)的事件。using System;using Syste

18、m.Threading;public class Barrier private volatile int m_count; private int m_originalCount; private EventWaitHandle m_oddEvent; private EventWaitHandle m_evenEvent; private volatile bool m_sense = false; / false=even, true=odd. public Barrier(int count) m_count = count; m_originalCount = count; m_od

19、dEvent = new ManualResetEvent(false); m_evenEvent = new ManualResetEvent(false); public void Await() bool sense = m_sense; / The last thread to signal also sets the event. if (m_count = 1 | Interlocked.Decrement(ref m_count) = 0) m_count = m_originalCount; m_sense = !sense; / Reverse the sense. if (

20、sense = true) / odd m_evenEvent.Reset(); m_oddEvent.Set(); else / even m_oddEvent.Reset(); m_evenEvent.Set(); else if (sense = true) m_oddEvent.WaitOne(); else m_evenEvent.WaitOne(); 為何需要兩個(gè)事件，其原因很難講清楚。一種方法是在 Await 中先執(zhí)行 Set 隨后立即執(zhí)行 Reset，但這很危險(xiǎn)，并會(huì)導(dǎo)致死鎖。原因有二：第一，另一線程的 m_count 可能已減少，但在依次快速調(diào)用 Set 和 Reset 時(shí)，

21、計(jì)數(shù)的值還缺乏以到達(dá)可在事件上調(diào)用 WaitOne。第二，雖然等待的線程可能已經(jīng)能夠調(diào)用 WaitOne，但可報(bào)警等待如 CLR 一貫使用的那些可以中斷等待，從而暫時(shí)從等待隊(duì)列中刪除等待的線程，以便運(yùn)行異步過程調(diào)用 (APC)。等待的線程將始終看不到事件處于設(shè)置 (set) 狀態(tài)。兩種情況均會(huì)導(dǎo)致事件喪失，并可能導(dǎo)致死鎖。針對(duì)奇數(shù)階段和偶數(shù)階段使用單獨(dú)的事件即可防止這種情況。您可能想繼續(xù)像 CountdownLatch 中那樣將旋轉(zhuǎn)添加到 Barrier。但如果您嘗試這樣做，可能會(huì)遇到一個(gè)問題：一般情況下，旋轉(zhuǎn)線程會(huì)開始旋轉(zhuǎn)直到 m_count 歸零。但通過上述實(shí)現(xiàn)，m_count 的值實(shí)際上

22、永遠(yuǎn)不會(huì)為零，除非最后一個(gè)線程將其重置為 m_originalCount。這種想當(dāng)然的方法將導(dǎo)致一個(gè)或多個(gè)線程進(jìn)行旋轉(zhuǎn)永遠(yuǎn)地，而其他線程那么會(huì)在下一階段阻塞也是永遠(yuǎn)地。解決方法很簡單。您旋轉(zhuǎn)，等待感應(yīng)發(fā)生變化public void Await() bool sense = m_sense; / The last thread to signal also sets the event. if (m_count = 1 | Interlocked.Decrement(ref m_count) = 0) m_count = m_originalCount; m_sense = !sense; /

23、Reverse the sense. if (sense = true) / odd m_evenEvent.Set(); m_oddEvent.Reset(); else / even m_oddEvent.Set(); m_evenEvent.Reset(); else SpinWait s = new SpinWait(); while (sense = m_sense) if (s.Spin() = s_spinCount) if (sense = true) m_oddEvent.WaitOne(); else m_evenEvent.WaitOne(); 由于所有線程都必須離開前一

24、階段的 Await 才可以完成下一階段，因此可以確定，所有線程要么會(huì)觀察到感應(yīng)發(fā)生變化，要么最終等待事件并被喚醒。阻塞隊(duì)列在共享內(nèi)存的體系結(jié)構(gòu)中，兩項(xiàng)或多項(xiàng)任務(wù)間唯一的同步點(diǎn)通常是一個(gè)位于中樞的共享集合的數(shù)據(jù)結(jié)構(gòu)。通常，一項(xiàng)或多項(xiàng)任務(wù)會(huì)負(fù)責(zé)為其他一項(xiàng)或多項(xiàng)任務(wù)生成要執(zhí)行的“工作，我們稱之為生產(chǎn)者/使用者關(guān)系 (producer/consumer)。這類數(shù)據(jù)結(jié)構(gòu)的簡單同步通常是非常簡單的 使用 Monitor 或 ReaderWriterLock 即可解決，但當(dāng)緩沖區(qū)為空時(shí)，任務(wù)間的協(xié)調(diào)就會(huì)變得比擬困難。要解決這個(gè)問題，通常需要使用阻塞隊(duì)列。實(shí)際上，阻塞隊(duì)列有幾種稍微不同的變體，包括簡單變體和復(fù)

25、雜變體。在簡單變量中，使用者僅在隊(duì)列為空時(shí)才會(huì)阻塞，而在復(fù)雜變量中，每個(gè)生產(chǎn)者都正好“配有一個(gè)使用者，也就是說，在使用者到達(dá)并開始處理排隊(duì)工程之前，生產(chǎn)者會(huì)被阻塞，同理，在生產(chǎn)者交付工程前，所有使用者也會(huì)被阻塞。這時(shí)通常使用 FIFO先進(jìn)先出排序，但我們并不總是有必要這么做。我們也可以對(duì)緩沖區(qū)進(jìn)行限制，這一點(diǎn)稍后會(huì)為大家介紹。由于稍后將要介紹的受限緩沖區(qū)包含更為簡單的“為空時(shí)阻塞(blocking-when-empty) 行為，因此我們這里僅對(duì)配對(duì)變量做一了解。要實(shí)現(xiàn)這個(gè)目的，我們只需封裝一個(gè)簡單的 Queue，上方保持同步。那么到底需要何種同步？每當(dāng)線程對(duì)元素進(jìn)行排隊(duì)時(shí)，在返回前都會(huì)等待使用

26、者取消元素排隊(duì)。當(dāng)線程取消元素排隊(duì)時(shí)，如果發(fā)現(xiàn)緩沖區(qū)為空，那么必須等待新元素的進(jìn)入。當(dāng)然，取消排隊(duì)后，使用者必須通知生產(chǎn)者已取到其工程請(qǐng)參見圖 5。Figure5阻塞隊(duì)列 復(fù)制代碼 class Cell internal T m_obj; internal Cell(T obj) m_obj = obj; public class BlockingQueue private QueueCell m_queue = new QueueCell(); public void Enqueue(T obj) Cell c = new Cell(obj); lock (m_queue) m_queue.

27、Enqueue(c); Monitor.Pulse(m_queue); Monitor.Wait(m_queue); public T Dequeue() Cell c; lock (m_queue) while (m_queue.Count = 0) Monitor.Wait(m_queue); c = m_queue.Dequeue(); Monitor.Pulse(m_queue); return c.m_obj; 請(qǐng)注意，我們可以在 Enqueue 方法中依次調(diào)用 Pulse 和 Wait，類似地，在 Dequeue 方法中依次調(diào)用 Wait 和 Pulse。只有在釋放監(jiān)視器之后，才會(huì)

28、對(duì)內(nèi)部事件進(jìn)行設(shè)置，而由于監(jiān)視器的這種實(shí)現(xiàn)方法，會(huì)導(dǎo)致某個(gè)線程調(diào)度 ping-pong。我們可能會(huì)想，也許可以使用 Win32 事件來構(gòu)建一個(gè)更加優(yōu)化的通知機(jī)制。但是，使用這類完善的 Win32 事件會(huì)帶來巨大開銷，尤其是使用它們時(shí)所進(jìn)行的本錢分配和內(nèi)核跳轉(zhuǎn) (kernel transitions)，因此還是考慮其他選擇吧。您可以像 CLR 的 ReaderWriterLock 那樣將這些時(shí)間集中到池中，也可以像 ThinEvent 類型稍后介紹一樣緩慢分配它們。這種實(shí)現(xiàn)方法也是有弊端的，即要為每個(gè)新元素分配對(duì)象，備選方法可能也會(huì)將這些對(duì)象參加到池中，但同樣會(huì)帶來其他麻煩。受限緩沖區(qū) (Bou

29、nded Buffer)某些類別的隊(duì)列中會(huì)出現(xiàn)資源使用問題。如果生產(chǎn)者任務(wù)創(chuàng)立工程的速度快于使用者處理工程的速度，那么系統(tǒng)的內(nèi)存使用將不受限制。為了說明這個(gè)問題，我們假設(shè)一個(gè)系統(tǒng)，單個(gè)生產(chǎn)者每秒鐘可將 50 個(gè)工程排入隊(duì)列，而使用者每秒鐘只能使用 10 個(gè)工程。首先，這樣會(huì)打亂系統(tǒng)的平衡性，而且對(duì)于一對(duì)一的生產(chǎn)者 使用者配置無法進(jìn)行很好的調(diào)整。只需一分鐘，就會(huì)有 2,400 個(gè)工程堆積在緩沖區(qū)中。假設(shè)這些工程中每個(gè)工程使用 10KB 內(nèi)存，那將意味著緩沖區(qū)本身就會(huì)占用 24MB 內(nèi)存。一小時(shí)后，這個(gè)數(shù)字將飆升至 1GB 以上。解決這一問題的一個(gè)方法是調(diào)整生產(chǎn)者線程與使用者線程的比例，在上述情況

30、中，要將比例調(diào)整為一個(gè)生產(chǎn)者對(duì)應(yīng)五個(gè)使用者。但是到達(dá)速度通常是不穩(wěn)定的，這會(huì)導(dǎo)致系統(tǒng)周期性的不穩(wěn)定，并帶來一些明顯的問題，簡單的固定比例是無法解決這個(gè)問題的。效勞器上的程序通常是長時(shí)間運(yùn)行的，人們希望程序能夠長期保持良好的運(yùn)行狀態(tài)，但如果內(nèi)存使用不受限，就會(huì)造成不可防止的混亂，還可能導(dǎo)致必須定期回收效勞器進(jìn)程的情況。受限緩沖區(qū)允許您對(duì)生產(chǎn)者強(qiáng)制阻塞前緩沖區(qū)可能到達(dá)的大小進(jìn)行限制。阻塞生產(chǎn)者可讓使用者有時(shí)機(jī)“趕上通過允許其線程接收調(diào)度時(shí)間片，同時(shí)還能夠解決內(nèi)存使用量增加的問題。我們的方法還是僅封裝 Queue，同時(shí)添加兩個(gè)等待條件和兩個(gè)事件通知條件：生產(chǎn)者在隊(duì)列滿時(shí)等待，直至隊(duì)列變?yōu)榉菨M，而使用

31、者在隊(duì)列空時(shí)等待，直至變?yōu)榉强眨簧a(chǎn)者在生產(chǎn)出工程后通知等待的使用者，而使用者也會(huì)在取走工程后通知生產(chǎn)者，如圖 6 中所示。Figure6 受限緩沖區(qū) (Bounded Buffer) 復(fù)制代碼 public class BoundedBuffer private Queue m_queue = new Queue(); private int m_consumersWaiting; private int m_producersWaiting; private const int s_maxBufferSize = 128; public void Enqueue(T obj) lock (

32、m_queue) while (m_queue.Count = (s_maxBufferSize - 1) m_producersWaiting+; Monitor.Wait(m_queue); m_producersWaiting-; m_queue.Enqueue(obj); if (m_consumersWaiting 0) Monitor.PulseAll(m_queue); public T Dequeue() T e; lock (m_queue) while (m_queue.Count = 0) m_consumersWaiting+; Monitor.Wait(m_queue

33、); m_consumersWaiting-; e = m_queue.Dequeue(); if (m_producersWaiting 0) Monitor.PulseAll(m_queue); return e; 這里我們?cè)僖淮问褂昧艘环N比擬天真的方法。但是我們確實(shí)優(yōu)化了對(duì) PulseAll 的調(diào)用因?yàn)樗鼈兿牡馁Y源很多，方法是使用 m_consumersWaiting 和 m_producersWaiting 這兩個(gè)計(jì)數(shù)器并僅就計(jì)數(shù)器值是否為零發(fā)出信號(hào)。除此以外，我們還可以再做進(jìn)一步的改良。例如，共享與此類似的單個(gè)事件可能會(huì)喚醒過多線程：如果使用者將隊(duì)列大小降為 0，并且生產(chǎn)者和使用者

34、同時(shí)處于等待狀態(tài)，顯然您只希望喚醒生產(chǎn)者至少開始時(shí)要這么做。此實(shí)現(xiàn)將以先進(jìn)先出的規(guī)那么處理所有等待者，這意味著您可能需要在任一生產(chǎn)者運(yùn)行前喚醒使用者，這樣做僅僅是為了讓使用者發(fā)現(xiàn)隊(duì)列為空，然后再次等待。令人欣慰的是，最終出現(xiàn)生產(chǎn)者和使用者同時(shí)等待的情況是很少見的，但其出現(xiàn)也是有規(guī)律的，而且受限區(qū)域較小。Thin 事件與 Monitor.Wait、Pulse 和 PulseAll 相比，Win32 事件有一個(gè)很大的優(yōu)勢(shì)：它們是“粘滯的(sticky)。也就是說，一旦有事件收到信號(hào)，任何后續(xù)等待都將立即取消阻止，即使線程在信號(hào)發(fā)出前尚未開始等待。如果沒有這個(gè)功能，您就需要經(jīng)常編寫一些代碼將所有等待

35、和信號(hào)通知嚴(yán)格地限制在臨界區(qū)域內(nèi)，由于 Windows 方案程序總是會(huì)提升已喚醒線程的優(yōu)先級(jí)，因此這些代碼的效率很低；如果不采取這種方法，您就必須使用某種技巧型很強(qiáng)、容易出現(xiàn)競(jìng)態(tài)條件的代碼。作為替代方法，您可以使用“thin 事件，這是一種可重用數(shù)據(jù)結(jié)構(gòu)，可在阻塞前短暫地旋轉(zhuǎn)，僅在必要時(shí)才緩慢分配 Win32 事件，否那么允許您執(zhí)行類似手動(dòng)重置事件的行為。換句話說，它可以對(duì)那些復(fù)雜的容易導(dǎo)致競(jìng)態(tài)條件的代碼進(jìn)行封裝，這樣您就不必在整個(gè)根本代碼內(nèi)散播它。此例如依賴于 Vance Morrison 的文章中描述的一些內(nèi)存模型保證，使用的時(shí)候要多加小心請(qǐng)參見圖 7。Figure7Thin 事件 復(fù)制代

36、碼 public struct ThinEvent private int m_state; / 0 means unset, 1 means set. private EventWaitHandle m_eventObj; private const int s_spinCount = 4000; public void Set() m_state = 1; Thread.MemoryBarrier(); / required. if (m_eventObj != null) m_eventObj.Set(); public void Reset() m_state = 0; if (m_e

37、ventObj != null) m_eventObj.Reset(); public void Wait() SpinWait s = new SpinWait(); while (m_state = 0) if (s.Spin() = s_spinCount) if (m_eventObj = null) ManualResetEvent newEvent = new ManualResetEvent(m_state = 1); if (Interlocked pareExchange( ref m_eventObj, newEvent, null) = null) / If someon

38、e set the flag before seeing the new / event obj, we must ensure its been set. if (m_state = 1) m_eventObj.Set(); else / Lost the race w/ another thread. Just use / its event. newEvent.Close(); m_eventObj.WaitOne(); 這根本上反映了 m_state 變量中的事件狀態(tài)，其中值為 0 表示未設(shè)置，值為 1 表示已設(shè)置?，F(xiàn)在，等待一個(gè)已設(shè)置事件消耗資源是很低的；如果 m_state 在 W

39、ait 例程的入口處值為 1，我們會(huì)立即返回，無需任何內(nèi)核跳轉(zhuǎn)。但如果線程在事件設(shè)置完畢之前進(jìn)入等待狀態(tài)，處理上就需要很多技巧。要等待的首個(gè)線程必須分配一個(gè)新的事件對(duì)象，并對(duì)其進(jìn)行比擬后交換至 m_eventObj 字段中；如果 CAS 失敗，那么意味著其他等待者對(duì)事件進(jìn)行了初始化，所以我們只可重復(fù)使用它；否那么就必須重新檢查自上次看到 m_state 后其值是否發(fā)生更改。不然的話，m_state 的值也許會(huì)為 1，那么 m_eventObj 就無法收到信號(hào)通知，這將導(dǎo)致在調(diào)用 WaitOne 時(shí)出現(xiàn)死鎖。調(diào)用 Set 的線程必須首先設(shè)置 m_state，隨后如果發(fā)現(xiàn)值為非空的 m_event

40、Obj，就會(huì)調(diào)用其上的 Set。這里需要兩個(gè)內(nèi)存屏障：需要注意的是切不可提前移動(dòng) m_state 的第二次讀取，我們會(huì)使用 Interlocked pareExchange 設(shè)置 m_eventObj 來保證這點(diǎn)；在寫入 m_eventObj 之前，不可移動(dòng) Set 中的對(duì) m_eventObj 的讀取這是一種在某些 Intel 和 AMD 處理器以及 CLR 2.0 內(nèi)存模型上出現(xiàn)的奇怪的合法轉(zhuǎn)換，并未顯式調(diào)用 Thread.MemoryBarrier。并行重置事件通常是不平安的，因此調(diào)用方需要進(jìn)行額外的同步化?，F(xiàn)在您可以輕松在其他地方使用它，例如在上述的 CountdownLatch 例如

41、和隊(duì)列例如中，而且通常這樣做可以大幅度提升性能，尤其是當(dāng)您可以巧妙地運(yùn)用旋轉(zhuǎn)時(shí)。我們上面介紹了一個(gè)技巧性很強(qiáng)的實(shí)現(xiàn)方式。請(qǐng)注意，您可以使用單個(gè)標(biāo)志和監(jiān)視器來實(shí)現(xiàn)自動(dòng)和手動(dòng)重置類型，這遠(yuǎn)比本例如簡單但效率方面有時(shí)會(huì)不及本例。無鎖定 LIFO 堆棧使用鎖定構(gòu)建線程平安的集合是相當(dāng)直接明了的，即使限制和阻塞方面會(huì)有些復(fù)雜如上面看到的。但是，當(dāng)所有協(xié)調(diào)均通過簡單的后進(jìn)先出 (LIFO) 堆棧數(shù)據(jù)結(jié)構(gòu)實(shí)現(xiàn)時(shí)，使用鎖定的開銷會(huì)比實(shí)際所需的高。線程的臨界區(qū)即保持鎖定的時(shí)間有開始點(diǎn)和結(jié)束點(diǎn)，其持續(xù)時(shí)間可能會(huì)跨越許多指令的有效期。保持鎖定可阻止其他線程同時(shí)進(jìn)行讀取和寫入操作。這樣做可以實(shí)現(xiàn)序列化，這當(dāng)然是我們所

42、想要的結(jié)果，但嚴(yán)格地講，這種序列化要比我們所需的序列化強(qiáng)很多。我們的目的是要在堆棧上推入和彈出元素，而這兩項(xiàng)操作只需通過常規(guī)讀取和單一的比擬交換寫入即可實(shí)現(xiàn)。我們可以利用這點(diǎn)來構(gòu)建伸縮性更強(qiáng)的無鎖定堆棧，從而不會(huì)強(qiáng)制線程進(jìn)行沒必要的等待。我們的算法工作方式如下。使用有鏈接的列表代表堆棧，其標(biāo)頭代表堆棧的頂端，并存儲(chǔ)在 m_head 字段中。在將一個(gè)新項(xiàng)推入堆棧時(shí)，要構(gòu)造一個(gè)新節(jié)點(diǎn)，其值等于您要推入堆棧的值，然后從本地讀取 m_head 字段，并將其存儲(chǔ)至新節(jié)點(diǎn)的 m_next 字段，隨后執(zhí)行不可再分的 Interlocked pareExchange 來替換堆棧當(dāng)前的頭部。如果頂端自首次讀取后

43、在序列中的任何點(diǎn)發(fā)生更改，CompareExchange 都會(huì)失敗，并且線程必須返回循環(huán)并再次嘗試整個(gè)序列。彈出同樣是比擬直截了當(dāng)?shù)?。讀取 m_head 并嘗試將其與 m_next 引用的本地副本交換；如果失敗，您只需一直嘗試，如圖 8 中所示。Win32 提供了一種名為 SList 的類比數(shù)據(jù)結(jié)構(gòu)，其構(gòu)建方法采用了類似的算法。Figure8無鎖定堆棧 復(fù)制代碼 public class LockFreeStack private volatile StackNode m_head; public void Push(T item) StackNode node = new StackNode

44、(item); StackNode head; do head = m_head; node.m_next = head; while (m_head != head | Interlocked pareExchange( ref m_head, node, head) != head); public T Pop() StackNode head; SpinWait s = new SpinWait(); while (true) StackNode next; do head = m_head; if (head = null) goto emptySpin; next = head.m_

45、next; while (m_head != head | Interlocked pareExchange( ref m_head, next, head) != head); break; emptySpin: s.Spin(); return head.m_value; class StackNode internal T m_value; internal StackNode m_next; internal StackNode(T val) m_value = val; 請(qǐng)注意，這是理想的并發(fā)控制的一種形式：無需阻止其他線程存取數(shù)據(jù)，只需抱著會(huì)在爭(zhēng)用中“勝出的信念繼續(xù)即可。如果事實(shí)證

46、明無法“勝出，您將會(huì)遇到一些變化不定的問題，例如活鎖。這種設(shè)計(jì)方案還說明您無法確保能夠?qū)崿F(xiàn) FIFO 調(diào)度。根據(jù)概率統(tǒng)計(jì)，系統(tǒng)中所有線程都將向前推進(jìn)。而實(shí)際上從整體來看，系統(tǒng)進(jìn)度總是向前推進(jìn)的，因?yàn)槠渲幸粋€(gè)線程的失敗總是意味著至少有一個(gè)其他線程是推進(jìn)的這是調(diào)用該“無鎖定的一個(gè)條件。有些情況下，當(dāng) CompareExchange 無法防止 m_head 大量爭(zhēng)用內(nèi)存時(shí)，使用指數(shù)回退算法會(huì)很有用。同時(shí)，我們還對(duì)堆棧變空的情況采取了相當(dāng)直截了當(dāng)?shù)姆椒āＮ覀冎皇且恢毙D(zhuǎn)，等待有新工程推入。將 Pop 重新寫入處于非等待狀態(tài)的 TryPop 是很簡單易懂的，但是要利用事件進(jìn)行等待那么會(huì)有一些復(fù)雜。這兩個(gè)

47、功能都是十分重要的，所以留給那些喜歡動(dòng)手的讀者作為練習(xí)之用。我們?yōu)槊總€(gè) Push 都分配了對(duì)象，這讓我們無需擔(dān)憂所謂的 ABA 問題。在內(nèi)部重復(fù)使用已從列表中彈出的節(jié)點(diǎn)就會(huì)導(dǎo)致 ABA 問題。開發(fā)人員有時(shí)會(huì)嘗試將節(jié)點(diǎn)集中到池中以減少對(duì)象分配的數(shù)目，但這樣做是有問題的：結(jié)果會(huì)是，即使對(duì) m_head 執(zhí)行了大量干擾性寫入操作，一個(gè)不可再分割的操作也可以實(shí)現(xiàn)，但實(shí)際上卻是不正確的。例如：線程 1 讀取節(jié)點(diǎn) A，然后由線程 2 將節(jié)點(diǎn) A 刪除并置入池中；線程 2 將節(jié)點(diǎn) B 作為新的頭推入，然后節(jié)點(diǎn) A 從池中返回到線程 2 并被推入；隨后，線程 1 會(huì)成功執(zhí)行 CompareExchange，但

48、現(xiàn)在作為頭的 A 已經(jīng)與先前所讀取的不同了。嘗試以本機(jī) C/C+ 編寫此算法時(shí)也會(huì)出現(xiàn)類似問題；因?yàn)橐坏┑刂繁会尫?，?nèi)存分配器就會(huì)立即重復(fù)使用這些地址，節(jié)點(diǎn)會(huì)被彈出并釋放，然后該節(jié)點(diǎn)的地址會(huì)被用于新的節(jié)點(diǎn)分配，結(jié)果導(dǎo)致與上面相同的問題。接下來我們不再討論 ABA，有關(guān)它的詳細(xì)介紹我們可以從其他的資源獲得。最后，可以使用類似的無鎖定技術(shù)編寫一個(gè) FIFO 隊(duì)列。它的優(yōu)勢(shì)在于并行推入和彈出的線程之間并不一定發(fā)生沖突，而在上述 LockFreeStack 中，推入者和彈出者始終會(huì)爭(zhēng)用同一 m_head 字段。然而，這種算法相當(dāng)復(fù)雜，如果您好奇的話，我推薦您閱讀 Maged M. Michael 和

49、Michael L. Scott 在 1996 年撰寫的文章“Simple, Fast, and Practical Non-Blocking and Blocking Concurrent Queue Algorithms。循環(huán)分塊循環(huán)分塊是指對(duì)循環(huán)的輸入范圍或數(shù)據(jù)進(jìn)行分區(qū)，并將每個(gè)分區(qū)分配給單獨(dú)的線程以實(shí)現(xiàn)并發(fā)。這是某些編程模型例如 OpenMP實(shí)現(xiàn)并行性的一項(xiàng)根本技術(shù)請(qǐng)參閱 Kang Su Gatlin 的?MSDN 雜志?文章，通常稱為并行 Forall 循環(huán)，是從高性能的 FORTRAN 術(shù)語中得到啟發(fā)而創(chuàng)造的。無論如何，范圍都只是一組索引：復(fù)制代碼 for (int i = 0; i c; i+) . 或者是一組數(shù)據(jù)： 復(fù)制代碼 foreach (T e in list) . 我們都可以設(shè)計(jì)分區(qū)技術(shù)以提供分塊的循環(huán)。 您可能想應(yīng)用多種特定于數(shù)據(jù)結(jié)構(gòu)的分區(qū)技術(shù)，這些技術(shù)確實(shí)很多，本專欄無法一一列舉。因此這里我們只關(guān)注一種常用技術(shù)，可用于將數(shù)組中各種非連續(xù)范圍的元素分