C++ 11 Thread: thread multipli in attesa di una variabile di condizione

Question

Attualmente sto lavorando su un problema che simula un modello esteso di Producer-Worker. In questo problema ci sono 3 lavoratori e 3 strumenti disponibili, e per i lavoratori al lavoro hanno bisogno di 2 strumenti (e materiali, ma quelli sono irrilevanti). Se ci sono> = 2 strumenti nel vault, un lavoratore prenderà 2. Altrimenti, essi aspetteranno una variabile di condizione che verrà segnalata quando ci sono> = 2.

Questo va bene con 2 lavoratori: uno funzionerà, quindi restituirà gli strumenti al vault e l'altro lavoratore in attesa verrà risvegliato e prenderà 2 strumenti. Il problema è che, con 3 lavoratori, ci sarà sempre uno affamato per ottenere gli strumenti.

Dopo alcuni test, ho notato che i thread in attesa di una variabile di condizione sono strutturati in pila. C'è comunque la possibilità di farlo in coda? (1 attende, 2 attende e 3 attende. Quando 1 è risvegliato e vuole fare un altro, deve attendere dietro 2 e 3.)

Ecco un esempio di output. Il codice è troppo lungo, quindi lo posterò se è davvero necessario. Esistono 3 thread di lavoro e 1 mutex di strumenti. Chi sta morendo di fame si differenzia a vicenda.

1 Tools taken. Remaining: 1 
2 Waiting on tools... 
3 Waiting on tools... 
1 Operator Product made. Tools returned. Tools now:3 
3 Tools taken. Remaining: 1 
1 Waiting on tools... 
3 Materials returned for switch. 
3 Operator Product made. Tools returned. Tools now:3 
1 Tools taken. Remaining: 1 
3 Waiting on tools... 
1 Materials returned for switch. 
1 Operator Product made. Tools returned. Tools now:3 
3 Tools taken. Remaining: 1 
1 Waiting on tools... 
3 Materials returned for switch. 
3 Operator Product made. Tools returned. Tools now:3 
1 Tools taken. Remaining: 1 
3 Waiting on tools... 
1 Materials returned for switch. 
1 Operator Product made. Tools returned. Tools now:3 
3 Tools taken. Remaining: 1 
1 Waiting on tools... 
3 Materials returned for switch. 
3 Operator Product made. Tools returned. Tools now:3 
1 Tools taken. Remaining: 1 
3 Waiting on tools... 
1 Materials returned for switch. 
... 

(Come si può vedere 2 non viene mai gli strumenti ...)

Aggiornamento: 2013/07/05 ho aggiunto un certo codice.

int tools = 3; //global 
string last; //current last product on output buffer 
mutex toolsMutex; 
mutex matSearchMutex; 

int main(){ 
//Initializing Producers 
    Producer prod1(1); 
    Producer prod2(2); 
     Producer prod3(3); 



    thread p1(processor,1); 
    thread p2(processor,2); 
    thread p3(processor,3); 

    p1.detach(); 
    p2.detach(); 
    p3.detach(); 

    while(true){//forever running 

    } 

    return 0; 
} 

Processore:

//Processor method 
void processor(int i){ 
    srand(time(NULL)); 

    while (true){ //forever running 


    bool hasTools = false; 
    bool productMade = false; 
    while (productMade == false){ //while product has yet to be made. 
     //choose what to make... 



     if (hasTools == false){ 
      thread matT(getMaterials,whatToMake); 
      thread toolT(getTools,i); 
      toolT.join();   
      matT.join(); 
      hasTools = true; 
     } 
     else{ //tools acquired but no materials 
      thread matT(getMaterials,whatToMake); 
      matT.join(); 
     } 

     if (recordedLast.compare(last) != 0){ 

      //return materials and acquire new ones the next run 

      continue; 
     } 
     else { 
      makeProduct(whatToMake); 
      unique_lock<mutex> locker(toolMutex); 
      tools = tools + 2; 
      cout << i << " Operator Product made. Tools returned. Tools now:" << tools << endl; 
      productMade = true; 
      if (tools >=2) toolsCV.notify_one(); 
     } 

    //done processing 

    } 


} 

} 

makeProducts:

void makeProduct(int i){ 
    unique_lock<mutex> mainMatLock(matSearchMutex); 
    // make product according to i 
    this_thread::sleep_for(chrono::milliseconds(rand() % 1000 + 10)); 
} 

getTools:

