Dotaz: C a pthreads - cekani na vice conditional variables

[Tomáš Skočdopole]

Ahoj,

chtel bych se zeptat jak lze udelat cekani na vice conditional variables soucasne a abych po uvolneni cekani byl schopny zjistit, kde se co prihodilo.

Zkousim udelat aplikaci, kde je hlavni vlakno (boss) a podvlakna (workers). Ted resim jak udelat komunikaci z podvlaken do hlavniho vlakna.

Hlavni vlakno by melo cekat na vysledky od podvlaken.

Dekuji predem za rady.

Tomas

Odpovědi

[Michal Kubeček]

Udělejte jednu společnou condition variable a pro ni pošlete signál ze všech jednotlivých "worker" vláken. V hlavním pak zkontrolujte, jestli jsou splněny podmínky pro pokračování. Pokud "worker" po provedení výpočtu skončí a vy potřebujete počkat, až skončí všichni, je nejjednodušší prostě zavolat postupně na všechny pthread_join() (na pořadí nezáleží).

[Tomáš Skočdopole]

Omlouvam se za nechapavost, ale nejak mi to stale neni jasne.

Pokud tedy hlavni vlakno (boss) ma cekat na podvlakna (workers), pouzije jedna condition variable a az kterekoliv vlakno je pripraveno odeslat sve vysledky, probudi hlavni vlakno (boss) pomoci teto spolecne condition variable.

Jak ale vlakno zjisti, ktery worker ho probudil? 

A jeste bych se chtel zeptat, zda pro predavani samotnych dat z jednotlivych vlaken pouzit jeden spolecny buffer, nebo pole bufferu pro jednotlive podvlakna (workers).

 

Dekuji!

[Michal Kubeček]

Condition variable sama o sobě nenese jinou informaci než že mohlo dojít ke změně stavu podmínky (splněna/nesplněna). Stejně pak musíte zkontrolovat, jestli je ta podmínka splněna nebo ne. Takže nechte každého workera po skončení nastavit příznak nebo dekrementovat proměnnou (počet zbývajících workerů) a boss bude kontrolovat, jestli jsou nastaveny všechny příznaky resp. jestli je ta proměnná nulová.

Co se týká předávání dat, každá z možností má své výhody a nevýhody, většinou to ale asi bude celkem jedno.

[Hop]

Dobre, 

takze jsem zatim udelal neco takoveho - predavani dat je pres spolecny buffer:

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

#define NUM_OF_WORKERS  3

#define NUM_WORKERS_JOB 5
#define JOB_MAX_TIME 10

#define MESSAGE_FOR_BOSS_LENGTH 80
char message_for_boss[MESSAGE_FOR_BOSS_LENGTH];

pthread_mutex_t message_for_boss_mutex;


void *worker_thread (void* t)
{
    int job;
    unsigned int random_time;

    printf ("W%i: Hello!\n", (unsigned int)t);

    for (job = 0; job < NUM_WORKERS_JOB; job++)
    {
        random_time = rand () % JOB_MAX_TIME;

        printf ("W%i: I will complete my job #%i after %i seconds.\n", (unsigned int)t, job, random_time);

        /* Simulate a job progress */
        sleep(random_time);

        /* Sent message for my Boss */
        pthread_mutex_lock (&message_for_boss_mutex);
        sprintf (message_for_boss, "Hi, there is W%i! My job #%i is completed! Processing time was %i seconds.", (unsigned int)t, job, random_time);
        pthread_mutex_unlock (&message_for_boss_mutex);

        printf ("W%i: I have completed my job #%i.\n", (unsigned int)t, job);
    }

    printf ("W%i: Good bye!\n", (unsigned int)t);

    pthread_exit (NULL);
}


int main (int argc, char **argv)
{
    unsigned int i;

    pthread_t threads[NUM_OF_WORKERS];
    pthread_attr_t attr;

    /* Initialize mutex */
    pthread_mutex_init(&message_for_boss_mutex, NULL);

    /* For portability, explicitly create threads in a joinable state */
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

    for (i = 0; i < NUM_OF_WORKERS; i++)
    {
        printf ("B: Creating worker thread W%i\n", i);
        pthread_create (&threads[i], &attr, worker_thread, (void *)i);
    }

    /* Wait for result from worker_thread */
    printf ("B: Waiting for messages from workers.\n");


    /* Wait for all threads to complete */
    for (i = 0; i < NUM_OF_WORKERS; i++) {
        pthread_join(threads[i], NULL);
    }

    printf ("B: All worker threads are joined!\n");

    /* Destroy mutex */
    pthread_mutex_destroy(&message_for_boss_mutex);

    return (0);
}


