Párhuzamosítás

Az eddigiekben arról volt szó, hogy hogyan is készítünk szálakat és hogyan menedzseljük azok életciklusát. Ezen ismereteket felhasználva már párhuzamosíthatnánk algoritmusokat, hogy gyorsítsuk azok futási idejét, de miért tennénk ilyet? Mármint manuálisan megírni azt, hogy mondjuk egy ciklus feldolgozását szétosszuk kettő vagy több szálra nem biztos, hogy a legjobb ötlet két okból. Az egyik ok, hogy ezt már más megírta helyettünk, ezért nem kell feltalálnunk a kereket. A másik ok pedig az, hogy ez tipikusan a könnyűnek tűnő, de a részleteiben végtelen komplexitású feladat.

.NET esetén egy algoritmus párhuzamosítására két lehetÅ‘ségünk van: a Parallel osztály használata és a PLINQ.

Parallel osztály

A Parallel osztály for és foreach ciklusok párhuzamosítását teheti lehetÅ‘vé. Fontos kiemelni a feltételes módot, hogy ezen metódusok egyike sem garantálja 100%-os biztonsággal, hogy a ciklus egynél több szálon fog futni. Arra vállalnak garanciát, hogy ha lehetséges, akkor több szálra lesz a munka elosztva. Az, hogy ez megtörténik-e vagy sem, attól függ, hogy a Threadpool-on van-e kellÅ‘ mennyiségű szabad szál a feladat elvégzéséhez. Ezen felül függ az iterációk számától is. Például egy 20 elemű tömb párhuzamos bejárásának nincs értelme a párhuzamosított foreach segítségével, mert az extra szálindítás költsége valószínűleg nagyobb, mint a nyereség a feldolgozáson. Ilyen esetekben megtörténhet az, hogy a párhuzamosításra irányuló kérelmünket felülbírálja a rendszer.

A feladat jellege sem mindegy, mert ha sok osztott erőforrást manipulál a ciklusunk, amik írásához szinkronizáció vagy egyéb módú zárolás kell, akkor finoman szólva is lábon lőttük magunkat. A feladat függvényében a Parallel osztály is dönthet úgy, hogy nem párhuzamosítja a kódunkat. Ez akkor fordulhat elő, ha az egy szálra jutó munka mennyisége nagyon pici.

Éppen ezért aranyszabály, hogy az ilyen optimalizációkat minden esetben mérjük ki, hogy volt-e értelmük. A mérés módjáról a könyv későbbi részében lesz szó.

Nézzünk egy példát a for ciklus párhuzamosítására:

using System;
using System.Threading.Tasks;

namespace PeldaParallel
{
    class Program
    {
        static void Main(string[] args)
        {
            for (int i = 0; i < 12; i++)
            {
                Console.Write("{0} ", i);
            }
            Console.ReadKey();
            Console.WriteLine();
            Parallel.For(0, 12, (i) =>
            {
                Console.Write("{0} ", i);
            });
            Console.ReadKey();
        }
    }
}

A program kimenete:

0 1 2 3 4 5 6 7 8 9 10 11
0 1 2 4 3 8 10 11 5 9 6 7

A szintaxisban eltérés azért van, mert a for kulcsszóval ellentétben a Parallel.For egy metódus, aminek az elsÅ‘ paramétere a ciklusváltozó kezdÅ‘ értéke, a második paramétere a ciklus változó maximális értéke, a harmadik paramétere pedig egy lambda, aminek a bemeneti paramétere a ciklusváltozó.

Hasonló a szintaxis eltérés a Parallel.ForEach esetén is:

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

namespace PeldaParallel2
{
    class Program
    {
        static void Main(string[] args)
        {
            var collection = new List<int>();
            for (int i = 0; i < 12; i++)
            {
                collection.Add(i);
            }

            foreach (var item in collection)
            {
                Console.WriteLine(item);
            }
            Console.ReadKey();
            Parallel.ForEach(collection, item =>
            {
                Console.WriteLine(item);
            });
            Console.ReadKey();
        }
    }
}

