聊一聊 Monitor.Wait 和 Pulse 的底層玩法

一：背景 1. 講故事在dump分析的過程中經常會看到很多線程卡在Monitor.Wait方法上，曾經也有不少人問我為什麼用 !syncblk 看不到 Monitor.Wait 上的鎖信息，剛好昨天有時間我就來研究一下。二：Monitor.Wait 底層怎麼玩的 1. 案例演示為了方便講述，先 ...

一：背景

1. 講故事

在dump分析的過程中經常會看到很多線程卡在Monitor.Wait方法上，曾經也有不少人問我為什麼用 !syncblk 看不到 Monitor.Wait 上的鎖信息，剛好昨天有時間我就來研究一下。

二：Monitor.Wait 底層怎麼玩的

1. 案例演示

為了方便講述，先上一段演示代碼，Worker1 在執行的過程中需要喚醒 Worker2 執行，當 Worker2 執行完畢之後自己再繼續執行，參考代碼如下：


    internal class Program
    {
        static Person lockObject = new Person();

        static void Main()
        {
            Task.Run(() => { Worker1(); });
            Task.Run(() => { Worker2(); });

            Console.ReadLine();
        }

        static void Worker1()
        {
            lock (lockObject)
            {
                Console.WriteLine($"{DateTime.Now} 1. 執行 worker1 的業務邏輯...");
                Thread.Sleep(1000);

                Console.WriteLine($"{DateTime.Now} 2. 等待 worker2 執行完畢...");
                Monitor.Wait(lockObject);

                Console.WriteLine($"{DateTime.Now} 4. 繼續執行 worker1 的業務邏輯...");
            }
        }

        static void Worker2()
        {
            Thread.Sleep(10);
            lock (lockObject)
            {
                Console.WriteLine($"{DateTime.Now} 3. worker2 的邏輯執行完畢...");
                Monitor.Pulse(lockObject);
            }
        }
    }

    public class Person { }

有了代碼和輸出之後，接下來就是分析底層玩法了。

2. 模型架構圖

研究來研究去總得有個結果，千言萬語繪成一張圖，截圖如下：

從圖中可以看到這地方會涉及到一個核心的數據結構 WaitEventLink，參考如下：


// Used inside Thread class to chain all events that a thread is waiting for by Object::Wait
struct WaitEventLink {
    SyncBlock         *m_WaitSB;	   // 當前對象的 syncblock
    CLREvent          *m_EventWait;    // 當前線程的 m_EventWait 
    PTR_Thread         m_Thread;       // Owner of this WaitEventLink.
    PTR_WaitEventLink  m_Next;         // Chain to the next waited SyncBlock.
    SLink              m_LinkSB;       // Chain to the next thread waiting on the same SyncBlock.
    DWORD              m_RefCount;     // How many times Object::Wait is called on the same SyncBlock.
};

代碼里對每一個欄位都做了表述，還是非常清楚的，也看到了這裡存在兩個隊列。

m_Next：當前線程要串聯的 SyncBlock 隊列，Node 是 WaitEventLink 結構。
m_LinkSB：當前同步塊串聯的 Thread 隊列，Node 是 m_LinkSB 地址。

3. 底層的源碼驗證

首先我們看下C#的 Monitor.Wait(lockObject) 底層是如何實現的，它對應著 coreclr 的 ObjectNative::WaitTimeout 方法，核心實現如下：


BOOL SyncBlock::Wait(INT32 timeOut)
{
	//步驟1
    WaitEventLink* walk = pCurThread->WaitEventLinkForSyncBlock(this);

	//步驟2
    CLREvent* hEvent = &(pCurThread->m_EventWait);

    waitEventLink.m_WaitSB = this;
    waitEventLink.m_EventWait = hEvent;
    waitEventLink.m_Thread = pCurThread;
    waitEventLink.m_Next = NULL;
    waitEventLink.m_LinkSB.m_pNext = NULL;
    waitEventLink.m_RefCount = 1;
    pWaitEventLink = &waitEventLink;
    walk->m_Next = pWaitEventLink;

    hEvent->Reset();

	//步驟3
    ThreadQueue::EnqueueThread(pWaitEventLink, this);

    isEnqueued = TRUE;
    PendingSync syncState(walk);

    OBJECTREF obj = m_Monitor.GetOwningObject();

    m_Monitor.IncrementTransientPrecious();

	//步驟4
    syncState.m_EnterCount = LeaveMonitorCompletely();

    isTimedOut = pCurThread->Block(timeOut, &syncState);

    return !isTimedOut;
}