jeste ale nechapu tu spolecnou condition variable - pokud bosse probudi cekani pthread_cond_wait na spolecnou podminku, tak jakym zpusobem se on potom dozvi, ktery worker ho probudil?

Jinak jeste bych zaroven poprosil o zbeznou kontrolu tohoto, co uz mam - jsem v programovani v podstate samouk, takze si nejsem jisty, zda to delam tak spravne.

Dekuji, Tomas

[Michal Kubeček]

Neprohlížel jsem to nijak podrobně, takže asi nevidím všechno, ale měl bych dvě poznámky. Za prvé: přinejmenším u pthread_create() se vyplatí vždy zkontrolovat návratovou hodnotu, jestli thread opravdu vznikl. Za druhé: připadá mi divné, že všechny "worker" thready zapisují zprávu pro "bosse" na stejné místo. Sice používáte mutex, takže mezi sebou kolidovat nebudou, ale ten, kdo by si zprávy chtěl přečíst (což ve vašem programu nikdo nedělá), uvidí vždy jen jednu (tu poslední).

S tou condition variable by to mohlo vypadat asi takto:

const unsigned n_workers = 10;

typedef struct {
  ...
} worker_data;
worker_data wdata_buffer[n_workers];

unsigned active_workers;
pthread_mutex_t mtx_aw;
pthread_cond_t cv_aw;

bool finish;
pthread_mutex_t mtx_finish;

void* worker_main(void* ptr)
{
  worker_data* const data = (worker_result*) ptr;

  while (true) {
    wait_for_data(data); 
    if (finish) break;
    process_data(data);

    pthread_mutex_lock(&mtx_aw);
    active_workers--;
    pthread_cond_broadcast(&cv_aw);
    pthread_mutex_unlock(&mtx_aw);
  }

  return NULL;
}

int main()
{
  ...

  finish = false;
  for (i=0; i<n_workers; i++) {
    R = pthread_create(&threads[i], NULL, worker_main, &wdata_buffer[i]);
    if (R) exit(1);
  }

  active_workers = 0;
  while(...) {
    for (i=0; i<n_workers; i++) {
      prepare_data(&wdata_buffer[i]);
      pthread_mutex_lock(&mtx_aw);
      active_workers++;
      pthread_mutex_unlock(&mtx_aw);
      notify_worker(&wdata_buffer[i]);  /* wait_for_data() waits for this */
    }

    pthread_mutex_lock(&mtx_aw);
    while (active_workers > 0)
      pthread_cond_wait(&cv_aw, &mtx_aw);
    pthread_mutex_unlock(&mtx_aw);

    for (i=0; i<n_workers; i++) {
      process_result(&wdata[i]);
    }
  }

  
  pthread_mutex_lock(&mtx_finish);
  finish = true;
  pthread_mutex_unlock(&mtx_finish);
  for (i=0; i<n_workers; i++)
    notify_worker(&wdata_buffer[i]);
  for (i=0; i<n_workers; i++)
    pthread_join(threads[i], NULL);

  ...
}

Píšu to z hlavy, takže tam budou určitě nějaké chyby, nejspíš to půjde i nějak vylepšit, záleží také na požadavcích konkrétní aplikace. Berte to spíš jako náznak principu.

[Tomáš Skočdopole]

Dekuji za reakci, je to pro me inspirujici.

Jenom bych se chtel zeptat, zda neni potreba osetrit mutextem pristup do pole worker_data wdata_buffer[n_workers]; i presto ze kazdy worker si bude zapisovat akorat do toho sveho indexu.

V tom mem ukazkovem prikladu jsem to pro jednoduchost napsal tak, ze kazdy worker vykonava stejny kod . Ve skutecnosti to bude tak, ze kazdy worker bude provadet uplne neco jineho.

