Vlákna v C# - 12. díl

Navážeme na minulý díl a dokončíme problematiku Wait a Pulse.

Fronta producent/spotřebitel

Jednoduchou aplikací, která Wait/Pulse využije, je fronta producent/spotřebitel (tu jsme si ukázali v 5. díle). Nějaký kód, označovaný jako producent, přidává do fronty zadání (typicky na hlavním vlákně), zatímco jeden nebo více spotřebitelů odebírají zadání jedno po druhém a plní je.

V našem příkladu použijeme na reprezentaci úkolu typ string, fronta tedy bude vypadat takto:

Queue<string> taskQ = new Queue<string>();

Protože budeme k frontě přistupovat z několika vláken, musíme všechen kód, který z ní čte nebo do ní zapisuje, obalit do locku. Takhle budeme přidávat úkoly do fronty:

lock (locker)
{
    taskQ.Enqueue("můj úkol");
    Monitor.PulseAll(locker);
}

Upravujeme potenciální blokovací podmínku, takže musíme zavolat Pulse. PulseAll namísto něj voláme, protože můžeme mít více spotřebitelů, tedy více čekajících vláken.

Je žádoucí, aby se spotřebitelé zablokovali, pokud nemají zrovna co dělat (pokud je fronta prázdná), aby zbytečně nezatěžovali počítač. Tento kód dělá přesně to, co chceme – kontroluje počet prvků ve frontě.

lock (locker)
    while (taskQ.Count == 0) Monitor.Wait(locker);

Dalším krokem je, aby spotřebitel mohl odebrat úkol z fronty a splnit ho:

lock (locker)
    while (taskQ.Count == 0) Monitor.Wait(locker);

string task;
lock (locker)
    task = taskQ.Dequeue();

Právě uvedený postup ale není thread-safe. Odstranění z fronty se totiž zakládá na staré informaci z uzamknutí locku. Představte si, co by se stalo, kdybychom spustili dvě spotřebitelská vlákna najednou s jedním předmětem umístěným ve frontě. Mohlo by se stát, že ani jedno by se uvnitř while cyklu nezablokovalo, protože by obě ve stejný moment viděla právě ten jeden předmět. Hned potom by se obě vlákna pokusila úkol z fronty odstranit a právě v této chvíli by došlo k chybě. Abychom tomuto předešli, budeme udržovat lock zamčený o chvilku déle:

string task;
lock (locker)
{
    while (taskQ.Count == 0) Monitor.Wait(locker);
    task = taskQ.Dequeue();
}

Po odstranění úkolu z fronty už nemusíme volat Pulse, protože žádný spotřebitel se neodblokuje jen kvůli tomu, že ubyly úkoly.

Jakmile je úkol z fronty pryč, není nutné pořád udržovat lock. Tím, že ho teď odemkneme, umožníme spotřebiteli provést časově náročný úkol, aniž by při tom blokoval ostatní vlákna.

Ukážeme si kompletní program, který jsme právě skládali dohromady. Stejně jako u verze, kde jsme použili AutoResetEvent v 5. dílu, použijeme null úkol, abychom oznámili spotřebiteli, že je konec. Protože aplikace podporuje více než jednoho spotřebitele, musíme do fronty zařadit odpovídající počet null úkolů, aby se práce ukončila u všech.

using System;
using System.Threading;
using System.Collections.Generic;

public class TaskQueue : IDisposable
{
    object locker = new object();
    Thread[] workers;
    Queue<string> taskQ = new Queue<string>();

    public TaskQueue(int workerCount)
    {
        workers = new Thread[workerCount];

        // Pro každého spotřebitele vytvoříme vlákno
        for (int i = 0; i < workerCount; i++)
            (workers[i] = new Thread(Consume)).Start();
    }

    public void Dispose()
    {
        // Do fronty zařadíme tolik null úkolů, kolik je vláken
        foreach (Thread worker in workers) EnqueueTask(null);
        foreach (Thread worker in workers) worker.Join();
    }

    public void EnqueueTask(string task)
    {
        lock (locker)
        {
            taskQ.Enqueue(task);
            Monitor.PulseAll(locker);
        }
    }

    void Consume()
    {
        while (true)
        {
            string task;
            lock (locker)
            {
                while (taskQ.Count == 0) Monitor.Wait(locker);
                task = taskQ.Dequeue();
            }
            if (task == null) return;   // konec
            Console.WriteLine(task);
            Thread.Sleep(1000); // Simulace časově náročného úkolu
        }
    }
}

A zde je metoda Main, která vše nastartuje, vytvoří dvě spotřebitelská vlákna a deset úkolů, které si mezi sebou rozdělí:

static void Main()
{
    using (TaskQueue q = new TaskQueue(2))
    {
        for (int i = 0; i < 10; i++)
            q.EnqueueTask("Úkol č. " + i);

        Console.WriteLine("Zařazeno 10 úkolů");
        Console.WriteLine("Čekám na splnění úkolů...");
    }

    // Použili jsme using, takže po skončení práce se zavolá Dispose()
    // na spotřebitelská vlákna
    Console.WriteLine("rnVšechny úkoly jsou hotovy!");
}

Zátěž Pulse

Řekneme si, jestli by se nějak dal zrychlit proces pulzování. Nejdříve si zopakujme kód z metody EnqueueTask výše:

lock (locker)
{
    taskQ.Enqueue(task);
    Monitor.PulseAll(locker);
}

Teoreticky bychom mohli volat PulseAll jen tehdy, je-li vůbec možné nějaké vlákno odblokovat:

lock (locker)
{
    taskQ.Enqueue(task);
    if (taskQ.Count <= workers.Length) Monitor.PulseAll(locker);
}

Ale pozor, moc bychom neušetřili (skoro vůbec nic), vzhledem k tomu, že už pulzování samotné je záležitost necelé mikrosekundy. Naopak bychom si mohli přitížit, podívejte se totiž na následující kód (a najděte rozdíl):

lock (locker)
{
    taskQ.Enqueue(task);
    if (taskQ.Count < workers.Length) Monitor.PulseAll(locker);
}

Přesně to je snad ten nejhorší typ bugů. Kompilátor chybu nenahlásí, vše klidně až do poslední chvíle funguje, jak má a kdo by pak v tom všem kódu hledal chybějící rovnítko?

Pulse nebo PulseAll?

Tady přichází na řadu další potencionální zúspornění kódu. Po zařazení úkolu do fronty bychom mohli zavolat namísto PulseAll „obyčejné“ Pulse a nic by se nestalo.

Zopakujme si rozdíly: když zavoláte Pulse, může se probudit k životu maximálně jedno vlákno. Pokud použijeme PulseAll, probudí se všechna. Když přidáváme do fronty jen jeden úkol, pouze jeden spotřebitel ho může zpracovat, takže nám stačí probudit jednoho pracovníka pomocí Pulse.

V příkladu níže máme jen dvě spotřebitelská vlákna, takže výkonnostní rozdíl mezi Pulse a PulseAll bude minimální. Pokud bychom ale měli takových vláken deset, bylo by o trošku výkonnostně výhodnější použít Pulse (i když ho volat desetkrát) než PulseAll:

lock (locker)
{
    taskQ.Enqueue("úkol 1");
    taskQ.Enqueue("úkol 2");
    Monitor.Pulse(locker);
    Monitor.Pulse(locker);
}

Cenou za rychlejší volání je zaseknutí pracovního vlákna. To je zase jeden z těžko odhalitelných bugů, projeví se až když je spotřebitel ve stavu Waiting. Dalo by se říct, že pokud jste na pochybách, jestli by k takovému bugu nemohlo dojít i ve vaší aplikaci, tak používejte PulseAll. Ztrátu výkonu nejspíš ani nezaznamenáte.

Použití timeoutu při Wait

Může se stát, že nebude vhodné zavolat Pulse hned, jakmile se splní blokovací podmínka. Příkladem takové situace je metoda, která získává informace pravidelným dotazováním se databáze. Pokud nám nevadí menší prodlevy, je řešení jednoduché – metodě Wait přidáme parametr timeout.

lock (locker)
{
    while (!podminka)
    Monitor.Wait(locker, timeout);
}

Tento postup přikáže znovu zkontrolovat platnost podmínky podle zadaného času a pak ještě jednou po obdržení pulzu. Čím je podmínka jednodušší, tím může být timeout menší při zachování účinnosti.

Timeouty se hodí i pokud by se mohlo stát, že se nebude moci zavolat Pulse (ať už kvůli bugu, je-li synchronizace složitá, nebo z jiného důvodu). Podmínka se pak zkontroluje i bez pulzování a aplikace může pokračovat.

Lock race a co s ním

Chceme-li zasignalizovat pracovní vlákno pětkrát za sebou, mohli bychom použít následující kód:

class Race
{
    static object locker = new object();
    static bool go;

    static void Main()
    {
        new Thread(SaySomething).Start();

        for (int i = 0; i < 5; i++)
        {
            lock (locker) { go = true; Monitor.Pulse(locker); }
        }
    }

    static void SaySomething()
    {
        for (int i = 0; i < 5; i++)
        {
            lock (locker)
            {
                while (!go) Monitor.Wait(locker);
                go = false;
            }
            Console.WriteLine("Co je?");
        }
    }
}

Očekávaný výstup by byla pětkrát otázka: „Co je?“. Ale to se nestane, místo toho se napíše jen jednou!