A háttérben mindkét metódus particionál a végrehajtás előtt, vagyis felosztja a munkamennyiséget az elérhető erőforrások között.

Mindkét metódusnak van Async végzÅ‘désű változata, amivel egy lambda helyett egy Task futtatható párhuzamosítva. Ami a ciklus vezérlést illeti, az is módosul, hiszen a break vagy continue kulcsszavak ebben az esetben nem azt teszik, mint szinkron környezetben.

A metódusoknak átadható lambda a ciklus változón kívül rendelkezhet egy második paraméterrel is, amelynek a típusa egy ParallelLoopState, ami a ciklus állapotát reprezentálja. Ezen az objektumon hívhatunk egy Break() metódust, ami megszakítja a ciklust. Létezik egy Stop() metódusa is, ami nem azonnali megállást eredményez, helyette az operációs rendszer ütemezÅ‘jére bízza, hogy a Stop() meghívása után az ütemezze be magának kényelmesen a ciklus megállítását.

var collection = new List<int> { 1, 2, 3, 4, 5};

Parallel.ForEach(collection, (item, loopstate) =>
{
    //item jelen esetben a kollekció egy eleme int típussal,
    Console.WriteLine(item);
    //a loopstate ParallelLoopState típusú, a keretrendszer ad neki értéket
    loopstate.Break();
});

//a loopstate hasonlóan alkalmazható párhuzamos for ciklusban is:
Parallel.For(0, 12, (i, loopstate) =>
{
    Console.Write("{0} ", i);
    loopstate.Break();
});

Kivételkezelés

A Parallel osztály metódusainak a használatakor ügyelni kell arra, hogy ha a ciklusmagban egy nem kezelt kivétel keletkezik egy elem feldolgozásakor, akkor az egész végrehajtás nem áll meg, mint szinkron esetben.

A kivétel keletkezésekor a keretrendszer további új, még el nem indított szálat nem fog elindítani, de a már futókat nem szakítja meg, ezért lényegében több kivétel is keletkezhet több külön szálon. Ezek eredményét egy AggregateException típusban kapjuk meg, aminek az InnerExceptions tulajdonságából kérdezhetjük le a keletkezett kivételeket. Az alábbi példa ezt a viselkedést szemlélteti:

try
{
    Parallel.For(0, 20, i =>
    {
        if (i == 3 || i == 7)
            throw new InvalidOperationException($"Hiba: {i}");

        Console.Write("{0}, ", i);
    });
}
catch (AggregateException ex)
{
    foreach (var inner in ex.InnerExceptions)
    {
        Console.WriteLine(inner.Message);
    }
}

A program kimenete valami hasonló lesz:

2, 1, 0, 5, 4, 6,
Hiba: 3
Hiba: 7

PLINQ

A PLINQ a Parallel LINQ rövidítése és LINQ kérések párhuzamosítására szolgál. Fontos kiemelni, hogy ez a fajta párhuzamosítás csak olyan LINQ kérések esetén alkalmazható, amelyek memóriában tárolt objektumok. Entity Framework és egyéb LINQ alapú megoldásoknál, ahol azok végrehajtásáért egy adatbázis felelÅ‘s, ott maga az adatbázis felel a kérés egyes részeinek párhuzamosításáért. Ez a párhuzamosítás a kódunkban az AsParallel() metódus meghívásával történik. Nézzünk is egy példát:

using System;
using System.Collections.Generic;
using System.Linq;

namespace PeldaParallel3
{
    class Program
    {
        static void Main(string[] args)
        {
            var collection = new List<int>();
            for (int i = 0; i < 444; i += 2)
            {
                collection.Add(i);
            }

            var q = from c in collection.AsParallel()
                    where c % 5 == 0
                    select c;

            q.ForAll(x => Console.Write("{0} ", x));

            Console.ReadKey();

        }
    }
}