Pokud jsem dobre pochopil Vasi ukazku, tak ze zacatku rozdate workerum data:

    for (i=0; i<n_workers; i++) {
      prepare_data(&wdata_buffer[i]);
      pthread_mutex_lock(&mtx_aw);
      active_workers++;
      pthread_mutex_unlock(&mtx_aw);
      notify_worker(&wdata_buffer[i]);  /* wait_for_data() waits for this */
    }

 A pak se ceka, az vsichni workeri odeslou vysledek:

   pthread_mutex_lock(&mtx_aw);
    while (active_workers > 0)
      pthread_cond_wait(&cv_aw, &mtx_aw);
    pthread_mutex_unlock(&mtx_aw);

Coz pro me neni moc vhodne, protoze jak jsem jiz uvedl, kazdy worker bude delat jiny kod a kazdemu to bude trvat jinou dobu, nez budou pripraveni odesilat data.

Dalsi veci co neni potreba je to opetovne odesilani pracovnich dat pro workery. Worker po vytvoreni bude vedet co ma delat po celou dobu jeho zivota. Kdybych dal jednoduchy priklad, jeden z workeru se vytvori s attributem www.abclinuxu.cz  a on bude mit na starost napriklad kazdou minutu pingnout www.abclinuxu.cz a treba v pripade pingu vetsiho jak 500ms notifikuje bose a preda mu tuto zpravu.

Boss vytvori na zacatku sve workery a pak jenom bude cekat az mu nektery z nich neco posle.Takze ja hledam techniku jak uspat bosse a jak ho z nejakeho workera probudit, a predat mu nejake data. Na ty data boss zareaguje a pak se znovu uspi.

Napadlo me napriklad udelat strukturu message_for_boss { int WorkerID; const char *Message } a worker by toto naplnil a kdyz se boss vzbudi, precetl by si WorkerID a vedel by tak od koho zprava je.

Mohlo by to tak byt?

Dekuji Tomas

[Michal Kubeček]

Jenom bych se chtel zeptat, zda neni potreba osetrit mutextem pristup do pole worker_data wdata_buffer[n_workers]; i presto ze kazdy worker si bude zapisovat akorat do toho sveho indexu.

Zamykání je nutné jen tam, kde hrozí, že se budou dva thready současně pokoušet přistupovat ke stejnému úseku paměti (a aspoň jeden z nich zapisovat), což by tady nastat nemělo, pokud jsem neudělal nějakou chybu.

Coz pro me neni moc vhodne, protoze jak jsem jiz uvedl, kazdy worker bude delat jiny kod a kazdemu to bude trvat jinou dobu, nez budou pripraveni odesilat data.

Pak se to samozřejmě musí řešit trochu jinak, např. udržováním fronty požadavků a poolu volných workerů.

Napadlo me napriklad udelat strukturu message_for_boss { int WorkerID; const char *Message } a worker by toto naplnil a kdyz se boss vzbudi, precetl by si WorkerID a vedel by tak od koho zprava je.

To bych nedporučil, protože probuzení se z pthread_cond_wait() nemusejí odpovídat jednotlivým signálům od workerů. Takže by se mohlo stát (a podle zákona schválnosti stávalo), že dva workeři skonči přibližně současně a vy uvidíte zprávu jen od toho druhého. Takže bych to podle okolností řešil buď frontou zpráv nebo polem, kde každý thread bude mít svůj prostor pro zprávu. Pole je trochu jednodušší na implementaci, ale má nevýhodu, že ho budete muset pokaždé prohledat.

[Tomáš Skočdopole]

Dobry den,

jeste bych se chtel zeptat ohledne predavani dat do vlaken. Podobne, jak mam v te ukazce, tak vytvarim vlakna ve smycce, kde jim jako parametr predam ukazatel na poradove cislo vytvoreni:

void* worker_thread (void* t)
{
    unsigned int my_id = *((unsigned int *)t);
        
    ...
}


...
for (unsigned int i = 0; i < NUM_OF_WORKERS; ++i)
{
    ... pthread_create (&T, &TA, worker_thread, (void *)&i);
}
...

Ovsem takto koukam ze to nejde, protoze obcas vice vlaken si mysli ze jsou se stejnym s poradovym cislem. Asi kdyz se uvnitr vlakna ziskava poradove promenna my_id, tak jeste tesne predtim cyklus uz jede dalsi smycku, takze hodnota i je uz jina.

