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

Tentokrát se zaměříme na třídy EventWaitHandle, Mutex a Semaphore.

Dnes to možná vypadá na krátký díl, ale zdání klame. Tento díl je zatím nejdelší a možná nejdelší, který v seriálu bude. Všechny tři třídy toho mají spolu hodně společného, proto jsem je raději dal do jednoho článku, délky se nebojte, uvidíte, že to není nic složitého.

Konstrukce lock (tedy Monitor.Enter a Monitor.Exit) je jen jedním z několika způsobů pro synchronizaci vláken. Locking je vhodný pro zajištění exkluzivního přístupu k nějakým resources nebo sekcím kódu, ale existují i způsoby, jak synchronizovat bez exkluzivního přístupu, a těmi se dnes budeme zabývat.

Win32 API má bohatou paletu různých synchronizačních schopností a několik z nich převzal i .NET Framework v podobě tříd EventEwaitHandle, Mutex a Semaphore. Navzájem se od sebe docela liší, například Mutex je jen hodně rozšířený lock, zatímco EventWaitHandle poskytuje unikátní funkcionalitu.

Všechny tři třídy jsou odvozené od abstraktní třídy WaitHandle a mají jednu společnou věc. Instancím můžeme přiřadit jméno a podle něj se identifikují mezi ostatními procesy. Co se tím myslí, si vysvětlíme za chvilku.

EventWaitHandle má dvě podtřídy: AutoResetEvent a ManualResetEvent (pozor, název mate, ani jedna nemá nic společného s událostmi (event), nebo delegáty). Rozdíl mezi těmito dvěma třídami je v tom, že každá volá konstruktor své bázové třídy EventWaitHandle s jiným argumentem.

Co se týče výkonu, vše za normálních okolností probíhá v jednotkách mikrosekund.

AutoResetEvent je nejužitečnější ze všech tříd odvozených od WaitHandle, společně s konstrukcí lock tvoří základy synchronizace.

AutoResetEvent

AutoResetEvent se svým chováním podobná třeba nějakému turniketu v metru. Strčíte tam lístek a ono to pustí jednoho člověka skrz. „Auto“ v názvu třídy odkazuje na skutečnost, že tento turniket se automaticky zavře, jakmile člověk projde. Vláknu přikážeme, aby u turniketu počkalo, pomocí metody WaitOne, a aby vložilo pomyslný lístek zase pomocí metody Set. Pokud více vláken zavolá WaitOne, za turniketem se vytvoří fronta. Jakékoliv nezablokované vlákno může zavolat metodu Set, ale vždy dojde k puštění prvního vlákna ve frontě (nebudeme se přece předbíhat :-)).

Pokud dojde k zavolání Set ve chvíli, kdy žádné vlákno ve frontě nečeká, tato metoda počká a spustí se ve chvíli, kdy dojde k zavolání WaitOne (tím se jejich efekt okamžitě „vykrátí“ a vlákno projde skrz turniket bez čekání). Ale pozor, tento efekt se nesčítá! I když zavoláte Set desetkrát, tak to neznamená, že bude puštěno bez čekání 10 vláken, vždy se to týká jen jednoho a ostatní volňásky „propadnou“.

WaitOne přijímá jeden nepovinný parametr související s časovým limitem. Metoda pak vrátí false, pokud by čas vypršel dřív, než došlo k zavolání Set.

Další z metod ve třídě AutoResetEvent je metoda Reset, to je právě ta, která je automaticky volána a vyresetuje nastavení „turniketu“ na výchozí hodnoty (zavřeno, není vložen žádný lístek).

AutoResetEvent můžeme vytvořit dvěma způsoby. Jeden je přes konstruktor:

EventWaitHandle wh = new AutoResetEvent (false);

Přijímá parametr typu bool, pokud bychom zadali true, okamžitě po vytvoření instance by došlo k zavolání Set. Druhým způsobem je vytvoření instance přes bázovou třídu:

EventWaitHandle wh = new EventWaitHandle (false, EventResetMode.AutoReset);

Kdybychom nastavili EventResetMode na ManualReset, nevytvořila by se instance třídy AutoResetEvent, nýbrž třídy ManualResetEvent (probereme níže, ale jak už asi tušíte, je to téměř totéž, jen se metoda Reset nevolá automaticky).

