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

V tomto dílu si povíme o thread poolingu (česky také jako fond vláken) a navazujícím tématu – asynchronních delegátech.

Hned na úvod se zmíním o překladu pojmu „thread pool“. Většinou se setkáme s významem „fond vláken“, od toho odvozené sloveso „thread pooling“ by mohlo znamenat něco jako „shromažďování vláken“. Jak se totiž vzápětí dozvíte, „fond vláken“ je několik dohromady spojených vláken. V článku se ale raději budu držet anglického názvu - thread poolingu.

Teď už ale k využití thread poolingu. Pokud ve své aplikaci používáte hodně vláken, která většinu svého života tráví zablokovaná pomocí Wait Handle, můžete snížit použité systémové prostředky právě pomocí thread pooolingu, jež spojí několik vláken do menšího počtu.

K použití thread poolu musíte zaregistrovat Wait Handle společně s metodou, která se zavolá, když dojde k signalizaci Wait Handlu. Toho docílíme pomocí ThreadPool.RegisterWaitForSingleObject, jako v tomto příkladu:

class Test
{
    static ManualResetEvent starter = new ManualResetEvent(false);

    public static void Main()
    {
        ThreadPool.RegisterWaitForSingleObject(starter, Go, "ahoj", -1, true);
        Thread.Sleep(2000);
        Console.WriteLine("Signalizace pracovnímu vláknu...");
        starter.Set();
        Console.ReadLine();
    }

    public static void Go(object data, bool timedOut)
    {
        Console.WriteLine("Započato - " + data);
        // Provedení práce
    }
}

Výsledkem bude dvouvteřinová pauza, vypsání věty „Signalizace pracovnímu vláknu“ a „Započato – ahoj“.

Popíšeme si jednotlivé parametry metody. Podle MSDN má metoda tento předpis:

public static RegisteredWaitHandle RegisterWaitForSingleObject(WaitHandle waitObject,
    WaitOrTimerCallback callBack,
    Object state,
    int millisecondsTimeOutInterval,
    bool executeOnlyOnce)

Teď následuje samotný popis parametrů:

• waitObject – Jakýkoliv WaitHandle kromě Mutexu.
• callBack – Přijímá delegát typu Threading.WaitOrTimerCallback, jež zastupuje metodu, která se má zavolat, když dojde k signalizaci waitObjectu.
• state – Objekt předaný metodě. Může to být cokoliv, následně se předá jako parametr delegované metodě (takže text „ahoj“ se předá jako parametr „data“ metodě „Go“) vlastně stejně jako u ParametrizedThreadStart (vzpomínáte na první díl?).
• milisecondsTimeOutInterval - Časový údaj v milisekundách. Má stejnou funkci, jako všechny ostatní timeouty (hodnota -1 znamená žádný časový limit).
• executeOnlyOnce – Pokud je true, znamená to, že vlákno nebude čekat na waitObject, jestliže už došlo k zavolání delegátu callBack. Naopak false indikuje, že bude operace probíhat pořád dokola, dokud neodregistrujete waitObject

Všechna vlákna ve fondu vláken pracují na pozadí (opakování: automaticky se zruší, jestliže přestanou existovat všechna vlákna běžící na popředí). Ovšem pokud bychom chtěli, aby se před ukončením aplikace nejdříve dokončila nějaká práce na vláknu ve fondu, zavolat metodu Join (jako bychom to udělali v běžné situaci - viz 2. díl) by nebylo řešením. Vlákna ve fondu totiž nikdy ve skutečnosti přirozeně neskončí! Místo toho se „recyklují“ (nefungují, ale nezmizí z paměti), přestanou existovat jedině tehdy, pokud přestane existovat i nadřízený proces. Takže abychom zjistili, jestli už vlákno v thread poolu dokončilo svoji práci, museli bychom odeslat nějaký signál, třeba pomocí jiného Wait Handlu.

Můžeme použít thread pool i bez Wait Handlu, a to pomocí metody QueueUserWorkItem, které předáme delegát, jež se má okamžitě zavolat. Sice si tím znemožníte sdílení jednotlivých vláken pro několik prací, ale tento postup má i jednu výhodu. Thread pool si kontroluje celkový počet vláken (výchozí počet je 25) a automaticky vytváří frontu, pokud vznikne více úloh, než je vláken. V následujícím příkladu máme 100 úloh a 25 z nich probíhá v jeden okamžik. Primární vlákno pak pomocí metod Wait a Pulse čeká, dokud všechna pracovní vlákna nedodělají zadání.