Dekuji za pomoc. Tomas

[Vojtěch Horký]

Buď si uděláte pole a předáte ukazatel na prvek toho pole nebo -- pokud víte, že velikost int je menší nebo rovna velikosti void * -- můžete "prasit" takhle:

pthread_create (&T, &TA, worker_thread, (void *)i);

[Tomáš Skočdopole]

Dobra, dekuji za info.

Jeste bych se chtel zeptat ohledne problematiky cekani bosse na zpravu od nejakeho workera. Mam to teda reseno tak, ze mam udelanou globalni promennou std::queue<tMessageFromWorker> a do ni workeri (za pomoc mutexu a condition variables) ukladaji sve zpravy.

        boss_notify_mtx.lock ();
        {
            MessagesForBoss.push (M);
            boss_notify_cond.broadcast ();
        }
        boss_notify_mtx.unlock ();

Vyzvednuti zpravy bossem pak probiha takto:

        boss_notify_mtx.lock ();
        {
            while (MessagesForBoss.empty ())
                boss_notify_cond.wait (&boss_notify_mtx);

            /* Take message from queue */
            MessageForMe = MessagesForBoss.front ();
            MessagesForBoss.pop ();
        }
        boss_notify_mtx.unlock ();


Chtel bych si jen overit, zda to tak mam spravne - snazil jsem se aby boss zbytecne neprudil dlouho uvnitr zamceneho mutexu boss_notify_mtx, takze on si jenom z te fronty vyzvedne svoji zpravu a pak uz pracuje s jeji kopii.

Snad by to tak mohlo byt.

Predem dekuji za cenne namitky a rady. Tomas

[Michal Kubeček]

Z toho fragmentu není patrné, jestli to máte ošetřené nebo ne, tak na to pro jistotu upozorním ještě jednou: může se stát, že při probuzení z čekání na condition variable na vás bude čekat víc zpráv najednou, takže je potřeba s tím počítat a převzít si všechny.

[Tomáš Skočdopole]

Ano, dekuji za upozorneni, toto bych mel mit snad osetrene.

 while (1)
    {
        tMessageForBoss MessageForMe;

        /* Wait for message from threads */
        boss_notify_mtx.lock ();
        {
            while (MessagesForBoss.empty ())
            {
                boss_notify_cond.wait (&boss_notify_mtx);
            }

            /* Take message from queue */
            MessageForMe = MessagesForBoss.front ();
            MessagesForBoss.pop ();
        }
        boss_notify_mtx.unlock ();


        /* Process message */
        ...

        /* Check for presence of active threads */
        if (PTM.ActivePluginsThreads.Get () == 0) break;
    }

Cili kdyz bude ve fronte vice zprav, volani pthread_cond_wait, ktere je v te vnitrni smycce while se provadet nebude, takze pri dalsim prubehu te vnejsi smycky while se cekani neprovadi a je vybrana dalsi zprava z fronty.

Alespon se mi zatim nestalo, ze by boss ztratil zpravu od vlaken.

[frr]

Za prvé si všimněte (přečtěte si dokumentaci), že při práci s "conditional variable" je potřeba ještě vedle této proměnné taky Mutex. Ten je tam právě proto, aby bylo možné nejenom signalizovat "něco někde se stalo", ale aby čekající vlákno mělo šanci si taky atomicky ověřit z dalších datových struktur, *co* se vlastně stalo. Volání pthread_cond_wait() se ukládá ke spánku a zároveň atomicky odemyká mutex, aby se jím mohla chránit další vlákna.

Souhlasím, že je třeba použít jedinou podmínkovou proměnnou, kterou může polechtat kterýkoli z workerů. A potažmo jediný doprovodný mutex.

Takže workeři budou mít každý nějaký svůj kus dat, který bude za provozu chráněný tímto "společným doprovodným mutexem podmínkové proměnné".

Worker:

pthread_mutex_lock(&mutex);
Worker_Save_Message_To_Boss(); // toto nemusi byt funkce - proste kus kodu
pthread_mutex_unlock(&mutex);
pthread_cond_signal(&boss_cond);
   (... možná že ta poslední dvě volání lze navzájem prohodit... )

Boss:

