多線程之旅之三——Windows內核對象同步機制

 

內核對象(kernel object)：

windows操作系統提供的最近本同步機制，這些對象是構建並發程序和基本並發數據結構的基礎。事實上，無論在代碼中是否直接使用了這些對象，在軟件的某個層次中都肯定會依賴它們。直接使用內核對象將會帶來代價很高的內核切換操作，因為內核對象通常是在內核內存中分布的，因此只有在內核態中運行的代碼才能訪問他們。用戶態抽象層通常使用內核對象來實現等待和出發操作，但同時包含了一些機制保證了盡量避免調用內核對象，避免進行真正的等待。

 

為什么要使用內核對象？

1.內核對象可以實現進程間的同步，也就是說同步線程的時候，線程可以來自多個進程。

2.內核對象可以用於在非托管代碼和托管代碼之間實現互操作。

 

內核對象的分類:

1.內核對象有很多種，有5種是專門用於同步的。其他的諸如I/O完成端口后面會有介紹。

分別是:Mutex, Semaphore, Auto-Reset Event , Manual-Reset Event , Waitable Timer

2.這幾種內核對象就是對不同情形的分類。

從狀態來說，這幾種內核對象都分為 Signaled和 Nonsignaled狀態。簡單來說，就是“是” 和“否”允許進入臨界域。因為創建這幾種內核對象本身是對不同情況的分類，所以這幾種內核對象切換”是“和”否“所需要的條件肯定是不同的。

如同前面所說，在等待某件事情發生時，采用自旋操作往往伴隨着對CPU運行周期的一種搶奪和浪費，有時候我們就需要線程”什么事都不干“，也就是線程阻塞。

 

“阻塞”和停止的區別:

只有Windows操作體統內核才能停止一個線程的運行。在用戶態模式中運行的線程可能會被操作系統搶占，但線程會以最快的速度再次調度。

因此理想的鎖應該是在沒有競爭的情況下可以快速的執行不會阻塞，但是如果出現了多個線程競爭，那么就應該被操作系統內核阻塞，這樣就可以避免CPU時間的浪費。

線程發生阻塞的原因有多種，比如: 執行I/O操作 、睡眠、掛起等。另外一個常見的原因就是等待內核對象切換成Signaled狀態。

其實說白了，為了維護次序肯定不能一擁而上，關鍵就是 要“等”。當然等這個抽象概念大家都知道，更具體一點來說就是等“一個”還是等“多個”？有可能等待一個條件的成熟，也有可能等待多個條件的成熟。再更具體點，在等多個的時候，是這多個中的任意一個呢，還是要等全部？

 

 

 

當線程阻塞時，CLR將使用一個通用的等待函數，而並不考慮這個等待是不是由調用了WaitHandler類的 WaitOne、 WaitAny、WaitAll等，或是在用戶態的混合鎖上的任何阻塞調用，如Monitor,ReadWriteLockSlim等。

 

 

 

內核對象API

當一個線程獲得了一個指向內核對象的引用，也就是說在代碼中new出來或者其他方式獲得一個內核對象時，我們可以調用.NET中的等待API在這個對象等待。多個線程可以同時等待一個內核對象，這樣多個線程也許都會進入阻塞狀態。根據不同的內核對象，喚醒阻塞的線程時也會有不同的情況，有的內核對象在狀態切換時，只喚醒多個等待線程中的一個，比如Pulse，有的是全部都喚醒,比如PulseAll。這兩種方式各有自己的好處和不足。比如PulseAll安全，不會造成喚醒遺失現象，但是會造成喚醒的線程重復等待。具體情況后面會有詳細說明。

 

既然說了各種內核對象只是為了應對不同的狀況而被人們分類的時候，我們先討論這幾種狀況：

 

1.當我們只允許一個線程進入臨界區的時候：

 

Mutex

在win32中，等待一個Mutex對象的API如下:

 

DWORD WINAPI WaitForSingleObject(HANDLE hHandle , DWORD dwMilliseconds)

 

