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

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

3、了解這些技巧如何應用于實際操作也有助于加深您對并行編程的整體理解。在開始講解前，我假定您對現有基元已經有了一個基本的了解。您也可以參閱MSDN® 雜志2005 年 8 月版的文章“關于多線程應用程序：每個開發人員都應了解的內容”，以全面了解其概念。一、倒計數鎖存 (Countdown Latch)Semaphore 之所以成為并發編程中一種較為知名的數據結構，原因是多方面的，而并不只是因為它在計算機科學領域有著悠久的歷史（可以追溯到 19 世紀 60 年代的操作系統設計）。Semaphore 只是一種帶有一個計數字段的數據結構，它只支持兩種操作：放置和取走（通常分別稱為 P 和 V）

4、。一次放置操作會增加一個 semaphore 計數，而一次取走操作會減少一個 semaphore 計數。當 semaphore 計數為零時，除非執行一項并發的放置操作使計數變為非零值，否則任何后續的取走嘗試都將阻塞（等待）。這兩種操作均為不可再分 (atomic) 操作，并發時不會產生錯誤，能夠確保并發的放置和取走操作有序地進行。Windows具有基礎內核和對 semaphore 對象的 Win32支持（請參閱 CreateSemaphore 和相關 API），并且在 .NET Framework 中這些對象可以通過 System.Threading.Semaphore 類公開到上層。Mute

5、x 和 Monitor 所支持的臨界區，通常被認為是一種特殊的 semaphore，其計數會在 0 和 1 之間來回切換，換句話說，是一個二進制的 semaphore。另外還有一種“反向 semaphore”也是非常有用。也就是說，有時您需要數據結構能夠等待數據結構計數歸零。Fork/join 式并行模式在數據并行編程中是極為常見的，其中由單個“主”線程控制執行若干“輔助”線程并等待線程執行完畢。在這類情況下，使用反向 semaphore 會很有幫助。大多數時候，您其實并不想喚醒線程來修改計數。因此在這種情況下，我們將結構稱為倒計數“鎖存”，用來表示計數的減少，同時還表明一旦設置為“Signa

6、led”狀態，鎖存將保持“signaled”（這是一個與鎖存相關的屬性）。遺憾的是，Windows 和 .NET Framework 均不支持這種數據結構。但令人欣慰的是，構建這種數據閉鎖并不難。要構建倒計數鎖存，只需將其計數器初始值設為 n，并讓每項輔助任務在完成時不可再分地將 n 減掉一個計數，這可以通過為減量操作加上“鎖”或調用 Interlocked.Decrement 來實現。接下來，線程可以不執行取走操作，而是減少計數并等待計數器歸零；而當線程被喚醒時，它就可以得知已經有 n 個信號向鎖存注冊。在 while (count != 0) 循環中，讓等待的線程阻塞通常是不錯的選擇（這種

7、情況下，您稍后將不得不使用事件），而不是使用旋轉。public 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_

8、remain) = 0) m_event.Set(); public void Wait() m_event.WaitOne(); 這看上去極為簡單，但要正確運用還需要技巧。稍后我們將通過一些示例來講解如何使用這種數據結構。請注意，此處所示基本實現還有很多可以改進地方，例如：在事件上調用 WaitOne 之前添加某種程度的旋轉等待、緩慢分配事件而不是在構造器中進行分配（以防足夠的旋轉會避免出現阻塞，如本專欄稍后介紹的 ThinEvent 演示的那樣）、添加重置功能以及提供 Dispose 方法（以便在不再需要內部事件對象時將對象關閉）。二、可重用旋轉等待 (Spin Wait)雖然忙碌等待 (

9、busy waiting) 更容易實現阻塞，但在某些情況下，您也許的確想在退回到真正的等待狀態前先旋轉 (spin) 一段時間。我們很難理解為何這樣做會有幫助，而大多數人之所以一開始就避免旋轉等待，是因為旋轉看上去像是在做無用功；如果上下文切換（每當線程等待內核事件時都會發生）需要幾千個周期（在 Windows 上確實是這樣），我們稱之為 c，并且線程所等待的條件出現的時間少于 2c 周期時間（1c 用于等待自身，1c 用于喚醒），則旋轉可以降低等待所造成的系統開銷和滯后時間，從而提升算法的整體吞吐量和可伸縮性。如果您決定使用旋轉等待，就必須謹慎行事。因為如果這樣做，您可能需要注意很多問題，比