Kód je totiž vadný. Cyklus for v metodě Main může projet svých pět iterací, když ještě pracovní vlákno nedrží zámek nad objektem locker. Možná se projde cyklem ještě dřív, než se vůbec stihne pracovní vlákno nastartovat! Příklad v kapitole Fronta producent/spotřebitel netrpěl tímhle neduhem, protože pokud se hlavní vlákno dostalo v práci před pracovní vlákno, každý požadavek se jednoduše zařadil do fronty. Ale v tomto případě musíme hlavní vlákno zablokovat v každé iteraci, pokud je pracovní vlákno pořád zaměstnáno svým předešlým úkolem.

Jednoduchým řešením je, aby hlavní vlákno po každé iteraci počkalo, dokud nebude proměnná go nastavena na false pracovním vláknem. Kvůli této změně musíme volat i Pulse.

class Acknowledged
{
    static object locker = new object();
    static bool go;

    static void Main()
    {
        new Thread(SaySomething).Start();

        for (int i = 0; i < 5; i++)
        {
            lock (locker) { go = true; Monitor.Pulse(locker); }
            lock (locker) { while (go) Monitor.Wait(locker); }
        }
    }

    static void SaySomething()
    {
        for (int i = 0; i < 5; i++)
        {
            lock (locker)
            {
                while (!go) Monitor.Wait(locker);
                go = false; Monitor.Pulse(locker);
            }
            Console.WriteLine("Co je?");
        }
    }
}

Důležitou vlastností tohoto programu je to, že pracovní vlákno uvolní zámek, než se vrhne na svůj (možná dlouhý) úkol (tady je úkolem zavolání Console.WriteLine).

V našem příkladu jen jedno vlákno (hlavní) signalizuje pracovnímu vláknu, aby pracovalo. Pokud bychom ale měli taková vlákna dvě a obě by volala kód podobný tomu v metodě Main, mohl by se následující řádek kódu zavolat dvakrát za sebou.

lock (locker) { go = true; Monitor.Pulse(locker); }

Výsledkem by bylo, že by druhá signalizace neměla žádný účinek, pokud by pracovní vlákno mezitím nestihlo svojí úlohu dokončit. Tohle můžeme ošetřit dvojicí proměnných – „ready“ a „go“. Ready indikuje, že je pracovní vlákno připraveno přijmout další zadání, zatímco go znamená příkaz k práci stejně jako předtím. Je to analogické k dřívějšímu příkladu se dvěma AutoResetEventy, jen flexibilnější. Takhle tedy bude vypadat kód:

public class Acknowledged
{
    object locker = new object();
    bool ready;
    bool go;

    public void NotifyWhenReady()
    {
        lock (locker)
        {
            // Čekat, pokud má pracovní vlákno práci
            while (!ready) Monitor.Wait(locker);
            ready = false;
            go = true;
            Monitor.PulseAll(locker);
        }
    }

    public void AcknowledgedWait()
    {
        // Řekneme, že jsme připraveni na další úkol
        lock (locker) { ready = true; Monitor.Pulse(locker); }

        lock (locker)
        {
            while (!go) Monitor.Wait(locker);      // Počkáme na "go" signál
            go = false; Monitor.PulseAll(locker);  // Přenastavíme "go", zapulzujeme
        }

        Console.WriteLine("Co je?");   // Splníme úkol
    }
}

Pro demonstraci použijeme dvě vlákna, každé z nich pošle pětkrát signál pracovnímu vláknu. Mezitím hlavní vlákno čeká na deset zpráv.

public class Test
{
    static Acknowledged a = new Acknowledged();

    static void Main()
    {
        new Thread(Notify5).Start();     // Spustíme
        new Thread(Notify5).Start();     // dvě vlákna
        Wait10();   // ... a jednoho čekatele.
        Console.ReadKey();
    }

    static void Notify5()
    {
        for (int i = 0; i < 5; i++)
            a.NotifyWhenReady();
    }

    static void Wait10()
    {
        for (int i = 0; i < 10; i++)
            a.AcknowledgedWait();
    }
}

V metodě NotifyWhenReady je proměnná ready nastavena na false před opuštěním zámku a na tom staví celý příklad! Zabraňuje to totiž dvěma vláknům poslat signál za sebou bez zkontrolování hodnoty ready. Pro zjednodušení ve stejném zámku nastavíme i proměnnou go a zavoláme PulseAll, i když bychom mohli tyto dvě věci přesunout do jiného locku a nic by se nestalo.

Simulace třídy WaitHandle

V předchozím kódu jste si mohli všimnout, že oba while cykly mají takovouhle strukturu:

lock (locker)
{
    // Čekat, pokud má pracovní vlákno práci
    while (!promenna) Monitor.Wait(locker);
    promenna = false;
    ...

Přičemž „promenna“ je nastavena na true v jiném vlákně. Tato struktura kódu kopíruje AutoResetEvent, třídu odvozenou od WaitHandle. Pokud bychom přehlédli kód "promenna = false;", dostali bychom ManualResetEvent. A pokud bychom místo boolean proměnné použili int, dostali bychom Semaphore. Ve skutečnosti jediná třída, jejíž funkci nemůžeme pomocí Wait a Pulse okopírovat, je Mutex, která dělá to samé co lock.

Simulovat statické metody, které fungují napříč několika Wait Handly je většinou jednoduché. Ekvivalentem k zavolání WaitAll na několik EventWaitHandlů není ve skutečnosti nic jiného než blokovací podmínka, která použije proměnné namísto Wait Handlů:

lock (locker) 
{
    while (!promenna1 && !promenna2 && !promenna3...) Monitor.Wait (locker);

Tento kód může být někdy užitečný, protože WaitAll je často nepoužitelný kvůli jeho COM historii. Simulování WaitAny je stejně jednoduché, jen nahraďte && operátorem ||.

SignalAndWait je už zapeklitější. Vzpomeňte si, že tato metoda signalizuje jeden handle, zatímco čeká na jiný, v jedné atomické operaci. Předpokládejme, že chceme poslat signál proměnné A, zatímco čekáme na proměnnou B. Museli bychom každou proměnnou rozdělit do dvou, ve výsledku by kód mohl vypadat následovně:

lock (locker) 
{
    Acast1 = true;
    Monitor.Pulse (locker);
    while (!Bcast1) Monitor.Wait (locker);

    Acast2 = true;
    Monitor.Pulse (locker);
    while (!Bcast2) Monitor.Wait (locker);
}

Setkávání vláken pomocí Wait a Pulse

Tak jako WaitHandle.SignalAndWait může být použito pro setkání dvou vláken v jednom bodě, můžeme využít i Wait a Pulse. V příkladu, na který se podíváme za chvíli, bychom mohli říct, že simulujeme dva ManualResetEventy (jinými slovy: definujeme dvě boolean proměnné) a pak nastavujeme jednu proměnnou, zatímco čekáme na tu druhou. V tomto případě nepotřebujeme atomický přístup k věci, takže se vyhneme rozdělení „jedné proměnné na dvě“. Dokud budeme nastavovat proměnnou na true a volat Wait v jednom locku, bude vše probíhat tak, jak má:

class Rendezvous
{
    static object locker = new object();
    static bool signal1, signal2;

    static void Main()
    {
        // Každé vlákno uspíme na náhodný čas
        Random r = new Random();
        new Thread(Mate).Start(r.Next(10000));
        Thread.Sleep(r.Next(10000));

        lock (locker)
        {
            signal1 = true;
            Monitor.Pulse(locker);
            while (!signal2) Monitor.Wait(locker);
        }
        Console.Write("Hej! ");
    }

    // Toto je voláno přes ParameterizedThreadStart
    static void Mate(object delay)
    {
        Thread.Sleep((int)delay);
        lock (locker)
        {
            signal2 = true;
            Monitor.Pulse(locker);
            while (!signal1) Monitor.Wait(locker);
        }
        Console.Write("Hej! ");
    }
}

Ve stejnou chvíli se na obrazovku vypíše dvakrát „Hej!“ – vlákna se úspěšně setkala.

Wait a Pulse vs. WaitHandle

Protože jsou Wait a Pulse rozhodně nejflexibilnější ze všech synchronizačních konstrukcí, najdou využití téměř v každé situaci. Přesto mají WaitHandle konstrukce dvě výhody:

• Mají schopnost pracovat napříč procesy.
• Jsou jednodušší na pochopení a těžší na „rozbití“.

Co se výkonu týče, pokud s Wait a Pulse postupujeme podle vzoru, který je:

lock (locker)
    while ( blokovací podmínka ) Monitor.Wait (locker);

Je to kvůli zamykání a znovuodemykání zámku, WaitHandle.WaitOne je prostě o trošičku rychlejší.

Pokud vezmeme v úvahu různé procesory, operační systémy, verze CLR a samotnou logiku aplikací, dostaneme se k rozdílu maximálně pár mikrosekund mezi Wait/Pulse a WaitHandle. Tento rozdíl téměř určitě žádnou škodu nenadělá, proto si můžete vybrat vhodný nástroj podle aktuální situace.

Blahopřeji, že jste to skrz tuto náročnou kapitolu zvládli až na její konec. Stejně jako problematika metod Pulse a Wait končí i dnešní díl. Uvidíme se příště u posledního dílu!