std::shared_mutex::unlock_shared() 即使 Windows 上没有活动的独占锁也会阻塞

Question

我的团队遇到了死锁，我怀疑这是 SRW 锁的 Windows 实现中的错误。下面的代码是真实代码的精炼版本。总结如下：

1. 主线程获取排它锁
2. 主线程创建N个子线程
3. 每个子线程
   1. 获取共享锁
   2. 旋转直到所有子进程都获得共享锁
   3. 释放共享锁
4. 主线程释放排它锁

是的，这可以通过 C++20 中的 std::latch 来完成。这不是重点。

此代码在大多数时间都有效。然而，大约五千分之一的循环会陷入死锁。当它死锁时，正好有 1 个子进程成功获取共享锁，并且 N-1 个子进程被困在

lock_shared()

中。在 Windows 上，此函数调用

RtlAcquireSRWLockShared

并阻止

NtWaitForAlertByThreadId

。

直接使用

std::shared_mutex

、

std::shared_lock

/

std::unique_lock

，或直接调用

SRW

函数时，可以观察到该行为。

2017 Raymond Chen 的帖子询问了这一确切行为，但归咎于用户错误。

对我来说这看起来像是一个 SRW bug。也许值得注意的是，如果一个孩子不尝试衔乳并喊“

unlock_shared

”，这将唤醒其受阻的兄弟姐妹。 

std::shared_lock

 或 

*SRW*

 的文档中没有任何内容表明即使没有活动的独占锁也允许阻塞。

在非 Windows 平台上尚未观察到此死锁。

示例代码：

#include <atomic>
#include <cstdint>
#include <iostream>
#include <memory>
#include <shared_mutex>
#include <thread>
#include <vector>

struct ThreadTestData {
    int32_t numThreads = 0;
    std::shared_mutex sharedMutex = {};
    std::atomic<int32_t> readCounter;
};

int DoStuff(ThreadTestData* data) {
    // Acquire reader lock
    data->sharedMutex.lock_shared();

    // wait until all read threads have acquired their shared lock
    data->readCounter.fetch_add(1);
    while (data->readCounter.load() != data->numThreads) {
        std::this_thread::yield();
    }

    // Release reader lock
    data->sharedMutex.unlock_shared();

    return 0;
}

int main() {
    int count = 0;
    while (true) {
        ThreadTestData data = {};
        data.numThreads = 5;

        // Acquire write lock
        data.sharedMutex.lock();

        // Create N threads
        std::vector<std::unique_ptr<std::thread>> readerThreads;
        readerThreads.reserve(data.numThreads);
        for (int i = 0; i < data.numThreads; ++i) {
            readerThreads.emplace_back(std::make_unique<std::thread>(DoStuff, &data));
        }

        // Release write lock
        data.sharedMutex.unlock();

        // Wait for all readers to succeed
        for (auto& thread : readerThreads) {
            thread->join();
        }

        // Cleanup
        readerThreads.clear();

        // Spew so we can tell when it's deadlocked
        count += 1;
        std::cout << count << std::endl;
    }

    return 0;
}

这是并行堆栈的图片。您可以看到主线程正确地阻塞在

thread::join

 上。一个读取器线程获取了锁并处于yield 循环中。四个读取器线程在 

lock_shared

 内被阻塞。

Answer 1

真正的错误在于您的代码逻辑中。您正在 SRW 锁内

wait。但锁不是为等待而设计的，它是为快速操作而设计的。线程进入锁后必须尽可能快地离开锁。

您的具体代码会发生什么以及为什么会发生？

当线程（
1. A）从独占访问中释放锁定时（ ReleaseSRWLockExclusive

   ）

并且“同时”另一个线程（
2. B）尝试获取共享锁（ AcquireSRWLockShared

   ）

另一个线程（至少还有一个）已经在等待锁了

当线程

B实际上获取独占访问锁时的可能情况，尽管他只要求共享。结果另一个共享等待者和新的共享请求将等待，直到线程 B 不离开 srw 锁。但在你的情况下它不会离开，直到另一个线程不进入部分，但它们无法进入，直到B离开..死锁。

但是，如果您将 srw 留给线程

B，则另一个线程会唤醒。

如何修改自我代码来测试这个？

入场前节省时间

while (data->readCounter.load() != data->numThreads)

循环。并检查循环时间。实际上，在 SRW 锁从独占访问中释放后，所有共享等待者必须“非常快”地进入锁定。如果在一段时间内这种情况没有发生（假设为 1 秒 - 在这种情况下确实是很长的时间） - 让现在处于 SRW 中的线程 - 退出循环并释放锁。僵局就会消失。

尝试下一个代码

int DoStuff(ThreadTestData* data) {
    // Acquire reader lock
    data->sharedMutex.lock_shared();

    ULONG64 time = GetTickCount64() + 1000;
    // wait until all read threads have acquired their shared lock
    // but no more 1000 ms !!
    data->readCounter.fetch_add(1);
    while (data->readCounter.load() != data->numThreads && GetTickCount64() < time) {
        std::this_thread::yield();
    }

    // Release reader lock
    data->sharedMutex.unlock_shared();

    return 0;
}

但是我更喜欢纯winapi并且为了更好的视觉效果，下一个代码：

struct ThreadTestData 
{
    HANDLE hEvent;
    SRWLOCK SRWLock = {};
    LONG numThreads = 1;
    LONG readCounter = 0;
    LONG done = 0;

    void EndThread()
    {
        if (!InterlockedDecrementNoFence(&numThreads))
        {
            if (!SetEvent(hEvent)) __debugbreak();
        }
    }

    void DoStuff()
    {
        AcquireSRWLockShared(&SRWLock);

        InterlockedDecrementNoFence(&readCounter);

        ULONG64 time = GetTickCount64() + 1000;

        while (readCounter)
        {
            if (GetTickCount64() > time)
            {
                if (InterlockedExchangeNoFence(&done, TRUE))
                {
                    MessageBoxW(0, 0, 0, MB_ICONHAND);
                    break;
                }
            }

            SwitchToThread();
        }

        ReleaseSRWLockShared(&SRWLock);

        EndThread();
    }

    static ULONG WINAPI _S_DoStuff(PVOID data)
    {
        reinterpret_cast<ThreadTestData*>(data)->DoStuff();
        return 0;
    }

    BOOL Test(ULONG n)
    {
        if (hEvent = CreateEventW(0, 0, 0, 0))
        {
            AcquireSRWLockExclusive(&SRWLock);

            do 
            {
                numThreads++;
                readCounter++;

                if (HANDLE hThread = CreateThread(0, 0, _S_DoStuff, this, 0, 0))
                {
                    CloseHandle(hThread);
                }
                else
                {
                    readCounter--;
                    numThreads--;
                }

            } while (--n);

            ReleaseSRWLockExclusive(&SRWLock);

            EndThread();

            if (WAIT_OBJECT_0 != WaitForSingleObject(hEvent, INFINITE))
            {
                __debugbreak();
            }

            CloseHandle(hEvent);
        }

        return done;
    }
};

BOOL DoSrwTest(ULONG nThreads)
{
    ThreadTestData data;
    return data.Test(nThreads);
}

ULONG DoSrwTest(ULONG nLoops, ULONG nThreads)
{
    while (!DoSrwTest(nThreads) && --nLoops);

    return nLoops;
}

完整的项目是

这里