使用方式是:

HANDLE hMutex = CreateMutex(...);
...
DWORD WINAPI WaitForSingleObject(hMutant , IFINITE);

 

由於方法參數用的是一個句柄，因此也可以用該方法在其他內核對象上等待。換句話說，就是 用方法來抽象了"等待"的概念。

 

.NET中對應的方法如下

Mutex mutex = new Mutex();
...
mutex.WaitOne();

 

 

類的繼承關系如下：

  WaitHandle 
      EventWaitHandle 
         AutoResetEvent 
         ManualResetEvent 
      Semaphore 
      Mutex

 

是用 對象間的關系來表達相應的邏輯。 用一個WaitHandle  類來抽象出”等待“的概念。

 

是用方法參數來抽象還是用類的繼承來抽象，可以看做這是C++中過程式風格和C#面向對象風格的區別

 

public abstract class WaitHandle : MarshalByRefObject, IDisposable { 
   public virtual void Close(); 
   public void Dispose(); 

   public virtual Boolean WaitOne(); 
   public virtual Boolean WaitOne(Int32 millisecondsTimeout); 

   public static Int32 WaitAny(WaitHandle[] waitHandles); 
   public static Int32 WaitAny(WaitHandle[] waitHandles, Int32 millisecondsTimeout); 

   public static Boolean WaitAll(WaitHandle[] waitHandles); 
   public static Boolean WaitAll(WaitHandle[] waitHandles, Int32 millisecondsTimeout); 

   public static Boolean SignalAndWait(WaitHandle toSignal, WaitHandle toWaitOn); 
   public static Boolean SignalAndWait(WaitHandle toSignal, WaitHandle toWaitOn, 
      Int32 millisecondsTimeout, Boolean exitContext) 

   public SafeWaitHandle SafeWaitHandle { get; set; } 

   // Returned from WaitAny if a timeout occurs 
   public const Int32 WaitTimeout = 0x102;  
}

 

 

2.當我們只允許指定數量的線程進入臨界區的情況以及應用

 

Semaphore

和mutex不同的是， Semaphore不應該被認為是由某個特定的線程所“擁有”。例如，一個線程可以在信號量上執行插入操作，而另一個線程可以在同一個線程上執行取走操作。通常，信號量用於保護在容量上有限的資源。比如一個固定大小的數據庫連接池需要被定期訪問，這樣同時請求的連接數量不應該超過可用連接的數量。同樣，還可能有一個共享的內存緩沖區，緩沖區的大小是變動的，但要確保同時訪問緩沖區的線程數量與緩沖區中的可用項數量一樣多。

因為允許多個線程進入臨界區，因此信號量並不能避免並發帶來的危害，通常需要配合其他的數據同步機制。 Semaphore本質上就是內核維護的一個計數器，計數值大於1的 Semaphore並不能保證互斥行為。

public sealed class SimpleWaitLock : IDisposable { 
   private Semaphore m_AvailableResources; 

   public SimpleWaitLock(Int32 maximumConcurrentThreads) { 
      m_AvailableResources =  
         new Semaphore(maximumConcurrentThreads, maximumConcurrentThreads); 
   } 

   public void Enter() { 
      // Wait efficiently in the kernel for resource access, then return 
      m_AvailableResources.WaitOne(); 
   } 

   public void Leave() { 
      // This thread doesn’t need access anymore; another thread can have it 
      m_ AvailableResources.Release(); 
   } 

   public void Dispose() { m_AvailableResources.Close(); } 
}

 

 

阻塞列隊

在我們操作一個列隊的時候，最先想到的就是從列隊中提取一個元素的時候如果沒有元素應該怎么辦。我們可以用一個等待條件使得列隊中有元素了再提取，否則就阻塞。看樣子就應該這樣的