class Test
{
    static object workerLocker = new object();
    static int runningWorkers = 100;

    public static void Main()
    {
        for (int i = 0; i < runningWorkers; i++)
        {
            ThreadPool.QueueUserWorkItem(Go, i);
        }
        Console.WriteLine("Čekám na pracovní vlákna, až dodělají práci...");
        lock (workerLocker)
        {
            while (runningWorkers > 0) Monitor.Wait(workerLocker);
        }
        Console.WriteLine("Hotovo!");
        Console.ReadLine();
    }

    public static void Go(object instance)
    {
        Console.WriteLine("Započato: " + instance);
        Thread.Sleep(1000);
        Console.WriteLine("Ukončeno: " + instance);
        lock (workerLocker)
        {
            runningWorkers--; Monitor.Pulse(workerLocker);
        }
    }
}

Na kódu by nemělo být co k nepochopení. Koneckonců, naprostou většinu jsme už dříve probrali.

Pokud bychom chtěli cílové metodě (v příkladu je to metoda Go) předat víc než jen jeden object parametr, máme několik možností. Můžeme použít anonymní metody. Kdyby metoda Go přijímala dva parametry typu int, mohli bychom delegát vytvořit takto:

ThreadPool.QueueUserWorkItem (delegate (object notUsed) { Go (23,34); });

Druhým způsob, jak se dostat do thread poolu, je přes asynchronní delegáty, na které se právě podíváme v následující kapitole.

Asynchronní delegáty

V prvním díle jsme si ukázali, jak předat data vláknu pomocí ParametrizedThreadStart. Někdy potřebujeme udělat opak – získat data od vlákna, jakmile dokončí svou práci. Asynchronní delegáty jsou pro toto nanejvýš vhodné, dovolují totiž předat libovolný počet argumentů v obou směrech. Navíc výjimky, které vzniknou na asynchronním delegátu jsou předány zpět volajícímu vláknu, to nám zajišťuje snadnější ošetření. Další výhoda byla zmíněna na konci předchozí kapitoly - díky asynchronním delegátům se můžeme dostat do thread poolu.

Nic není zadarmo, a tak i tady musíme zaplatit nějakou cenu. Tím je samotný asynchronní model, který je logicky o něco komplikovanější. Abyste lépe pochopili, o čem teď mluvím, ukážeme si jeden příklad vyřešený jak synchronně, tak asynchronně. Pokud bychom třeba chtěli porovnat obsah dvou stránek, jako první řešení by nás napadlo asi toto:

static void ComparePages()
{
    WebClient wc = new WebClient();
    string s1 = wc.DownloadString("http://www.programujte.com");
    string s2 = wc.DownloadString("http://chrasty.cz");
    // Na provedení chvíli počkáme...
    Console.WriteLine(s1 == s2 ? "Jsou stejné" : "Liší se");
}

Samozřejmě by bylo rychlejší stáhnout obě stránky naráz. Ale jak na to, když se další příkaz zavolá, až jakmile se dokončí ten předchozí? Ideální by bylo, kdyby to šlo následovně:

1. Zavoláme DownloadString.
2. Zatímco pracuje, budeme vykonávat jinou operaci, třeba stahování jiné stránky.
3. Řekneme si metodě DownloadString o výsledky.

Třetí krok je místo, kde jsou asynchronní delegáty užitečné. Volající metoda se setká s pracující metodou, společně vrátí nějaký výsledek a zároveň znovu vyhodí výjimky (pokud k nim v průběhu práce došlo), díky tomu je můžeme na tomto místě snadno ošetřit. Bez třetího bodu by se použití asynchronních delegátů nijak nelišilo od využití „obyčejného“ multithreadingu. Následující kód řeší stejný problém jako nahoře, jen asynchronně:

delegate string DownloadString(string uri);

static void ComparePages()
{
    // Vytvoříme instance delegátu DownloadString
    DownloadString download1 = new WebClient().DownloadString;
    DownloadString download2 = new WebClient().DownloadString;

    // Začneme stahovat
    IAsyncResult cookie1 = download1.BeginInvoke("http://programujte.com", null, null);
    IAsyncResult cookie2 = download2.BeginInvoke("http://www.programujte.com", null, null);

    // Získáme výsledky stahování, pokud je nutné, počkáme na dopočítání
    // Zde také dojde k vyhození výjimek
    string s1 = download1.EndInvoke(cookie1);
    string s2 = download2.EndInvoke(cookie2);

    Console.WriteLine(s1 == s2 ? "Jsou stejné" : "Liší se");
}

