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

以下は、mutex と条件変数を使用した単純なリーダー/ライター ロックの擬似コードです。ミューテックス API は一目瞭然です。条件変数には、wait(Mutex&)(アトミックに!) ミューテックスを削除し、条件が通知されるのを待つメンバーがあると想定されます。条件は、 1 つのウェイター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()では機能しません。したがって、これを修正するには、読みやすさと書き込みやすさのために別々の条件変数が必要です。

公平性がない。読者は作家を飢えさせる

保留中の読み取りロックと書き込みロックの数を追跡し、保留中の書き込みロックがあると読み取りロックの取得を停止するか (ただし、リーダーを飢えさせます!)、すべてのリーダーまたは 1 つのライターのいずれかをランダムにウェイクアップすることで修正できます (別の条件変数を使用します。上記のセクションを参照してください)。

ロックは要求された順序で処理されません

これを保証するには、実際の待機キューが必要です。たとえば、各ウェイターに 1 つの条件変数を作成し、ロックを解放した後、両方ともキューの先頭にあるすべてのリーダーまたは単一のライターにシグナルを送ることができます。

純粋な読み取りワークロードでもミューテックスが原因で競合が発生する

これは修正するのが難しいです。1 つの方法は、アトミック命令を使用して読み取りロックまたは書き込みロックを取得することです (通常は比較交換)。ロックが取得されたために取得が失敗した場合は、ミューテックスにフォールバックする必要があります。ただし、それを正しく行うのは非常に困難です。さらに、依然として競合が発生します。特に多数のコアを搭載したマシンでは、アトミック命令は無料ではありません。

結論

同期プリミティブを正しく実装するのは困難です。効率的で公正な同期プリミティブを実装するのはさらに 困難です。そして、それが報われることはめったにありません。Linux 上の pthreads には、たとえば、futex とアトミック命令の組み合わせを使用するリーダー/ライター ロックが含まれているため、おそらく数日間の作業で思いつくものよりも優れています。

于 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

ここからウィキペディアの正確なアルゴリズムに従って、リーダー/ライター ロックを実装できます(私が書きました)。

#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 に答える