Na WaitHandle bychom měli zavolat metodu Close, pokud ho už nebudeme dále potřebovat, abychom uvolnili systémové prostředky. Ale většinou využíváme funkci „turniketu“ po celou dobu života aplikace, takže Close volat nemusíme. V následujícím příkladu si ukážeme funkci EventWaitHandle v praxi:

class BasicWaitHandle
{
    static EventWaitHandle wh = new AutoResetEvent(false);

    static void Main()
    {
        new Thread(Waiter).Start();
        Thread.Sleep(5000);     // Chvilku počkáme
        wh.Set();               // Pustíme vlákno dál
    }
    static void Waiter()
    {
        Console.WriteLine("Čekám...");
        wh.WaitOne();                        // Čekat na propustku
        Console.WriteLine("Propuštěn!");
    }
}

Tento kód vypíše do konzole text „Čekám…“ a o pět vteřin později text „Propuštěn!“, jako důkaz toho, co jsme si řekli výše.

Cross-Process EventWaitHandle

Rychle pro ty, co nevědí, co to cross-process znamená. Z názvu se dá odvodit, že je to proces, který dokáže pracovat napříč spuštěnými procesy (samozřejmě jen těmi, které vědí, jak spolu komunikovat). Docílíme toho pomocí konstruktoru EventWaitHandle, který umožňuje dát instanci této třídy takové „jméno“. Toto jméno je jen nějaký řetězec a může to být cokoliv, co se nedostane do konfliktu s ostatními procesy (ideální je ve tvaru MojeSpolečnost.MojeAplikace.Jméno).

EventWaitHandle wh = new EventWaitHandle (false, EventResetMode.AutoReset,
  "MojeSpolecnost.MojeAplikace.NejakeJmeno");

Pokud dvě aplikace v sobě spustí tento kód, budou si jejich vlákna moci mezi sebou posílat signály.

Využitelnost

EventWaitHandle využijeme v situacích, kdy budeme chtít provádět úkoly na pozadí, bez nutnosti vytvářet nové vlákno pro každou operaci. Toho můžeme docílit i jednoduchým uzavřením vlákna do smyčky – počká na přidělení úkolu, splní ho, počká na další, splní ho, … Tento postup je poměrně častý, navíc se zamezí riziku v podobě interakce s jiným vláknem a další spotřebou systémových zdrojů.

Musíme se ale nějak rozhodnout, co udělat, když je pracovní vlákno už zaměstnáno jinou činností a do toho dostane další úkol. Logicky druhé vlákno zablokujeme, aby počkalo, než první vlákno dokončí to, co zrovna dělá. Ale jak dáme druhému vláknu vědět, že první vlákno už dokončilo vše, co chtělo? Právě zde přichází ke slovu AutoResetEvent. Ukážeme si to na příkladu s jednou proměnnou typu string (deklarovanou pomocí klíčového slova volatile, které zajišťuje, že obě vlákna vždy uvidí stejnou verzi – probereme v budoucnu):

class VyuzitiWaitHandle
{
    static EventWaitHandle pripraven = new AutoResetEvent(false);
    static EventWaitHandle makej = new AutoResetEvent(false);
    static volatile string ukol;

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

        // 5x pošleme signál pracovnímu vláknu
        for (int i = 1; i <= 5; i++)
        {
            pripraven.WaitOne();    // Počkáme, než je pracovní vlákno připravené
            ukol = "a".PadRight(i, 'h');    // Přidělíme úkol
            makej.Set();    // Přikážeme pracovnímu vláknu, aby pracovalo
        }

        // Řekneme pracovnímu vláknu, aby přestalo pracovat (pomocí "žádného" úkolu)
        pripraven.WaitOne(); ukol = null; makej.Set();
    }

    static void Pracuj()
    {
        while (true)
        {
            pripraven.Set();    // Indikace toho, že je vlákno připravené
            makej.WaitOne();    // Čekat na propuštění
            if (ukol == null) return;   // Konec
            Console.WriteLine(ukol);
        }
    }
}

Jako výsledek dostaneme to, co je na obrázku. Pro lepší pochopení, jak to celé pracuje, doporučuju dopsat si do kódu různá čekání na stisk kláves atd.

Všimněte si, že práci vlákna ukončuje pomocí null úkolu, ale ne pomocí Interrupt nebo Abort. Je pravda, že by to fungovalo naprosto stejně, jen bychom museli ošetřovat výjimky, které obě metody vytváří, a to je zbytečný kód navíc.

