仅使用关键部分的 Win32 读/写锁答案

【问题标题】：Win32 Read/Write Lock Using Only Critical Sections仅使用关键部分的 Win32 读/写锁
【发布时间】：2010-11-03 18:28:30
【问题描述】：

作为工作项目的一部分，我必须使用 Win32 api 在 C++ 中实现读/写锁。所有现有的解决方案都使用需要在执行期间进行上下文切换的内核对象（信号量和互斥体）。这对我的应用程序来说太慢了。

如果可能的话，我希望只使用关键部分来实现。锁不必是进程安全的，只有线程安全的。关于如何解决这个问题的任何想法？

【问题讨论】：

标签： c++ multithreading winapi critical-section

【解决方案1】：

如果您可以针对 Vista 或更高版本，您应该使用内置的SRWLock's。它们像临界区一样是轻量级的，在没有争用时完全是用户模式。

Joe Duffy 的博客最近有一些entries 介绍了实现不同类型的非阻塞读/写锁。这些锁确实会旋转，因此如果您打算在持有锁的同时做大量工作，它们将不合适。代码是 C#，但应该可以直接移植到本机。

您可以使用临界区和事件来实现读取器/写入器锁 - 您只需保持足够的状态以仅在必要时发出事件信号以避免不必要的内核模式调用。

【讨论】：

【解决方案2】：

我认为如果不使用至少一个内核级对象（Mutex 或 Semaphore）就无法做到这一点，因为您需要内核的帮助才能使调用进程阻塞，直到锁可用为止。

关键部分确实提供了阻塞，但 API 太有限了。例如你不能抓住一个 CS，发现读锁可用但没有写锁，然后等待其他进程完成读取（因为如果其他进程有临界区，它会阻塞其他错误的读取器，如果它那么你的进程不会阻塞，而是旋转，消耗 CPU 周期。）

但是，您可以做的是使用自旋锁并在出现争用时回退到互斥锁。临界区本身就是以这种方式实现的。我会采用现有的临界区实现，并用单独的读取器和写入器计数替换 PID 字段。

【讨论】：

【解决方案3】：

老问题，但这应该可行。它不依赖于争论。如果读者很少或没有争用，他们会产生有限的额外成本，因为 SetEvent 被懒惰地调用（查看编辑历史以获取没有此优化的更重量级版本）。

#include <windows.h>

typedef struct _RW_LOCK {
    CRITICAL_SECTION countsLock;
    CRITICAL_SECTION writerLock;
    HANDLE noReaders;
    int readerCount;
    BOOL waitingWriter;
} RW_LOCK, *PRW_LOCK;

void rwlock_init(PRW_LOCK rwlock)
{
    InitializeCriticalSection(&rwlock->writerLock);
    InitializeCriticalSection(&rwlock->countsLock);

    /*
     * Could use a semaphore as well.  There can only be one waiter ever,
     * so I'm showing an auto-reset event here.
     */
    rwlock->noReaders = CreateEvent (NULL, FALSE, FALSE, NULL);
}

void rwlock_rdlock(PRW_LOCK rwlock)
{
    /*
     * We need to lock the writerLock too, otherwise a writer could
     * do the whole of rwlock_wrlock after the readerCount changed
     * from 0 to 1, but before the event was reset.
     */
    EnterCriticalSection(&rwlock->writerLock);
    EnterCriticalSection(&rwlock->countsLock);
    ++rwlock->readerCount;
    LeaveCriticalSection(&rwlock->countsLock);
    LeaveCriticalSection(&rwlock->writerLock);
}

int rwlock_wrlock(PRW_LOCK rwlock)
{
    EnterCriticalSection(&rwlock->writerLock);
    /*
     * readerCount cannot become non-zero within the writerLock CS,
     * but it can become zero...
     */
    if (rwlock->readerCount > 0) {
        EnterCriticalSection(&rwlock->countsLock);

        /* ... so test it again.  */
        if (rwlock->readerCount > 0) {
            rwlock->waitingWriter = TRUE;
            LeaveCriticalSection(&rwlock->countsLock);
            WaitForSingleObject(rwlock->noReaders, INFINITE);
        } else {
            /* How lucky, no need to wait.  */
            LeaveCriticalSection(&rwlock->countsLock);
        }
    }

    /* writerLock remains locked.  */
}

