老實說，沒有哪個開發人員愿意在其編碼時還要考慮線程同步。更糟糕的情況是，編寫線程同步代碼一點也不好玩。稍一不慎，就會導致共享資源狀態不一致，從而引發程序未預期行為。此外，當我們添加線程同步代碼時還會導致程序運行變慢，損害性能和可伸縮性。從這點上來看，線程同步簡直一無是處?？上?#xff0c;這也是現實生活中必要的一部分。尤其在多核CPU成為主流的今天。

考慮下這種情況：只有一個線程試圖訪問某個資源。在此種狀況下，那些線程同步代碼純粹是額外開銷。那么如何保護性能呢？

首先，不要讓線程從用戶模式轉到內核模式。在Windows上，這是個非常昂貴的操作。比如大家熟知的Critical Section(在.NET上我們叫它Monitor)在等待持有鎖時，便在用戶模式先旋轉等待一段時間，實在不行才轉入內核模式。

其次，當線程進入等待狀態時，它便停止運行。很顯然，停止運行的線程還有啥性能可言。所幸的是，假如線程進入內核模式，它也就只能進入等待狀態。所以，如果你遵守第一條規則(避免進入內核)，那么第二條規則(不要讓線程進入等待狀態)將會自動遵守。我們應當讓線程大多數時間在運行而不是在等待。

最后，盡可能用盡量少的線程來完成任務。最理想的情況是，從不讓存在的線程數多于你的計算機上的CPU個數。因此，當你的機器只有四個CPU的時候，最好也不要超過四個線程的數量。為什么？因為當超過四個線程的時候，操作系統就會以每隔幾毫秒~幾十毫秒的頻率調度線程到CPU。在Windows上，這個數字大概是20毫秒。線程上下文切換花費不少時間，而且這純粹是額外開銷。在Windows上，線程上下文切換需要幾千個時鐘周期。如果進行切換的線程屬于另一個進程，開銷還會更大，因為操作系統還要切換虛擬地址空間。如果可運行的線程數量等于或小于CPU數量的話，操作系統只運行這些線程，而不會發生上下文切換。在這種情況下，當然性能最佳。

但是，這僅在理想的世界中才可行。所以，我們通?？梢钥吹綑C器上存在的線程數量遠多于CPU數量。事實上，在我的單核CPU機器上，目前就有378個線程活動，這跟理想情況差了豈止十萬八千里。Windows上每個進程都有自己的線程，這主要是為了進程彼此隔離。假如一個進程進入死循環，其他進程還可以繼續執行?，F在我的機器上運行著29個進程，也就是說，至少有29個線程在運行了。那剩下的349個線程呢？很多程序通常還會用多線程來進行隔離。比如，CLR就專門有一個GC線程來負責回收資源，而不管其他線程在干什么。

當然，還有其他眾多原因。其中一個比較重要的原因就是相當多的(或者我大膽的說絕大多數)開發人員并不知道如何有效的使用線程。多年來，我們從各種知識資源(甚至不少講解多線程的教材)上學習到的編程方式，實際上對構建高性能，高可伸縮性程序有害。比如在《Windows System Programming, Third Edition》中，作者多處提到他偏愛使用線程而不是系統提供的異步方式。很明顯的一個例子就是作者采用線程編寫的同步I/O API要快于系統異步I/O API。雖然異步和線程有著很親密的關系，但我們得到的暗示是鼓勵使用線程而不是異步?？紤]下當有大量線程爭用I/O時的情景：我們的CPU忙著線程上下文切換，磁盤的磁頭數量有限，不夠這些線程來回搗騰，而每一個I/O操作卻又耗時良多。再比如MSDN上Microsoft提供的Windows API示例。它們大多數均使用順序思維編寫。大多數情況下這無可厚非，而且符合人類直覺。然而這卻讓我們學習模仿，甚至變成了習慣。在并行程序設計上相當有害。

如果你發現思考下列問題，那么很可能你錯誤的架構了你的程序。而且需要重新架構它。

Windows支持的最大線程數是多少？

在一個單CPU上，我能最大創建多少線程？

在線程池中，我能最大擁有多少線程？

我能增加線程池中最大線程數嗎？

我應該在線程和用戶間進行1對1通信嗎？我知道很少有服務端程序每個用戶或客戶端用掉一個線程。

接下來，我們進入正題，講解SpinWait。那么SpinWait是什么呢？