10、如：要確保在旋轉循環內調用 Thread.SpinWait，以提高 Intel 超線程技術的計算機上硬件對其他硬件線程的可用性；偶爾使用參數 1 而非 0 來調用 Thread.Sleep，以避免優先級反向問題；通過輕微的回退 (back-off) 來引入隨機選擇，從而改善訪問的局部性（假定調用方持續重讀共享狀態）并可能避免活鎖；當然，在單 CPU 的計算機最好不要采用這種方法（因為在這種環境下旋轉是非常浪費資源的）。SpinWait 類需要被定義為值類型，以便分配起來更加節省資源。現在，我們可以使用此算法來避免前述 CountdownLatch 算法中出現的阻塞。public struct

11、SpinWait private int m_count; private static 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 coun

12、ter is always / a multiple of s_yieldFrequency, 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) T

13、hread.SpinWait(int)(1 + (countModFrequency * 0.05f); else Thread.Sleep(m_count <= s_yieldOneFrequency ? 0 : 1); return oldCount; private void Yield() Thread.Sleep(m_count < s_yieldOneFrequency ? 0 : 1); private const int s_spinCount = 4000;public void Wait() SpinWait s = new SpinWait();while (

14、m_remain > 0) if (s.Spin() >= s_spinCount) m_event.WaitOne(); 不可否認，選擇頻率和旋轉計數是不確定的。與 Win32 臨界區旋轉計數類似，我們應該根據測試和實驗的結果來選擇合理的數值，而且即使合理的數值在不同系統中也會發生變化。例如，根據 Microsoft Media Center 和 Windows kernel 團隊的經驗，MSDN 文檔建議臨界區旋轉計數為 4,000 ，但您的選擇可以有所不同。理想的計數取決于多種因素，包括在給定時間等待事件的線程數和事件出現的頻率等。大多數情況下，您會希望通過等待事件來消除顯式

15、讓出時間，如鎖存的示例中所示。您甚至可以選擇動態調整計數：例如，從中等數量的旋轉開始，每次旋轉失敗就增加計數。一旦計數達到預定的最大值，就完全停止旋轉并立即發出 WaitOne。邏輯如下所示：您希望立即增加達到預定的最大周期數，但卻無法超過最大周期數。如果您發現此最大值不足以阻止上下文切換，那么立即執行上下文切換總的算來占用的資源更少。慢慢您就會希望旋轉計數能夠達到一個穩定的值。三、屏障 (Barrier)屏障，又稱集合點，是一種并發性基元，它無需另一“主”線程控制即可實現各線程之間簡單的互相協調。每個線程在到達屏障時都會不可再分地發出信號并等待。僅當所有 n 都到達屏障時，才允許所有線程繼續

16、。這種方法可用于協調算法 (cooperative algorithms)，該算法廣泛應用于科學、數學和圖形領域。很多計算中都適合使用屏障，實際上，甚至 CLR 的垃圾收集器都在使用它們。屏障只是將較大的計算分割為若干較小的協作階段 (cooperative phase)，例如：const int P = .;Barrier barrier = new Barrier(P);Data partitions = new DataP;/ Running on P separate threads in parallel:public void Body(int myIndex) FillMyPar

17、tition(partitionsmyIndex); barrier.Await(); ReadOtherPartition(partitionsP myIndex - 1); barrier.Await(); / .您會很快發現在這種情況下是可以使用倒計數鎖存的。每個線程都可以在調用 Signal 后立即調用 Wait，而不是調用 Await；在到達屏障后，所有線程都會被釋放。但這里存在一個問題：前面所講的鎖存并不支持多次重復使用同一對象，而這卻是所有屏障都支持的一個常用屬性。實際上，上述示例就需要使用此屬性。您可以通過單獨的屏障對象來實現這一點，但這樣做非常浪費資源；而由于所有線程每次只出

18、現在一個階段，因此根本無需多個屏障對象。要解決這個問題，您可以使用相同的基礎倒計數鎖存算法來處理減少計數器計數、發信號指示事件、等待等方面的問題，從而將其擴展為可重復使用。要實現這一點，您需要使用所謂的感應反向屏障 (sense reversing barrier)，該屏障需要在“偶數”和“奇數”階段之間交替。在交替階段需要使用單獨的事件。using System;using System.Threading;public class Barrier private volatile int m_count; private int m_originalCount; private Event

19、WaitHandle 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_oddEvent = new ManualResetEvent(false); m_evenEvent = new ManualResetEvent(false); public void Await() bool

20、 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 (sense = true) / odd m_evenEvent.Reset(); m_oddEvent.Set(); else / even m_oddEvent.Reset(); m_evenEvent.Se

21、t(); else if (sense = true) m_oddEvent.WaitOne(); else m_evenEvent.WaitOne(); 為何需要兩個事件，其原因很難講清楚。一種方法是在 Await 中先執行 Set 隨后立即執行 Reset，但這很危險，并會導致死鎖。原因有二：第一，另一線程的 m_count 可能已減少，但在依次快速調用 Set 和 Reset 時，計數的值還不足以達到可在事件上調用 WaitOne。第二，雖然等待的線程可能已經能夠調用 WaitOne，但可報警等待（如 CLR 一貫使用的那些）可以中斷等待，從而暫時從等待隊列中刪除等待的線程，以便運行異步

22、過程調用 (APC)。等待的線程將始終看不到事件處于設置 (set) 狀態。兩種情況均會導致事件丟失，并可能導致死鎖。針對奇數階段和偶數階段使用單獨的事件即可避免這種情況。您可能想繼續像 CountdownLatch 中那樣將旋轉添加到 Barrier。但如果您嘗試這樣做，可能會遇到一個問題：一般情況下，旋轉線程會開始旋轉直到 m_count 歸零。但通過上述實現，m_count 的值實際上永遠不會為零，除非最后一個線程將其重置為 m_originalCount。這種想當然的方法將導致一個或多個線程進行旋轉（永遠地），而其他線程則會在下一階段阻塞（也是永遠地）。解決方法很簡單。您旋轉，等待感應

23、發生變化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 (sense = true) / odd m_evenEvent.Set(); m_oddEvent.Reset(); else / even m_oddE

24、vent.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(); 由于所有線程都必須離開前一階段的 Await 才可以完成下一階段，因此可以確定，所有線程要么會觀察到感應發生變化，要么最終等待事件并被喚醒。阻塞隊列在共享內存的體系結構中，兩項或多項任務間唯一的同步點通常是一個位于中樞

25、的共享集合的數據結構。通常，一項或多項任務會負責為其他一項或多項任務生成要執行的“工作”，我們稱之為生產者/使用者關系 (producer/consumer)。這類數據結構的簡單同步通常是非常簡單的 使用 Monitor 或 ReaderWriterLock 即可解決，但當緩沖區為空時，任務間的協調就會變得比較困難。要解決這個問題，通常需要使用阻塞隊列。實際上，阻塞隊列有幾種稍微不同的變體，包括簡單變體和復雜變體。在簡單變量中，使用者僅在隊列為空時才會阻塞，而在復雜變量中，每個生產者都正好“配有”一個使用者，也就是說，在使用者到達并開始處理排隊項目之前，生產者會被阻塞，同理，在生產者交付項目前

26、，所有使用者也會被阻塞。這時通常使用 FIFO（先進先出）排序，但我們并不總是有必要這么做。我們也可以對緩沖區進行限制，這一點稍后會為大家介紹。由于稍后將要介紹的受限緩沖區包含更為簡單的“為空時阻塞”(blocking-when-empty) 行為，因此我們這里僅對配對變量做一了解。要實現這個目的，我們只需封裝一個簡單的 Queue<T>，上方保持同步。那么到底需要何種同步？每當線程對元素進行排隊時，在返回前都會等待使用者取消元素排隊。當線程取消元素排隊時，如果發現緩沖區為空，則必須等待新元素的進入。當然，取消排隊后，使用者必須通知生產者已取到其項目（請參見圖 5）。 F

27、igure 5 阻塞隊列 復制代碼 class Cell<T> internal T m_obj; internal Cell(T obj) m_obj = obj; public class BlockingQueue<T> private Queue<Cell<T>> m_queue = new Queue<Cell<T>>(); public void Enqueue(T obj) Cell<T> c = new Cell<T>(obj); lock (m_queue) m

28、_queue.Enqueue(c); Monitor.Pulse(m_queue); Monitor.Wait(m_queue); public T Dequeue() Cell<T> 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; 請注意，我們可以在 Enqueue 方法中依次調用 Pulse 和 Wait，類似地，在 Dequeue 方法中依次調用 Wait 和 Pul

29、se。只有在釋放監視器之后，才會對內部事件進行設置，而由于監視器的這種實現方法，會導致某個線程調度 ping-pong。我們可能會想，也許可以使用 Win32 事件來構建一個更加優化的通知機制。但是，使用這類完善的 Win32 事件會帶來巨大開銷，尤其是使用它們時所進行的成本分配和內核跳轉 (kernel transitions)，因此還是考慮其他選擇吧。您可以像 CLR 的 ReaderWriterLock 那樣將這些時間集中到池中，也可以像 ThinEvent 類型（稍后介紹）一樣緩慢分配它們。這種實現方法也是有弊端的，即要為每個新元素分配對象，備選方法可能也會將這些對象加入到池中，但同樣

30、會帶來其他麻煩。受限緩沖區 (Bounded Buffer)某些類別的隊列中會出現資源使用問題。如果生產者任務創建項目的速度快于使用者處理項目的速度，則系統的內存使用將不受限制。為了說明這個問題，我們假設一個系統，單個生產者每秒鐘可將 50 個項目排入隊列，而使用者每秒鐘只能使用 10 個項目。首先，這樣會打亂系統的平衡性，而且對于一對一的生產者 使用者配置無法進行很好的調整。只需一分鐘，就會有 2,400 個項目堆積在緩沖區中。假設這些項目中每個項目使用 10KB 內存，那將意味著緩沖區本身就會占用 24MB 內存。一小時后，這個數字將飆升至 1GB 以上。解決這一問題的一個方法是調整生產者

31、線程與使用者線程的比例，在上述情況中，要將比例調整為一個生產者對應五個使用者。但是到達速度通常是不穩定的，這會導致系統周期性的不穩定，并帶來一些明顯的問題，簡單的固定比例是無法解決這個問題的。服務器上的程序通常是長時間運行的，人們希望程序能夠長期保持良好的運行狀態，但如果內存使用不受限，就會造成不可避免的混亂，還可能導致必須定期回收服務器進程的情況。受限緩沖區允許您對生產者強制阻塞前緩沖區可能達到的大小進行限制。阻塞生產者可讓使用者有機會“趕上”（通過允許其線程接收調度時間片），同時還能夠解決內存使用量增加的問題。我們的方法還是僅封裝 Queue<T>，同時添加兩個等待條件和兩個事

32、件通知條件：生產者在隊列滿時等待，直至隊列變為非滿，而使用者在隊列空時等待，直至變為非空；生產者在生產出項目后通知等待的使用者，而使用者也會在取走項目后通知生產者，如圖 6 中所示。 Figure 6  受限緩沖區 (Bounded Buffer) 復制代碼 public class BoundedBuffer<T> private Queue<T> m_queue = new Queue<T>(); private int m_consumersWaiting; private int m_producersWaiting;

33、private const int s_maxBufferSize = 128; public void Enqueue(T obj) lock (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; loc

34、k (m_queue) while (m_queue.Count = 0) m_consumersWaiting+; Monitor.Wait(m_queue); m_consumersWaiting-; e = m_queue.Dequeue(); if (m_producersWaiting > 0) Monitor.PulseAll(m_queue); return e; 這里我們再一次使用了一種比較天真的方法。但是我們確實優化了對 PulseAll 的調用（因為它們耗費的資源很多），方法是使用 m_consumersWaiting 和 m_producersWaiting 這兩個

35、計數器并僅就計數器值是否為零發出信號。除此以外，我們還可以再做進一步的改進。例如，共享與此類似的單個事件可能會喚醒過多線程：如果使用者將隊列大小降為 0，并且生產者和使用者同時處于等待狀態，顯然您只希望喚醒生產者（至少開始時要這么做）。此實現將以先進先出的規則處理所有等待者，這意味著您可能需要在任一生產者運行前喚醒使用者，這樣做僅僅是為了讓使用者發現隊列為空，然后再次等待。令人欣慰的是，最終出現生產者和使用者同時等待的情況是很少見的，但其出現也是有規律的，而且受限區域較小。Thin 事件與 Monitor.Wait、Pulse 和 PulseAll 相比，Win32 事件有一個很大的優勢：它們

36、是“粘滯的”(sticky)。也就是說，一旦有事件收到信號，任何后續等待都將立即取消阻止，即使線程在信號發出前尚未開始等待。如果沒有這個功能，您就需要經常編寫一些代碼將所有等待和信號通知嚴格地限制在臨界區域內，由于 Windows 計劃程序總是會提升已喚醒線程的優先級，因此這些代碼的效率很低；如果不采取這種方法，您就必須使用某種技巧型很強、容易出現競態條件的代碼。作為替代方法，您可以使用“thin 事件”，這是一種可重用數據結構，可在阻塞前短暫地旋轉，僅在必要時才緩慢分配 Win32 事件，否則允許您執行類似手動重置事件的行為。換句話說，它可以對那些復雜的容易導致競態條件的代碼進行封裝，這樣您

37、就不必在整個基本代碼內散播它。此示例依賴于 Vance Morrison 的文章中描述的一些內存模型保證，使用的時候要多加小心（請參見圖 7）。 Figure 7 Thin 事件 復制代碼 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.Mem

38、oryBarrier(); / required. if (m_eventObj != null) m_eventObj.Set(); public void Reset() m_state = 0; if (m_eventObj != 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

39、ManualResetEvent(m_state = 1); if (Interlocked.CompareExchange<EventWaitHandle>( ref m_eventObj, newEvent, null) = null) / If someone 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

40、 / its event. newEvent.Close(); m_eventObj.WaitOne(); 這基本上反映了 m_state 變量中的事件狀態，其中值為 0 表示未設置，值為 1 表示已設置。現在，等待一個已設置事件耗費資源是很低的；如果 m_state 在 Wait 例程的入口處值為 1，我們會立即返回，無需任何內核跳轉。但如果線程在事件設置完畢之前進入等待狀態，處理上就需要很多技巧。要等待的首個線程必須分配一個新的事件對象，并對其進行比較后交換至 m_eventObj 字段中；如果 CAS 失敗，則意味著其他等待者對事件進行了初始化，所以我們只可重復使用它；否則就必須重新檢查

41、自上次看到 m_state 后其值是否發生更改。不然的話，m_state 的值也許會為 1，那么 m_eventObj 就無法收到信號通知，這將導致在調用 WaitOne 時出現死鎖。調用 Set 的線程必須首先設置 m_state，隨后如果發現值為非空的 m_eventObj，就會調用其上的 Set。這里需要兩個內存屏障：需要注意的是切不可提前移動 m_state 的第二次讀取，我們會使用 Interlocked.CompareExchange 設置 m_eventObj 來保證這點；在寫入 m_eventObj 之前，不可移動 Set 中的對 m_eventObj 的讀取（這是一種在某些

42、Intel 和 AMD 處理器以及 CLR 2.0 內存模型上出現的奇怪的合法轉換，并未顯式調用 Thread.MemoryBarrier）。并行重置事件通常是不安全的，因此調用方需要進行額外的同步化。現在您可以輕松在其他地方使用它，例如在上述的 CountdownLatch 示例和隊列示例中，而且通常這樣做可以大幅度提升性能，尤其是當您可以巧妙地運用旋轉時。我們上面介紹了一個技巧性很強的實現方式。請注意，您可以使用單個標志和監視器來實現自動和手動重置類型，這遠比本示例簡單（但效率方面有時會不及本例）。無鎖定 LIFO 堆棧使用鎖定構建線程安全的集合是相當直接明了的，即使限制和阻塞方面會有些復

43、雜（如上面看到的）。但是，當所有協調均通過簡單的后進先出 (LIFO) 堆棧數據結構實現時，使用鎖定的開銷會比實際所需的高。線程的臨界區（即保持鎖定的時間）有開始點和結束點，其持續時間可能會跨越許多指令的有效期。保持鎖定可阻止其他線程同時進行讀取和寫入操作。這樣做可以實現序列化，這當然是我們所想要的結果，但嚴格地講，這種序列化要比我們所需的序列化強很多。我們的目的是要在堆棧上推入和彈出元素，而這兩項操作只需通過常規讀取和單一的比較交換寫入即可實現。我們可以利用這點來構建伸縮性更強的無鎖定堆棧，從而不會強制線程進行沒必要的等待。我們的算法工作方式如下。使用有鏈接的列表代表堆棧，其標頭代表堆棧的頂

44、端，并存儲在 m_head 字段中。在將一個新項推入堆棧時，要構造一個新節點，其值等于您要推入堆棧的值，然后從本地讀取 m_head 字段，并將其存儲至新節點的 m_next 字段，隨后執行不可再分的 Interlocked.CompareExchange 來替換堆棧當前的頭部。如果頂端自首次讀取后在序列中的任何點發生更改，CompareExchange 都會失敗，并且線程必須返回循環并再次嘗試整個序列。彈出同樣是比較直截了當的。讀取 m_head 并嘗試將其與 m_next 引用的本地副本交換；如果失敗，您只需一直嘗試，如圖 8 中所示。Win32 提供了一種名為 SList 的類比數據結構

45、，其構建方法采用了類似的算法。 Figure 8 無鎖定堆棧 復制代碼 public class LockFreeStack<T> private volatile StackNode<T> m_head; public void Push(T item) StackNode<T> node = new StackNode<T>(item); StackNode<T> head; do head = m_head; node.m_next = head; while (m_head != head | I

46、nterlocked.CompareExchange( ref m_head, node, head) != head); public T Pop() StackNode<T> head; SpinWait s = new SpinWait(); while (true) StackNode<T> next; do head = m_head; if (head = null) goto emptySpin; next = head.m_next; while (m_head != head | Interlocked.CompareExchange( ref m_h

47、ead, next, head) != head); break; emptySpin: s.Spin(); return head.m_value; class StackNode<T> internal T m_value; internal StackNode<T> m_next; internal StackNode(T val) m_value = val; 請注意，這是理想的并發控制的一種形式：無需阻止其他線程存取數據，只需抱著會在爭用中“勝出”的信念繼續即可。如果事實證明無法“勝出”，您將會遇到一些變化不定的問題，例如活鎖。這種設計方案還表明您無法確保能夠

48、實現 FIFO 調度。根據概率統計，系統中所有線程都將向前推進。而實際上從整體來看，系統進度總是向前推進的，因為其中一個線程的失敗總是意味著至少有一個其他線程是推進的（這是調用該“無鎖定”的一個條件）。有些情況下，當 CompareExchange 無法避免 m_head 大量爭用內存時，使用指數回退算法會很有用。同時，我們還對堆棧變空的情況采取了相當直截了當的方法。我們只是一直旋轉，等待有新項目推入。將 Pop 重新寫入處于非等待狀態的 TryPop 是很簡單易懂的，但是要利用事件進行等待則會有一些復雜。這兩個功能都是十分重要的，所以留給那些喜歡動手的讀者作為練習之用。我們為每個 Push

49、都分配了對象，這讓我們無需擔心所謂的 ABA 問題。在內部重復使用已從列表中彈出的節點就會導致 ABA 問題。開發人員有時會嘗試將節點集中到池中以減少對象分配的數目，但這樣做是有問題的：結果會是，即使對 m_head 執行了大量干擾性寫入操作，一個不可再分割的操作也可以實現，但實際上卻是不正確的。（例如：線程 1 讀取節點 A，然后由線程 2 將節點 A 刪除并置入池中；線程 2 將節點 B 作為新的頭推入，然后節點 A 從池中返回到線程 2 并被推入；隨后，線程 1 會成功執行 CompareExchange，但現在作為頭的 A 已經與先前所讀取的不同了。）嘗試以本機 C/C+ 編寫此算法時

50、也會出現類似問題；因為一旦地址被釋放，內存分配器就會立即重復使用這些地址，節點會被彈出并釋放，然后該節點的地址會被用于新的節點分配，結果導致與上面相同的問題。接下來我們不再討論 ABA，有關它的詳細介紹我們可以從其他的資源獲得。最后，可以使用類似的無鎖定技術編寫一個 FIFO 隊列。它的優勢在于并行推入和彈出的線程之間并不一定發生沖突，而在上述 LockFreeStack 中，推入者和彈出者始終會爭用同一 m_head 字段。然而，這種算法相當復雜，如果您好奇的話，我推薦您閱讀 Maged M. Michael 和 Michael L. Scott 在 1996 年撰寫的文章“Simple,

51、Fast, and Practical Non-Blocking and Blocking Concurrent Queue Algorithms”。循環分塊循環分塊是指對循環的輸入范圍或數據進行分區，并將每個分區分配給單獨的線程以實現并發。這是某些編程模型（例如 OpenMP）實現并行性的一項基本技術（請參閱 Kang Su Gatlin 的MSDN 雜志文章），通常稱為并行 Forall 循環，是從高性能的 FORTRAN 術語中得到啟發而創造的。無論如何，范圍都只是一組索引：復制代碼 for (int i = 0; i < c; i+) . 或者是一組數據： 復制代碼 foreach (T e in list) . 我們都可以設計分區技術以提供分塊的循環。 您可能想應用多種特定于數據結構的分區技術，這些技術確實很多，本專欄無法一一列舉。因此這里我們只關注一種常用技術，可用于將數組中各種非連續范圍的元素分配到每個分區。我們只需計算跨距的值（它大約等于元素數目除以分區數目的值），并用它來計算連續范圍（參見圖 9）。當然盡管其他方法是有效、有用并在某些情況下有必需的，但當輸入內容為值類型數組時，此方法可以實現較好的空間局部性。