pthread_mutex_lock(&mutex);
pthread_cond_wait(&boss_cond,&mutex);
Boss_Check_Messages_From_All_Workers(); // toto nemusi byt funkce - proste kus kodu
pthread_mutex_unlock(&mutex);

Jakým konkrétně způsobem budou workeři říkat bossovi, co se přesně stalo, to je čistě na uvážení programátora. Může mít třeba každý worker svůj řádek v nějaké tabulce (poli), ve kterém může signalizovat, že "já něco potřebuju". A mutexem je chráněna manipulace s tabulkou.  Nebo libovolně složitější struktura, třeba nějaký dynamický index per-worker objektů, které mohou být v C++ třeba polymorfní. Nebo může komunikační roli plnit fronta zpráv, jak už jste naznačil - což je mimochodem možná optimální řešení co do nároků na čas procesoru, protože není třeba vyčerpávajícím způsobem kontrolovat všechny workery.

Zajímavým a odděleným problémem pak je, kdo může přidávat/odebírat záznamy v indexu workerů - zda boss, nebo workeři, nebo kdo vlastně, a jak to přesně zařídit, aby nedocházelo k přístupu do objektů, které již byly dealokovány/destruovány... (v případě ukončení workerů). Jistě je možné pojmout to nejjednodušším možným způsobem na úrovni základních operací s POSIXovými vlákny (pthread_join() a spol). 
Upozorňuji taky na možnost vykašlat se na pthread_join(), plodit workery jako "detached" a při ukončení celého programu prostě rozeslat dohodnutý signál a počkat, až workeři sami pochcípou^W^W^W odejdou na svačinu (počítat si někde, kolik je jich naživu, a počkat, až jejich celkový počet klesne na nulu).

Pro svižný chod vícevláknového programu je potřeba, aby se v tom "podmínkovém doprovodném mutexu" žádný worker ani boss dlouho nezamysleli - konkrétně aby se nemohlo stát, že se někdo zablokuje v syscallu. V kritické sekci je potřeba řešit jenom opravdu nezbytné věci. Pokud je potřeba řešit něco s rizikem "zamyšlení se", je potřeba si tuto práci odložit na dobu po uvolnění mutexu (třeba odstranit objekt z indexu a podržet pointer přes pomocnou lokální proměnnou). Práce s dynamickými datovými strukturami by měla být ještě zhruba OK - paměťové alokace/dealokace nejsou až tak rizikové (záleží na požadované době odezvy aplikace).

Předávání dat workerům řeším typicky tak, že pro každé vlákno vytvářím instanci nějakého objektu, který pro toto vlákno drží pohromadě všechna relevantní data. Pokud jsou workeři různých typů, tak ten objekt může být polymorfní. Boss vytvoří instanci objektu, přidá ho do nějakého svého globálního indexu, nastartuje workera, kterému předá odkaz na objekt argumentem void* skrz funkci pthread_create() - od toho tam ten argument je. V tu chvíli (v okamžiku startu vlákna) vzniká zajímavý dílčí problém typu "vejce a slepice" - thread ID vznikne teprve ve chvíli, kdy je worker úspěšně nastartován, Boss se ho dozví teprve po návratu funkce pthread_create() - v tu chvíli už se Worker možná rozběhl a někam kus utekl, aniž by v globálním indexu bylo poznamenáno jeho thread ID. Obecně není možno předvídat, zda se po pthread_create() vzbudí dřív Boss nebo zrovna narozený Worker. Proto je vhodné tuto počáteční inicializaci ošetřit opět mutexem. Tato kritická sekce při startu workera je věcně oddělený problém od výše zmíněné "komunikační" podmínkové proměnné a jejího doprovodného mutexu - proto typicky používám pro start workerů oddělený mutex.

 Boss:

tmp_worker = new Worker_obj();
pthread_mutex_lock(&thr_start_mutex);
tmp_worker->ID = pthread_create(..., (void*)tmp_worker);
add_worker_to_index(tmp_worker);
pthread_mutex_unlock(&thr_start_mutex);

Worker:

worker_thr(...)
{
   pthread_mutex_lock(&thr_start_mutex);
   pthread_mutex_unlock(&thr_start_mutex);
   ...
}

