Efficient Consumer-Thread mit mehreren Herstellern

Question

Ich versuche, ein Erzeuger/Verbraucher-Thread Situation effizienter durch Überspringen teure Veranstaltung Operationen, wenn nötig mit so etwas wie zu machen:

//cas(variable, compare, set) is atomic compare and swap 
//queue is already lock free 

running = false 


// dd item to queue – producer thread(s) 

if(cas(running, false, true)) 
{ 
    // We effectively obtained a lock on signalling the event 
    add_to_queue() 
    signal_event() 
} 
else 
{ 
    // Most of the time if things are busy we should not be signalling the event 
    add_to_queue() 

    if(cas(running, false, true)) 
    signal_event() 
} 

... 

// Process queue, single consumer thread 

reset_event() 

while(1) 
{ 
    wait_for_auto_reset_event() // Preferably IOCP 

    for(int i = 0; i &lt SpinCount; ++i) 
    process_queue() 

    cas(running, true, false) 

    if(queue_not_empty()) 
    if(cas(running, false, true)) 
     signal_event() 
} 

Offensichtlich versuchen, diese Dinge zu bekommen richtig ist ein wenig knifflig (!) also ist der obige Pseudocode korrekt? Eine Lösung, die das Ereignis mehr signalisiert, als genau benötigt wird, ist in Ordnung, aber nicht für jedes Element.

Answer 1

Ging durch eine Reihe von Fällen, kann kein Problem sehen. Aber es ist irgendwie kompliziert. Ich dachte, du könntest vielleicht ein Problem mit queue_not_empty/add_to_queue haben. Aber es sieht so aus, als ob der post-dominierende CAS in beiden Fällen diesen Fall abdeckt.

CAS ist teuer (nicht so teuer wie Signal). Wenn Sie das Signal gemeinsam sein erwarten Überspringen, würde ich die CAS-Code wie folgt:

bool cas(variable, old_val, new_val) { 
    if (variable != old_val) return false 
    asm cmpxchg 
} 

Lock-freie Strukturen wie dies das Zeug ist, dass Jinx (das Produkt, das ich auf Arbeit) bei Tests sehr gut. Daher sollten Sie möglicherweise eine Evaluierungslizenz verwenden, um die Warteschlangen- und Signaloptimierungslogik zu testen.

---

Edit: vielleicht können Sie diese Logik vereinfachen.

running = false 

// add item to queue – producer thread(s) 
add_to_queue() 
if (cas(running, false, true)) { 
    signal_event() 
} 

// Process queue, single consumer thread 

reset_event() 

while(1) 
{ 
    wait_for_auto_reset_event() // Preferably IOCP 

    for(int i = 0; i &lt SpinCount; ++i) 
     process_queue() 

    cas(running, true, false) // this could just be a memory barriered store of false 

    if(queue_not_empty()) 
     if(cas(running, false, true)) 
     signal_event() 
} 

Jetzt, da die cas/Signal immer nebeneinander stehen, können sie in ein Unterprogramm verschoben werden.

Answer 2

Warum verbinden Sie nicht einfach eine bool mit dem Ereignis? Verwenden Sie cas, um es auf True zu setzen, und wenn das cas erfolgreich ist, dann melden Sie das Ereignis, da das Ereignis gelöscht sein muss. Der Kellner kann dann löschen Sie einfach die Fahne, bevor es

bool flag=false; 

// producer 
add_to_queue(); 
if(cas(flag,false,true)) 
{ 
    signal_event(); 
} 

// consumer 
while(true) 
{ 
    while(queue_not_empty()) 
    { 
     process_queue(); 
    } 
    cas(flag,true,false); // clear the flag 
    if(queue_is_empty()) 
     wait_for_auto_reset_event(); 
} 

Auf diese Weise wartet, müssen Sie nur warten, wenn es keine Elemente in der Warteschlange sind, und Sie nur das Ereignis einmal für jede Charge von Gegenständen signalisieren.

Answer 3

glaube ich, wollen Sie so etwas wie in dieser Frage erreichen:
WinForms Multithreading: Execute a GUI update only if the previous one has finished. Es ist spezifisch auf C# und WinForms, aber die Struktur kann auch für Sie gelten.

Answer 4

Dies fällt in die Unterkategorie "Stoppt herum und kehrt zur Arbeit zurück", bekannt als "vorzeitige Optimierung". :-)

Wenn die „teueren“ Event-Operationen einen erheblichen Teil der Zeit in Anspruch nehmen sind, Ihr Design ist falsch, und anstatt einen Erzeuger/Verbraucher zu verwenden, sollten Sie einen kritischen Abschnitt/Mutex verwenden und tun nur die Arbeit von der aufrufende Thread.

Ich schlage vor, Sie profilieren Ihre Anwendung, wenn Sie wirklich besorgt sind.

Aktualisiert:

Richtige Antwort:

Produzent

ProducerAddToQueue(pQueue,pItem){ 

    EnterCriticalSection(pQueue->pCritSec) 
     if(IsQueueEmpty(pQueue)){ 
      SignalEvent(pQueue->hEvent) 
     } 

     AddToQueue(pQueue, pItem) 
    LeaveCriticalSection(pQueue->pCritSec) 
} 

Consumer

nCheckQuitInterval = 100; // Every 100 ms consumer checks if it should quit. 

ConsumerRun(pQueue) 
{ 
    while(!ShouldQuit()) 
    { 
     Item* pCurrentItem = NULL; 
     EnterCriticalSection(pQueue-pCritSec); 
      if(IsQueueEmpty(pQueue)) 
      { 
       ResetEvent(pQueue->hEvent) 
      } 
      else 
      { 
       pCurrentItem = RemoveFromQueue(pQueue); 
      } 
     LeaveCriticalSection(pQueue->pCritSec); 

     if(pCurrentItem){ 
      ProcessItem(pCurrentItem); 
      pCurrentItem = NULL; 
     } 
     else 
     { 
      // Wait for items to be added. 
      WaitForSingleObject(pQueue->hEvent, nCheckQuitInterval); 
     } 

    } 
} 

Hinweise:

• Das Ereignis ist ein Ereignis zum manuellen Zurücksetzen.
• Die durch den kritischen Abschnitt geschützten Vorgänge sind schnell. Das Ereignis wird nur gesetzt oder zurückgesetzt, wenn die Warteschlange in den leeren Zustand übergeht. Es muss innerhalb des kritischen Bereichs gesetzt/zurückgesetzt werden, um eine Wettlaufsituation zu vermeiden.
• Dies bedeutet, dass der kritische Abschnitt nur für kurze Zeit gehalten wird. so wird Konkurrenz selten sein.
• Kritische Abschnitte blockieren nicht, es sei denn, sie sind strittig. Daher werden Kontextwechsel selten sein.

Annahmen:

• Das ist ein echtes Problem, keine Hausaufgaben.
• Hersteller und Verbraucher verbringen die meiste Zeit damit, andere Dinge zu tun, d. H. Die Artikel für die Warteschlange bereit zu machen und sie nach dem Entfernen aus der Warteschlange zu verarbeiten.
• Wenn sie die meiste Zeit die eigentlichen Warteschlangenoperationen ausführen, sollten Sie keine Warteschlange verwenden. Ich hoffe, das ist offensichtlich.