Похоже, у вас есть только мьютекс и condition_variable в качестве примитивов синхронизации. поэтому я пишу здесь блокировку чтения-записи, которая голодает читателей. он использует один мьютекс, два условных_переменных и три целых числа.

readers - readers in the cv readerQ plus the reading reader
writers - writers in cv writerQ plus the writing writer
active_writers - the writer currently writing. can only be 1 or 0.

Так голодают читатели. Если есть несколько писателей, желающих писать, у читателей никогда не будет возможности прочитать, пока все писатели не закончат писать. Это потому, что более поздние читатели должны проверить переменную writers. В то же время переменная active_writers будет гарантировать, что одновременно может писать только один писатель.

class RWLock {
public:
    RWLock()
    : shared()
    , readerQ(), writerQ()
    , active_readers(0), waiting_writers(0), active_writers(0)
    {}

    void ReadLock() {
        std::unique_lock<std::mutex> lk(shared);
        while( waiting_writers != 0 )
            readerQ.wait(lk);
        ++active_readers;
        lk.unlock();
    }

    void ReadUnlock() {
        std::unique_lock<std::mutex> lk(shared);
        --active_readers;
        lk.unlock();
        writerQ.notify_one();
    }

    void WriteLock() {
        std::unique_lock<std::mutex> lk(shared);
        ++waiting_writers;
        while( active_readers != 0 || active_writers != 0 )
            writerQ.wait(lk);
        ++active_writers;
        lk.unlock();
    }

    void WriteUnlock() {
        std::unique_lock<std::mutex> lk(shared);
        --waiting_writers;
        --active_writers;
        if(waiting_writers > 0)
            writerQ.notify_one();
        else
            readerQ.notify_all();
        lk.unlock();
    }

private:
    std::mutex              shared;
    std::condition_variable readerQ;
    std::condition_variable writerQ;
    int                     active_readers;
    int                     waiting_writers;
    int                     active_writers;
};

На первый взгляд мне показалось, что я узнал этот ответ как тот же алгоритм, который представил Александр Терехов. Но изучив его, я считаю, что он ошибочен. Два записывающих устройства могут одновременно ожидать m_exclusive_cond. Когда один из этих модулей записи просыпается и получает эксклюзивную блокировку, он устанавливает exclusive_waiting_blocked = false на unlock, тем самым переводя мьютекс в несогласованное состояние. После этого мьютекс, скорее всего, будет закрыт.

N2406, который первым предложенный std::shared_mutex содержит частичную реализацию, которая повторяется ниже с обновленным синтаксисом.

class shared_mutex
{
    mutex    mut_;
    condition_variable gate1_;
    condition_variable gate2_;
    unsigned state_;

    static const unsigned write_entered_ = 1U << (sizeof(unsigned)*CHAR_BIT - 1);
    static const unsigned n_readers_ = ~write_entered_;

public:

    shared_mutex() : state_(0) {}

// Exclusive ownership

    void lock();
    bool try_lock();
    void unlock();

// Shared ownership

    void lock_shared();
    bool try_lock_shared();
    void unlock_shared();
};

// Exclusive ownership

void
shared_mutex::lock()
{
    unique_lock<mutex> lk(mut_);
    while (state_ & write_entered_)
        gate1_.wait(lk);
    state_ |= write_entered_;
    while (state_ & n_readers_)
        gate2_.wait(lk);
}

bool
shared_mutex::try_lock()
{
    unique_lock<mutex> lk(mut_, try_to_lock);
    if (lk.owns_lock() && state_ == 0)
    {
        state_ = write_entered_;
        return true;
    }
    return false;
}

void
shared_mutex::unlock()
{
    {
    lock_guard<mutex> _(mut_);
    state_ = 0;
    }
    gate1_.notify_all();
}

// Shared ownership

void
shared_mutex::lock_shared()
{
    unique_lock<mutex> lk(mut_);
    while ((state_ & write_entered_) || (state_ & n_readers_) == n_readers_)
        gate1_.wait(lk);
    unsigned num_readers = (state_ & n_readers_) + 1;
    state_ &= ~n_readers_;
    state_ |= num_readers;
}

bool
shared_mutex::try_lock_shared()
{
    unique_lock<mutex> lk(mut_, try_to_lock);
    unsigned num_readers = state_ & n_readers_;
    if (lk.owns_lock() && !(state_ & write_entered_) && num_readers != n_readers_)
    {
        ++num_readers;
        state_ &= ~n_readers_;
        state_ |= num_readers;
        return true;
    }
    return false;
}