Globální index workerů může být třeba map<pthread_t, Worker_obj*>. Takových indexů může být víc. Čímž se dostáváme také k zajímavým zádrhelům v oblasti konstrukce C++ map<> šablon obsahujících pointery v roli cílů i klíčů (pokud je pthread_t nebo jiný porovnávací identifikátor memberem Worker_obj) - dají se použít obalové objektíky, overloadnutý operator<(), náhrada porovnávacího functoru apod...
Teď mě napadl domácí úkol: zkuste se zamyslet, jak by bylo potřeba upravit Bosse, pokud by se funkce add_worker_to_index() volala z konstruktoru Worker_obj - potažmo konstruktor Worker_obj by měl jako povinný argument "pthread_t ID" Vlastně by ta mapa mohla být statickou member proměnnou Worker_obj...
Jojo - je to už delší doba, co jsem si na to naposledy sáhl :->

Ještě jednu poznámku k tomu: když už máme datový objekt, který odpovídá konkrétnímu běžícímu posixovému vláknu, proč to nezařídit tak, aby vlákno bylo member metodou? Aby si mohlo šahat přímo do member proměnných svého objektu přes implicitní "this."... Tedy proč ne: protože posixová funkce pthread_create() je určena pro holé Cčko, a není si vědoma nějakého "this". "Datový vozík vlákna" se dá předat jedině skrz "void* data" poslední argument. Ale je tu samozřejmý způsob, jak tuto "neschopnost" pthred_create() obejít: nadefinovat jednoduchou obalovou funkci (možno jako statickou metodu), která se spustí skrz pthread_create() a jediné co udělá je, že spustí "výkonná střeva", deklarovaná již v podobě klasické member metody. Tj.

// static
void* Worker_obj::worker_thr(void* data)
{
    return ((Worker_obj*)data)->do_the_job();
}

Potažmo u polymorfních workerů se nabízí, aby Boss měl jenom jednu spouštěcí smyčku a rozlišení podtypů workerů se dělo různými variantami virtuální metody do_the_job()...

Jasně, pro použití posixových vláken jsou v C++ hotové knihovny... ale to je nuda, ne?

[Tomáš Skočdopole]

Dobry den,

mnohokrat dekuji za reakci, je pro me nesmirne poucna. Tu komunikaci se ted snazim resit pomoci FIFO front nejak takto: std::queue<pthread_t, std::queue<tMessage> > MessagesQueues.

Jeste bych se chtel zeptat, zda je korektni volat v hlavnim vlakne programu (main fce) pthread_self a pouzivat ziskanou hodnotu pro identifikaci vlakna Bosse?

Dekuji

[frr]

 

Díky za hřejivou reakci :-)

std::queue<pthread_t, std::queue<tMessage> >

 

To je na mě dost silné kafe Tuhle stezku bludištěm jsem nezkoumal, neporadím... V C++ jsem jenom mírně pokročilý.

 

Jeste bych se chtel zeptat, zda je korektni volat v hlavnim vlakne programu (main fce) pthread_self a pouzivat ziskanou hodnotu pro identifikaci vlakna Bosse?

Jako pro posílání signálů? No jasně, jak taky jinak

 

[Tomáš Skočdopole]

 

 

Jeste bych se chtel zeptat, zda je korektni volat v hlavnim vlakne programu (main fce) pthread_self a pouzivat ziskanou hodnotu pro identifikaci vlakna Bosse?

Jako pro posílání signálů? No jasně, jak taky jinak  

Aha, tak to jsem nevedel, ja myslel ze obycejny program (jednovlaknovy) nema svoje thread ID (pthread_t), pouze PID.

 

Kazdopadne opet dekuji za reakci. Tomas

[Michal Kubeček]

"Obyčejný" jednovláknový program je prostě program, který má jedno vlákno. :-)

Ono by asi bylo dost nepraktické (pro autora implementace i pro autory programů) dělat to tak, že by hlavní vlákno procesu (to, které vznikne při jeho vytvoření) nemělo thread ID a ostatní ano.

[frr]

Přemejšlím, jestli má smysl bavit se o pthread_t jediného vlákna jednovláknového programu, pokud tento byl zkompilován bez #include <pthread.h> a nebyl slinkován s libpthread... To je u mě spíš program bez podpory vláken Ale je možné, že v kernelu příslušné datové struktury běžícího procesu tuto podporu beztak obsahují.