void getTools(int i){ 
    unique_lock<mutex> locker(toolMutex); 
    if (tools <2){ 
     cout << i << " Waiting on tools..." << endl; 
     toolsCV.wait(locker);} 
    tools = tools - 2;//tools acquired 
    cout << i <<" Tools taken. Remaining: " << tools << endl; 

} 

grazie a coloro che hanno risposto. Proverò a implementare una coda di attesa stasera utilizzando più variabili di condizione.

(PS C'è qualche modo migliore per fare formattazione del codice qui su Stack Overflow? Altro che i quattro spazi ...

Answer 1

std::condition_variable non specifica quale thread di attesa viene svegliato quando si chiama notify_one. Dovresti quindi scrivere un codice a cui non interessa quale thread venga svegliato. Lo schema standard prevede che qualsiasi thread venga svegliato, che il thread faccia il lavoro che deve essere fatto.

Se è necessario che i thread vengano risvegliati in un ordine specifico, utilizzare un meccanismo diverso. Ad esempio, è possibile avere un std::condition_variable separato per ogni thread e quindi inserire i thread in una coda quando hanno bisogno di strumenti. Come thread inserisce gli strumenti, può quindi segnalare la variabile di condizione corrispondente al thread nella parte anteriore della coda. Quel thread verrà quindi svegliato e gli altri rimarranno a dormire (modulo wake up spurie).

Answer 2

Quando ci sono diversi thread in attesa su una condizione, l'ordine in cui sono risvegliato (notify_all) o che si è risvegliato (notify_one) è non specificato Se avete bisogno di un qualche tipo dell'ordine, è necessario utilizzare notify_all, e attuarlo te è possibile mantenere una coda dei thread in attesa:.. prima attesa (ma dopo aver acquisito il mutex), spingere il thread ID su alla fine della coda. "questo thread su parte anteriore della coda e gli strumenti necessari disponibili". Quando si ricevono gli strumenti , rimuovere l'ID dalla parte anteriore della coda e chiamare di nuovo notify_all.

Answer 3

Un approccio potrebbe essere l'utilizzo di un semaforo completo che è condiviso tra i thread anziché una variabile di condizione. In questo modo, puoi aspettare un conteggio specifico.

#include <mutex> 
#include <thread> 
#include <condition_variable> 

using std::mutex; 
using std::condition_variable; 

class Semaphore 
{ 
public: 
    /** 
    * Construct a counting semaphore with an initial value 
    * @param cnt The value of the initial semaphore count 
    */ 
    Semaphore(unsigned int cnt); 

    /** 
    * acquire a semaphore count 
    * @param numRes The number of count ressources to acquire 
    */ 
    void acquire(unsigned int numRes = 1); 

    /** 
    * try to acquire a semaphore count. 
    * @param numRes The number of count ressources that the method tries to acquire 
    * @return true, if numRes could be aquired 
    *   false, otherwise 
    */ 
    bool tryAcquire(unsigned int numRes = 1); 

    /** 
    * release one semaphore cnt 
    * @param numRes The number of count ressources to release 
    */ 
    void release(unsigned int numRes = 1); 

private: 
    unsigned int cnt; 
    mutex mut; 
    condition_variable cond; 
}; 

Attuazione assomiglia:

void Semaphore::acquire(unsigned int numRes) 
{ 
    unique_lock<mutex> lock(mut); 
    while (cnt < numRes) 
    { 
     cond.wait(lock); 
    } 

    cnt-=numRes; 
} 

bool Semaphore::tryAcquire(unsigned int numRes) 
{ 
    unique_lock<mutex> lock(mut); 
    if (cnt>=numRes) 
    { 
     cnt -= numRes; 
     return true; 
    } 
    return false; 
} 

void Semaphore::release(unsigned int numRes) 
{ 
    { 
     unique_lock<mutex> lock(mut); 
     cnt += numRes; 
    } 
    // notify <numRes> waiting entities 
    for (unsigned int i = 0; i<numRes; ++i) 
    { 
     cond.notify_one(); 
    } 
} 

Answer 4

Il vero problema qui è che se si dispone di thread di lavoro e limitata quantità di risorse necessarie, non si deve la cura che infila in realtà viene attivato, si deve solo cura che il lavoro sia fatto. L'unica differenza qui è nella registrazione. Il numero di thread che hai definito è il numero di thread che possono essere eseguiti in parallelo, che è limitato dalle risorse a uno.

Se ciò non va bene per te, allora devi agire tu stesso come spiegato in altre risposte.