Der einfachste Weg, atomar zu entketten?

Question

Ich habe eine Reihe von Daten, die gleichzeitig verarbeitet werden müssen mit Multi-Threading, die Anzahl der Daten ist angeblich größer als die Anzahl der Threads. Ich entschied mich dafür, die Daten in eine Art von Warteschlange zu stellen, so dass jeder freie Thread seinen Teil platzen und verarbeiten kann, bis die Warteschlange leer ist. Ich könnte eine einfache STL-Warteschlange verwenden und sie durch einen Mutex sperren, wenn ich ein Element daraus entfernen möchte, aber ich möchte einen Lock-Free-Ansatz versuchen. Zur gleichen Zeit ist mein Projekt zu klein, um von einer Drittanbieter-Bibliothek abhängig zu sein, die blockfreie Strukturen bietet, tatsächlich brauche ich nur eine atomare Auslagerung. Also habe ich beschlossen, meine eigene Warteschlange zu implementieren, basierend auf einem Vektor mit einem Zeiger auf den „Kopf“ und erhöhen diesen Zeiger atomar:

template <typename T> 
class AtomicDequeueable 
{ 
public: 

    // Assumption: data vector never changes 
    AtomicDequeueable(const std::vector<T>& data) : 
     m_data(data), 
     m_pointer(ATOMIC_VAR_INIT(0)) 
    {} 

    const T * const atomicDequeue() 
    { 
     if (std::atomic_load(&m_pointer) < m_data.size()) 
     { 
      return &m_data 
      [ 
       std::atomic_fetch_add(&m_pointer, std::size_t(1)) 
      ]; 
     } 

     return nullptr; 
    } 

private: 

    AtomicDequeueable(const AtomicDequeueable<T>&) {} 

    std::atomic_size_t m_pointer; 
    const std::vector<T>& m_data; 
}; 

Themen Funktion sieht wie folgt aus:

void f(AtomicDequeueable<Data>& queue) 
{ 
    while (auto dataPtr = queue.atomicDequeue()) 
    { 
     const Data& data = *dataPtr; 
     // processing data... 
     std::this_thread::sleep_for(std::chrono::milliseconds(1)); 
    } 
} 

Meine Erfahrung mit Lock-Free-Strukturen und Primitive ist wirklich arm, also frage ich mich: Wird mein Ansatz richtig funktionieren? Sicherlich habe ich es auf Ideone getestet, aber ich weiß nicht, wie es sich mit echten Daten verhalten wird.

Answer 1

Momentan hat Ihre atomicDequeue-Funktion ein Datenrennen: Es ist für 2 Threads möglich, dass beide die erste atomic-Anweisung ausgeführt haben, bevor die zweite ausgeführt wird. Dies kann jedoch festgelegt werden, wie Sie wirklich nur 1 Atom-Betrieb benötigen, wie in der folgenden Änderung:

const T * const atomicDequeue() 
{ 
    auto myIndex = std::atomic_fetch_add(&m_pointer, std::size_t(1)); 

    if(myIndex >= m_data.size()) 
     return nullptr; 

    return &m_data[myIndex]; 
} 

Dieser nichts ändert den Eingangsvektor während Fadenbetrieb vorgesehen funktioniert.

Answer 2

Ihr Code ist sehr fehlerhaft. Lassen Sie mich hier in meiner Empfehlung sehr ehrlich:

• "Actum Ne Agas.   Sie etwas nicht tun bereits getan"   Sie haben viel von bereits vorhandenen C++ - Klassen zur Verfügung, die zuverlässig Warteschlangen und Interprozesskommunikation (IPC) im Allgemeinen durchführen können. Benutze einen von ihnen.

• Mach dir keine Sorgen über "sperren frei." Das ist, was Schlösser sind für, und sie sind zuverlässig, schnell und billig.

Ihre Vorstellung von „unter Verwendung einer Warteschlange“ ist solide, aber Sie unnötige Arbeit gehen ... und Wanzen ... und wenn man einfach „etwas aus dem Regal greifen.“ Sie wissen, dass der Standardteil korrekt funktioniert, weil andere Personen ihn zu Tode getestet haben.