同步下的资源互斥：停运保护（Run-Down Protection）机制

Posted 2021-01-07 liaoguifa

背景

近期在学习ProcessHacker的源码，Process Hacker是一个免费的、功能强大的“任务管理器”，可用于监听系统资源的使用情况，调试软件以及检测恶意程序。使用中你会发现其可以与Sysinternals开发的Process Explorer相媲美。最重要的它是开源的，源码均可以在Github上查看，这使得我们有机会深入了解其实现原理和窥探一些重要的Windows系统接口。我的计划是结合《深入解析windows操作系统》这本书籍学习一些Windows系统原理的相关知识。

关于停运保护（Run-Down Protection）机制

关于停运保护（暂且这样翻译）的介绍，发哥我翻找了官方的资料，边理解边做了部分的翻译，如有错误或模棱两可之处，还请你高抬贵手帮忙指出：

从WindowsXP开始，内核驱动就支持停运保护机制。驱动通过停运机制可以安全地访问在系统内存中的对象，通常这些对象是由其他内核驱动创建和销毁的。

当对一个对象的所有访问操作已经完成并且不再允许其他新的操作请求，那么就可以将这个对象视为停运的。比如说一个共享对象可能需要被停运，这样的话它就可以被清理然后用新的对象替换它。

拥有共享对象的驱动允许其他驱动对该对象请求并实施停运保护机制。当停运保护生效时，除对象的所有者外，其他驱动可以访问该对象而不用担心在访问结束前该对象会被其所有者删除。在访问开始之前，要访问的驱动会提出对目标对象实施停运保护的请求。对于一个存活周期较长的对象来说，这类请求几乎都是被允许的。当访问结束时，执行访问的驱动会卸除之前对对象实施的停运保护。

常规的停运保护流程

要想共享一个对象，拥有该对象的驱动要调用ExInitializeRundownProtection函数以初始化停运保护机制，在这之后，其他要访问此对象的驱动就可以对其实施和撤销停运保护功能。

要访问共享对象的驱动通过调用ExAcquireRundownProtection函数来请求对该对象的停运保护，当访问结束后，驱动通过调用 ExReleaseRundownProtection 来取消停运保护。

如果对象拥有者打算删除共享对象，它将调用ExWaitForRundownProtectionRelease来等待对象停运。在这期间，驱动调用线程会被阻塞，该函数会一直等待直至在之前被允许的所有停运保护被释放，同时拒绝新的停运保护请求。直到最后一次的访问结束并且所有停运保护被释放后，ExWaitForRundownProtectionRelease方才返回，这时对象的拥有者就可以安全地删除该对象了。为了防止等待阻塞过长时间，访问对象的驱动线程在实施停运保护的过程中应避免出现延缓的情况。

适合使用场景

停运机制很适合用于那些经常有效可用但不知何时会突然被删除或替换的共享对象，访问共享对象数据的驱动或者是调用线程在对象被删除后需确保不再尝试访问该对象，否则这些非法访问可能会造成无法预料的行为后果比如数据损坏，更严重点甚至会出现系统崩溃。

举个例子，典型的病毒防御驱动在操作系统运行时需要长时间加载到内存中。运行期间，其他驱动会发送IO请求到防御驱动以访问驱动中的数据和函数，但有时驱动需要被卸载和更新，为避免驱动还在处理IO请求时过早地被卸载，在发送IO请求之前，一个内核组件如文件系统过滤管理器，可以请求停运保护，当IO请求完成后，停运保护被释放，这时再卸载和更新就安全了。

停运保护不支持串行访问共享对象，如果两个或两个以上的驱动同时对同一对象实施停运保护并且要求必须要串行访问的话，那么一些其他的防护措施比如说互斥锁就需要派上用场了。

相对于锁

停运保护是众多用于保证安全访问共享对象的方式之一，而另外一种方式是使用互斥软件锁。如果一个驱动需要访问一个已被其他驱动上锁的对象，那么前者必须要等待后者释放锁才可以对其进行访问。然而，请求和释放锁会造成性能上的瓶颈，并且会消耗大量的内存。如果使用不正确，锁可能还会对同时进行资源竞争的驱动造成死锁的局面，但为检测和避免死锁，往往也需要耗费大量的计算资源。

原文翻译自：MSDN官方原文链接

实现细则

需要一个结构EX_RUNDOWN_REF用于追踪共享对象停运保护的状态，该结构内容是不透明的（也就是不对外开放的），停运保护机制的相关接口都以指向该结构的指针类型作为传入参数类型，该结构记录当前在共享对象上实施的停运保护的次数。

1. 拥有者调用ExInitializeRundownProtection将共享对象绑定到EX_RUNDOWN_REF结构；
2. 其他要访问的驱动使用EX_RUNDOWN_REF结构值调用 ExAcquireRundownProtection 和ExReleaseRundownProtection 来请求和释放针对该对象的停运保护；
3. 拥有者调用ExWaitForRundownProtectionRelease 来等待对象被释放以此确保对象可以被安全地删除。

代码解析

摘自 phlibincludephbasesup.h 文件

#define PH_RUNDOWN_ACTIVE 0x1
#define PH_RUNDOWN_REF_SHIFT 1
#define PH_RUNDOWN_REF_INC 0x2

typedef struct _PH_RUNDOWN_PROTECT
{  
    /*
    1. 存储PH_RUNDOWN_WAIT_BLOCK类型变量的地址；
    2. 停运保护是否激活的标志位
    */
    ULONG_PTR Value;
} PH_RUNDOWN_PROTECT, *PPH_RUNDOWN_PROTECT;

#define PH_RUNDOWN_PROTECT_INIT { 0 }