A program kimenete valami hasonló lesz:

340 350 360 370 380 390 400 410 420 430 440 230 240 250 260 270 280 290 300 310 320 330 0 120 130 140 150 160 170 180 190 200 210 220 10 20 30 40 50 60 70 80 90 100 110

Az AsParallel() a motorháztetÅ‘ alatt a kapott IEnumerable<T> kollekciónkat átkonvertálja egy ParallelEnumerable típusba, ami a háttérben elvégzi a kollekció particionálását, ami ahhoz kell, hogy az elemek a szálak között szétoszthatóak legyenek. Ez a program kimenetén szépen látszik.

A kiíratást a ForAll metódussal mutatja be a példa, ami arra való, hogy egy az eredményeket párhuzamosítva be tudjuk járni. Ez jelen kontextusban, hogy a Console.Write van benne használva picit antipélda, mivel ez egy osztott erÅ‘forrás és a tankönyvi példán kívül éles környezetben ilyen jellegű alkalmazása nem ajánlott.

Az AsParallel után ha valamiért olyan műveletet szeretnénk elvégezni, aminek csak egy szálon van értelme, akkor az AsSequential hívással visszaválthatunk szinkron végrehajtásra.

Fontos megjegyezni, hogy a párhuzamosan futó lekérdezés a kapott kollekció eredeti sorrendjét nem veszi figyelembe alapértelmezetten. Éppen ezért ha fontos a sorrend, akkor az AsParallel() hívás után egy AsOrdered() hívással érdemes rendezni a kollekciót:

using System;
using System.Collections.Generic;
using System.Linq;

namespace PeldaParallel4
{
    class Program
    {
        static void Main(string[] args)
        {
            var collection = new List<int>();
            for (int i = 0; i < 444; i += 2)
            {
                collection.Add(i);
            }

            var q = from c in collection.AsParallel().AsOrdered()
                    where c % 5 == 0
                    select c;

            foreach (var item in q)
            {
                Console.Write("{0} ", item);
            }

            Console.ReadKey();
        }
    }
}

A program kimenete így valami hasonló lesz:

0 10 20 30 40 50 60 70 80 90 100 110 120 130 140 150 160 170 180 190 200 210 220 230 
240 250 260 270 280 290 300 310 320 330 340 350 360 370 380 390 400 410 420 430 440

Az AsParallel() egyfajta varázslatnak tűnhet, mert igen pici módosítással nagy elÅ‘nyt tudunk nyerni, de mint máshol is, itt is két oldala van az éremnek. Ebben az esetben a kevésbé szép, vagy csúnya oldala a dolognak, hogy ez is egyfajta kérésnek fogható fel és nem utasítás. Vagyis a végrehajtó keretrendszer dönthet úgy, hogy nem éri meg párhuzamosítani a kérést. Itt is azok a szabályok érvényesek, mint amelyekrÅ‘l szó volt a Parallel osztály esetén. Ezeken felül az alábbi esetekben biztos, hogy egy szálon fog futni a kérésünk:

• Ha az Select, indexelt Where, indexelt SelectMany vagy ElementAt záradékot tartalmaz egy rendezÅ‘ vagy szűrÅ‘ operátor után, amely eltávolította vagy átrendezte az eredeti indexeket.
• Ha az Take, TakeWhile, Skip vagy SkipWhile operátort tartalmaz, és ahol a forrás sorozat indexei nincsenek az eredeti sorrendben.
• Ha az Zip vagy SequenceEqual operátort tartalmaz. Ez alól kivétel az az eset, amikor az egyik adatforrás eredetileg rendezett indexekkel rendelkezik, és a másik adatforrás indexelhetÅ‘, azaz tömb vagy lista.
• Ha az Concat vagy Reverse operátort tartalmaz. Ez alól kivételt képez az az eset, amikor ezt indexelhetÅ‘ adatforrásokra alkalmazzuk.