代碼邏輯非常簡單，大概步驟如下：

從當前線程的 m_WaitEventLink 所指向的隊列中尋找 SyncBlock 節點，如果沒有就返回尾部節點。
將當前節點拼接到尾部。
新節點通過 EnqueueThread 方法送入到 m_LinkSB 所指向的隊列，這裡有一個小技巧，它只存放 WaitEventLink->m_LinkSB 地址，後續會通過 -0x20 來反推 WaitEventLink 結構首地址，從而來獲取線程等待事件，參考代碼如下：


inline PTR_WaitEventLink ThreadQueue::WaitEventLinkForLink(PTR_SLink pLink)
{
    LIMITED_METHOD_CONTRACT;
    SUPPORTS_DAC;
    return (PTR_WaitEventLink) (((PTR_BYTE) pLink) - offsetof(WaitEventLink, m_LinkSB));
}

使用 LeaveMonitorCompletely 方法將 AwareLock 鎖給釋放掉，從而讓等待這個 lock 的線程進入方法，即當前的 Worker2，簡化後代碼如下：


LONG LeaveMonitorCompletely()
{
    return m_Monitor.LeaveCompletely();
}

void Signal()
{
    m_SemEvent.SetMonitorEvent();
}

void CLREventBase::SetMonitorEvent(){
    Set();
}

總而言之，Monitor.Wait 主要還是用來將Node追加到兩大隊列，接下來研究下 Monitor.Pulse 的內部實現，這個就比較簡單了，無非就是在 m_LinkSB 指向的隊列中提取一個Node而已，核心代碼如下：


void SyncBlock::Pulse()
{
    WaitEventLink* pWaitEventLink;

    if ((pWaitEventLink = ThreadQueue::DequeueThread(this)) != NULL)
        pWaitEventLink->m_EventWait->Set();
}

// Unlink the head of the Q.  We are always in the SyncBlock's critical
// section.
/* static */
inline WaitEventLink *ThreadQueue::DequeueThread(SyncBlock *psb)
{
    WaitEventLink* ret = NULL;
    SLink* pLink = psb->m_Link.m_pNext;

    if (pLink)
    {
        psb->m_Link.m_pNext = pLink->m_pNext;
        ret = WaitEventLinkForLink(pLink);
    }
    return ret;
}

inline PTR_WaitEventLink ThreadQueue::WaitEventLinkForLink(PTR_SLink pLink)
{
    return (PTR_WaitEventLink)(((PTR_BYTE)pLink) - offsetof(WaitEventLink, m_LinkSB));
}

class SyncBlock
{
  protected:
    SLink m_Link;
}

上面的代碼邏輯還是非常清楚的，從 SyncBlock.m_Link 所串聯的 WaitEventLink 隊列中提取第一個節點，但這個節點保存的是 WaitEventLink.m_LinkSB 地址，所以需要反向 -0x20 取到 WaitEventLink 首地址，可以用 windbg 來驗證一下。


0:017> dt coreclr!WaitEventLink
   +0x000 m_WaitSB         : Ptr64 SyncBlock
   +0x008 m_EventWait      : Ptr64 CLREvent
   +0x010 m_Thread         : Ptr64 Thread
   +0x018 m_Next           : Ptr64 WaitEventLink
   +0x020 m_LinkSB         : SLink
   +0x028 m_RefCount       : Uint4B

取到首地址之後就就可以將當前線程的 m_EventWait 喚醒，這就是為什麼調用 Monitor.Pulse(lockObject); 之後另一個線程喚醒的內部邏輯，有些朋友好奇那 Monitor.PulseAll 是不是會把這個隊列中的所有 Node 上的 m_EventWait 都喚醒呢？哈哈，真聰明，源碼如下：


void SyncBlock::PulseAll()
{
    WaitEventLink* pWaitEventLink;

    while ((pWaitEventLink = ThreadQueue::DequeueThread(this)) != NULL)
        pWaitEventLink->m_EventWait->Set();
}

眼尖的朋友會有一個疑問，這個隊列數據提取了，那另一個隊列的數據是不是也要相應的改動，這個確實，它的邏輯是在Wait方法的 PendingSync syncState(walk); 析構函數里，感興趣的朋友可以看一下內部的void Restore(BOOL bRemoveFromSB) 方法即可。