void
shared_mutex::unlock_shared()
{
    lock_guard<mutex> _(mut_);
    unsigned num_readers = (state_ & n_readers_) - 1;
    state_ &= ~n_readers_;
    state_ |= num_readers;
    if (state_ & write_entered_)
    {
        if (num_readers == 0)
            gate2_.notify_one();
    }
    else
    {
        if (num_readers == n_readers_ - 1)
            gate1_.notify_one();
    }
}

Алгоритм взят из старой записи группы новостей Александра Терехова. Она не уморит голодом ни читателей, ни писателей.

Есть двое «ворот», gate1_ и gate2_. Читатели и писатели должны пройти gate1_ и могут быть заблокированы при попытке сделать это. Как только читатель проходит gate1_, он блокирует чтение мьютекса. Читатели могут пройти дальше gate1_, пока не будет достигнуто максимальное количество читателей с правом собственности, и пока писатель не превысит gate1_.

Только один писатель за раз может пройти дальше gate1_. И писатель может преодолеть gate1_, даже если читатели владеют им. Но после gate1_ писатель по-прежнему не имеет права собственности. Сначала он должен пройти gate2_. Писатель не может пройти дальше gate2_, пока все читатели, владеющие им, не откажутся от него. Напомним, что новые читатели не могут пройти дальше gate1_, пока писатель ждет в gate2_. И новый писатель не может пройти дальше gate1_, пока писатель ждет в gate2_.

То, что и читатели, и писатели блокируются в gate1_ с (почти) одинаковыми требованиями, предъявляемыми для его преодоления, делает этот алгоритм справедливым как для читателей, так и для писателей, не голодая ни для одного из них.

«Состояние» мьютекса намеренно сохранено в одном слове, чтобы предположить, что частичное использование атомарности (в качестве оптимизации) для определенных изменений состояния возможно (т. Е. Для неоспариваемого «быстрого пути»). Однако эта оптимизация здесь не демонстрируется. Например, если поток записи может атомарно изменить state_ с 0 на write_entered, тогда он получит блокировку без необходимости блокировать или даже блокировать/разблокировать mut_. И unlock() можно реализовать с атомарным хранилищем. И т. д. Эти оптимизации не показаны здесь, потому что их гораздо сложнее правильно реализовать, чем это кажется простым описанием.

Параллельное чтение данных, защищенных мьютексом, довольно распространено, а запись — редко.

Это звучит как идеальный сценарий для RCU пользовательского пространства:

URCU аналогичен своему аналогу ядра Linux, обеспечивая замену блокировки чтения-записи, среди прочего. Это сходство продолжается и в том, что считыватели не синхронизируются напрямую с программами обновления RCU, что делает пути кода на стороне чтения RCU чрезвычайно быстрыми, а также позволяет программам чтения RCU продвигаться вперед, даже если они работают одновременно с программами обновления RCU, и наоборот.

Есть несколько хороших трюков, которые вы можете использовать, чтобы помочь.

Во-первых, хорошая производительность. VxWorks отличается очень хорошим временем переключения контекста. Какое бы решение блокировки вы ни использовали, оно, скорее всего, будет включать семафоры. Я бы не боялся использовать для этого семафоры (множественное число), они довольно хорошо оптимизированы в VxWorks, а быстрое время переключения контекста помогает минимизировать снижение производительности из-за оценки многих состояний семафора и т. д.

Также я бы забыл об использовании семафоров POSIX, которые просто будут располагаться поверх собственных семафоров VxWork. VxWorks предоставляет собственные счетные, двоичные и мьютексные семафоры; использование того, который подходит, делает все немного быстрее. Бинарные иногда могут быть весьма полезными; отправлено много раз, никогда не превышайте значение 1.

Во-вторых, запись важнее чтения. Когда у меня были такие требования в VxWorks и я использовал семафор(ы) для управления доступом, я использовал приоритет задачи, чтобы указать, какая задача важнее и должна получить доступ к ресурсу первой. Это работает довольно хорошо; буквально все в VxWorks это задача (ну и поток) как и любая другая, включая все драйвера устройств и т.д.

VxWorks также разрешает инверсии приоритетов (такие вещи, которые ненавидит Линус Торвальдс). Таким образом, если вы реализуете блокировку с помощью семафора(ов), вы можете положиться на планировщик ОС, который будет блокировать операции чтения с более низким приоритетом, если они блокируют запись с более высоким приоритетом. Это может привести к гораздо более простому коду, и вы также получите максимальную отдачу от ОС.