void rwlock_rdunlock(PRW_LOCK rwlock)
{
    EnterCriticalSection(&rwlock->countsLock);
    assert (rwlock->readerCount > 0);
    if (--rwlock->readerCount == 0) {
        if (rwlock->waitingWriter) {
            /*
             * Clear waitingWriter here to avoid taking countsLock
             * again in wrlock.
             */
            rwlock->waitingWriter = FALSE;
            SetEvent(rwlock->noReaders);
        }
    }
    LeaveCriticalSection(&rwlock->countsLock);
}

void rwlock_wrunlock(PRW_LOCK rwlock)
{
    LeaveCriticalSection(&rwlock->writerLock);
}

您可以通过使用单个CRITICAL_SECTION 来降低读者的成本：

countsLock 在 rdlock 和 rdunlock 中替换为 writerLock
rwlock->waitingWriter = FALSE 在 wrunlock 中被删除

wrlock的身体变成了

EnterCriticalSection(&rwlock->writerLock);
rwlock->waitingWriter = TRUE;
while (rwlock->readerCount > 0) {
    LeaveCriticalSection(&rwlock->writerLock);
    WaitForSingleObject(rwlock->noReaders, INFINITE);
    EnterCriticalSection(&rwlock->writerLock);
}
rwlock->waitingWriter = FALSE;

/* writerLock remains locked.  */

但是这不公平，所以我更喜欢上面的解决方案。

【讨论】：

@KindDragon 我已经包含了一个新版本的算法，它对读者来说更便宜。

【解决方案4】：

看看“Concurrent Programming on Windows”一书，里面有很多不同的读写器锁参考示例。

【讨论】：

【解决方案5】：

查看英特尔Thread Building Blocks 中的spin_rw_mutex ...

spin_rw_mutex 严格位于用户区并使用自旋等待阻塞

【讨论】：

好的链接——文章指出Vista有一个内置的rw锁msdn.microsoft.com/en-us/library/aa904937(VS.85).aspx

【解决方案6】：

这是一个老问题，但也许有人会觉得这很有用。我们开发了一个高性能的open-source RWLock for Windows，如果可用，它会自动使用 Vista+ SRWLock Michael mentioned，否则会退回到用户空间实现。

作为额外的奖励，它有四种不同的“风味”（尽管您可以坚持基本的，这也是最快的），每种都提供更多的同步选项。它从基本的RWLock() 开始，它是不可重入的，仅限于单进程同步，并且不会将读/写锁交换为具有重入支持和读/写删除的成熟的跨进程 IPC RWLock海拔。

如前所述，它们会动态切换到 Vista+ slim 读写锁以获得最佳性能，但您完全不必担心这一点，因为它会退回到 Windows 上完全兼容的实现XP 及其同类产品。

【讨论】：

【解决方案7】：

如果您已经知道仅使用互斥锁的解决方案，您应该能够将其修改为使用临界区。

我们使用两个关键部分和一些计数器滚动我们自己的。它符合我们的需求 - 我们的作者数量非常少，作者优先于读者，等等。我无权发布我们的，但可以说没有互斥锁和信号量是可能的。

【讨论】：

问题是使用标准互斥体，任何人都可以锁定/解锁它。这不适用于关键部分，这使我的解决方案不起作用。
不清楚你的意思 - 只有拥有 mutext 的线程才能解锁它。一旦互斥锁被解锁，任何人都可以锁定它。关键部分也是如此。
对不起，我的意思是信号量。我需要一个解决方案，任何人都可以减少信号量，这在关键部分是不可能的。
根据信号量的用途，您可以使用通过 InterlockedIncrement/InterlockedDecrement 修改的计数器。

【解决方案8】：

这是我能想到的最小解决方案：

http://www.baboonz.org/rwlock.php

并逐字粘贴：

/** A simple Reader/Writer Lock.

This RWL has no events - we rely solely on spinlocks and sleep() to yield control to other threads.
I don't know what the exact penalty is for using sleep vs events, but at least when there is no contention, we are basically
as fast as a critical section. This code is written for Windows, but it should be trivial to find the appropriate
equivalents on another OS.

**/
class TinyReaderWriterLock
{
public:
    volatile uint32 Main;
    static const uint32 WriteDesireBit = 0x80000000;

    void Noop( uint32 tick )
    {
        if ( ((tick + 1) & 0xfff) == 0 )     // Sleep after 4k cycles. Crude, but usually better than spinning indefinitely.
            Sleep(0);
    }

