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

可重復(fù)使用的并行數(shù)據(jù)結(jié)構(gòu)和算法精品文檔9種可重復(fù)使用的并行數(shù)據(jù)結(jié)構(gòu)和算法目錄倒計數(shù)鎖存 (CountdownLatch)可重用旋轉(zhuǎn)等待 (SpinWait)屏障(Barrier)阻塞隊列受限緩沖區(qū) (BoundedBuffer)Thin事件無鎖定LIFO堆棧循環(huán)分塊并行分拆總結(jié)本專欄并未涉及很多公共語言運行庫 (CLR)功能的機制問題，而是更多介紹了如何有效使用您手頭所具有的工具。身為一名程序員，必須做出很多決策，而選擇正確的數(shù)據(jù)結(jié)構(gòu)和算法無疑是最常見的，也是最重要的決策之一。錯誤的選擇可能導(dǎo)致程序無法運行，而大多數(shù)情況下，則決定了性能的好壞。鑒于并行編程通常旨在改進性能，并且要難于串行編程，因此所作的選擇對您程序的成功就更為重要。在本專欄中，我們將介紹九種可重復(fù)使用的數(shù)據(jù)結(jié)構(gòu)和算法，這些結(jié)構(gòu)和算法是許多并行程序所常用的，您應(yīng)該能夠輕松將它們應(yīng)用到自己的 .NET軟件中。專欄中每個示例隨附的代碼都是可用的，但尚未經(jīng)過完全定型、測試和優(yōu)化。這里列舉的模式雖然并不詳盡，但卻代表了一些較為常見的模式。如您所見，很多示例都是互為補充的。在開始前，我想還是先介紹一些相關(guān)內(nèi)容。 Microsoft?.NETFramework提供了幾個現(xiàn)有的并發(fā)基元。雖然我要為您講解如何構(gòu)建自己的基元，但實際上現(xiàn)有基元是足以應(yīng)付大多數(shù)情況的。我只是想說某些可選的方案有時也是有參收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔考價值的。此外，了解這些技巧如何應(yīng)用于實際操作也有助于加深您對并行編程的整體理解。在開始講解前，我假定您對現(xiàn)有基元已經(jīng)有了一個基本的了解。您也可以參閱《MSDN?雜志》2005年8月版的文章“關(guān)于多線程應(yīng)用程序：每個開發(fā)人員都應(yīng)了解的內(nèi)容 ”，以全面了解其概念。一、倒計數(shù)鎖存 (CountdownLatch)Semaphore之所以成為并發(fā)編程中一種較為知名的數(shù)據(jù)結(jié)構(gòu)，原因是多方面的，而并不只是因為它在計算機科學(xué)領(lǐng)域有著悠久的歷史（可以追溯到 19世紀60年代的操作系統(tǒng)設(shè)計）。 Semaphore只是一種帶有一個計數(shù)字段的數(shù)據(jù)結(jié)構(gòu)，它只支持兩種操作：放置和取走（通常分別稱為 P和V）。一次放置操作會增加一個semaphore計數(shù)，而一次取走操作會減少一個 semaphore計數(shù)。當(dāng)semaphore計數(shù)為零時，除非執(zhí)行一項 并發(fā)的放置操作使計數(shù)變?yōu)榉橇阒?，否則任何后續(xù)的取走嘗試都將阻塞（等待）。這兩種操作均為不可再分 (atomic)操作，并發(fā)時不會產(chǎn)生錯誤，能夠確保并發(fā)的放置和取走操作有序地進行。Windows具有基礎(chǔ)內(nèi)核和對 semaphore對象的Win32支持（請參閱CreateSemaphore和相關(guān)API），并且在.NETFramework中這些對象可以通過System.Threading.Semaphore類公開到上層。Mutex和Monitor所支持的臨界區(qū)，通常被認為是一種特殊的 semaphore，其計數(shù)會在0和1之間來回切換，換句話說，是一個二進制的 semaphore。另外還有一種“反向semaphore”也是非常有用。也就是說，有時您需要數(shù)據(jù)結(jié)構(gòu)能夠等待數(shù)據(jù)結(jié)構(gòu)計數(shù)歸零。 Fork/join式并行模式在數(shù)據(jù)并行編程中是極為常見的，其中由單個“主”線程控制執(zhí)行若干“輔助”線程并等待線程執(zhí)行完收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔畢。在這類情況下，使用反向 semaphore會很有幫助。大多數(shù)時候，您其實并不想喚醒線程來修改計數(shù)。因此在這種情況下，我們將結(jié)構(gòu)稱為倒計數(shù) “鎖存”，用來表示計數(shù)的減少，同時還表明一旦設(shè)置為 “Signaled狀”態(tài)，鎖存將保持“signaled（”這是一個與鎖存相關(guān)的屬性）。遺憾的是， Windows和.NETFramework均不支持這種數(shù)據(jù)結(jié)構(gòu)。但令人欣慰的是，構(gòu)建這種數(shù)據(jù)閉鎖并不難。要構(gòu)建倒計數(shù)鎖存，只需將其計數(shù)器初始值設(shè)為 n，并讓每項輔助任務(wù)在完成時不可再分地將 n減掉一個計數(shù)，這可以通過為減量操作加上 “鎖”或調(diào)用Interlocked.Decrement來實現(xiàn)。接下來，線程可以不執(zhí)行取走操作，而是減少計數(shù)并等待計數(shù)器歸零；而當(dāng)線程被喚醒時，它就可以得知已經(jīng)有 n個信號向鎖存注冊。在while(count!=0)循環(huán)中，讓等待的線程阻塞通常是不錯的選擇（這種情況下，您稍后將不得不使用事件），而不是使用旋轉(zhuǎn)。publicclassCountdownLatch{privateintm_remain;privateEventWaitHandlem_event;publicCountdownLatch(intcount){m_remain=count;m_event=newManualResetEvent(false);}publicvoidSignal(){Thelastthreadtosignalalsosetstheevent.if(Interlocked.Decrement(refm_remain)==0)m_event.Set();}收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔publicvoidWait(){m_event.WaitOne();}}這看上去極為簡單，但要正確運用還需要技巧。稍后我們將通過一些示例來講解如何使用這種數(shù)據(jù)結(jié)構(gòu)。請注意，此處所示基本實現(xiàn)還有很多可以改進地方，例如：在事件上調(diào)用 WaitOne之前添加某種程度的旋轉(zhuǎn)等待、緩慢分配事件而不是在構(gòu)造器中進行分配（以防足夠的旋轉(zhuǎn)會避免出現(xiàn)阻塞，如本專欄稍后介紹的ThinEvent演示的那樣）、添加重置功能以及提供 Dispose方法（以便在不再需要內(nèi)部事件對象時將對象關(guān)閉）。二、可重用旋轉(zhuǎn)等待 (SpinWait)雖然忙碌等待(busywaiting)更容易實現(xiàn)阻塞，但在某些情況下，您也許的確想在退回到真正的等待狀態(tài)前先旋轉(zhuǎn) (spin)一段時間。我們很難理解為何這樣做會有幫助，而大多數(shù)人之所以一開始就避免旋轉(zhuǎn)等待，是因為旋轉(zhuǎn)看上去像是在做無用功；如果上下文切換（每當(dāng)線程等待內(nèi)核事件時都會發(fā)生）需要幾千個周期（在 Windows上確實是這樣），我們稱之為 c，并且線程所等待的條件出現(xiàn)的時間少于 2c周期時間（1c用于等待自身，1c用于喚醒），則旋轉(zhuǎn)可以降低等待所造成的系統(tǒng)開銷和滯后時間，從而提升算法的整體吞吐量和可伸縮性。如果您決定使用旋轉(zhuǎn)等待，就必須謹慎行事。因為如果這樣做，您可能需要注意很多問題，比如：要確保在旋轉(zhuǎn)循環(huán)內(nèi)調(diào)用 Thread.SpinWait，以提高Intel超線程技術(shù)的計算機上硬件對其他硬件線程的可用性；偶爾使用參數(shù)1而收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔非0來調(diào)用Thread.Sleep，以避免優(yōu)先級反向問題；通過輕微的回退 (back-off)來引入隨機選擇，從而改善訪問的局部性（假定調(diào)用方持續(xù)重讀共享狀態(tài)）并可能避免活鎖；當(dāng)然，在單 CPU的計算機最好不要采用這種方法（因為在這種環(huán)境下旋轉(zhuǎn)是非常浪費資源的）。SpinWait類需要被定義為值類型，以便分配起來更加節(jié)省資源?，F(xiàn)在，我們可以使用此算法來避免前述 CountdownLatch算法中出現(xiàn)的阻塞。publicstructSpinWait{privateintm_count;privatestaticreadonlybools_isSingleProc=(Environment.ProcessorCount==1);privateconstints_yieldFrequency=4000;privateconstints_yieldOneFrequency=3*s_yieldFrequency;publicintSpin(){intoldCount=m_count;//Onasingle-CPUmachine,weensureourcounterisalways//amultipleof ‘s_yieldFrequency ’,soweyieldeverytime.//Else,wejustincrementbyone.m_count+=(s_isSingleProc?s_yieldFrequency:1);//Ifnotamultipleof ‘s_yieldFrequency ’spin(w/backoff).intcountModFrequency=m_count%s_yieldFrequency;if(countModFrequency>0)Thread.SpinWait((int)(1+(countModFrequency*0.05f)));elseThread.Sleep(m_count<=s_yieldOneFrequency?0:1);returnoldCount;}privatevoidYield(){收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔Thread.Sleep(m_count<s_yieldOneFrequency?0:1);}}privateconstints_spinCount=4000;publicvoidWait(){SpinWaits=newSpinWait();while(m_remain>0){if(s.Spin()>=s_spinCount)m_event.WaitOne();}}不可否認，選擇頻率和旋轉(zhuǎn)計數(shù)是不確定的。與 Win32臨界區(qū)旋轉(zhuǎn)計數(shù)類似，我們應(yīng)該根據(jù)測試和實驗的結(jié)果來選擇合理的數(shù)值，而且即使合理的數(shù)值在不同系統(tǒng)中也會發(fā)生變化。例如，根據(jù) MicrosoftMediaCenter和Windowskernel團隊的經(jīng)驗，MSDN文檔建議臨界區(qū)旋轉(zhuǎn)計數(shù)為 4,000，但您的選擇可以有所不同。理想的計數(shù)取決于多種因素，包括在給定時間等待事件的線程數(shù)和事件出現(xiàn)的頻率等。大多數(shù)情況下，您會希望通過等待事件來消除顯式讓出時間，如鎖存的示例中所示。您甚至可以選擇動態(tài)調(diào)整計數(shù)：例如，從中等數(shù)量的旋轉(zhuǎn)開始，每次旋轉(zhuǎn)失敗就增加計數(shù)。一旦計數(shù)達到預(yù)定的最大值，就完全停止旋轉(zhuǎn)并立即發(fā)出WaitOne。邏輯如下所示：您希望立即增加達到預(yù)定的最大周期數(shù)，但卻無法超過最大周期數(shù)。如果您發(fā)現(xiàn)此最大值不足以阻止上下文切換，那么立即執(zhí)行上下文切換總的算來占用的資源更少。慢慢您就會希望旋轉(zhuǎn)計數(shù)能夠達到一個穩(wěn)定的值。收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔三、屏障(Barrier)屏障，又稱集合點，是一種并發(fā)性基元，它無需另一 “主”線程控制即可實現(xiàn)各線程之間簡單的互相協(xié)調(diào)。每個線程在到達屏障時都會不可再分地發(fā)出信號并等待。僅當(dāng)所有 n都到達屏障時，才允許所有線程繼續(xù)。這種方法可用于協(xié)調(diào)算法(cooperativealgorithms)，該算法廣泛應(yīng)用于科學(xué)、數(shù)學(xué)和圖形領(lǐng)域。很多計算中都適合使用屏障，實際上，甚至 CLR的垃圾收集器都在使用它們。屏障只是將較大的計算分割為若干較小的協(xié)作階段 (cooperativephase)，例如：constintP=...;Barrierbarrier=newBarrier(P);Data[]partitions=newData[P];Runningon‘P’separatethreadsinparallel:publicvoidBody(intmyIndex){FillMyPartition(partitions[myIndex]);barrier.Await();ReadOtherPartition(partitions[P–myIndex-1]);barrier.Await();//...}您會很快發(fā)現(xiàn)在這種情況下是可以使用倒計數(shù)鎖存的。每個線程都可以在調(diào)用Signal后立即調(diào)用Wait，而不是調(diào)用Await；在到達屏障后，所有線程都會被釋放。但這里存在一個問題：前面所講的鎖存并不支持多次重復(fù)使用同一對象，而這卻是所有屏障都支持的一個常用屬性。實際上，上述示例就需要使用此屬性。您可以通過單獨的屏障對象來實現(xiàn)這一點，但這樣做非常浪費資源；而由于所有線程每次只出現(xiàn)在一個階段，因此根本無需多個屏障對象。收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔要解決這個問題，您可以使用相同的基礎(chǔ)倒計數(shù)鎖存算法來處理減少計數(shù)器計數(shù)、發(fā)信號指示事件、等待等方面的問題，從而將其擴展為可重復(fù)使用。要實現(xiàn)這一點，您需要使用所謂的感應(yīng)反向屏障 (sensereversingbarrier)，該屏障需要在“偶數(shù)”和“奇數(shù)”階段之間交替。在交替階段需要使用單獨的事件。usingSystem;usingSystem.Threading;publicclassBarrier{privatevolatileintm_count;privateintm_originalCount;privateEventWaitHandlem_oddEvent;privateEventWaitHandlem_evenEvent;privatevolatileboolm_sense=false;//false==even,true==odd.publicBarrier(intcount){m_count=count;m_originalCount=count;m_oddEvent=newManualResetEvent(false);m_evenEvent=newManualResetEvent(false);}publicvoidAwait(){boolsense=m_sense;//Thelastthreadtosignalalsosetstheevent.if(m_count==1||Interlocked.Decrement(refm_count)==0){m_count=m_originalCount;m_sense=!sense;//Reversethesense.if(sense==true){//oddm_evenEvent.Reset();m_oddEvent.Set();}else{//evenm_oddEvent.Reset();m_evenEvent.Set();收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔}}else{if(sense==true)m_oddEvent.WaitOne();else m_evenEvent.WaitOne();}}}為何需要兩個事件，其原因很難講清楚。一種方法是在 Await中先執(zhí)行Set隨后立即執(zhí)行Reset，但這很危險，并會導(dǎo)致死鎖。原因有二：第一，另一線程的m_count可能已減少，但在依次快速調(diào)用Set和Reset時，計數(shù)的值還不足以達到可在事件上調(diào)用WaitOne。第二，雖然等待的線程可能已經(jīng)能夠調(diào)用WaitOne，但可報警等待（如 CLR一貫使用的那些）可以中斷等待，從而暫時從等待隊列中刪除等待的線程，以便運行異步過程調(diào)用 (APC)。等待的線程將始終看不到事件處于設(shè)置 (set)狀態(tài)。兩種情況均會導(dǎo)致事件丟失，并可能導(dǎo)致死鎖。針對奇數(shù)階段和偶數(shù)階段使用單獨的事件即可避免這種情況。您可能想繼續(xù)像 CountdownLatch中那樣將旋轉(zhuǎn)添加到 Barrier。但如果您嘗試這樣做，可能會遇到一個問題：一般情況下，旋轉(zhuǎn)線程會開始旋轉(zhuǎn)直到 m_count歸零。但通過上述實現(xiàn)， m_count的值實際上永遠不會為零，除非最后一個線程將其重置為m_originalCount。這種想當(dāng)然的方法將導(dǎo)致一個或多個線程進行旋轉(zhuǎn)（永遠地），而其他線程則會在下一階段阻塞（也是永遠地）。解決方法很簡單。您旋轉(zhuǎn)，等待感應(yīng)發(fā)生變化publicvoidAwait(){boolsense=m_sense;收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔//Thelastthreadtosignalalsosetstheevent.if(m_count==1||Interlocked.Decrement(refm_count)==0){m_count=m_originalCount;m_sense=!sense;//Reversethesense.if(sense==true){//oddm_evenEvent.Set();m_oddEvent.Reset();}else{//evenm_oddEvent.Set();m_evenEvent.Reset();}}else{SpinWaits=newSpinWait();while(sense==m_sense){if(s.Spin()>=s_spinCount){if(sense==true)m_oddEvent.WaitOne();else m_evenEvent.WaitOne();}}}}由于所有線程都必須離開前一階段的 Await才可以完成下一階段，因此可以確定，所有線程要么會觀察到感應(yīng)發(fā)生變化，要么最終等待事件并被喚醒。阻塞隊列在共享內(nèi)存的體系結(jié)構(gòu)中，兩項或多項任務(wù)間唯一的同步點通常是一個位于中樞的共享集合的數(shù)據(jù)結(jié)構(gòu)。通常，一項或多項任務(wù)會負責(zé)為其他一項或多項任務(wù)生成要執(zhí)行的“工作”，我們稱之為生產(chǎn)者/使用者關(guān)系(producer/consumer)。這類數(shù)據(jù)結(jié)構(gòu)的簡單同步通常是非常簡單的 —使用Monitor或收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔ReaderWriterLock即可解決，但當(dāng)緩沖區(qū)為空時，任務(wù)間的協(xié)調(diào)就會變得比較困難。要解決這個問題，通常需要使用阻塞隊列。實際上，阻塞隊列有幾種稍微不同的變體，包括簡單變體和復(fù)雜變體。在簡單變量中，使用者僅在隊列為空時才會阻塞，而在復(fù)雜變量中，每個生產(chǎn)者都正好“配有”一個使用者，也就是說，在使用者到達并開始處理排隊項目之前，生產(chǎn)者會被阻塞，同理，在生產(chǎn)者交付項目前，所有使用者也會被阻塞。這時通常使用FIFO（先進先出）排序，但我們并不總是有必要這么做。我們也可以對緩沖區(qū)進行限制，這一點稍后會為大家介紹。由于稍后將要介紹的受限緩沖區(qū)包含更為簡單的“為空時阻塞”(blocking-when-empty)行為，因此我們這里僅對配對變量做一了解。要實現(xiàn)這個目的，我們只需封裝一個簡單的 Queue<T>，上方保持同步。那么到底需要何種同步？每當(dāng)線程對元素進行排隊時，在返回前都會等待使用者取消元素排隊。當(dāng)線程取消元素排隊時，如果發(fā)現(xiàn)緩沖區(qū)為空，則必須等待新元素的進入。當(dāng)然，取消排隊后，使用者必須通知生產(chǎn)者已取到其項目（請參見 圖5）。Figure5阻塞隊列復(fù)制代碼classCell<T>{internalTm_obj;internalCell(Tobj){m_obj=obj;}}publicclassBlockingQueue<T>{privateQueue<Cell<T>>m_queue=newQueue<Cell<T>>();收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔publicvoidEnqueue(Tobj){Cell<T>c=newCell<T>(obj);lock(m_queue){m_queue.Enqueue(c);Monitor.Pulse(m_queue);Monitor.Wait(m_queue);}}publicTDequeue(){Cell<T>c;lock(m_queue){while(m_queue.Count==0)Monitor.Wait(m_queue);c=m_queue.Dequeue();Monitor.Pulse(m_queue);}returnc.m_obj;}}請注意，我們可以在 Enqueue方法中依次調(diào)用 Pulse和Wait，類似地，在Dequeue方法中依次調(diào)用 Wait和Pulse。只有在釋放監(jiān)視器之后，才會對內(nèi)部事件進行設(shè)置，而由于監(jiān)視器的這種實現(xiàn)方法，會導(dǎo)致某個線程調(diào)度 ping-pong。我們可能會想，也許可以使用 Win32事件來構(gòu)建一個更加優(yōu)化的通知機制。但是，使用這類完善的 Win32事件會帶來巨大開銷，尤其是使用它們時所進行的成本分配和內(nèi)核跳轉(zhuǎn) (kerneltransitions)，因此還是考慮其他選擇吧。您可以像CLR的ReaderWriterLock那樣將這些時間集中到池中，也可以像ThinEvent類型（稍后介紹）一樣緩慢分配它們。這種實現(xiàn)方法也是有弊端的，即要為每個新元素分配對象，備選方法可能也會將這些對象加入到池中，但同樣會帶來其他麻煩。收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔受限緩沖區(qū)(BoundedBuffer)某些類別的隊列中會出現(xiàn)資源使用問題。如果生產(chǎn)者任務(wù)創(chuàng)建項目的速度快于使用者處理項目的速度，則系統(tǒng)的內(nèi)存使用將不受限制。為了說明這個問題，我們假設(shè)一個系統(tǒng)，單個生產(chǎn)者每秒鐘可將 50個項目排入隊列，而使用者每秒鐘只能使用 10個項目。首先，這樣會打亂系統(tǒng)的平衡性，而且對于一對一的生產(chǎn)者 —使用者配置無法進行很好的調(diào)整。只需一分鐘，就會有2,400個項目堆積在緩沖區(qū)中。假設(shè)這些項目中每個項目使用 10KB內(nèi)存，那將意味著緩沖區(qū)本身就會占用 24MB內(nèi)存。一小時后，這個數(shù)字將飆升至1GB以上。解決這一問題的一個方法是調(diào)整生產(chǎn)者線程與使用者線程的比例，在上述情況中，要將比例調(diào)整為一個生產(chǎn)者對應(yīng)五個使用者。但是到達速度通常是不穩(wěn)定的，這會導(dǎo)致系統(tǒng)周期性的不穩(wěn)定，并帶來一些明顯的問題，簡單的固定比例是無法解決這個問題的。服務(wù)器上的程序通常是長時間運行的，人們希望程序能夠長期保持良好的運行狀態(tài)，但如果內(nèi)存使用不受限，就會造成不可避免的混亂，還可能導(dǎo)致必須定期回收服務(wù)器進程的情況。受限緩沖區(qū)允許您對生產(chǎn)者強制阻塞前緩沖區(qū)可能達到的大小進行限制。阻塞生產(chǎn)者可讓使用者有機會“趕上”（通過允許其線程接收調(diào)度時間片），同時還能夠解決內(nèi)存使用量增加的問題。我們的方法還是僅封裝Queue<T>，同時添加兩個等待條件和兩個事件通知條件：生產(chǎn)者在隊列滿時等待，直至隊列變?yōu)榉菨M，而使用者在隊列空時等待，直至變?yōu)榉强?；生產(chǎn)者在生產(chǎn)出項目后通知等待的使用者，而使用者也會在取走項目后通知生產(chǎn)者，如 圖6中所示。收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔Figure6受限緩沖區(qū)(BoundedBuffer)復(fù)制代碼publicclassBoundedBuffer<T>{privateQueue<T>m_queue=newQueue<T>();privateintm_consumersWaiting;privateintm_producersWaiting;privateconstints_maxBufferSize=128;publicvoidEnqueue(Tobj){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);}}publicTDequeue(){Te;lock(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);}收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔returne;}}這里我們再一次使用了一種比較天真的方法。但是我們確實優(yōu)化了對 PulseAll的調(diào)用（因為它們耗費的資源很多），方法是使用 m_consumersWaiting和m_producersWaiting這兩個計數(shù)器并僅就計數(shù)器值是否為零發(fā)出信號。除此以外，我們還可以再做進一步的改進。例如，共享與此類似的單個事件可能會喚醒過多線程：如果使用者將隊列大小降為 0，并且生產(chǎn)者和使用者同時處于等待狀態(tài)，顯然您只希望喚醒生產(chǎn)者（至少開始時要這么做）。此實現(xiàn)將以先進先出的規(guī)則處理所有等待者，這意味著您可能需要在任一生產(chǎn)者運行前喚醒使用者，這樣做僅僅是為了讓使用者發(fā)現(xiàn)隊列為空，然后再次等待。令人欣慰的是，最終出現(xiàn)生產(chǎn)者和使用者同時等待的情況是很少見的，但其出現(xiàn)也是有規(guī)律的，而且受限區(qū)域較小。Thin事件與Monitor.Wait、Pulse和PulseAll相比，Win32事件有一個很大的優(yōu)勢：它們是“粘滯的”(sticky)。也就是說，一旦有事件收到信號，任何后續(xù)等待都將立即取消阻止，即使線程在信號發(fā)出前尚未開始等待。如果沒有這個功能，您就需要經(jīng)常編寫一些代碼將所有等待和信號通知嚴格地限制在臨界區(qū)域內(nèi)，由于Windows計劃程序總是會提升已喚醒線程的優(yōu)先級，因此這些代碼的效率很低；如果不采取這種方法，您就必須使用某種技巧型很強、容易出現(xiàn)競態(tài)條件的代碼。作為替代方法，您可以使用 “thin事件”，這是一種可重用數(shù)據(jù)結(jié)構(gòu)，可在阻塞前短暫地旋轉(zhuǎn)，僅在必要時才緩慢分配 Win32事件，否則允許您執(zhí)行類似手動收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔重置事件的行為。換句話說，它可以對那些復(fù)雜的容易導(dǎo)致競態(tài)條件的代碼進行封裝，這樣您就不必在整個基本代碼內(nèi)散播它。此示例依賴于 VanceMorrison的文章中描述的一些內(nèi)存模型保證，使用的時候要多加小心（請參見圖7）。Figure7Thin事件復(fù)制代碼publicstructThinEvent{privateintm_state;//0meansunset,1meansset.privateEventWaitHandlem_eventObj;privateconstints_spinCount=4000;publicvoidSet(){m_state=1;Thread.MemoryBarrier();//required.if(m_eventObj!=null)m_eventObj.Set();}publicvoidReset(){m_state=0;if(m_eventObj!=null)m_eventObj.Reset();}publicvoidWait(){SpinWaits=newSpinWait();while(m_state==0){if(s.Spin()>=s_spinCount){if(m_eventObj==null){ManualResetEventnewEvent=newManualResetEvent(m_state==1);收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔if(Interlocked.CompareExchange<EventWaitHandle>(refm_eventObj,newEvent,null)==null){//Ifsomeonesettheflagbeforeseeingthenew//eventobj,wemustensureit set’.sbeenif(m_state==1)m_eventObj.Set();}else{Losttheracew/anotherthread.Justuseitsevent.newEvent.Close();}}m_eventObj.WaitOne();}}}}這基本上反映了 m_state變量中的事件狀態(tài)，其中值為 0表示未設(shè)置，值為 1表示已設(shè)置?，F(xiàn)在，等待一個已設(shè)置事件耗費資源是很低的；如果 m_state在Wait例程的入口處值為 1，我們會立即返回，無需任何內(nèi)核跳轉(zhuǎn)。但如果線程在事件設(shè)置完畢之前進入等待狀態(tài)，處理上就需要很多技巧。要等待的首個線程必須分配一個新的事件對象，并對其進行比較后交換至 m_eventObj字段中；如果CAS失敗，則意味著其他等待者對事件進行了初始化，所以我們只可重復(fù)使用它；否則就必須重新檢查自上次看到 m_state后其值是否發(fā)生更改。不然的話，m_state的值也許會為1，那么m_eventObj就無法收到信號通知，這將導(dǎo)致在調(diào)用WaitOne時出現(xiàn)死鎖。調(diào)用 Set的線程必須首先設(shè)置 m_state，隨后如果發(fā)現(xiàn)值為非空的 m_eventObj，就會調(diào)用其上的 Set。這里需要兩個內(nèi)存屏障：需要注意的是切不可提前移動 m_state的第二次讀取，我們會使用收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔Interlocked.CompareExchange設(shè)置m_eventObj來保證這點；在寫入 m_eventObj之前，不可移動 Set中的對m_eventObj的讀?。ㄟ@是一種在某些 Intel和AMD處理器以及CLR2.0內(nèi)存模型上出現(xiàn)的奇怪的合法轉(zhuǎn)換，并未顯式調(diào)用Thread.MemoryBarrier）。并行重置事件通常是不安全的，因此調(diào)用方需要進行額外的同步化?，F(xiàn)在您可以輕松在其他地方使用它，例如在上述的 CountdownLatch示例和隊列示例中，而且通常這樣做可以大幅度提升性能，尤其是當(dāng)您可以巧妙地運用旋轉(zhuǎn)時。我們上面介紹了一個技巧性很強的實現(xiàn)方式。請注意，您可以使用單個標志和監(jiān)視器來實現(xiàn)自動和手動重置類型，這遠比本示例簡單（但效率方面有時會不及本例）。無鎖定LIFO堆棧使用鎖定構(gòu)建線程安全的集合是相當(dāng)直接明了的，即使限制和阻塞方面會有些復(fù)雜（如上面看到的）。但是，當(dāng)所有協(xié)調(diào)均通過簡單的后進先出 (LIFO)堆棧數(shù)據(jù)結(jié)構(gòu)實現(xiàn)時，使用鎖定的開銷會比實際所需的高。線程的臨界區(qū)（即保持鎖定的時間）有開始點和結(jié)束點，其持續(xù)時間可能會跨越許多指令的有效期。保持鎖定可阻止其他線程同時進行讀取和寫入操作。這樣做可以實現(xiàn)序列化，這當(dāng)然是我們所想要的結(jié)果，但嚴格地講，這種序列化要比我們所需的序列化強很多。我們的目的是要在堆棧上推入和彈出元素，而這兩項操作只需通過常規(guī)讀取和單一的比較－交換寫入即可實現(xiàn)。我們可以利用這點來構(gòu)建伸縮性更強的無鎖定堆棧，從而不會強制線程進行沒必要的等待。收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔我們的算法工作方式如下。使用有鏈接的列表代表堆棧，其標頭代表堆棧的頂端，并存儲在m_head字段中。在將一個新項推入堆棧時，要構(gòu)造一個新節(jié)點，其值等于您要推入堆棧的值，然后從本地讀取 m_head字段，并將其存儲至新節(jié)點的m_next字段，隨后執(zhí)行不可再分的 Interlocked.CompareExchange來替換堆棧當(dāng)前的頭部。如果頂端自首次讀取后在序列中的任何點發(fā)生更改，CompareExchange都會失敗，并且線程必須返回循環(huán)并再次嘗試整個序列。彈出同樣是比較直截了當(dāng)?shù)?。讀取 m_head并嘗試將其與 m_next引用的本地副本交換；如果失敗，您只需一直嘗試，如 圖8中所示。Win32提供了一種名為SList的類比數(shù)據(jù)結(jié)構(gòu)，其構(gòu)建方法采用了類似的算法。Figure8無鎖定堆棧復(fù)制代碼publicclassLockFreeStack<T>{privatevolatileStackNode<T>m_head;publicvoidPush(Titem){StackNode<T>node=newStackNode<T>(item);StackNode<T>head;do{head=m_head;node.m_next=head;}while(m_head!=head||Interlocked.CompareExchange(refm_head,node,head)!=head);}publicTPop(){StackNode<T>head;SpinWaits=newSpinWait();收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔while(true){StackNode<T>next;do{head=m_head;if(head==null)gotoemptySpin;next=head.m_next;}while(m_head!=head||Interlocked.CompareExchange(refm_head,next,head)!=head);break;emptySpin:s.Spin();}returnhead.m_value;}}classStackNode<T>{internalTm_value;internalStackNode<T>m_next;internalStackNode(Tval){m_value=val;}}請注意，這是理想的并發(fā)控制的一種形式：無需阻止其他線程存取數(shù)據(jù)，只需抱著會在爭用中“勝出”的信念繼續(xù)即可。如果事實證明無法 “勝出”，您將會遇到一些變化不定的問題，例如活鎖。這種設(shè)計方案還表明您無法確保能夠?qū)崿F(xiàn)FIFO調(diào)度。根據(jù)概率統(tǒng)計，系統(tǒng)中所有線程都將向前推進。而實際上從整體來看，系統(tǒng)進度總是向前推進的，因為其中一個線程的失敗總是意味著至少有一個其他線程是推進的（這是調(diào)用該 “無鎖定”的一個條件）。有些情況下，當(dāng)收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔CompareExchange無法避免m_head大量爭用內(nèi)存時，使用指數(shù)回退算法會很有用。同時，我們還對堆棧變空的情況采取了相當(dāng)直截了當(dāng)?shù)姆椒?。我們只是一直旋轉(zhuǎn)，等待有新項目推入。將 Pop重新寫入處于非等待狀態(tài)的 TryPop是很簡單易懂的，但是要利用事件進行等待則會有一些復(fù)雜。這兩個功能都是十分重要的，所以留給那些喜歡動手的讀者作為練習(xí)之用。我們?yōu)槊總€Push都分配了對象，這讓我們無需擔(dān)心所謂的 ABA問題。在內(nèi)部重復(fù)使用已從列表中彈出的節(jié)點就會導(dǎo)致 ABA問題。開發(fā)人員有時會嘗試將節(jié)點集中到池中以減少對象分配的數(shù)目，但這樣做是有問題的：結(jié)果會是，即使對m_head執(zhí)行了大量干擾性寫入操作，一個不可再分割的操作也可以實現(xiàn)，但實際上卻是不正確的。（例如：線程 1讀取節(jié)點A，然后由線程2將節(jié)點A刪除并置入池中；線程2將節(jié)點B作為新的頭推入，然后節(jié)點A從池中返回到線程2并被推入；隨后，線程1會成功執(zhí)行CompareExchange，但現(xiàn)在作為頭的A已經(jīng)與先前所讀取的不同了。）嘗試以本機 C/C++編寫此算法時也會出現(xiàn)類似問題；因為一旦地址被釋放，內(nèi)存分配器就會立即重復(fù)使用這些地址，節(jié)點會被彈出并釋放，然后該節(jié)點的地址會被用于新的節(jié)點分配，結(jié)果導(dǎo)致與上面相同的問題。接下來我們不再討論 ABA，有關(guān)它的詳細介紹我們可以從其他的資源獲得。最后，可以使用類似的無鎖定技術(shù)編寫一個 FIFO隊列。它的優(yōu)勢在于并行推入和彈出的線程之間并不一定發(fā)生沖突，而在上述 LockFreeStack中，推入者和彈出者始終會爭用同一 m_head字段。然而，這種算法相當(dāng)復(fù)雜，如果您好奇的話，我推薦您閱讀 MagedM.Michael和MichaelL.Scott在1996年撰寫的文收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔章“Simple,Fast,andPracticalNon-BlockingandBlockingConcurrentQueueAlgorithms”。循環(huán)分塊循環(huán)分塊是指對循環(huán)的輸入范圍或數(shù)據(jù)進行分區(qū)，并將每個分區(qū)分配給單獨的線程以實現(xiàn)并發(fā)。這是某些編程模型（例如 OpenMP）實現(xiàn)并行性的一項基本技術(shù)（請參閱KangSuGatlin的《MSDN雜志》文章），通常稱為并行 Forall循環(huán)，是從高性能的 FORTRAN術(shù)語中得到啟發(fā)而創(chuàng)造的。無論如何，范圍都只是一組索引：復(fù)制代碼for(inti=0;i<c;i++){...}或者是一組數(shù)據(jù)：復(fù)制代碼foreach(Teinlist){...}我們都可以設(shè)計分區(qū)技術(shù)以提供分塊的循環(huán)。您可能想應(yīng)用多種特定于數(shù)據(jù)結(jié)構(gòu)的分區(qū)技術(shù)，這些技術(shù)確實很多，本專欄無法一一列舉。因此這里我們只關(guān)注一種常用技術(shù)，可用于將數(shù)組中各種非連續(xù)范圍的元素分配到每個分區(qū)。我們只需計算跨距的值（它大約等于元素數(shù)目除以分區(qū)數(shù)目的值），并用它來計算連續(xù)范圍（參見 圖9）。當(dāng)然盡管其他方法是有效、有用并在某些情況下有必需的，但當(dāng)輸入內(nèi)容為值類型數(shù)組時，此方法可以實現(xiàn)較好的空間局部性。Figure9循環(huán)分塊收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔復(fù)制代碼publicstaticvoidForAll(intfrom,intto,Action<int>a,intp){ForAll<int>(null,from,to,null,a,p);}publicstaticvoidForAll<T>(IList<T>data,Action<T>a,intp){ForAll<T>(data,0,data.Count,a,null,p);}privatestaticvoidForAll<T>(IList<T>data,intfrom,intto,Action<T>a0,Action<int>a1,intp){intsize=from-to;intstride=(size+p-1)/p;CountdownLatchlatch=newCountdownLatch(p);for(inti=0;i<p;i++){intidx=i;ThreadPool.QueueUserWorkItem(delegate{intend=Math.Min(size,stride*(idx+1));for(intj=stride*idx;j<end;j++){if(data!=null)a0(data[j]);else a1(j);}latch.Signal();});}latch.Wait();}我們這里提供了兩種公共版的 ForAll：一種接受一定范圍內(nèi)的數(shù)字，另一種接受IList<T>（類似于C#中的Foreach循環(huán)）。兩者都轉(zhuǎn)發(fā)到同一幫助程序重收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔載，重載可調(diào)用從給定索引的列表傳遞元素這一操作，或者傳遞索引本身。您應(yīng)使用第一個重載（通常在此加入一個普通 For循環(huán)）。例如，下面的代碼復(fù)制代碼for(inti=0;i<10;i++){S;}將變?yōu)椋簭?fù)制代碼Parallel.ForAll(0,10,delegate(inti){S;},Environment.ProcessorCount);現(xiàn)在可以使用第二個重載（通常在此加入一個 C#foreach循環(huán)），這樣，復(fù)制代碼List<T>list=...;foreach(Teinlist){S;}將變?yōu)椋簭?fù)制代碼Parallel.ForAll(list,delegate(Te){S;},Environment.ProcessorCount);您需要謹慎以確保 S中的語句不會寫入共享內(nèi)存；否則您需要向并行版本添加合適的同步操作。當(dāng)然可以編寫相應(yīng)版本來支持 IEnumerable<T>、以其他方式對迭代空間進行分區(qū)等（為了節(jié)省空間，本專欄省略了這些內(nèi)容）。在本示例中，在n項子任務(wù)的持續(xù)時間內(nèi)調(diào)用線程被 “浪費”了。更好的方法是使用調(diào)用線程來運行其中一項任務(wù)，然后在其完成時加入其他任務(wù)。沒有必要通過擴展ForAll方法來執(zhí)行此操作。收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔并行分拆有一類操作可使用分拆（又稱為折疊或聚合）來執(zhí)行，其中會以某種方式將許多值進行組合以便生成單一輸出。一般分拆的工作方式如下。使用一個二元運算符（即包含兩個參數(shù)的函數(shù)），并對照向量或一組大小為 n的元素從左至右對其進行計算。對于 j=0至n–1，調(diào)用該二元運算符，作為輸入傳遞至 jth迭代，并將在元素 j–1上調(diào)用運算符得到的輸出作為第一個參數(shù)，將 jth元素本身作為第二個參數(shù)。由于沒有預(yù)先給定的值可用，因此會將一個特殊的種子值用作第0個元素的第一個參數(shù)。然后使用最終（可選）結(jié)果選擇器將中間值轉(zhuǎn)換為最終結(jié)果。我們來看一個示例。如果二元運算符為 +，輸入為包含5個元素{1,2,3,4,5}的向量，那么展開的計算應(yīng)類似于 ((((1+2)+3)+4)+5)。如果將此展開式轉(zhuǎn)換為函數(shù)調(diào)用格式，則類似于以下形式（假定種子值為0）：+(+(+(+(+(0,1),2),3),4),5)換句話說，您只需計算所有輸入數(shù)字的總和即可。這稱為總和分拆。有種直接的方法可以將這種一般化的算法轉(zhuǎn)換為串行算法，如下所示：復(fù)制代碼delegateTFunc<T>(Targ0,Targ1);TReduce<T>(T[]input,Tseed,Func<T>r){Tresult=seed;foreach(Teininput)result=r(result,e);returnresult;}調(diào)用它時，您只需求得一組數(shù)字的和即可，如下所示（在 C#3.0中）：收集于網(wǎng)絡(luò)，如有侵權(quán)請聯(lián)系管理員刪除精品文檔復(fù)制代碼int[]nums=...somesetofnumbers...;intsum=Reduce(nums,0,(x,y)=>x+y;);所有這些內(nèi)容都很抽象，但除了求和之外，還有