Таким образом, потенциальное решение состоит в том, чтобы иметь один счетный семафор VxWorks, защищающий ресурс, инициализированный значением, равным количеству читателей. Каждый раз, когда читатель хочет прочитать, он берет семафор (уменьшая счетчик на 1. Каждый раз, когда выполняется чтение, он отправляет семафор, увеличивая счетчик на 1. Каждый раз, когда пишущий хочет записать, он берет семафор n (n = количество читателей) раз и публикует его n раз, когда закончите. Наконец, сделайте задачу записи более приоритетной, чем любой из читателей, и полагайтесь на быстрое время переключения контекста ОС и инверсию приоритета.

Помните, что вы программируете на ОС жесткого реального времени, а не на Linux. Получение/публикация собственного семафора VxWorks не требует такого же количества времени выполнения, как аналогичное действие в Linux, хотя даже Linux в наши дни довольно хорош (я использую PREEMPT_RT в настоящее время). На поведение планировщика VxWorks и всех драйверов устройств можно положиться. Вы даже можете сделать свою задачу записи наивысшим приоритетом во всей системе, если хотите, даже выше, чем у всех драйверов устройств!

Чтобы помочь делу, также подумайте, что делает каждый из ваших потоков. VxWorks позволяет указать, что задача использует или не использует FPU. Если вы используете собственные подпрограммы VxWorks TaskSpawn вместо pthread_create, у вас есть возможность указать это. Это означает, что если ваш поток/задача не выполняет математических вычислений с плавающей запятой, и вы сказали об этом в своем вызове TaskSpawn, время переключения контекста будет еще быстрее, потому что планировщик не будет заботиться о сохранении / восстановить состояние FPU.

Это имеет разумные шансы стать лучшим решением для платформы, на которой вы разрабатываете. Он использует сильные стороны ОС (быстрые семафоры, быстрое время переключения контекста) без добавления дополнительного кода для воссоздания альтернативного (и, возможно, более элегантного) решения, обычно используемого на других платформах.

В-третьих, застрял со старым GCC и старым Boost. По сути, я не могу помочь, кроме малоценных предложений о том, чтобы позвонить в WindRiver и обсудить покупку обновления. Лично говоря, когда я программировал для VxWorks, я использовал собственный API VxWork, а не POSIX. Итак, код не очень переносимый, но в итоге он стал быстрым; В любом случае POSIX — это просто слой поверх собственного API, и это всегда будет замедлять работу.

Тем не менее, семафоры подсчета и мьютекса POSIX очень похожи на собственные семафоры подсчета и мьютекса VxWork. Вероятно, это означает, что слои POSIX не очень толстые.

Общие примечания о программировании для VxWorks

Отладка Я всегда старался использовать инструменты разработки (Tornado), доступные для Solaris. Это, безусловно, лучшая многопоточная среда отладки, с которой я когда-либо сталкивался. Почему? Это позволяет запускать несколько сеансов отладки, по одному для каждого потока/задачи в системе. В итоге вы получаете окно отладки для каждого потока, и вы индивидуально и независимо отлаживаете каждый из них. Перешагнув операцию блокировки, это окно отладки будет заблокировано. Переместите фокус мыши на другое окно отладки, пройдитесь по операции, освобождающей блок, и наблюдайте, как первое окно завершает свой шаг.

В итоге вы получаете множество окон отладки, но это, безусловно, лучший способ отладки многопоточных приложений. Это упростило написание очень сложных вещей и возможность видеть проблемы. Вы можете легко исследовать различные динамические взаимодействия в своем приложении, потому что у вас есть простой и мощный контроль над тем, что делает каждый поток в любое время.

По иронии судьбы версия Tornado для Windows не позволяла вам этого сделать; одно жалкое одиночное окно отладки на систему, как и любая другая скучная старая IDE, такая как Visual Studio и т. д. Я никогда не видел, чтобы даже современные IDE приближались к тому, чтобы быть такими же хорошими, как Tornado на Solaris для многопоточной отладки.

Жесткие диски Если ваши программы чтения и записи используют файлы на диске, учтите, что VxWorks 5.5 устарел. Такие вещи, как NCQ, не будут поддерживаться. В этом случае мое предложенное решение (изложенное выше) может быть лучше реализовано с одним семафором мьютекса, чтобы предотвратить спотыкание нескольких читателей друг о друга в их борьбе за чтение разных частей диска. Это зависит от того, что именно делают ваши читатели, но если они читают непрерывные данные из файла, это позволит избежать тряски головки чтения/записи туда и обратно по поверхности диска (очень медленно).

В моем случае я использовал этот трюк для формирования трафика через сетевой интерфейс; каждая задача отправляла разные типы данных, а приоритет задачи отражал приоритет данных в сети. Это было очень элегантно, ни одно сообщение никогда не фрагментировалось, но важные сообщения получали львиную долю доступной пропускной способности.

Как всегда, лучшее решение будет зависеть от деталей. Возможно, вам нужна спин-блокировка для чтения и записи, но другие подходы, такие как копирование-обновление, как было предложено выше, может быть решением, хотя на старой встроенной платформе использование дополнительной памяти может быть проблемой. С редкими операциями записи я часто организую работу, используя систему задач, так что запись может происходить только тогда, когда нет операций чтения из этой структуры данных, но это зависит от алгоритма.

Один из алгоритмов для этого, основанный на семафорах и мьютексах, описан в Concurrent Control with Читатели и писатели; П.Дж. Куртуа, Ф. Хейманс и Д.Л. Парнас; Исследовательская лаборатория MBLE; Брюссель, Бельгия.

Это упрощенный ответ, основанный на моих заголовках Boost (я бы назвал Boost одобренным способом). Для этого требуются только переменные условия и мьютексы. Я переписал его, используя примитивы Windows, потому что считаю их описательными и очень простыми, но рассматриваю это как псевдокод.

Это очень простое решение, которое не поддерживает такие вещи, как обновление мьютекса или операции try_lock(). Я могу добавить их, если хотите. Я также убрал некоторые излишества, такие как отключение прерываний, которые не являются строго необходимыми.

Кроме того, стоит проверить boost\thread\pthread\shared_mutex.hpp (это основано на этом). Это удобочитаемо для человека.

class SharedMutex {
  CRITICAL_SECTION m_state_mutex;
  CONDITION_VARIABLE m_shared_cond;
  CONDITION_VARIABLE m_exclusive_cond;

  size_t shared_count;
  bool exclusive;

  // This causes write blocks to prevent further read blocks
  bool exclusive_wait_blocked;

  SharedMutex() : shared_count(0), exclusive(false)
  {
    InitializeConditionVariable (m_shared_cond);
    InitializeConditionVariable (m_exclusive_cond);
    InitializeCriticalSection (m_state_mutex);
  }

  ~SharedMutex() 
  {
    DeleteCriticalSection (&m_state_mutex);
    DeleteConditionVariable (&m_exclusive_cond);
    DeleteConditionVariable (&m_shared_cond);
  }

  // Write lock
  void lock(void)
  {
    EnterCriticalSection (&m_state_mutex);
    while (shared_count > 0 || exclusive)
    {
      exclusive_waiting_blocked = true;
      SleepConditionVariableCS (&m_exclusive_cond, &m_state_mutex, INFINITE)
    }
    // This thread now 'owns' the mutex
    exclusive = true;

    LeaveCriticalSection (&m_state_mutex);
  }

  void unlock(void)
  {
    EnterCriticalSection (&m_state_mutex);
    exclusive = false;
    exclusive_waiting_blocked = false;
    LeaveCriticalSection (&m_state_mutex);
    WakeConditionVariable (&m_exclusive_cond);
    WakeAllConditionVariable (&m_shared_cond);
  }

  // Read lock
  void lock_shared(void)
  {
    EnterCriticalSection (&m_state_mutex);
    while (exclusive || exclusive_waiting_blocked)
    {
      SleepConditionVariableCS (&m_shared_cond, m_state_mutex, INFINITE);
    }
    ++shared_count;
    LeaveCriticalSection (&m_state_mutex);
  }

  void unlock_shared(void)
  {
    EnterCriticalSection (&m_state_mutex);
    --shared_count;

    if (shared_count == 0)
    {
      exclusive_waiting_blocked = false;
      LeaveCriticalSection (&m_state_mutex);
      WakeConditionVariable (&m_exclusive_cond);
      WakeAllConditionVariable (&m_shared_cond);
    }
    else
    {
      LeaveCriticalSection (&m_state_mutex);
    }
  }
};

Поведение

Хорошо, есть некоторая путаница в поведении этого алгоритма, так что вот как он работает.

Во время блокировки записи: блоки чтения и записи блокируются.

В конце блокировки записи — потоки чтения и один поток записи будут соревноваться, чтобы определить, какой из них начнется.

Во время блокировки чтения — модули записи заблокированы. Читатели также блокируются, если и только если блокируется писатель.

После снятия окончательной блокировки чтения — потоки чтения и один поток записи будут соревноваться, чтобы определить, какой из них начнется.

Это может привести к тому, что читатели будут голодать при записи, если контекст процессора часто переключается на m_shared_cond поток перед m_exclusive_cond во время уведомления, но я подозреваю, что эта проблема является теоретической, а не практической, поскольку это алгоритм Boost.

Теперь, когда Microsoft открыла исходный код .NET, вы можете посмотреть их ReaderWRiterLockSlim.

Я не уверен, что более простые примитивы, которые они используют, доступны вам, некоторые из них также являются частью библиотеки .NET, и их код также доступен.

Microsoft потратила довольно много времени на повышение производительности своих механизмов блокировки, так что это может стать хорошей отправной точкой.