Estructura de datos de un solo productor y un solo consumidor con doble búfer en C++

Problema muy interesante! Mucho más complicado de lo que pensé :-) Me gustan las soluciones sin bloqueo, así que he tratado de elaborar una a continuación.

Hay muchas maneras de pensar acerca de este sistema. Puedes modelar como un buffer/cola circular de tamaño fijo (con dos entradas), pero luego se pierde la capacidad de actualizar el siguiente valor disponible para el consumo, ya que no sabes si el consumidor ha comenzado a leer el más reciente publicado valor o está todavía (potencialmente) leyendo el anterior. Tan estado extra es necesario más allá de la de un búfer de anillo estándar con el fin de alcanzar una más óptima solución.

Primero tenga en cuenta que siempre hay una celda en la que el productor puede escribir de forma segura en cualquier momento dado; si una celda está siendo leída por el consumidor, el se puede escribir a otros. Vamos a llamar a la celda que se puede escribir con seguridad a la celda" activa " (la celda de la que se puede leer potencialmente es cualquier celda que no sea el activo). La celda activa solo se puede cambiar si el otra célula no es actualmente se está leyendo de.

A diferencia de la celda activa, en la que siempre se puede escribir, la celda no activa puede solo se puede leer de si contiene un valor; una vez que ese valor se consume, desaparece. (Esto significa que livelock se evita en el caso de un productor agresivo; en algunos punto, el consumidor habrá vaciado una célula y dejará de tocar las células. Una vez eso sucede, el productor definitivamente puede publicar un valor, mientras que antes de ese punto, solo puede publicar un valor (cambiar la celda activa) si el consumidor no está en medio de una lectura.)

Si es un valor que está listo para ser consumido, solo el consumidor puede cambiar eso fact (para la célula inactiva, de todos modos); las producciones posteriores pueden cambiar qué célula está activo y el valor publicado, pero un valor siempre estará listo para ser leído hasta se consume.

Una vez que el productor haya terminado de escribir en la celda activa, puede "publicar" este valor cambiar qué celda es la activa uno (intercambiar el índice), siempre que el consumidor no en medio de la lectura de la otra celda. Si el consumidor está en el medio de al leer la otra celda, el intercambio no puede ocurrir, pero en ese caso el consumidor puede intercambiar después de es hecho de leer el valor, siempre que el productor no está en el medio de un escribe (y si lo es, el productor cambiará una vez que esté hecho). De hecho, en general, el consumidor siempre puede intercambiar después de haber terminado de leer (si es el único acceso al sistema) porque los swaps espurios por parte del consumidor son benignos: si hay algo en la otra celda, entonces el intercambio hará que se lea a continuación, y si no lo hay, el intercambio no afecta nada.

Por lo tanto, necesitamos una variable compartida para rastrear lo que es la celda activa, y también necesitamos una tanto para el productor como para el consumidor para indicar si están en el medio de un operación. Podemos almacenar estas tres piezas de estado en una variable atómica en orden para poder afectarlos a todos a la vez (atómicamente). También necesitamos una manera para que el consumidor compruebe si hay algo en la celda no activa en primer lugar, y para que ambos subprocesos modifiquen ese estado según proceda. Probé algunos otros enfoques, pero al final el más fácil fue solo incluir esta información en la otra variable atómica también. Esto hace que las cosas sean mucho más fácil de razonar, ya que todos los cambios de estado en el sistema son atómicos de esta manera.

Se me ocurrió una implementación sin espera (sin bloqueo, y todo operaciones completadas en un número limitado de instrucciones).

¡Hora código!

#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;
};

Tenga en cuenta que la semántica es tal que el consumidor nunca puede leer un valor dado dos veces, y un valor que lee siempre es más nuevo que el último valor que leyó. También es bastante eficiente en el uso de la memoria (dos buffers, como su solución original). Evité los bucles CAS porque generalmente son menos eficientes que una sola operación atómica bajo contención.

Si decide utilizar el código anterior, I sugiera que escriba primero algunas pruebas unitarias completas (enhebradas) para ello. Y puntos de referencia adecuados. Lo probé, pero apenas. Avísame si encuentras algún error: -)

Mi prueba unitaria:

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();

En cuanto a su implementación original, solo la miré de forma superficial (es mucho más divertido diseñar cosas nuevas, je), pero David.la respuesta de pfx parece abordar esa parte de tu pregunta.