我們知道一般的鎖(比如Mutex)在當前線程無法持有時，會使得線程在內核等待。Critical Section/Monitor則有所不同。它大概需要以下幾個步驟(以下簡稱其鎖為CS)：

執行EnterCriticalSection()或者Monitor.Enter()/lock的線程會首先測試CS的鎖定位。如果該位為OFF(解鎖)，那么線程就會在該位上設置一下(加鎖)，且不需要等待便繼續。這通常只需執行1~2個機器指令。

如果CS被鎖定，線程就會進入一個旋轉等待持有鎖。而線程在旋轉期間會反復測試鎖定位。單處理器系統會立即放棄，而在SMP或多核CPU系統上則旋轉一段時間才會放棄。在此之前，線程都在用戶模式下運行。

一旦線程放棄測試鎖定位(在單處理器上立即如此)，線程使用信號量在內核進入等待狀態。

執行LeaveCriticalSection()/Monitor.Exit()或代碼退出了lock塊。如果存在等待線程，則使用ReleaseSemaphore()通知內核。

在第二步，我們看到會有一個旋轉等待。這就是本文的主旨：SpinWait。CS正是使用它在用戶模式下旋轉等待的。值得注意的是SpinWait只有在SMP或多核CPU下才具有使用意義。在單處理器下，旋轉非常浪費CPU時間，沒有任何意義。

下面讓我們實現一個SpinWait。

using?System;
using?System.Threading;

namespace?Lucifer.DataStructure
{
????public?struct?SpinWait
????{
????????internal?const?int?yieldFrequency?=?4000;

????????private?volatile?int?count;

????????public?void?Spin()
????????{
????????????this.count?=?++count?%?Int32.MaxValue;
????????????if?(IsSingleProcessor)
????????????{
????????????????SwitchToThread();
????????????}
????????????else
????????????{
????????????????int?numeric?=?this.count?%?yieldFrequency;
????????????????if?(numeric?>?0)
????????????????{
????????????????????Thread.SpinWait((int)(1f?+?(numeric?*?0.032f)));
????????????????}
????????????????else
????????????????{
????????????????????SwitchToThread();
????????????????}
????????????}
????????}

????????public?void?Reset()
????????{
????????????this.count?=?0;
????????}

????????public?int?Count
????????{
????????????get
????????????{
????????????????return?this.count;
????????????}
????????}
????????/**//*
?????????*??判斷是否單處理器系統。
?????????*/
????????private?static?readonly?bool?IsSingleProcessor?=?
????????????(Environment.ProcessorCount?==?1);

????????[DllImport("Kernel32",?ExactSpelling?=?true)]
????????private?static?extern?void?SwitchToThread();
????}
}

注意，我們封裝了非托管API-SwitchToThread()來處理單處理器情況，以便線程調度可以平滑進行。假如使用Sleep(0)則需要注意在某些地方優先級反向，且性能也不如SwitchToThread高。

此外，我們使用Thread.SpinWait()方法來處理超線程問題。超線程是一項在Xeon，P4及以后CPU上的技術。超線程CPU上有兩個邏輯CPU。每個邏輯CPU都有自己的架構狀態(CPU寄存器)，但是各個邏輯CPU共享一個單獨的執行資源集合(例如CPU緩存)。當一個邏輯CPU暫停時(由于緩存失效，分支預測失敗或者等待上一指令的結果)，芯片就會切換到另一個邏輯CPU上，并讓它運行。據Intel報告說超線程CPU可以獲得10%~30%的性能提升。x86架構提供PAUSE匯編指令來完成上述情況，同時它等效于REP NOP;匯編指令。這個指令可以兼容于那些早期未支持超線程技術的IA-32 CPU。而在Windows上，Microsoft定義了YieldProcessor()宏來完成這些工作。.NET則對其進行了封裝。Thread.SpinWait(int iterations)在x86平臺上實際上相當于一下代碼：

void?Thread.SpinWait(Int32?iterations)
{
????for?(Int32?i?=?0;?i?<?iterations;?i++)
????{
????????YieldProcessor();
????}
}

由此，我們也應該知道SpinWait(0)沒有意義，調用它必須使用大于0之數。

我們為什么要使用SpinWait呢？因為它運行在用戶模式下。某些情況下，我們想要在退回到真正等待狀態前先旋轉一段時間?？雌饋磉@似乎很難理解。這根本在做無用功嘛。大多數人在一開始就避免旋轉等待。我們知道線程上下文切換需要花費幾千個周期(在Windows上確實如此)。我們暫且稱其為C。假如線程所等待的時間小于2C(1C用于等待自身，1C用于喚醒)，則旋轉等待可以降低等待所造成的系統開銷和滯后時間，從而提升算法的整體吞吐量和可伸縮性。

