45

我有一组需要用读/写锁保护的数据结构。我知道 boost::shared_lock，但我想有一个使用 std::mutex、std::condition_variable 和/或 std::atomic 的自定义实现，以便我可以更好地理解它是如何工作的（稍后再调整它） .

每个数据结构（可移动，但不可复制）都将继承自一个名为 Commons 的类，该类封装了锁定。我希望公共界面看起来像这样：

class Commons {
public:
    void read_lock();
    bool try_read_lock();
    void read_unlock();

    void write_lock();
    bool try_write_lock();
    void write_unlock();
};

...这样它就可以被某些人公开继承：

class DataStructure : public Commons {};

我正在编写科学代码，通常可以避免数据竞争；这个锁主要是为了防止我以后可能会犯的错误。因此，我的优先级是低读取开销，因此我不会过多地妨碍正确运行的程序。每个线程可能会在自己的 CPU 内核上运行。

你能告诉我（伪代码可以）一个读者/作者锁吗？我现在拥有的应该是防止作家饥饿的变体。到目前为止，我的主要问题是 read_lock 在检查读取是否安全与实际增加读取器计数之间存在差距，之后 write_lock 知道要等待。

void Commons::write_lock() {
    write_mutex.lock();
    reading_mode.store(false);
    while(readers.load() > 0) {}
}

void Commons::try_read_lock() {
    if(reading_mode.load()) {
        //if another thread calls write_lock here, bad things can happen
        ++readers; 
        return true;
    } else return false;
}

我对多线程有点陌生，我真的很想了解它。在此先感谢您的帮助！

4

4 回答 4

55

这是使用互斥锁和条件变量的简单读/写锁的伪代码。互斥 API 应该是不言自明的。假设条件变量有一个成员wait(Mutex&)（原子地！）删除互斥锁并等待条件发出信号。signal()用唤醒一个服务员或signal_all()唤醒所有服务员的情况发出信号。

read_lock() {
  mutex.lock();
  while (writer)
    unlocked.wait(mutex);
  readers++;
  mutex.unlock();
}

read_unlock() {
  mutex.lock();
  readers--;
  if (readers == 0)
    unlocked.signal_all();
  mutex.unlock();
}

write_lock() {
  mutex.lock();
  while (writer || (readers > 0))
    unlocked.wait(mutex);
  writer = true;
  mutex.unlock();
}

write_unlock() {
  mutex.lock();
  writer = false;
  unlocked.signal_all();
  mutex.unlock();
}

但是，这种实现有很多缺点。

每当锁可用时唤醒所有服务员

如果大多数等待者都在等待写锁，这是浪费 - 毕竟大多数等待者将无法获取锁，并继续等待。简单地使用signal()是行不通的，因为您确实想唤醒所有等待读取锁解锁的人。因此，要解决这个问题，您需要单独的条件变量来提高可读性和可写性。

没有公平。读者饿死作家

您可以通过跟踪挂起的读写锁的数量来解决这个问题，或者一旦出现挂起的写锁就停止获取读锁（尽管你会饿死读者！），或者随机唤醒所有读者或一个写者（假设您使用单独的条件变量，请参阅上面的部分）。

锁没有按照请求的顺序处理

为了保证这一点，您需要一个真正的等待队列。例如，您可以为每个等待者创建一个条件变量，并在释放锁后向队列头部的所有读取器或单个写入器发出信号。

由于互斥锁，即使是纯读取工作负载也会导致争用

这个很难修复。一种方法是使用原子指令来获取读或写锁（通常是比较和交换）。如果由于锁定而导致获取失败，您将不得不退回到互斥锁。但是，正确地做到这一点非常困难。另外，仍然会有争论——原子指令远非免费，尤其是在具有大量内核的机器上。

结论

正确实现同步原语是困难的。实现高效和公平的同步原语更加 困难。它几乎没有回报。linux 上的 pthreads，例如包含一个读/写锁，它使用 futexes 和原子指令的组合，因此它可能优于你在几天的工作中所能想到的任何东西。

于 2012-09-29T22:59:12.950 回答
5

检查这个类：

//
// Multi-reader Single-writer concurrency base class for Win32
//
// (c) 1999-2003 by Glenn Slayden (glenn@glennslayden.com)
//
//


#include "windows.h"

class MultiReaderSingleWriter
{
private:
    CRITICAL_SECTION m_csWrite;
    CRITICAL_SECTION m_csReaderCount;
    long m_cReaders;
    HANDLE m_hevReadersCleared;

public:
    MultiReaderSingleWriter()
    {
        m_cReaders = 0;
        InitializeCriticalSection(&m_csWrite);
        InitializeCriticalSection(&m_csReaderCount);
        m_hevReadersCleared = CreateEvent(NULL,TRUE,TRUE,NULL);
    }

