Abstrakt: Paralelní zpracování dat je v soucˇasné dobeˇ velmi aktuální téma. Jeden z používaných postup˚u je prˇevod vstupních dat na datové proudy a zpracování teˇchto proud˚u pomocí operátor˚u, které mohou být vyhodnocovány paralelneˇ. Protože pro specifikaci vzájemného propojení operátor˚u jsou beˇžné programovací jazyky nevhodné, vznikla pro tento úcˇel celá rˇada doménoveˇ specifických jazyk˚u. Jazyk Bobolang je jedním z nich. Kromeˇ beˇžných vlastností, ale prˇidává navíc neˇkteré syntaktické a sémantické prvky, které znacˇneˇ usnadnˇují intra-operátorovou paralelizaci. Díky tomu je možné snadno vytvárˇet vysoce škálovatelné aplikace zpracovávající proudová data. Paralelní systémy jsou v dnešní dobeˇ témeˇrˇ standardem. Z tohoto d˚uvodu se neustále hledají nové cesty jak co nejefektivneˇji využít tyto prostrˇedky. Bohužel, vývoj paralelních aplikací je nárocˇný a náchylný k chybám. Proto vznikají r˚uzné knihovny, nástroje, systémy a metody, jak tuto cˇinnost maximálneˇ usnadnit a zefektivnit. Neˇkteré tyto nástroje se snaží být co nejobecneˇjší, tzn. programátorovi pomáhají s vytvárˇením a synchronizací vláken [5], neˇkteré poskytují množství knihovních funkcí pro snazší paralelizaci neˇkterých typ˚u algoritm˚u [19], eˇnkteré rozšiˇrují samotný jazyk o direktivy urcˇené pro snazší vývoj [7]. Kromeˇ toho existují doménoveˇ specifické nástroje, které jsou urcˇeny pro konkrétní druhy aplikací. Systémy pro zpracování proudových dat jsou jedním z nich [20, 18, 21, 8, 4, 17, 2]. Tyto systémy pracují s tzv. datovými proudy, tj. v podstateˇ s posloupnostmi n-tic. Tyto proudy jsou pr˚ubeˇžneˇ (tj. tak jak n-tice prˇichází) zpracovávány operátory, které transformují vstupní proudy na výstupní. Vývoj aplikací pro takové systémy se skládá ze dvou fází:
1. Implementace potrˇebné množiny operátor˚u, tj. jak prˇetransformovat vstupní proud/proudy na výstupní proud/proudy. 2. Vytvorˇení exekucˇního plánu, tj. pospojování operátor˚u orientovanými hranami, které urcˇují datové proudy mezi nimi.
Zatímco implementaci operátor˚u lze provést v témeˇrˇ libovolném beˇžném programovacím jazyku (C/C++, Java,
C# apod.), pro specifikaci exekucˇního plánu se tyto jazyky prˇíliš nehodí. Exekucˇní plán totiž odpovídá orientovanému grafu, tj. je zadán seznamem vrchol˚u a hran. To sice lze snadno vyjádrˇit v beˇžných programovacích jazycích, ale takový kód je obtížneˇ cˇitelný a modifikovatelný. Proto vznikají jazyky, které vytvárˇení exekucˇního plánu usnadnˇují, at’ už speciální syntaxí, nebo grafickou vizualizací plánu.
Bobolang se zameˇrˇuje na druhou fázi vývoje. Kromeˇ cˇitelné syntaxe se ale snaží i o pomoc prˇi intra-operátorové paralelizaci, kdy provádí neˇkteré transformace zvyšující paralelismus exekucˇního plánu automaticky. Tím se odlišuje od ostatních podobných jazyk˚u.
Zbytek cˇlánku je rozdeˇlen následovneˇ: Kapitola 2 prˇedstavuje používané jazyky urcˇené pro systémy zpracování proudových dat, kapitola 3 popisuje systém Bobox, pro který vznikla pilotní implementace jazyka Bobolang. V kapitole 4 rozebíráme možnosti a postupy paralelizace operátor˚u. Strucˇný popis jazyku Bobolang je uveden v kapitole 5 a neˇkolik prˇíklad˚u jeho použití uvádíme v kapitole 6. Celý cˇlánek pak shrnujeme v kapitole 7.
Soucˇasné jazyky zameˇrˇující se na proudové zpracování dat se dají rozdeˇlit do neˇkolika skupin podle jejich zameˇrˇení.
Brook [
Lucid [
Jazyk X-Language [
Systém Bobox je jedna z implementací systému pro zpracování proudových dat. Bobox poskytuje beˇhové prostrˇedí pro vyhodnocování exekucˇních plán˚u v paralelním prostrˇedí. Systém podporuje jak acyklické exekucˇní plány, tak i plány obsahující cykly. Protože jazyk Bobolang je navržen práveˇ pro Bobox a využívá neˇkteré jeho vlastnosti, uvádíme v této kapitole neˇkteré technické detaily tohoto systému.
Datové proudy v Boboxu jsou reprezentované jako proud tzv. datových obálek. Každá obálka obsahuje seznam tzv. datový sloupc˚u. Tyto datové sloupce obsahují samotná data. Každý sloupec musí obsahovat data pouze jednoho typu, ale jedna obálka m˚uže obsahovat sloupce r˚uzných typ˚u. Dále platí, že všechny sloupce v jedné obálce mají vždy stejnou délku, takže se m˚užeme na obálku dívat jako na posloupnost n-tic. Kromeˇ datových obálek existují tzv. otrávené obálky, jejichž úkolem je signalizovat konec datového proudu.
V soucˇasné dobeˇ podporuje Bobox pouze sharedmemory systémy, takže operátory si mohou vzájemneˇ posílat pouze ukazatele na obálky. Tato implementace znacˇneˇ urychluje operátory jako naprˇ. broadcast (viz kapitola 4), nebot’ data nejsou nijak klonována a v pameˇti se nacházejí pouze v jedné instanci. Navíc je celkový pocˇet ˇrádk˚u v obálce je zvolen s ohledem na velikost vyrovnávacích pameˇtí v procesoru, tak aby komunikace mezi operátory probíhala bez nutnosti prˇístupu do hlavní pameˇti.
Každý exekucˇní plán, musí obsahovat dva speciální operátory: • init – tento operátor je vždy první v topologickém usporˇádání exekucˇního plánu. Jeho úkolem je nastartovat výpocˇet tím, že na sv˚uj výstup odešle otrávenou obálku. • term – tento operátor je vždy poslední v topologickém usporˇádání. Ve chvíli, kdy prˇijme na svém vstupu otrávenou obálku, oznámí systému, že exekucˇní plán byl vyhodnocen.
D˚uležitou soucˇástí systému je plánovacˇ, jehož
úkolem je prˇideˇlovat výpocˇetní cˇas jednotlivým operátor˚um.
Obecneˇ se plánovacˇ rˇídí dostupností datových obálek na
vstupech operátor˚u, tj. pokud má operátor neprázdnou
frontu vstupních obálek, je vložen do fronty operátor˚u
prˇipravených ke spušteˇní. Na základeˇ r˚uzných kritérií [
Ale i prˇes jednovláknovost operátor˚u, je možné
dosáhnout paralelního vyhodnocování exekucˇního plánu, nebot’
nezávislé operátory mohou být spoušteˇny paralelneˇ.
Rozlišujeme trˇi typy paralelism˚u [
Taskový paralelismus je pevneˇ zakódovaný v exekucˇním plánu, takže prˇináší pouze omezenou škálovatelnost. Datový a pipelinový paralelismus lze ale za urcˇitých okolností zvýšit a tím zvýšit škálovatelnost celého systému.
Pipelinový paralelismus m˚užeme zvýšit (resp. zavést) tím, že urcˇitý operátor rozdeˇlíme na posloupnost dílcˇích operátor˚u, kde každý vykoná nad proudemcˇást práce (viz Obrázek 1). Bohužel ne všechny operátory lze takto dekomponovat a u teˇch, u kterých to lze, to m˚uže být nevýhodné z d˚uvodu zvýšení režie nutné pro prˇenos datového proudu.
operator op1 op2 op3 op4 Obrázek 1: Rozklad operátoru pro zvýšení pipelinového paralelismu.
Datový paralelismus m˚užeme do plánu zavést tak, že vstupní proud rozdeˇlíme na neˇkolik dílcˇích proud˚u, ty zpracujeme paralelneˇ a poté je opeˇt spojíme do výsledného proudu. V následujících dvou podkapitolách rozebereme dva postupy, jak toho dosáhnout.
Bezestavové operátory si neudržují vnitrˇní stav. To znamená, že zpracování jedné n-tice je zcela nezávislé na ostatních. Typickou ukázkou je naprˇ. operátor filter, který z proudu n-tic odstraní ty, které nesplnˇují urcˇitou podmínku, nebot’ vyhodnocení podmínky pro jednu n-tici nezávisí na jiných n-ticích.
Protože zpracování jedné n-tice nezávisí na ostatních, nezávisí ani zpracování celé obálky na ostatních obálkách. M˚užeme tedy použít schéma naznacˇené na Obrázku 2. Operátor rr_dispatch jednoduše prˇeposílá vstupní obálky na své výstupy metodou round-robin, operátor rr_consolidate metodou round-robin odebírá výsledné obálky z jednotlivých operátor˚u a vytvárˇí tak výsledný proud.
rr_dispatch
Protože rr_dispatch a rr_consolidate pouze manipulují se ukazateli na obálky (viz kapitola 3), pracují oba operátory velmi rychle a celý výpocˇet zpomalují pouze zanedbatelneˇ.
Se stavovými operátory je situace složiteˇjší, nebot’ abychom zpracovali jednu obálku, musíme znát stav odvozený z obsahu prˇedchozích obálek. U neˇkterých operátor˚u m˚užeme použít postup nazncˇaený v této podkapitole. Prˇedpokládejme, že teˇlo stavového operátoru vypadá obecneˇ takto:
S ← iniciální stav while not konec do
zpracuj vstupní data pomocí S a zárovenˇ aktualizuj S end while Obcˇas ale lze toto schéma upravit do následující podoby: S ← iniciální stav while not konec do zpracuj vstupní data pomocí S aktualizuj S end while
Pokud je aktualizace stavu S rychlejší než zpracování dat, m˚užeme vytvorˇit N paralelních operátor˚u a ocˇíslovat je cˇísly 0 až N − 1 (toto cˇíslo budeme v dalším textu oznacˇovat jako PID – Parallel ID). Každý operátor pak bude pracovat následujícím zp˚usobem:
S ← inciální stav fáze ← 0 while not konec do if fáze mod N = PID then
zpracuj cˇást vstupu pomocí S end if aktualizuj S fáze ← fáze + 1 end while
V tuto chvíli se operátory strˇídají ve zpracování vstupních dat, takže pro paralelizaci m˚užeme použít schéma znázorneˇné na Obrázku 3. Efektivita paralelizace závisí na složitosti aktualizace stavu. Je zrˇejmé, že by meˇla být alesponˇ N-krát rychlejší než zpracování dat. Problém nastává, pokud je aktualizace stavu netriviální, nebot’ v takovém prˇípadeˇ se pocˇítá N-krát totéž. Rˇ ešením je dedikovat samostatný operátor pro aktualizaci stavu, který by všem ostatním posílal aktuální stav.
operator[0]
operator[
Pomocí jazyka je rovneˇž možné z množiny hotových operátor˚u (naimplementovaných v jazyku C++ nebo v jazyku Bobolang) poskládat samostatný operátor, který lze poté použít v exekucˇním plánu. K tomu slouží následující syntaxe: o p e r a t o r p r o c e s s ( i n t ) − >( i n t ) { p r e p r o p c e s s ( i n t ) − >( i n t ) p r e ; p o s t p r o c e s s ( i n t ) − >( i n t ) p o s t ; i n p u t −> p r e ; p r e −> p o s t ; p o s t −> o u t p u t ;
Rˇ ádek operator process(int)->(int) rˇíká, že chceme vytvorˇit operátor se jménem process, který transformuje proud celých cˇísel na proud celých cˇísel. Následuje teˇlo operátoru, které se skládá z instancí operátor˚upreprocess a postprocess. Kromeˇ explicitneˇ uvedených instancí operátor˚u (pre a post), obsahuje každé teˇlo implicitneˇ operátory input a output. Ty reprezentují vstup/výstup celého operátoru. Takže rˇádek input -> pre rˇíká, že vstup operátoru process je prˇeposílán na vstup operátoru pre. Podobneˇ funguje operátor output.
Exekucˇní plán se specifikuje stejnou syntaxí. Jak bylo uvedeno v kapitole 3, exekucˇní plán se skládá ze dvou speciálních operátor˚uinit a term a teˇla exekucˇního plánu. Na teˇlo exekucˇního plánu se tedy m˚užeme dívat jako na operátor, který má jeden vstup (k neˇmu je prˇipojen operátor init) a jeden výstup (k neˇmu je prˇipojen operátor term).
Aby interpretr jazyka poznal, který operátor reprezentuje exekucˇní plán, musí být vždy pojmenován jako main.
Pokud bychom chteˇli vyrobit aplikaci, která zpracovává posloupnost celých cˇísel, napíšeme následující kód: s o u r c e () − >( i n t ) s o u r c e ; p r o c e s s ( i n t ) − >( i n t ) op ; s i n k ( i n t ) − >() s i n k ; } i n p u t −> s o u r c e ; s o u r c e −> op ; op −> s i n k ; s i n k −> o u t p u t ;
Pokud prˇedáme systému Bobox tento kód, interpretr Bobolangu instanciuje operátor main, vytvorˇí operátory init a term a vytvorˇí exekucˇní plán, který je znázorneˇný na Obrázku 4.
Pokud má operátor více vstup˚u/výstup˚u, jsou tyto
operátory cˇíslovány od nuly a cˇíslo vstupu/výstupu musí být
uvedeno. Pokud má vstup/výstup pouze jeden, nemusí být
toto cˇíslo uvedeno. Viz naprˇ. použití operátoru merge, kdy
main
process
init
source
pre
post
sink
term
Obrázek 4: Ukázka plneˇ instanciovaného exekucˇního
plánu.
navíc musí rozeslat otrávenou obálku z operátoru init do
operátor˚usource (viz Obrázek 5).
b r o a d c a s t ( ) − > ( ) , ( ) b c a s t ;
s o u r c e () − >( i n t ) s r c 1 , s r c 2 ;
merge ( i n t ) , ( i n t ) − >( i n t ) merge ;
s i n k ( i n t ) − >() s i n k ;
i n p u t −> b c a s t ;
b c a s t [ 0 ] −> s r c 1 −> [ 0 ] merge ;
b c a s t [
main init bcast merge sink term Obrázek 5: Ukázka práce s operátory, které mají více vstup˚u/výstup˚u. 5.2
Každý operátor m˚uže mít libovolný nenulový pocˇet vstup˚u a výstup˚u. Kromeˇ toho ale m˚uže být každý vstup/výstup tzv. násobný. Implicitneˇ je každý vstup/výstup jednonásobný, násobnost se musí zapisovat explicitneˇ, tj. naprˇ.: b r o a d c a s t ( ) − > ( ) {N} b c a s t ;
kde N znacˇí násobnost. N m˚uže být bud’ prˇirozené cˇíslo nebo znak *. Cˇ íslo N prˇesneˇ urcˇuje násobnost vstupu/výstupu, zatímco * nechává toto rozhodnutí na Bobolangu, který dosadí vhodné cˇíslo (v pilotní implementaci shodné s pocˇtem vláken v systému).
Bobolang umožnˇuje vzájemneˇ propojit výstup libovolné násobnosti na vstup libovolné násobnosti, pokud taková operace nevede k logické chybeˇ.
Pokud je prˇipojen vícenásobný výstup na jednonásobný vstup, dojde k automatické replikaci cílového operátoru podle násobnosti výstupu a prˇipojení výstup˚u na jednotlivé vstupy replikovaných operátor˚u.
Spojení jednonásobného výstupu s jednonásobným vstupem zp˚usobí, že cílový operátor je replikovaný práveˇ tolikrát, kolikrát je replikovaný zdrojový operátor a jednotlivé výstupy jsou napojeny na jednotlivé vstupy.
Aby spojení jednonásobného výstupu na vícenásobný vstup bylo korektní, musí být zdrojový operátor replikovaný. Pokud je tato podmínka splneˇna, je vytvorˇena jedna instance cílového operátoru a jednotlivé výstupy jsou prˇipojeny na vstup tohoto operátoru.
Spojení vícenásobného výstupu a vícenásobného vstupu je rovneˇž povoleno, pokud je zdrojový operátor replikovaný. V takovém prˇípadeˇ je cílový operátor replikovaný tolikrát, kolikrát je replikovaný zdrojový operátor. V prˇípadeˇ operátoru -> je 1. operátor napojen na 1. podvstup cílových operátor˚u, 2. operátor na 2. podvstup, atd.
Následující zdrojový kód, který pokrývá všechny
uvedené možnosti, bude interpretován tak, jak je znázorneˇno
na Obrázku 6.
op ( ) − > ( ) { ∗ } op1 ;
op ( ) − > ( ) op2 ;
op ( ) − > ( ) { ∗ } op3 ;
op ( ) { ∗ } − > ( ) op4 ;
op ( ) { ∗ } − > ( ) op5 ;
}
}
i n p u t −> op1 −> op2 −> op3 ;
op3 −> op4 −> op5 −> o u t p u t ;
op2[
Aby bylo možné vytvárˇet znovupoužitelné operátory (naprˇ. operátor trˇídící celá a desetinná cˇísla bude mít pravdeˇpodobneˇ identickou vnitrˇní strukturu), obsahuje Bobolang klícˇové slovo typename. To je inspirované stejným slovem v jazyku C++ a je možné jej použít v deklaraci operátoru naprˇ. v prˇípadeˇ trˇídeˇní takto: o p e r a t o r s o r t ( typename T) − >(T ) { s o m e _ o p e r a t o r ( T) − >(T ) op ; . . .
V teˇle operátoru pak m˚užeme používat typT, jako jakýkoliv jiný beˇžný typ. Pokud instanciujeme tento operátor naprˇ. následujícím zp˚usobem:
dosadí se za T typ (int,int), tj. dvojice celých cˇísel. Pokud by vstupní a výstupní typ byl r˚uzný, dojde k chybeˇ prˇi vyhodnocování.
Zápis intra-operátorové paralelizace je nyní snadný. Pokud máme bezestavový operátor, stacˇí zapouzdrˇit jej následovneˇ: o p e r a t o r p a r a l l e l _ s t a t e l e s s ( typename T) − >( typename U) r r _ d i s p a t c h ( T) − >(T ) { ∗ } d i s p ; s t a t e l e s s _ o p e r a t o r ( T) − >(U) op ; r r _ c o n s o l i d a t e (U){∗} − >(U) c o n s ; i n p u t −> d i s p −> op ; op −> c o n s −> o u t p u t ;
Instanciací operátoru dostaneme stejné schéma jako v podkapitole 4.1 (viz Obrázek 2).
Paralelizovatelný operátor má identické schéma, jenom místo operátoru rr_dispatch, použijeme operátor broadcast. Aby programátor nemusel bezestavové a paralelizovatelné operátory takto paralelizovat rucˇneˇ, provádí tuto úpravu Bobolang sám. Stacˇí oznacˇit operátor jako bezestavový (klícˇovým slovem stateless) nebo paralelizovatelný (klícˇovým slovem parallel). V ostatních prˇípadech se žádná modifikace neprovádí.
Pokud se jedná o komplexneˇjší paralelizaci operátoru, je nutné zapsat schéma operátoru rucˇneˇ, nicméneˇ Bobolang tuto cˇinnost znacˇneˇ usnadnˇuje, viz kapitola 6. 6 6.1
Nested-loops join je velmi snadný algoritmus pro paralelizaci. Máme-li naimplementovaný operátor, který vykonává nested-loops join nad vstupními daty, m˚užeme paralelizovat operátor tak, že vytvorˇíme N instancí toho operátoru a do jednoho vstupu operátor˚u prˇepošleme celý první vstup, zatímco do druhého pouze jednu N-tinu druhého vstupu (N-tiny musí být samozrˇejmeˇ disjunktní). Výsledný proud pak získáme jako sjednocení výsledku replikovaných operátor˚u.
V Bobolangu tento algoritmus zapíšeme následujícím
zp˚usobem (dispatch má z úkol rozdeˇlit proud na N cˇástí,
union spojit N proud˚u do jednoho)
{
b r o a d c a s t ( L) − >(L ) { ∗ } b c a s t ;
r r _ d i s p a t c h ( R) − >(R) { ∗ } d i s p ;
n e s t e d _ l o o p s _ j o i n ( L ) , ( R) − >(T ) j o i n ;
u n i o n ( T){∗} − >(T ) u n i o n ;
i n p u t [ 0 ] −> b c a s t −> [ 0 ] j o i n ;
i n p u t [
Instanciovaný operátor je zobrazen na Obrázku 7. Více
detail˚u vcˇetneˇ experiment˚u lze nalézt vcˇlánku [
Obrázek 7: Paralelizovaný nested-loops join.
}
}
6.2 Trˇídeˇní
Problémem trˇídeˇní v systémech proudového zpracování
dat jsme se zabývali v prˇedchozí práci [
Protože je merge oznacˇen jako parallel, vloží se prˇed tento operátor automaticky broadcast a za neˇj rr_consolidate. Pokud použijeme operátor parallel_sort v exekucˇním plánu, rozvine se do tvaru znázorneˇného na Obrázku 8.
sort[
parallel_sort
broadcast[
Instanciovaný operátor je znázorneˇn na Obrázku 9.
Bližší podrobnosti vcˇetneˇ detailní implementace operátoru
preprocess a join lze nalézt v cˇlánku [
broadcast
broadcast
parallel_join
join[
Do budoucna plánujeme rozšírˇit Bobolang tak, aby podporoval rovneˇž distribuované systémy. Bude tedy možné snadno specifikovat, jak rozdistribuovat exekucˇní plán mezi více uzl˚u, prˇípadneˇ nechat interpret jazyka rozdistribuovat plán automaticky.