Форум программистов [Powered by Invision Power Board]

Версия для печати темы
Нажмите сюда для просмотра этой темы в оригинальном формате

Форум программистов > C/C++: Общие вопросы > Потокобезопасная очередь

Автор: xbarmaglot 6.5.2016, 10:31

Необходимо очередь, которая удовлетворяет следующим условиям:

1. Потокобезопасная (N писателей и один читатель)
2. Должна быть без блокировок (большие потоки и может начаться гонка за ресурсы)
3. Должна быть с ограничением на количество элементов (с отказом на запись, если она заполнена)
4. Должна быть с ожиданием на чтение (при попытке чтения ожидать заданное количество времени, если она пустая, чтоб не нагружать поток чтения)

Есть ли готовое решение? Может что в boost-e?

P.S. Желательно шаблонная и чтобы элементы хранились по значению, т.к. new/delete при больших потоках - это тормоза

Автор: Alexeis 6.5.2016, 12:32

Цитата(xbarmaglot @ 6.5.2016, 11:31

)

1. Потокобезопасная (N писателей и один читатель)
2. Должна быть без блокировок (большие потоки и может начаться гонка за ресурсы)

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

http://www.boost.org/doc/libs/1_59_0/doc/html/atomic/usage_examples.html
Wait-free ring buffer
A wait-free ring buffer provides a mechanism for relaying objects from one single "producer" thread to one single "consumer" thread without any locks. The operations on this data structure are "wait-free" which means that each operation finishes within a constant number of steps. This makes this data structure suitable for use in hard real-time systems or for communication with interrupt/signal handlers

Автор: Sartorius 6.5.2016, 12:46

Цитата(Alexeis @ 6.5.2016, 13:32

)

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

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

Автор: Alexeis 6.5.2016, 12:46

Там ниже по тексту есть даже пример Wait-free multi-producer queue

Цитата(xbarmaglot @ 6.5.2016, 11:31

)

т.к. new/delete при больших потоках - это тормоза

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

Добавлено через 6 минут и 35 секунд

Цитата(Sartorius @ 6.5.2016, 13:46

)

Даже если только один производитель и один потребитель, блокировки все равно нужны.
Тот самый пример из буста

Код


#include <boost/atomic.hpp>

template<typename T, size_t Size>
class ringbuffer {
public:
  ringbuffer() : head_(0), tail_(0) {}

  bool push(const T & value)
  {
    size_t head = head_.load(boost::memory_order_relaxed);
    size_t next_head = next(head);
    if (next_head == tail_.load(boost::memory_order_acquire))
      return false;
    ring_[head] = value;
    head_.store(next_head, boost::memory_order_release);
    return true;
  }
  bool pop(T & value)
  {
    size_t tail = tail_.load(boost::memory_order_relaxed);
    if (tail == head_.load(boost::memory_order_acquire))
      return false;
    value = ring_[tail];
    tail_.store(next(tail), boost::memory_order_release);
    return true;
  }
private:
  size_t next(size_t current)
  {
    return (current + 1) % Size;
  }
  T ring_[Size];
  boost::atomic<size_t> head_, tail_;
};

Автор: Sartorius 6.5.2016, 13:16

Цитата(Alexeis @ 6.5.2016, 13:46

)

Даже если только один производитель и один потребитель, блокировки все равно нужны.

В общем случае - да. Но при возможности атомарной записи (хотя бы указателей) - нет. Такую атомарность обычно можно обеспечить просто выравниванием данных.

Автор: Alexeis 6.5.2016, 17:23

Одного выравнивания точно не достаточно. Я думал, что потребуется еще как минимум volatile. Ведь при включении оптимизации переменные часто оптимизируются и компилятор думает, что они меняются только из текущего потока. Т.е. фактически потребитель может вообще не видеть наполнение очереди. Я подобное поведение наблюдал на практике и volatile решало вопрос.
Однако, оказалось, что даже volatile недостаточно. Автор утверждает, что это костыль, который работает не всегда.

Цитата(https://habrahabr.ru/post/174369/)

Одновременный доступ к памяти

Если текущее значение в некоторой ячейке памяти равно 100, и один из потоков в данный момент пишет в нее 200, а другой поток читает ее значение (допустим, что кроме этих двух потоков в системе других потоков нет). Какое значение прочитает второй поток? Кажется разумным, что он прочитает либо 100, либо 200. Однако, это не так. В С++ описанный сценарий приводит к неопределенному поведению, и второй поток теоретически может прочитать 42. До C++11 люди использовали volatile для переменных, к которым был возможен одновременный доступ, но на самом деле volatile для этой цели использован никогда быть не должен.

и отсылка к https://software.intel.com/en-us/blogs/2007/11/30/volatile-almost-useless-for-multi-threaded-programming

То что иногда лок-фри работает, я сам проверял, но какие условия, в общем случае, следует соблюдать не ясно.

Автор: Sartorius 6.5.2016, 19:10

Alexeis, ОК. Что б исключить оптимизацию компилятора, можно asm вставками критические моменты реализовать. Но тогда можно и инструкции чтение-изменение-запись с префиксом LOCK использовать smile

. На них фактически и построены системные средства межпоточной синхронизации. Зато тут нет системного вызова, и работать все это должно быстрее.

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

Автор: Alexeis 8.5.2016, 01:26

Цитата(Sartorius @ 6.5.2016, 20:10

)

Но тогда можно и инструкции чтение-изменение-запись с префиксом LOCK использовать

Я почитал еще по этой теме, оказалось, что не все типы процессоров синхронизируют кеш ядер так как это делают 0х86 е. Даже при том, что компилятор не оптимизирует код, на многоядерных системах типа ARM кеш автоматически не сбрасывается.
Если брать указанный пример из буста, то получается, что строчка

Код


ring_[head] = value;

запишет данные в кеш процессора, при этом 2й поток, не сумеет эти данные прочитать(иногда), даже если head_ и tail_ будут помечены volatie. Т.е. поток потребителя увидит, что добавился элемент, при этом нет гарантии, что сам элемент будет ему виден, поскольку volatile гарантирует отсутствие оптимизации только для самой переменной. Строчка, которая идет ниже
head_.store(next_head, boost::memory_order_release);
служит гарантией, что после ее выполнения, записанные выше значения будут синхронизированы с памятью. Но любопытно, что этого также недостаточно для корректной работы, поскольку операции чтения также кешируются. Поэтому потребитель вынужден убеждаться, что то что он прочтет будет памятью, а не кешем. Строчка
tail == head_.load(boost::memory_order_acquire)
стоящая перед чтением элемента массива как раз это делает.

Те же операции делают(гарантируют) критические секции и семафоры при выполнении lock/unlock. К сожалению мне еще не приходилось писать на многоядерных АРМ процессорах, более того пишут, что для того, чтобы избежать указанной проблемы Microsoft компиляторы также вставляют указанные инструкции при записи/чтении в volatile переменные. Поэтому, собственно, мой lock-free код работал корректно.