typedef struct _PH_RUNDOWN_WAIT_BLOCK
{
    /*共享对象的请求此处，表明共享对象正在被访问*/
    ULONG_PTR Count;
    /*
    事件抛出表明所有对共享对象的访问已结束，
    所有者发起的等待函数将返回，意味着接下来可以对共享对象进行删除或替换
    */
    PH_EVENT WakeEvent;
} PH_RUNDOWN_WAIT_BLOCK, *PPH_RUNDOWN_WAIT_BLOCK;

摘自 phlibsync.c 文件

VOID FASTCALL PhfInitializeRundownProtection(
    _Out_ PPH_RUNDOWN_PROTECT Protection
    )
{
    Protection->Value = 0;
}

BOOLEAN FASTCALL PhfAcquireRundownProtection(
    _Inout_ PPH_RUNDOWN_PROTECT Protection
    )
{
    ULONG_PTR value;

    // Increment the reference count only if rundown has not started.

    while (TRUE)
    {
        value = Protection->Value;

        if (value & PH_RUNDOWN_ACTIVE)
            return FALSE;
        /*原子操作：对比后满足相等条件则进行赋值，函数返回目标参数的原有值*/
        if ((ULONG_PTR)_InterlockedCompareExchangePointer(
            (PVOID *)&Protection->Value,
            /*每次请求对象共享则增加引用计数，每次都加2(PH_RUNDOWN_REF_INC)*/
            (PVOID)(value + PH_RUNDOWN_REF_INC),
            (PVOID)value
            ) == value)
            return TRUE;
    }
}

VOID FASTCALL PhfReleaseRundownProtection(
    _Inout_ PPH_RUNDOWN_PROTECT Protection
    )
{
    ULONG_PTR value;

    while (TRUE)
    {
        value = Protection->Value;
        /*如果停运保护没被激活，value不可能为奇数，PH_RUNDOWN_ACTIVE的值为1*/
        if (value & PH_RUNDOWN_ACTIVE)
        {  /*停运保护已被激活*/
            PPH_RUNDOWN_WAIT_BLOCK waitBlock;

            // Since rundown is active, the reference count has been moved to the waiter's wait
            // block. If we are the last user, we must wake up the waiter.
           /*一旦停运保护激活后，Protection->Value将改变原有的意义，现在存储的是等待块的地址*/
            waitBlock = (PPH_RUNDOWN_WAIT_BLOCK)(value & ~PH_RUNDOWN_ACTIVE);

            if (_InterlockedDecrementPointer(&waitBlock->Count) == 0)
            {
                PhSetEvent(&waitBlock->WakeEvent);
            }

            break;
        }
        else
        {
            // Decrement the reference count normally.

            if ((ULONG_PTR)_InterlockedCompareExchangePointer(
                (PVOID *)&Protection->Value,
                (PVOID)(value - PH_RUNDOWN_REF_INC),
                (PVOID)value
                ) == value)
                break;
        }
    }
}

VOID FASTCALL PhfWaitForRundownProtection(
    _Inout_ PPH_RUNDOWN_PROTECT Protection
    )
{
    ULONG_PTR value;
    ULONG_PTR count;
    PH_RUNDOWN_WAIT_BLOCK waitBlock;
    BOOLEAN waitBlockInitialized;

    // Fast path. If the reference count is 0 or rundown has already been completed, return.
    value = (ULONG_PTR)_InterlockedCompareExchangePointer(
        (PVOID *)&Protection->Value,
        (PVOID)PH_RUNDOWN_ACTIVE,
        (PVOID)0
        );

    if (value == 0 || value == PH_RUNDOWN_ACTIVE)
        return;

    waitBlockInitialized = FALSE;

    while (TRUE)
    {
        value = Protection->Value;
        /*
        向右移一位，有两个作用：
        1. 消除 PH_RUNDOWN_ACTIVE 的影响；
        2. 之前每次请求共享对象时都是加2，现在右移1位相当于除以2，得到的是真正的引用次数！
        */
        count = value >> PH_RUNDOWN_REF_SHIFT;

        // Initialize the wait block if necessary.
        if (count != 0 && !waitBlockInitialized)
        {
            PhInitializeEvent(&waitBlock.WakeEvent);
            waitBlockInitialized = TRUE;
        }

        // Save the existing reference count.
        waitBlock.Count = count;
        /*
           为什么要不厌其烦地使用原子操作？
           因为怕在执行此循环的每一条语句时有请求插入，改变Protection->Value的值
        */
        if ((ULONG_PTR)_InterlockedCompareExchangePointer(
            (PVOID *)&Protection->Value,
            (PVOID)((ULONG_PTR)&waitBlock | PH_RUNDOWN_ACTIVE),
            (PVOID)value
            ) == value)
        {
            /*有共享对象的访问还没结束，要等待，触发事件见 PhfReleaseRundownProtection 函数*/
            if (count != 0) 
                PhWaitForEvent(&waitBlock.WakeEvent, NULL);

            break;
        }
    }
}

总结

看别人的代码就像是在游历一个世界，阅读让批判思维和共情能力显得如此重要。这段代码看得出编码的人是花了心思进行多番重构的，可借鉴的点：

1. 同一变量存储的值的意义切换；
2. 原子操作Interlocked系列函数的使用；
3. 看似简单的奇偶位标识。

通俗的讲，停运保护的机制就比如：一座博物馆，平日敞开大门供游客参观，现在突然说要装修，然后把大门关了，只准出不许入，而博物馆的人不能驱逐里面的游客游客，只能等着，直到所有在里面的游客都出去了，然后才能开始装修。