聊一聊 Monitor.Wait 和 Pluse 的底层玩法

monitor,wait,pluse · 浏览次数 : 0

小编点评

本文研究了 .NET 中 Monitor.Wait 方法的底层实现原理。通过案例演示和模型架构图，我们可以清晰地了解 Monitor.Wait 方法的工作原理。同时，本文还分析了 Monitor.Pulse 和 Monitor.PulseAll 方法的实现。 1. 案例演示本文首先提供了一个简单的案例，其中 Worker1 和 Worker2 通过锁对象 lockObject 进行同步。Worker1 在执行过程中需要唤醒 Worker2，当 Worker2 执行完毕后，Worker1 再继续执行。通过这种方式，我们可以观察到 Monitor.Wait 方法在等待线程唤醒时的行为。 2. 模型架构图本文给出了一个模型架构图，展示了 Monitor.Wait 方法涉及的关键数据结构和队列。从图中我们可以看到，Monitor.Wait 方法主要涉及到两个队列：一个是等待事件队列 m_Next，另一个是等待线程队列 m_LinkSB。这两个队列共同构成了 Monitor.Wait 方法的底层实现原理。 3. 底层源码验证本文分析了 Monitor.Wait 方法的底层实现，对应着 coreclr 的 ObjectNative::WaitTimeout 方法。通过源码验证，我们可以看到 Monitor.Wait 方法的主要步骤包括：从当前线程的 m_WaitEventLink 队列中寻找 SyncBlock 节点，将当前节点拼接到尾部，以及将新节点通过 EnqueueThread 方法送入到 m_LinkSB 队列。 4. Monitor.Pulse 和 Monitor.PulseAll 实现本文还分析了 Monitor.Pulse 和 Monitor.PulseAll 方法的实现。Monitor.Pulse 方法的作用是在等待事件队列中提取一个节点，并将其 m_EventWait 标记为已唤醒。而 Monitor.PulseAll 方法的作用是将等待事件队列中的所有节点都标记为已唤醒。这两个方法都是通过调用 ThreadQueue::DequeueThread 方法来实现的。通过本文的研究，我们对 Monitor.Wait 方法的底层实现有了更深入的了解，这对于理解和优化多线程程序的性能具有很大的帮助。

正文

一：背景

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) 方法即可。

三：总结

花了半天研究这东西还是挺有意思的，重点还是要理解下那张图，理解了之后我相信你对 Monitor.Pluse 方法注释中所指的 waiting queue 会有一个新的体会。

图片名称