當我們采用旋轉等待時，必須謹慎行事。比如：要確保在旋轉循環內調用Thread.SpinWait()，以提高 Intel 超線程技術的計算機上硬件對其他硬件線程的可用性；通過輕微的回退 (back-off) 來引入隨機選擇，從而改善訪問的局部性（假定調用方持續重讀共享狀態）并可能避免活鎖；當然，在單 CPU 的計算機最好不要采用這種方法（因為在這種環境下旋轉非常浪費資源）。

不可否認，選擇頻率和旋轉計數是不確定的。與 Win32 臨界區旋轉計數類似，我們應該根據測試和實驗的結果來選擇合理的數值，而且即使合理的數值在不同系統中也會發生變化。例如，根據 Microsoft Media Center 和 Windows kernel 團隊的經驗，MSDN 文檔建議臨界區旋轉計數為 4,000 ，但您的選擇可以有所不同。理想的計數取決于多種因素，包括在給定時間等待事件的線程數和事件出現的頻率等。一個經過實踐的數字大概在4000~12000之間。

接下來，我們通過《并發數據結構：Stack》和本文學習到的知識構造ConcurrentStack，來學習SpinWait如何使用，同時也以它作為本文結束。

代碼如下：

????public?class?ConcurrentStack<T>
????{
????????private?volatile?ListNode<T>?topOfStack;

????????public?bool?IsEmpty()
????????{
????????????return?topOfStack?==?null;
????????}

????????public?void?Push(T?item)
????????{
????????????ListNode<T>?newTopOfStack?=?new?ListNode<T>(item);
????????????SpinWait?wait?=?new?SpinWait();
????????????while?(true)
????????????{
????????????????ListNode<T>??oldTopOfStack?=?topOfStack;
????????????????newTopOfStack.next?=?oldTopOfStack;

????????????????if?(Interlocked.CompareExchange<ListNode<T>>(ref?topOfStack,?newTopOfStack,?oldTopOfStack)?==?oldTopOfStack)
????????????????{
????????????????????return;
????????????????}

????????????????wait.Spin();
????????????}
????????}
????????/**////?<summary>
????????///?考慮到在多線程環境中，這里不拋出異常。我們需要人為判斷其是否為空。
????????///?</summary>
????????///?<returns></returns>
????????public?bool?TryPop(out?T?result)
????????{
????????????ListNode<T>?oldTopOfStack;
????????????ListNode<T>?newTopOfStack;
????????????SpinWait?wait?=?new?SpinWait();

????????Spin:
????????????oldTopOfStack?=?topOfStack;
????????????if?(oldTopOfStack?==?null)
????????????{
????????????????result?=?default(T);
????????????????return?false;
????????????}
????????????newTopOfStack?=?topOfStack.next;

????????????if?(Interlocked.CompareExchange<ListNode<T>>(ref?topOfStack,?newTopOfStack,?oldTopOfStack)?!=?oldTopOfStack)
????????????{
????????????????wait.Spin();
????????????????goto?Spin;
????????????}
????????????result?=?oldTopOfStack.item;
????????????return?true;
????????}

????????public?bool?TryPeek(out?T?result)
????????{
????????????ListNode<T>?head?=?topOfStack;
????????????if?(head?==?null)
????????????{
????????????????result?=?default(T);
????????????????return?false;
????????????}
????????????result?=?head.item;
????????????return?true;
????????}

????????/**//*?*****************************************
?????????*?簡單的單向鏈表實現
?????????*?*****************************************/
????????class?ListNode<T>
????????{
????????????internal?T?item;
????????????internal?ListNode<T>?next;

????????????public?ListNode(T?initItem,?ListNode<T>?initNext)
????????????{
????????????????this.item?=?initItem;
????????????????this.next?=?initNext;
????????????}

????????????public?ListNode(T?initItem)
????????????????:?this(initItem,?null)
????????????{
????????????}
????????}
????} 最后，重申一下，SpinWait僅適合在線程等待時間較短，且在SMP或多核環境下。單處理器系統使用沒有意義。

轉載于:https://www.cnblogs.com/lucifer1982/archive/2008/06/01/1211471.html

總結

以上是生活随笔為你收集整理的并发数据结构 : SpinWait的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。