Producent/spotřebitel

Častým využitím vláken je, že pracovní vlákno zpracovává úkoly, které stojí za sebou ve frontě, v tzv. frontě producent/spotřebitel (Producer/consumer queue). Producent „vytváří“ úkoly a spotřebitelem je to vlákno, které ty úkoly plní (spotřebitelů může být z jedné fronty i více, dobré pro využití potenciálu víceprocesorových systémů).

Princip je podobný jako v předchozí kapitolce u turniketu, jen s tím rozdílem, že volající kód se nezablokuje, pokud je pracovní vlákno už zaměstnané, ale zařadí úkol do fronty a jde „vyrábět“ další úkol.

V následujícím příkladu využijeme AutoResetEvent pro posílání signálů pracovnímu vláknu, když nemá co dělat (což nastane jen tehdy, když je fronta úkolů prázdná). Fronta úkolů se ukládá do generické kolekce Queue, ke které musíme postupovat pod zámkem, aby se zajistila thread-safety. Nakonec celou práci ukončíme předáním null úkolu.

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

class ProducerConsumerQueue : IDisposable
{
    EventWaitHandle wh = new AutoResetEvent(false);
    Thread pracovniVlakno;
    object zamek = new object();
    Queue<string> ukoly = new Queue<string>();

    public ProducerConsumerQueue()
    {
        pracovniVlakno = new Thread(Pracuj);
        pracovniVlakno.Start();
    }

    public void ZaraditUkolDoFronty(string task)
    {
        lock (zamek) ukoly.Enqueue(task);
        wh.Set();
    }

    public void Dispose()
    {
        ZaraditUkolDoFronty(null);  // null úkol=konec práce
        pracovniVlakno.Join();  // Počkáme, než pracovní vlákno dodělá to, co dělá.
        wh.Close(); // Uvolníme systémové prostředky
    }

    void Pracuj()
    {
        while (true)
        {
            string ukol = null;
            lock (zamek)
                if (ukoly.Count > 0)
                {
                    ukol = ukoly.Dequeue();
                    if (ukol == null) return;
                }
            if (ukol != null)
            {
                Console.WriteLine("Provádím úkol: " + ukol);
                Thread.Sleep(1000); // Ať to celé není tak rychlé...
            }
            else
                wh.WaitOne();   // Žádné další úkoly
        }
    }
}

Jako poslední dílek skládačky nám chybí metoda Main, pomocí které vše otestujeme:

class Test
{
    static void Main()
    {
        using (var q = new ProducerConsumerQueue())
        {
            q.ZaraditUkolDoFronty("Ahoj");
            for (int i = 0; i < 10; i++) q.ZaraditUkolDoFronty("číslo " + i);
            q.ZaraditUkolDoFronty("Nashle!");
        }
        // Díky použití "using" dojde na závěr automaticky k zavolání
        // metody Dispose.
    }
}

Vše zkompilujeme a jako výstup dostaneme to, co je na obrázku:

ManualResetEvent

V tomto článku jsem už řekl, že ManualResetEvent funguje téměř stejně jako jeho automatický bratr, jen s tím rozdílem, že zde musíme volat metodu „Reset“ my, nedělá se to automaticky.

Instance této třídy se někdy používají, chceme-li nějakému jinému vláknu oznámit, že jedno vlákno dokončilo nějakou operaci, nebo říct, že je připravené k práci.

Mutex

Tak jsme dokončili část článku, zabývající se třídou EventWaitHandle, teď nás čeká třída Mutex.

Samotná funkce Mutexu je naprosto zbytečná, protože ji do písmene kopíruje konstrukce lock. Jediná výhoda Mutexu oproti zamykání je v tom, že dokáže pracovat mezi více procesy, zatímco lock jen v té jedné aplikaci.

Mutex je sám o sobě rychlý, ale lock je stokrát rychlejší! Uzamknutí pomocí Mutexu trvá pár mikrosekund, ale pomocí locku jsou to desetiny nanosekund, i kvůli výkonu tedy používejte lock!

Třída Mutex, stejně jako EventWaitHandle, obsahuje metodu WaitOne, která zajišťuje zámek a všechno blokování. „Odemčení“ dosáhneme pomocí metody ReleaseMutex, stejně jako u „locku“, odemknout zámek může jen to vlákno, které ho zamklo.