public class BlockingQueueWithAutoResetEvents <T>
    {
        private Queue <T> m_queue = new Queue <T>();
        private Mutex m_mutex = new Mutex ();
        private AutoResetEvent m_event = new AutoResetEvent (false );
 
        public void Enqueue(T obj)
        {
            // Enter the critical region and insert into our queue.
            m_mutex.WaitOne();
            try
            {
                m_queue.Enqueue(obj);
            }
            finally
            {
                m_mutex.ReleaseMutex();
            }
 
            // Note that an item is available, possibly waking a consumer.
            m_event.Set();
        }
 
        public T Dequeue()
        {
            // Dequeue the item from within our critical region.
            T value;
            bool taken = true ;
            m_mutex.WaitOne();
            try
            {
                // If the queue is empty, we will need to exit the
                // critical region and wait for the event to be set.
                while (m_queue.Count == 0)
                {
                    taken = false ;
                    WaitHandle .SignalAndWait(m_mutex, m_event);
                    m_mutex.WaitOne();
                    taken = true ;
                }
 
                value = m_queue.Dequeue();
            }
            finally
            {
                if (taken)
                {
                    m_mutex.ReleaseMutex();
                }
            }
 
            return value;
        }
    }

 

 

阻塞/有界列隊

但是如果對於一個動態的生產者/消費者的情況中，這樣的只對消費者限制，對生產者沒限制的列隊，隨着時間的推移，生產者和消費者之間的比率就會不平衡。

我們想要實現的功能是：如果試圖從一個空的隊列中取數據，那么線程將阻塞，知道有新的元素加入。試圖將數據放入到一個容量已滿的列隊同樣也會導致線程阻塞，直到有空間騰出來。

 

現在我們采用 Semaphore+Mutex的方式實現這個數據結構。Mutex用來實現臨界域的互斥行為，確保對狀態的修改能夠安全的進行； Semaphore用於實現控制同步。 Semaphore使得這個工作變得相對容易，因為對容量有限的資源進行保護正是當初為什么要划分出這么一個類的原因。值得注意的是，管理 Semaphore和Mutex產生的內核切換開銷可能是這個結構最大的性能問題。WaitOne會使得計數器減1，當計數為0時，再調用WaitOne就會阻塞。

 

        public class BlockingBoundedQueue<T>
        {
            private Queue<T> m_queue = new Queue<T>();
            private Mutex m_mutex = new Mutex();
            private Semaphore m_producerSemaphore;
            private Semaphore m_consumerSemaphore;
            public BlockingBoundedQueue(int capacity)
            {
                m_producerSemaphore = new Semaphore(capacity, capacity); 
　　　　　　　　　 m_consumerSemaphore = new Semaphore(0, capacity);
            }
            public void Enqueue(T obj)
            {
                m_producerSemaphore.WaitOne();
                try { m_queue.Enqueue(obj); }
                finally { m_mutex.ReleaseMutex(); }
                m_consumerSemaphore.Release();
            }
            public T Dequeue()
            {
                m_consumerSemaphore.WaitOne();
                T value;
                m_mutex.WaitOne();
                try { value = m_queue.Dequeue(); }
                finally { m_mutex.ReleaseMutex(); }
                m_producerSemaphore.Release();
                return value;
            }
        }

 

 

因為其實 Semaphore說白了就是一個計數器，因此我們用一個int類型來計數+mutex也同樣能實現這個功能。后面我會只用混合構造鎖Monitor這一個類來實現。

其實在一些情況下，如AutoResteEvent和計數為1的 Semaphore, 感覺起來就和 Mutex差不多。

 

如果單純的使用內核對象作為同步手段因為需要付出內核態切換的代價所以並不是一個好的選擇。

我們接下來就看看結合自旋機制和內核對象的抽象層度更高的混合構造鎖。

多線程之旅

1.多線程之旅——從概念開始

2.多線程之旅二——線程

3.多線程之旅之三——Windows內核對象同步機制

4.多線程之旅之四——淺談內存模型和用戶態同步機制

5.多線程之旅之五——線程池與I/O完成端口

6.多線程之旅六——異步編程模式，自己實現IAsyncResult

7.多線程之旅七——再談內存模型

8.多線程之旅八——多線程下的數據結構