    ~MultiReaderSingleWriter()
    {
        WaitForSingleObject(m_hevReadersCleared,INFINITE);
        CloseHandle(m_hevReadersCleared);
        DeleteCriticalSection(&m_csWrite);
        DeleteCriticalSection(&m_csReaderCount);
    }


    void EnterReader(void)
    {
        EnterCriticalSection(&m_csWrite);
        EnterCriticalSection(&m_csReaderCount);
        if (++m_cReaders == 1)
            ResetEvent(m_hevReadersCleared);
        LeaveCriticalSection(&m_csReaderCount);
        LeaveCriticalSection(&m_csWrite);
    }

    void LeaveReader(void)
    {
        EnterCriticalSection(&m_csReaderCount);
        if (--m_cReaders == 0)
            SetEvent(m_hevReadersCleared);
        LeaveCriticalSection(&m_csReaderCount);
    }

    void EnterWriter(void)
    {
        EnterCriticalSection(&m_csWrite);
        WaitForSingleObject(m_hevReadersCleared,INFINITE);
    }

    void LeaveWriter(void)
    {
        LeaveCriticalSection(&m_csWrite);
    }
};

我没有机会尝试它，但代码看起来不错。

于 2014-05-31T04:52:49.253 回答
1

您可以按照此处的确切 Wikipedia 算法实现 Readers-Writers 锁（我写了它）：

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>

int g_sharedData = 0;
int g_readersWaiting = 0;
std::mutex mu;
bool g_writerWaiting = false;
std::condition_variable cond;

void reader(int i)
{
    std::unique_lock<std::mutex> lg{mu};
    while(g_writerWaiting)
        cond.wait(lg);
    ++g_readersWaiting;
    // reading
    std::cout << "\n reader #" << i << " is reading data = " << g_sharedData << '\n';
    // end reading
    --g_readersWaiting;
    while(g_readersWaiting > 0)
        cond.wait(lg);
    cond.notify_one();
}

void writer(int i)
{
    std::unique_lock<std::mutex> lg{mu};
    while(g_writerWaiting)
        cond.wait(lg);
    // writing
    std::cout << "\n writer #" << i << " is writing\n";
    g_sharedData += i * 10;
    // end writing
    g_writerWaiting = true;
    while(g_readersWaiting > 0)
        cond.wait(lg);
    g_writerWaiting = false;
    cond.notify_all();
}//lg.unlock()


int main()
{
    std::thread reader1{reader, 1};
    std::thread reader2{reader, 2};
    std::thread reader3{reader, 3};
    std::thread reader4{reader, 4};
    std::thread writer1{writer, 1};
    std::thread writer2{writer, 2};
    std::thread writer3{writer, 3};
    std::thread writer4{reader, 4};

    reader1.join();
    reader2.join(); 
    reader3.join();
    reader4.join();
    writer1.join();
    writer2.join();
    writer3.join();
    writer4.join();

    return(0);
}
于 2019-09-19T20:34:21.900 回答
1

我相信这就是您正在寻找的：

class Commons {
    std::mutex write_m_;
    std::atomic<unsigned int> readers_;

public:
    Commons() : readers_(0) {
    }

    void read_lock() {
        write_m_.lock();
        ++readers_;
        write_m_.unlock();
    }

    bool try_read_lock() {
        if (write_m_.try_lock()) {
            ++readers_;
            write_m_.unlock();
            return true;
        }
        return false;
    }

    // Note: unlock without holding a lock is Undefined Behavior!
    void read_unlock() {
        --readers_;
    }

    // Note: This implementation uses a busy wait to make other functions more efficient.
    //       Consider using try_write_lock instead! and note that the number of readers can be accessed using readers()
    void write_lock() {
        while (readers_) {}
        if (!write_m_.try_lock())
            write_lock();
    }

    bool try_write_lock() {
        if (!readers_)
            return write_m_.try_lock();
        return false;
    }

    // Note: unlock without holding a lock is Undefined Behavior!
    void write_unlock() {
        write_m_.unlock(); 
    }

    int readers() { 
        return readers_; 
    }
};

对于 C++17 以来的记录，我们有 std::shared_mutex，请参阅：https ://en.cppreference.com/w/cpp/thread/shared_mutex

于 2022-01-16T22:09:16.157 回答