[Michal Kubeček]

IMHO to thread ID mít pořád bude, jen v tom programu nepůjde zjistit a i kdyby šlo zjistit, stejně nepůjde k ničemu rozumnému použít. Taková Schrödingerova kočka… :-)

[frr]

 

std::queue<pthread_t, std::queue<tMessage> >

Tenhle zádrhel mi nějak nedává spát

 

Zaprvé jsem měl vždycky pocit, že std::queue<něco> má jenom jeden argument, a to je typ objektu, který se do té FIFO fronty vkládá. Ale protože věci vždycky nejsou opravdu tak, jak na první pohled vypadají, kouknul jsem se na svém linuxu do

/usr/include/c++/4.1.1/bits/stl_queue.h

a uzřel jsem, že ty dva argumenty máte v zásadě správně, protože celý šablonový prototyp v této konkrétní implementaci vypadá následovně:

  template<typename _Tp, typename _Sequence = deque<_Tp> >
    class queue;

Tj. druhý argument je defaultem nastaven na deque<něco>. Šablona queue je v G++ STL jakýsi tenký wrapper okolo šablony deque (Double Ended Queue) - tj. jako obousměrné FIFO, což už je "elementární" typ G++ STL.

Nicméně jsem nepronikl tak daleko, abych pochopil, k čemu je dobré mít ve vnější frontě jako prvek pthread_t, a v "zabalené" interní frontě prvek tMessage (nějaká Vaše zpráva, pokud tomu správně rozumím). Tj. ten prvek ve frontě je navenek jiný než uvnitř... Navíc ta "zabalená" interní fronta je opět typu "queue", tj. expanduje to na

std::queue<pthread_t, std::queue<tMessage, std::deque<tMessage> > >

pokud to správně píšu. K čemu je *tohle* dobré, to ví jenom pánbůh...

Nemyslel jste to takhle?

std::map<pthread_t, std::queue<tMessage> >

[Tomáš Skočdopole]

Nemyslel jste to takhle?

std::map<pthread_t, std::queue<tMessage> >

Ano, presne tak jsem to myslel, dekuji za opravu

[frr]

Jasně. V tom případě už mám jenom jednu poznámku: všechny tyhle kontejnery jako queue nebo map pracují zásadně stylem "pass by value". Tj. při vkládání hodnot toho typu, který jste deklaroval, ty hodnoty kopírují (nedělají si pointery nebo reference na předloženou instanci). Tj. výše uvedená deklarace znamená, že při vkládání hodnot nejdřív uděláte pair<pthread_t, queue<tMessage>>  (takže se volá konstruktor pro intermediate instanci queue<tMessage>) a tento pár následně vložíte do mapy, takže se volá copy-constructor pro tento pair, tj. taky copy-constructor pro queue, která je memberem toho pairu... (hrozná čeština, já vim). Ještě že v té čerstvé frontě zatím nejsou žádné zprávy, kopírovaly by se taky. A pozor při odkazování na ty zprávy ve frontě v mapě, abyste se nevědomky nedopustil copy-konstrukce, a pak neprováděl úpravy té fronty na *kopii* s úmyslem měnit *originál*... No při správné práci s iterátory by se to stávat nemělo - přesto mi tento přístup moc pod nos nejde

Osobně mám při práci se složitějšími objekty "ukládanými do šablon" nutkání používat pointery - aby se při těch frontových a indexových operacích kopírovaly jenom pointery, ne celé instance složitých objektů nastojato. Tohle moje nutkání je asi projevem mého Cčkového zpátečnictví (Cčko je takový lepší asembler) = projevem mého začátečnictví v C++. Jasně, znamená to, že si pak musím nějak "po svém" zařídit košer dealokaci těch odkazovaných objektů, možná na bázi počítání odkazů apod. Zas na druhou stranu mi to umožňuje udržovat více indexů (map) podle různých kritérií na tytéž instance tlustých objektů.

Použít pointer v roli "value", to je ještě v klidu. Horší je, pokud jako klíč nemůžete použít nějaký elementární Cčkový typ (třeba pthread_t je dodnes ve skutečnosti integer), ale potřebujete použít nějaké složitější třídící kritérium. Pak má smysl vložit jako "key" do mapy jenom lehoučký "meziobjekt", který bude mít nadefinovanou správnou porovnávací funkci operator<(), a třeba bude držet jenom pointer na skutečný tlustý objekt v pozadí.