Typickým využitím Mutexu je zajištění toho, že v jednu chvíli může běžet jen jedna instance programu. Následující kód zkuste zkompilovat a výsledný .exe spusťte dvakrát. Uvidíme, že program bude mít námitky.

using System;
using System.Threading;

class PustSeJednou
{
    // Ujistěte se, že je název aplikace unikátní, použijte třeba URL adresu vašeho webu
    static Mutex mutex = new Mutex(false, "chrasty.cz PustSeJednou");

    static void Main()
    {
        // 5 vteřin počká, pak ukončí aplikaci
        if (!mutex.WaitOne(TimeSpan.FromSeconds(5), false))
        {
            Console.WriteLine("Jiná instance této aplikace běží! Konec");
            return;
        }
        try
        {
            Console.WriteLine("Aplikace spuštěna - Stiskněte Enter pro ukončení");
            Console.ReadLine();
        }
        finally { mutex.ReleaseMutex(); }
    }
}

Dobré, ne? Toto je jedna z věcí, které můžete okamžitě uvést do praxe i bez nějakých hlubších znalostí vláken.

Dobrou funkcí Mutexu je, že pokud zapomeneme zámek uvolnit pomocí metody ReleaseMutex před vypnutím aplikace, CLR to udělá za nás.

Semaphore

I tuto třídu si přirovnáme k něčemu z reálného světa, třeba ke klubu. Klub má danou kapacitu, kolik dokáže pojmout lidí (podle místa uvnitř). Jakmile je plno, lidé už nemohou dovnitř a před vchodem se tvoří fronta. Jakmile někdo z klubu odejde, jeden člověk ze začátku fronty může dovnitř. Konstruktor třídy Semaphore přijímá dva parametry – počet míst v klubu, která jsou momentálně volná, a celkovou kapacitu klubu.

Jakékoliv vlákno může zavolat metodu Release na instanci třídy Semaphore (v tom se liší od Mutex a lockingu, kdy jen vlákno, které blokuje, může propustit blokované vlákno).

Následující příklad vytvoří deset vláken, každé spustí smyčku s metodou Sleep uprostřed. A právě třída Semaphore zajistí, že Sleep nezavolají víc než tři vlákna najednou.

class SemaphoreTest
{
    static Semaphore s = new Semaphore(3, 3);  // Dostupná kapacita=3; Celková=3

    static void Main()
    {
        for (int i = 0; i < 10; i++) new Thread(Pracuj).Start();
    }

    static void Pracuj()
    {
        while (true)
        {
            s.WaitOne();
            Thread.Sleep(100);  // Maximálně tři vlákna najednou se sem dostanou
            s.Release();
        }
    }
}

WaitAny, WaitAll a SignalAndWait

Na závěr tohoto dílu se podíváme na tři metody, které jsou uvedené v nadpisu. Vedle metod Set a WaitOne, které už znáte, existuje právě tato trojka statických metod ve třídě WaitHandle, které slouží pro rozlousknutí složitějších synchronizačních oříšků.

Nejužitečnější je asi metoda SignalAndWait – zavolá WaitOne na jeden WaitHandle a Set na druhý. To můžeme využít na dvojici EventWaitHandlerů, abychom zařídili, že se dvě vlákna setkají v jednu chvíli na jednom místě. První vlákno zavolá

WaitHandle.SignalAndWait (wh1, wh2);

A druhé zavolá opak:

WaitHandle.SignalAndWait (wh2, wh1);

WaitHandle.WaitAny čeká na jakýkoliv ze zadaného pole Wait Handlerů; WaitHandle.WaitAll čeká na všechny, než podnikne nějakou akci. Máme-li několik turniketů, obě tyto metody vytváří frontu za všemi turnikety. U metody WaitAny půjdou lidé skrz první turniket, který se otevře, u WaitAll půjdou teprve až se otevřou všechny.

Zdá se, že je to pro dnešek všechno. Snad jste si toho co nejvíc odnesli a uvidíme se u dalšího dílu. Pro dnešek platí více než jindy jedna věc (ač jsem zkoušel co nejvíc chybek opravit, určitě mi nějaké unikly kvůli délce článku) – pokud jsem něco v článku vysvětlil nejasně nebo nepřesně, ozvěte se v komentářích.