天天看點

C#的并發機制優秀在哪?

作者 | 馬超 責編 | 張紅月

出品 | CSDN部落格

上次用C#寫.Net代碼差不多還是10多年以前,由于當時Java已經頗具王者風範,.Net幾乎被打得潰不成軍。是以當時筆者對于這個.Net的項目态度比較敷衍了事,沒有對其中一些優秀機制有很深的了解,在去年寫《C和Java沒那麼香了,高并發時代誰能稱王》時都沒給.Net以一席之地,不過最近恰好機緣巧合,我又接手了一個Windows方面的項目,這也讓我有機會重新審視一下自己關于.Net架構的相關知識。

項目原型要實作的功能并不複雜,主要就是記錄移動儲存設備中檔案拷出的記錄,而且需要盡可能少的占用系統資源,而在開發過程中我無意中加了一行看似沒有任何效果的代碼,使用Invoke方法記錄檔案拷出情況,這樣的操作卻讓程式執行效率明顯會更高,這背後的原因特别值得總結。

C#的并發機制優秀在哪?

一行沒用的代碼卻提高了效率?

由于筆者需要記錄的檔案拷出資訊并沒有回顯在UI的需要,是以也就沒考慮并發沖突的問題,在最初版本的實作中,我對于filesystemwatcher的回調事件,都是直接處理的,如下:

private void DeleteFileHandler(object sender, FileSystemEventArgs e)              {              if(files.Contains(e.FullPath))              {              files.Remove(e.FullPath);              //一些其它操作              }              }           

這個程式的處理效率在普通的辦公PC上如果同時拷出20個檔案,那麼在拷貝過程中,U盤監測程式的CPU使用率大約是0.7%。

C#的并發機制優秀在哪?

但是一個非常偶然的機會,我使用了Event/Delegate的Invoke機制,結果發現這樣一個看似的廢操作,卻讓程式的CPU占用率下降到0.2%左右

private void UdiskWather_Deleted(object sender, FileSystemEventArgs e)              {              if(this.InvokeRequired)              {              this.Invoke(new DeleteDelegate(DeleteFileHandler), new object { sender,e }); }              else              {              DeleteFileHandler(sender, e);              }              }           
C#的并發機制優秀在哪?

在我最初的認識中.net中的Delegate機制在調用過程中是要進行拆、裝箱操作的,是以這不拖慢操作就不錯了,但實際的驗證結果卻相反。

C#的并發機制優秀在哪?

看似沒用的Invoke到底有什麼用

這裡先給出結論,Invoke能提升程式執行效率,其關鍵還是在于線程在多核之間切換的消耗要遠遠高于拆、裝箱的資源消耗,我們知道我們程式的核心就是操作files這個共享變量,每次在被檢測的U盤目錄中如果發生檔案變動,其回調通知函數可能都運作在不同的線程,如下:

C#的并發機制優秀在哪?

Invoke機制的背後其實就是保證所有對于files這個共享變量的操作,全部都是由一個線程執行完成的。

C#的并發機制優秀在哪?

目前.Net的代碼都開源的,下面我們大緻講解一下Invoke的調用過程,不管是BeginInvoke還是Invoke背後其實都是調用的MarshaledInvoke方法來完成的,如下:

public IAsyncResult BeginInvoke(Delegate method, params Object[] args) {              using (new MultithreadSafeCallScope) {              Control marshaler = FindMarshalingControl;              return(IAsyncResult)marshaler.MarshaledInvoke(this, method, args, false);              }              }           

MarshaledInvoke的主要工作是建立ThreadMethodEntry對象,并把它放在一個連結清單裡進行管理,然後調用PostMessage将相關資訊發給要通信的線程,如下:

private Object MarshaledInvoke(Control caller, Delegate method, Object[] args, bool synchronous) {              if (!IsHandleCreated) {              throw new InvalidOperationException(SR.GetString(SR.ErrorNoMarshalingThread));              }               ActiveXImpl activeXImpl = (ActiveXImpl)Properties.GetObject(PropActiveXImpl);              if (activeXImpl != ) {              IntSecurity.UnmanagedCode.Demand;              }              // We don't want to wait if we're on the same thread, or else we'll deadlock.              // It is important that syncSameThread always be false for asynchronous calls.              //              bool syncSameThread = false;              int pid; // ignored              if (SafeNativeMethods.GetWindowThreadProcessId(new HandleRef(this, Handle), out pid) == SafeNativeMethods.GetCurrentThreadId) {              if (synchronous)              syncSameThread = true;              }              // Store the compressed stack information from the thread that is calling the Invoke              // so we can assign the same security context to the thread that will actually execute              // the delegate being passed.              //              ExecutionContext executionContext = ;              if (!syncSameThread) {              executionContext = ExecutionContext.Capture;              }              ThreadMethodEntry tme = new ThreadMethodEntry(caller, this, method, args, synchronous, executionContext);              lock (this) {              if (threadCallbackList == ) {              threadCallbackList = new Queue;              }              }              lock (threadCallbackList) {              if (threadCallbackMessage == 0) {              threadCallbackMessage = SafeNativeMethods.RegisterWindowMessage(Application.WindowMessagesVersion + "_ThreadCallbackMessage");              }              threadCallbackList.Enqueue(tme);              }              if (syncSameThread) {              InvokeMarshaledCallbacks;              } else {              //              UnsafeNativeMethods.PostMessage(new HandleRef(this, Handle), threadCallbackMessage, IntPtr.Zero, IntPtr.Zero);              }              if (synchronous) {              if (!tme.IsCompleted) {              WaitForWaitHandle(tme.AsyncWaitHandle);              }              if (tme.exception != ) {              throw tme.exception;              }              return tme.retVal;              }              else {              return(IAsyncResult)tme;              }              }           

Invoke的機制就保證了一個共享變量隻能由一個線程維護,這和GO語言使用通信來替代共享記憶體的設計是暗合的,他們的理念都是 "讓同一塊記憶體在同一時間内隻被一個線程操作" 。這和現代計算體系結構的多核CPU(SMP)有着密不可分的聯系,

這裡我們先來科普一下CPU之間的通信MESI協定的内容。我們知道現代的CPU都配備了高速緩存,按照多核高速緩存同步的MESI協定約定,每個緩存行都有四個狀态,分别是E(exclusive)、M(modified)、S(shared)、I(invalid),其中:

M:代表該緩存行中的内容被修改,并且該緩存行隻被緩存在該CPU中。這個狀态代表緩存行的資料和記憶體中的資料不同。

E:代表該緩存行對應記憶體中的内容隻被該CPU緩存,其他CPU沒有緩存該緩存對應記憶體行中的内容。這個狀态的緩存行中的資料與記憶體的資料一緻。

I:代表該緩存行中的内容無效。

S:該狀态意味着資料不止存在本地CPU緩存中,還存在其它CPU的緩存中。這個狀态的資料和記憶體中的資料也是一緻的。不過隻要有CPU修改該緩存行都會使該行狀态變成 I 。

四種狀态的狀态轉移圖如下:

C#的并發機制優秀在哪?

我們上文也提到了,不同的線程是有大機率是運作在不同CPU核上的,在不同CPU操作同一塊記憶體時,站在CPU0的角度上看,就是CPU1會不斷發起remote write的操作,這會使該高速緩存的狀态總是會在S和I之間進行狀态遷移,而一旦狀态變為I将耗費比較多的時間進行狀态同步。

C#的并發機制優秀在哪?

是以我們可以基本得出 this.Invoke(new DeleteDelegate(DeleteFileHandler), new object { sender,e }); ;這行看似無關緊要的代碼之後,無意中使files共享變量的維護操作,由多核多線程共同操作,變成了衆多子線程向主線程通信,所有維護操作均由主線程進行,這也使最終的執行效率有所提高。

C#的并發機制優秀在哪?

深度解讀,為何要加兩把鎖

在目前使用通信替代共享記憶體的大潮之下,鎖其實是最重要的設計。

我們看到在.Net的Invoke實作中,使用了兩把鎖lock (this) 與lock (threadCallbackList)。

lock (this) {              if (threadCallbackList == ) {              threadCallbackList = new Queue;              }              }              lock (threadCallbackList) {              if (threadCallbackMessage == 0) {              threadCallbackMessage = SafeNativeMethods.RegisterWindowMessage(Application.WindowMessagesVersion + "_ThreadCallbackMessage");              }              threadCallbackList.Enqueue(tme);              }           

在.NET當中lock關鍵字的基本可以了解為提供了一個近似于CAS的鎖(Compare And Swap)。CAS的原理不斷地把"期望值"和"實際值"進行比較,當它們相等時,說明持有鎖的CPU已經釋放了該鎖,那麼試圖擷取這把鎖的CPU就會嘗試将"new"的值(0)寫入"p"(交換),以表明自己成為spinlock新的owner。僞代碼示範如下:

void CAS(int p, int old,int new)              {              if *p != old              do nothing              else               *p ← new              }           

基于CAS的鎖效率沒問題,尤其是在沒有多核競争的情況CAS表現得尤其優秀,但CAS最大的問題就是不公平,因為如果有多個CPU同時在申請一把鎖,那麼剛剛釋放鎖的CPU極可能在下一輪的競争中擷取優勢,再次獲得這把鎖,這樣的結果就是一個CPU忙死,而其它CPU卻很閑,我們很多時候诟病多核SOC“一核有難,八核圍觀”其實很多時候都是由這種不公平造成的。

為了解決CAS的不公平問題,業界大神們又引入了TAS(Test And Set Lock)機制,個人感覺還是把TAS中的T了解為Ticket更好記一些,TAS方案中維護了一個請求該鎖的頭尾索引值,由"head"和"tail"兩個索引組成。

struct lockStruct{              int32 head;              int32 tail;              } ;           

"head"代表請求隊列的頭部,"tail"代表請求隊列的尾部,其初始值都為0。

最一開始時,第一個申請的CPU發現該隊列的tail值是0,那麼這個CPU會直接擷取這把鎖,并會把tail值更新為1,并在釋放該鎖時将head值更新為1。

在一般情況下當鎖被持有的CPU釋放時,該隊列的head值會被加1,當其他CPU在試圖擷取這個鎖時,鎖的tail值擷取到,然後把這個tail值加1,并存儲在自己專屬的寄存器當中,然後再把更新後的tail值更新到隊列的tail當中。接下來就是不斷地循環比較,判斷該鎖目前的"head"值,是否和自己存儲在寄存器中的"tail"值相等,相等時則代表成功獲得該鎖。

TAS這類似于使用者到政務大廳去辦事時,首先要在叫号機取号,當從業人員廣播叫到的号碼與你手中的号碼一緻時,你就擷取了辦事櫃台的所有權。

但是TAS卻存在一定的效率問題,根據我們上文介紹的MESI協定,這個lock的頭尾索引其實是在各個CPU之間共享的,是以tail和head頻繁更新,還是會引發調整緩存不停的invalidate,這會極大的影響效率。

是以我們看到在.Net的實作中幹脆就直接引入了threadCallbackList的隊列,并不斷将tme(ThreadMethodEntry)加入隊尾,而接收消息的程序,則不斷從隊首擷取消息。

C#的并發機制優秀在哪?
lock (threadCallbackList) {              if (threadCallbackMessage == 0) {              threadCallbackMessage = SafeNativeMethods.RegisterWindowMessage(Application.WindowMessagesVersion + "_ThreadCallbackMessage");              }              threadCallbackList.Enqueue(tme);              }           

當隊首指向這個tme時,消息才被發送,其實是一種類似于MAS的實作,當然MAS實際是為每個CPU都建立了一個專屬的隊列,和Invoke的設計略有不同,不過基本的思想是一緻的。

很多時候年少時不是品不出很多東西背後味道的,這也讓我錯過了很多非常值得總結的技術要點,是以在春節假期總結一下最近使用C#的心得,以飨讀者,順祝大家新春愉快!

作者簡介:馬超,金融科技專家,人民大學高禮金融研究院校外雙聘導師,阿裡雲MVP,華為2020年十大開發者之星,CSDN約稿專欄作者,著名的金融科技的布道者。衆多國産開源項目的推動者及貢獻人。
C#的并發機制優秀在哪?

《新程式員003》正式上市,50餘位技術專家共同創作,雲原生和數字化的開發者們的一本技術精選圖書。内容既有發展趨勢及方法論結構,華為、阿裡、位元組跳動、網易、快手、微軟、亞馬遜、英特爾、西門子、施耐德等30多家知名公司雲原生和數字化一手實戰經驗!

C#的并發機制優秀在哪?
C#的并發機制優秀在哪?