    TinyReaderWriterLock()                 { Main = 0; }
    ~TinyReaderWriterLock()                { ASSERT( Main == 0 ); }

    void EnterRead()
    {
        for ( uint32 tick = 0 ;; tick++ )
        {
            uint32 oldVal = Main;
            if ( (oldVal & WriteDesireBit) == 0 )
            {
                if ( InterlockedCompareExchange( (LONG*) &Main, oldVal + 1, oldVal ) == oldVal )
                    break;
            }
            Noop(tick);
        }
    }

    void EnterWrite()
    {
        for ( uint32 tick = 0 ;; tick++ )
        {
            if ( (tick & 0xfff) == 0 )                                     // Set the write-desire bit every 4k cycles (including cycle 0).
                _InterlockedOr( (LONG*) &Main, WriteDesireBit );

            uint32 oldVal = Main;
            if ( oldVal == WriteDesireBit )
            {
                if ( InterlockedCompareExchange( (LONG*) &Main, -1, WriteDesireBit ) == WriteDesireBit )
                    break;
            }
            Noop(tick);
        }
    }

    void LeaveRead()
    {
        ASSERT( Main != -1 );
        InterlockedDecrement( (LONG*) &Main );
    }
    void LeaveWrite()
    {
        ASSERT( Main == -1 );
        InterlockedIncrement( (LONG*) &Main );
    }
};

【讨论】：

【解决方案9】：

我只使用关键部分编写了以下代码。

class ReadWriteLock {
    volatile LONG writelockcount;
    volatile LONG readlockcount;
    CRITICAL_SECTION cs;
public:
    ReadWriteLock() {
        InitializeCriticalSection(&cs);
        writelockcount = 0;
        readlockcount = 0;
    }
    ~ReadWriteLock() {
        DeleteCriticalSection(&cs);
    }
    void AcquireReaderLock() {        
    retry:
        while (writelockcount) {
            Sleep(0);
        }
        EnterCriticalSection(&cs);
        if (!writelockcount) {
            readlockcount++;
        }
        else {
            LeaveCriticalSection(&cs);
            goto retry;
        }
        LeaveCriticalSection(&cs);
    }
    void ReleaseReaderLock() {
        EnterCriticalSection(&cs);
        readlockcount--;
        LeaveCriticalSection(&cs);
    }
    void AcquireWriterLock() {
        retry:
        while (writelockcount||readlockcount) {
            Sleep(0);
        }
        EnterCriticalSection(&cs);
        if (!writelockcount&&!readlockcount) {
            writelockcount++;
        }
        else {
            LeaveCriticalSection(&cs);
            goto retry;
        }
        LeaveCriticalSection(&cs);
    }
    void ReleaseWriterLock() {
        EnterCriticalSection(&cs);
        writelockcount--;
        LeaveCriticalSection(&cs);
    }
};

要执行旋转等待，请使用 Sleep(0) 注释这些行。

【讨论】：

【解决方案10】：

在这里查看我的实现：

https://github.com/coolsoftware/LockLib

VRWLock 是一个 C++ 类，它实现了单写 - 多读逻辑。

看看测试项目TestLock.sln。

UPD。下面是读写器的简单代码：

LONG gCounter = 0;

// reader

for (;;) //loop
{
  LONG n = InterlockedIncrement(&gCounter); 
  // n = value of gCounter after increment
  if (n <= MAX_READERS) break; // writer does not write anything - we can read
  InterlockedDecrement(&gCounter);
}
// read data here
InterlockedDecrement(&gCounter); // release reader

// writer

for (;;) //loop
{
  LONG n = InterlockedCompareExchange(&gCounter, (MAX_READERS+1), 0); 
  // n = value of gCounter before attempt to replace it by MAX_READERS+1 in InterlockedCompareExchange
  // if gCounter was 0 - no readers/writers and in gCounter will be MAX_READERS+1
  // if gCounter was not 0 - gCounter stays unchanged
  if (n == 0) break;
}
// write data here
InterlockedExchangeAdd(&gCounter, -(MAX_READERS+1)); // release writer

VRWLock 类支持自旋计数和线程特定的引用计数，允许释放终止线程的锁。

【讨论】：

请不要随便扔链接。包含相关代码并解释它是如何回答问题的。