Na začátku deklarujeme a vytvoříme instance delegátu DownloadString pro metody, které chceme spustit asynchronně. V tomto případě potřebujeme dvě instance kvůli dvěma souborům, které chceme stáhnout.

Pak zavoláme BeginInvoke. To provede daný delegát a okamžitě vrátí kontrolu nad aplikací nám. Metodě BeginInvoke musíme předat tři parametry: cestu k souboru (stránce, která se má stáhnout), nepovinný callback (tedy metodu, která se má zavolat při volání delegátu) a jako třetí parametr můžeme zadat cokoliv – je typu object. Posledním dvěma se často nastavuje hodnota null, nejen v příkladu nahoře, protože většinou nejsou potřeba. Metoda BeginInvoke vrací objekt typu IASynchResult, jenž využijeme zároveň jako cookie (tedy nějaký vzorek dat) při volání EndInvoke. Objekt IASynchResult zároveň disponuje vlastností IsCompleted, díky které můžeme monitorovat průběh stahování.

Následně zavoláme zmiňovanou metodu EndInvoke s „cookie“ parametrem na delegáty, abychom získali výsledky. Pokud je to nutné, EndInvoke počká, dokud jeho metoda nedokončí práci, pak vrátí její výslednou hodnotu. Typ vrácených dat jsme nastavili v hlavičce delegátu (v našem případě tedy bude hodnota typu string, takže ji musíme uložit do proměnné typu string).

Na závěr se vyhodí případné výjimky, ke kterým došlo během asynchronního volání, a zde je můžeme jednoduše ošetřit (vím, tuto vymoženost zmiňuji už potřetí).

Pokud metoda, kterou spouštíme asynchronně, nemá žádnou návratovou hodnotu, teoreticky nemusíme volat EndInvoke. Pak ale budeme muset případné výjimky ošetřit už na pracovním vláknu.

Asynchronní metody

Některé .NET typy poskytují asynchronní verze svých metod, typicky jejich název začíná na „Begin“ a „End“. Těmto metodám se říká asynchronní metody a mají podobné signatury jako asynchronní delegáty, ale používáme je pro řešení mnohem složitějšího problému – abychom mohli najednou provádět více operací, než máme k dispozici vláken. Například takový TCP socket server dokáže zpracovávat stovky všech možných požadavků a dotazů najednou, pokud použijeme metody NetworkStream.BeginRead a NetworkStream.BeginWrite, a přitom disponuje třeba jen několika vlákny v thread poolu.

Jestliže ale nejste v takto extrémní situaci, měli byste se použití asynchronních metod vyhnout hned z několika důvodů:

• Na rozdíl od asynchronních delegátů nemusí async. metody vždy běžet paralelně s tím, kdo je zavolal.
• Kód se za chvíli stane velmi složitým (jen si představte: synchronizace jednotlivých požadavků, zpracování, …), až se může stát, že výhoda async. metod za chvíli vymizí úplně.

Pokud vám jde jen o prosté paralelní spouštění metod, raději byste měli používat synchronní verze těchto metod pomocí asynchronních delegátů nebo použít BackgroundWorker anebo prostě vytvořit nové vlákno.

Asynchronní události

S asynchronními metodami se setkáme i zde, ve spojitosti s asynchronními událostmi. Z běžného programování znáte typickou dvojici – událost a metodu, která se zavolá, když dojde k události. To samé existuje i v asynchronních verzích, podle konvencí končí název takové metody na „Async“ a název události na „Completed“. Můžeme se s tím setkat například ve třídě WebClient, která definuje metodu DownloadStringAsync. Využití je následující: nejdříve zpracujeme událost DownloadStringCompleted, pak zavoláme DownloadStringAsync. Jakmile ta dokončí svou práci, zavolá se obsah zpracovatele události DownloadStringCompleted.

Máme k dispozici i několik událostí pro zpravování uživatele o průběhu nebo o zrušení akce, možná si pamatujete, že tyhle různé „Async“ a „Completed“ věci jsme viděli už v BackgroundWorkeru. Nebylo to nic jiného než asynchronní události.

To by pro dnešek stačilo. Ukázali jsme si několik asynchronních koutů multithreadingu a příště nás čeká třída Timer, Local Storage a úvod do neblokujících synchronizačních konstrukcí.