Gibt es ein Objekt, das ich als FIFO mit Ereignis zum Auslesen von Daten verwenden kann?

Question

Ich brauche ein FIFO-Objekt, das, wenn es ein Element hat, ein Ereignis generiert oder einen Rückruf zurückgibt, um zu informieren, dass ein Element verfügbar ist. Soweit ich das sehen kann, unterstützt std: queue dies nicht.

In meinem Fall habe ich zwei Threads, die ein Thread Daten generieren und andere Thread benötigt, um sie zu verbrauchen.

Die Geschwindigkeit, mit der der erste Thread Daten erzeugt, ist nicht fest und daher muss ich einen Puffer haben, der die Daten speichert, damit der andere Thread sie lesen und relativ konstant verarbeiten kann.

Ich weiß, wie man den Schreiber implementiert, aber auf der Leserseite, Wenn ich die Warteschlange abfrage, dann verliere ich etwas Verarbeitungsleistung, um den Warteschlangenstatus zu überprüfen, und ich frage mich, ob es eine bessere Möglichkeit gibt, dies zu tun?

Edit 1

Es ist der Warteschlange nicht über Thread-Sicherheit, aber std :: Warteschlange basierend auf Abruf arbeitet, aber ich brauche etwas, das Ereignis basiert. std :: queue ist nicht ereignisbasiert und führt keinen Rückruf aus, wenn neue Daten verfügbar sind.

Answer 1

Wenn ich Ihre Frage richtig verstanden habe, können Sie sehr gut C++ std::condition_variables für die Benachrichtigung zwischen Threads über die Verfügbarkeit von Element in einer Warteschlange verwenden und dann den Rückruf aufrufen.

Der Code sieht ungefähr so ​​aus. Hier ist der Hauptthread dient als Generator Faden und Empfangen Thema fungiert als Verbraucher Faden

std::condition_variable Cv_; 
std::mutex Mutex_; 
std::queue<int> qVal; 

void callback() { 
    cout << "Callback Called with queue Value =>" << qVal.front() << endl; 
    qVal.pop(); 
} 
void ReceiveThread() { 

    while (true) { 
     std::unique_lock<mutex> Lock(Mutex_); 
     Cv_.wait(Lock); 
     callback(); 
    } 
} 

int main() { 
    thread thrd(ReceiveThread); 
    int pushVal = 1; 
    while (1) { 
     this_thread::sleep_for(std::chrono::seconds(1)); 
     qVal.push(pushVal); 
     cout << "Signalling for callback with value = " << pushVal<< endl; 
     pushVal++; 
     Cv_.notify_all(); 
    } 
} 

ich keine Exit-Bedingung in der while-Schleife hinzugefügt haben, die Sie vielleicht haben.

Hoffe, das hilft.

Answer 2

Die std::queue::push() Funktion haben keine Platzhalter im Inneren, so dass wir einfach unsere call_back function Referenz setzen kann, so dass nach dem erfolgreichen Element Einsetzen innerhalb queue die call_back function aufgerufen erhalten würde.

std::queue::push(data) 
{ 
    //add data to internal container 

    //placeholder section-------- 
    //invoke call_back function or event. 
    //placeholder section-------- 
} 

So in Ermangelung eines solchen Platzhalter können wir automatische Aufruf von call_back function oder ein Ereignis versuchen RAII verwenden.

Angenommen, wir wickeln unsere tatsächlichen Daten in eine struct, die uns in Benachrichtigungen helfen wird. Dann müssen wir indirekt über Objekte dieser struct auf die eigentlichen Daten zugreifen. Jetzt

struct data_notifier 
{ 
    //the true data. 
    int actual_data; 

    data_notifier(int data) : actual_data(data) 
    { 
     //signal event queue_full  
     //or 
     //call a call_back function. 
    } 
} 

int actual_data = 90; 
std::queue<data_notifier*> q; 
q.push(new data_notifier(actual_data)); 

, das einzige Problem ist: bevor eine Instanz von data_notifier richtig in queue als Referenz/Zeiger, unsere call_back oder Ereignis aufgerufen erhalten würde eingefügt wird. Wenn das Ereignis aufgerufen wird, versucht der Leser, die Daten zu lesen, aber die Daten werden nicht aus der Warteschlange abgerufen, weil die Daten in der Warteschlange noch nicht gespeichert sind. Also, eine solche Garantie, dass Daten innerhalb der Warteschlange dauerhaft erhalten bleiben, ist nur möglich, nachdem die Funktion std::queue::push() zurückgibt, was innerhalb der Writer Funktion passieren kann und wird.

//event_full is a manual event which needs to be signalled and non-signalled manually. 

void Writer() 
{ 
    while(1) 
    { 
     //[1] wait for mutex_queue 
     //[2] myqueue.push(data); 
     //[3] data is now persisted so signal event_full 
     //[4] release mutex_queue 
    } 
} 

void Reader() 
{ 
    while(1) 
    { 
     //[1] wait for event_full (so no polling) 

     //[2] wait for mutex_queue 

     //[3] --- access queue --- 
     if(myqueue.size() != 0) 
     { 
      //process myqueue.front() 
      //myqueue.pop(); 
     } 
     if(myqueue.size() == 0) 
     { 
      //reset event_full  
      //so as long as queue has data, Reader can process it else needs to wait. 
     } 
     //[3] --- access queue --- 

     //[4] release mutex_queue 
    } 
}