[Tomáš Skočdopole]

Jeste bych se chtel zeptat ohledne pthread_cond_timedwait.

Dle meho ocekavani (a dle man) to funguje tak, ze pthread_cond_timedwait se "uvolni" bud po vyprseni timeoutu a nebo po prijeti signal nebo broadcast.

Ovsem ve skutecnosti se tomu tak deje pouze po vyprseni timeoutu (v podstate se to chova jako ekvivalent k sleep(TIMEOUT)).

Tak bych se chtel zeptat, zda mam ja neco blbe, nebo zda to tak opravdu ma byt.

 

Ted jeste radeji zkusim napsat maly sample programek, na kterem to vyzkousim.

Tomas

[Tomáš Skočdopole]

Tak problem budu mit asi jinde.

Napsal jsem jednoduchou ukazku:

#include <pthread.h>
#include <stdio.h>
#include <sys/time.h>
#include <errno.h>
#include <unistd.h>

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t  cond  = PTHREAD_COND_INITIALIZER;

int MyAlarm;

void PrintMessage (const char* Message)
{
    printf (Message);
    fflush (stdout);

    return;
}

int WaitForAlarm (void)
{
    int rc;

    struct timespec   ts;
    struct timeval    tp;

    gettimeofday (&tp, NULL);

    ts.tv_sec  = tp.tv_sec;
    ts.tv_nsec = tp.tv_usec * 1000;
    ts.tv_sec += 10;

    pthread_mutex_lock (&mutex);
    while (MyAlarm == 0)
    rc = pthread_cond_timedwait (&cond, &mutex, &ts);
    pthread_mutex_unlock (&mutex);
 
    return (rc);

}

void SetAlarm (void)
{
    pthread_mutex_lock (&mutex);

    MyAlarm = 1;
    pthread_cond_broadcast (&cond);

    pthread_mutex_unlock (&mutex);
 
    return;
}

void *ThreadMain (void* args)
{
    PrintMessage ("Cekam na alarm\n");

    WaitForAlarm ();

    PrintMessage ("Prestavam cekat\n");

    pthread_exit (NULL);
}

int main(int argc, char **argv)
{
    pthread_t thread;

    PrintMessage ("Cekani skonci timeoutem\n");

    MyAlarm = 0;

    pthread_create (&thread, NULL, ThreadMain, NULL);

    sleep (15);

    PrintMessage ("Zapinam alarm\n");
    SetAlarm ();

    pthread_join (thread, NULL);


    PrintMessage ("Cekani skonci podminkou\n");

    MyAlarm = 0;

    pthread_create (&thread, NULL, ThreadMain, NULL);

    sleep (5);

    PrintMessage ("Zapinam alarm\n");
    SetAlarm ();

    pthread_join (thread, NULL);


    return (0);
}


Jedna se o to, ze je vytvoreno vlakno, ktere ceka na alarm max 10 sekund (pomoci cond_timed_wait).

V prvnim je alarm nastaven az za 15s, takze dojde k timeoutu. V druhem pripade je alarm nastaven za 5s cili cekani je ukonceno splnenim podminky

Program funguje jak ma, akorat mi neni jasne, proc v prvni casti hned po vyprseni timeoutu (desata sekunda) do nastaveni alarmu (patnacta sekunda) vytizeni procesoru vzroste na 100 procent.

Mohl bych nekoho z Vas pozadat o pomoc / vysvetleni?

Dekuji mnohokrat. Tomas

[Michal Kubeček]

akorat mi neni jasne, proc v prvni casti hned po vyprseni timeoutu (desata sekunda) do nastaveni alarmu (patnacta sekunda) vytizeni procesoru vzroste na 100 procent.

Protože po vypršení timeoutu se vám funkce pthread_cond_timewait() vrátí okamžitě (máte tam nastavený pořád stejný - a tedy už prošlý - čas) a vy ji pořád dokola voláte v nekonečné smyčce.

[Michal Kubeček]

Tedy samozřejmě ne úplně nekonečné, ta smyčka samozřejmě skončí, ale až po dalších pěti sekundách.