параллелизм — один производитель, структура данных с одним потребителем и двойной буфер в переполнении стека

Очень интересная проблема! Намного хитрее, чем я думал сначала 🙂
Мне нравятся решения без блокировок, поэтому я постарался найти решение ниже.

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

Сначала обратите внимание, что всегда есть ячейка, в которую производитель может безопасно написать
в любой данный момент времени; если одна ячейка читается потребителем,
другие могут быть написаны. Давайте назовем ячейку, которую можно безопасно записать в
«активная» ячейка (ячейка, из которой потенциально можно прочитать, является любой ячейкой не
активный). Активная ячейка может быть переключена, только если другая ячейка не
в настоящее время читается из.

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

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

Как только продюсер завершит запись в активную ячейку, он может «опубликовать» это значение
изменение какой ячейки является активной (замена индекса), при условии, что потребитель
не в середине чтения другой клетки. Если потребитель является во время
читая другую ячейку, обмен не может произойти, но в этом случае потребитель может поменяться
после его закончил чтение значения, при условии, что производитель не находится в середине
напишите (и если это так, продюсер поменяется, как только это будет сделано).
На самом деле, в общем, потребитель всегда может поменяться местами после того, как прочитает (если это единственный
доступ к системе), потому что ложные обмены потребителем являются доброкачественными: если есть
что-то в другой ячейке, то перестановка приведет к тому, что будет прочитано следующее, и если
нет, обмен не влияет ни на что.

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

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

Кодовое время!

#include <atomic>
#include <cstdint>

template <typename T>
class ProducerConsumerDoubleBuffer {
public:
ProducerConsumerDoubleBuffer() : m_state(0) { }
~ProducerConsumerDoubleBuffer() { }

// Never returns nullptr
T* start_writing() {
// Increment active users; once we do this, no one
// can swap the active cell on us until we're done
auto state = m_state.fetch_add(0x2, std::memory_order_relaxed);
return &m_buf[state & 1];
}

void end_writing() {
// We want to swap the active cell, but only if we were the last
// ones concurrently accessing the data (otherwise the consumer
// will do it for us when *it's* done accessing the data)

auto state = m_state.load(std::memory_order_relaxed);
std::uint32_t flag = (8 << (state & 1)) ^ (state & (8 << (state & 1)));
state = m_state.fetch_add(flag - 0x2, std::memory_order_release) + flag - 0x2;
if ((state & 0x6) == 0) {
// The consumer wasn't in the middle of a read, we should
// swap (unless the consumer has since started a read or
// already swapped or read a value and is about to swap).
// If we swap, we also want to clear the full flag on what
// will become the active cell, otherwise the consumer could
// eventually read two values out of order (it reads a new
// value, then swaps and reads the old value while the
// producer is idle).
m_state.compare_exchange_strong(state, (state ^ 0x1) & ~(0x10 >> (state & 1)), std::memory_order_release);
}
}

// Returns nullptr if there appears to be no more data to read yet
T* start_reading() {
m_readState = m_state.load(std::memory_order_relaxed);
if ((m_readState & (0x10 >> (m_readState & 1))) == 0) {
// Nothing to read here!
return nullptr;
}

// At this point, there is guaranteed to be something to
// read, because the full flag is never turned off by the
// producer thread once it's on; the only thing that could
// happen is that the active cell changes, but that can
// only happen after the producer wrote a value into it,
// in which case there's still a value to read, just in a
// different cell.

m_readState = m_state.fetch_add(0x2, std::memory_order_acquire) + 0x2;

// Now that we've incremented the user count, nobody can swap until
// we decrement it
return &m_buf[(m_readState & 1) ^ 1];
}

void end_reading() {
if ((m_readState & (0x10 >> (m_readState & 1))) == 0) {
// There was nothing to read; shame to repeat this
// check, but if these functions are inlined it might
// not matter. Otherwise the API could be changed.
// Or just don't call this method if start_reading()
// returns nullptr -- then you could also get rid
// of m_readState.
return;
}

// Alright, at this point the active cell cannot change on
// us, but the active cell's flag could change and the user
// count could change. We want to release our user count
// and remove the flag on the value we read.

auto state = m_state.load(std::memory_order_relaxed);
std::uint32_t sub = (0x10 >> (state & 1)) | 0x2;
state = m_state.fetch_sub(sub, std::memory_order_relaxed) - sub;
if ((state & 0x6) == 0 && (state & (0x8 << (state & 1))) == 1) {
// Oi, we were the last ones accessing the data when we released our cell.
// That means we should swap, but only if the producer isn't in the middle
// of producing something, and hasn't already swapped, and hasn't already
// set the flag we just reset (which would mean they swapped an even number
// of times).  Note that we don't bother swapping if there's nothing to read
// in the other cell.
m_state.compare_exchange_strong(state, state ^ 0x1, std::memory_order_relaxed);
}
}

private:
T m_buf[2];

// The bottom (lowest) bit will be the active cell (the one for writing).
// The active cell can only be switched if there's at most one concurrent
// user. The next two bits of state will be the number of concurrent users.
// The fourth bit indicates if there's a value available for reading
// in m_buf[0], and the fifth bit has the same meaning but for m_buf[1].
std::atomic<std::uint32_t> m_state;

std::uint32_t m_readState;
};

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

Если вы решите использовать приведенный выше код, я предлагаю вам сначала написать для него несколько комплексных (многопоточных) модульных тестов.
И правильные ориентиры. Я проверял это, но только с трудом. Дайте мне знать, если вы найдете какие-либо ошибки 🙂

Мой юнит тест:

ProducerConsumerDoubleBuffer<int> buf;
std::thread producer([&]() {
for (int i = 0; i != 500000; ++i) {
int* item = buf.start_writing();
if (item != nullptr) {      // Always true
*item = i;
}
buf.end_writing();
}
});
std::thread consumer([&]() {
int prev = -1;
for (int i = 0; i != 500000; ++i) {
int* item = buf.start_reading();
if (item != nullptr) {
assert(*item > prev);
prev = *item;
}
buf.end_reading();
}
});
producer.join();
consumer.join();

Что касается вашей первоначальной реализации, я только посмотрел на нее (гораздо интереснее
разрабатывать новые вещи, хе), но ответ david.pfx, кажется, касается этой части вашего вопроса.