C++ 11 coda thread-safe

Question

Un progetto su cui sto lavorando utilizza più thread per lavorare su una raccolta di file. Ogni thread può aggiungere file all'elenco dei file da elaborare, quindi ho messo insieme (quello che pensavo fosse) una coda thread-safe. parti rilevanti seguono:

// qMutex is a std::mutex intended to guard the queue 
// populatedNotifier is a std::condition_variable intended to 
//     notify waiting threads of a new item in the queue 

void FileQueue::enqueue(std::string&& filename) 
{ 
    std::lock_guard<std::mutex> lock(qMutex); 
    q.push(std::move(filename)); 

    // Notify anyone waiting for additional files that more have arrived 
    populatedNotifier.notify_one(); 
} 

std::string FileQueue::dequeue(const std::chrono::milliseconds& timeout) 
{ 
    std::unique_lock<std::mutex> lock(qMutex); 
    if (q.empty()) { 
     if (populatedNotifier.wait_for(lock, timeout) == std::cv_status::no_timeout) { 
      std::string ret = q.front(); 
      q.pop(); 
      return ret; 
     } 
     else { 
      return std::string(); 
     } 
    } 
    else { 
     std::string ret = q.front(); 
     q.pop(); 
     return ret; 
    } 
} 

Tuttavia, sono occasionalmente segfault all'interno del blocco if (...wait_for(lock, timeout) == std::cv_status::no_timeout) { } e ispezione gdb indica che i segfaults si verificano perché la coda è vuota. Com'è possibile? Sono stato a conoscenza del fatto che wait_for restituisce solo cv_status::no_timeout quando è stato notificato, e ciò dovrebbe avvenire solo dopo che FileQueue::enqueue ha appena inserito un nuovo elemento nella coda.

Answer 1

Secondo lo standard condition_variables, è possibile attivare spuriemente, anche se l'evento non si è verificato. In caso di sveglia spuria restituirà cv_status::no_timeout (poiché si è svegliato invece di scadere), anche se non è stato notificato. La soluzione corretta per questo è, naturalmente, per verificare se il wakeup era effettivamente legittimo prima di procedere.

I dettagli sono specificati nella norma §30.5.1 [thread.condition.condvar]:

-La funzione sbloccherà quando viene segnalato da una chiamata a notify_one(), una chiamata a notify_all() , scadenza del timeout assoluto (30.2.4) specificato da abs_time o spurio.

...

Returns: cv_status :: timeout se il timeout assoluto (30.2.4) speci fi edby abs_time scaduto, altri-ise cv_status :: no_timeout.

Answer 2

Questo è probabilmente il modo si dovrebbe fare:

void push(std::string&& filename) 
{ 
    { 
     std::lock_guard<std::mutex> lock(qMutex); 

     q.push(std::move(filename)); 
    } 

    populatedNotifier.notify_one(); 
} 

bool try_pop(std::string& filename, std::chrono::milliseconds timeout) 
{ 
    std::unique_lock<std::mutex> lock(qMutex); 

    if(!populatedNotifier.wait_for(lock, timeout, [this] { return !q.empty(); })) 
     return false; 

    filename = std::move(q.front()); 
    q.pop(); 

    return true;  
} 

Answer 3

vorrei riscrivere la funzione dequeue come:

std::string FileQueue::dequeue(const std::chrono::milliseconds& timeout) 
{ 
    std::unique_lock<std::mutex> lock(qMutex); 
    while(q.empty()) { 
     if (populatedNotifier.wait_for(lock, timeout) == std::cv_status::timeout) 
      return std::string(); 
    } 
    std::string ret = q.front(); 
    q.pop(); 
    return ret; 
} 

è più breve e non ha il codice duplicato come la vostra ha fatto. Solo il rilascio può attendere più a lungo il timeout. Per evitare che sia necessario ricordare l'ora di inizio prima del ciclo, controllare il timeout e regolare di conseguenza il tempo di attesa. O specificare il tempo assoluto in condizione di attesa.

Answer 4

Basta guardarlo, quando si controlla una variabile di condizione è meglio usare un ciclo while (in modo che se si sveglia e non sia ancora valido si ricontrolla). Ho appena scritto un modello per una coda asincrona, spero che questo aiuti.

#ifndef SAFE_QUEUE 
#define SAFE_QUEUE 

#include <queue> 
#include <mutex> 
#include <condition_variable> 

// A threadsafe-queue. 
template <class T> 
class SafeQueue 
{ 
public: 
    SafeQueue(void) 
    : q() 
    , m() 
    , c() 
    {} 

    ~SafeQueue(void) 
    {} 

    // Add an element to the queue. 
    void enqueue(T t) 
    { 
    std::lock_guard<std::mutex> lock(m); 
    q.push(t); 
    c.notify_one(); 
    } 

    // Get the "front"-element. 
    // If the queue is empty, wait till a element is avaiable. 
    T dequeue(void) 
    { 
    std::unique_lock<std::mutex> lock(m); 
    while(q.empty()) 
    { 
     // release lock as long as the wait and reaquire it afterwards. 
     c.wait(lock); 
    } 
    T val = q.front(); 
    q.pop(); 
    return val; 
    } 

private: 
    std::queue<T> q; 
    mutable std::mutex m; 
    std::condition_variable c; 
}; 
#endif 

Answer 5

ho qualche tempo fa mi sono imbattuto lo stesso problema e poi scrivere un modello di thread-safe-asincrono-coda GNU in C++ 11. Ho pubblicato nel mio blog:

http://gnodebian.blogspot.com.es/2013/07/a-thread-safe-asynchronous-queue-in-c11.html

Answer 6

Aggiungendo alla risposta accettata, direi che l'attuazione di una corretta coda a più produttori/multi-consumatori è difficile (più facile visto che il C++ 11, però)

Ti suggerisco di provare il (molto buono) lock free boost library, la struttura "coda" farà quello che vuoi, con le garanzie di assenza di attesa/blocco e without the need for a C++11 compiler.

sto aggiungendo questa risposta oggi perché la libreria senza blocchi è abbastanza nuovo per aumentare (dal 1,53 credo)

Answer 7

C'è anche una soluzione GLib per questo caso, non ho ancora provato, ma credo è una buona soluzione. https://developer.gnome.org/glib/2.36/glib-Asynchronous-Queues.html#g-async-queue-new