Distribuované programování

Distribuované programování je paradigma, při kterém se výpočetní úloha rozkládá na více nezávislých výpočetních jednotek (procesů, vláken, uzlů, serverů), které mezi sebou komunikují a spolupracují na dosažení společného cíle. Na rozdíl od klasického sekvenčního programu, který běží v jednom vlákně na jednom stroji, distribuovaný systém řeší problémy škálování, odolnosti proti výpadkům, konzistence dat a složitosti komunikace.

Typy škálování

Škálování je schopnost systému zvládat rostoucí zátěž. Rozlišujeme dva základní přístupy – vertikální a horizontální – a dvě strategie podle toho, zda aktivně reagujeme na zátěž.

Vertikální škálování (Scale Up)

Vertikální škálování znamená zvýšení výkonu existujícího stroje – přidání CPU jader, RAM, rychlejších disků (SSD místo HDD), výkonnější GPU.

Horizontální škálování (Scale Out)

Horizontální škálování znamená přidání dalších strojů do systému a rozložení zátěže mezi ně.

Reaktivní škálování

Reaktivní (též elastické) škálování je strategie, kdy systém automaticky přidává nebo odebírá zdroje podle aktuální zátěže. Běžně se používá v cloudovém prostředí (AWS Auto Scaling, Kubernetes HPA/VPA).

• Scale Out: Přidání nové instance/kontejneru při zvýšené zátěži.
• Scale In: Odebrání přebytečné instance při nízké zátěži.
• Scale Up: Zvětšení CPU/RAM limitu stávající instance (např. v Kubernetes).
• Scale Down: Snížení alokovaných zdrojů.

CAP teorém

CAP teorém (též Brewerův teorém) je fundamentální princip distribuovaných systémů, který říká, že v distribuovaném datovém úložišti lze garantovat nejvýše dvě z následujících tří vlastností současně:

graph TD
    CAP[CAP teorém]
    CAP --> C[Konzistence<br/>Consistency]
    CAP --> A[Dostupnost<br/>Availability]
    CAP --> P[Odolnost proti rozdělení<br/>Partition Tolerance]
    C -.-> CP[CP systém]
    A -.-> AP[AP systém]
    P -.-> CP
    P -.-> AP

Proč nejde mít všechny tři?

Představ si dva uzly databáze \(A\) a \(B\), mezi kterými se přerušila síť (partition). V tento moment máme dvě možnosti:

1. Zachovat konzistenci (C): Odmítnout zápisy na uzel \(B\), dokud se síť neobnoví – obětovali jsme dostupnost (A). Výsledek: CP systém.
2. Zachovat dostupnost (A): Povolit zápisy na oba uzly. Po obnovení sítě budou mít \(A\) a \(B\) rozdílná data – obětovali jsme konzistenci (C). Výsledek: AP systém.

Klasifikace systémů podle CAP

Koncepce paralelního a distribuovaného programování

Při přechodu od sekvenčního programu k distribuovanému se musíme vypořádat s novými výzvami: jak rozdělit práci, jak si předávat data, jak synchronizovat výpočet a jak se zotavit z chyb.

Paměťový model

Základní dělení podle toho, jak procesy sdílejí data:

Sdílená paměť (Shared Memory)

Všechna vlákna nebo procesy sdílejí společný adresní prostor. Komunikují čtením a zápisem do stejné paměti. Synchronizace se řeší pomocí mutexů, semaforů a atomických operací.

• Výhody: Rychlá komunikace (paměť je blízko), jednodušší myšlenkový model.
• Nevýhody: Souběh (race conditions), obtížné ladění, nelze škálovat mimo jeden stroj.
• Příklad: Vlákna v Javě, OpenMP, std::thread v C++.

Distribuovaná paměť (Distributed Memory / Message Passing)

Každý proces má vlastní izolovanou paměť. Komunikace probíhá výhradně posíláním zpráv přes síť. Žádná data nejsou sdílena implicitně – co si procesy nepošlou, to nevidí.

• Výhody: Lze škálovat na tisíce strojů, izolace pádů, jasné rozhraní.
• Nevýhody: Režie serializace/deserializace zpráv, latence sítě, složitější návrh.
• Příklad: MPI (Message Passing Interface), ZeroMQ, gRPC, RabbitMQ, Apache Kafka.

Programovací modely

Master–Worker (Task Farm)

Jeden hlavní proces (master) rozděluje práci mezi více podřízených procesů (workerů) a sbírá výsledky. Master se stará o rozdělení úlohy, distribuci, sledování postupu a agregaci výsledků.

• Výhody: Jednoduchý model, snadné škálování, přirozený pro mnoho úloh.
• Nevýhody: Master se může stát úzkým hrdlem, single point of failure.
• Příklad: MapReduce, distribuované renderování, vědecké výpočty s MPI.

Peer-to-Peer

Všechny uzly jsou si rovnocenné. Každý uzel může být klientem i serverem zároveň. Neexistuje centrální autorita.

• Výhody: Odolné proti výpadkům, žádné úzké hrdlo, decentralizované.
• Nevýhody: Složitější koordinace, obtížné zajištění globální konzistence.
• Příklad: BitTorrent, blockchain, distribuované hashovací tabulky (Kademlia, Chord).

Pipeline

Data protékají sekvencí fází, kde každá fáze provádí specifickou operaci a předává výsledek dál. Každá fáze běží paralelně na jiných datech.

• Výhody: Vysoká propustnost, přirozené pro streamingové zpracování.
• Nevýhody: Latence je součtem latencí všech fází; pomalá fáze brzdí celý pipeline.
• Příklad: Unixové roury (cat | grep | sort | uniq), ETL pipeline, kompilátor (lexer → parser → optimizer → codegen).

Producer–Consumer

Producenti generují data a vkládají je do fronty. Konzumenti data z fronty odebírají a zpracovávají. Komunikace je nepřímá – přes frontu, nikoliv napřímo.

• Výhody: Decoupling – producent a konzument se navzájem neznají; vyrovnávání zátěže.
• Nevýhody: Režie fronty, nutnost řešit backpressure (spotřebitel nestíhá).
• Příklad: Zpracování logů, objednávkový systém, streamování událostí.

SPMD (Single Program, Multiple Data)

Všechny procesy spouští stejný program, ale každý pracuje na jiné části dat. Každý proces má své ID (rank), podle kterého se rozhoduje, kterou část dat zpracuje.

• Výhody: Jednoduchý a jednotný kód pro všechny uzly.
• Nevýhody: Vyžaduje explicitní větvení v kódu podle ranku; méně flexibilní.
• Příklad: MPI programy (mpirun -np 4 ./program), CUDA kernely.

MPI (Message Passing Interface)

MPI je standardizovaný, přenositelný systém pro předávání zpráv, který je páteří většiny distribuovaných výpočtů v HPC (High-Performance Computing). Standard definuje API pro C, C++ a Fortran; mezi hlavní implementace patří OpenMPI, MPICH a Intel MPI.

Komunikátory

Komunikátor je základním konceptem MPI – představuje skupinu procesů a kontext, ve kterém probíhá komunikace. Každá zpráva je odeslána v rámci určitého komunikátoru a nemůže ho opustit.

• Výchozí komunikátor MPI_COMM_WORLD zahrnuje všechny procesy spuštěné při startu.
• Intra-komunikátor: Procesy uvnitř jedné skupiny – běžný případ (např. všechny procesy výpočtu).
• Inter-komunikátor: Propojení dvou různých skupin procesů – používá se např. při dynamickém připojování workerů.

int rank, color;
MPI_Comm_rank(MPI_COMM_WORLD, &rank);
color = rank % 2;  // sudé procesy do skupiny 0, liché do 1

MPI_Comm new_comm;
MPI_Comm_split(MPI_COMM_WORLD, color, rank, &new_comm);

Rank a velikost

• Rank (hodnost): Jednoznačný identifikátor procesu v rámci komunikátoru – celé číslo od 0 do size-1. Proces s rankem 0 se konvenčně označuje jako root a typicky řídí rozdělování práce a sběr výsledků.
• Velikost (size): Celkový počet procesů v komunikátoru, zjištěný voláním MPI_Comm_size.

#include <mpi.h>
#include <stdio.h>

int main(int argc, char** argv) {
    MPI_Init(&argc, &argv);

    int rank, size;
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    MPI_Comm_size(MPI_COMM_WORLD, &size);

    printf("Jsem proces %d z %d\n", rank, size);

    MPI_Finalize();
    return 0;
}

Základní sada operací

MPI poskytuje bohatou sadu operací, které lze rozdělit na point-to-point (komunikace mezi dvěma procesy) a kolektivní (komunikace zahrnující všechny procesy v komunikátoru).

Point-to-point operace

Kolektivní operace

Synchronizace

V MPI existuje několik úrovní synchronizace mezi procesy:

• Bariéra (MPI_Barrier): Nejjednodušší synchronizační primitivum – všechny procesy v komunikátoru se zastaví a čekají, dokud na bariéru nedorazí i poslední z nich. Používá se typicky mezi fázemi výpočtu, kde další fáze závisí na výsledcích všech procesů z fáze předchozí.
• Kolektivní operace jako synchronizační body: Většina kolektivních operací (MPI_Bcast, MPI_Reduce, …) implicitně synchronizuje – volání se dokončí až poté, co ho provedou všechny procesy.
• Point-to-point synchronizace: Dva procesy se synchronizují vzájemně pomocí send/recv dvojice – příjemce čeká na zprávu, odesílatel čeká na potvrzení.

Multicore vs. cluster mode

MPI podporuje dva základní režimy spouštění, které se liší způsobem komunikace mezi procesy:

• Multicore / Shared memory mode: Všechny MPI procesy běží na jednom fyzickém stroji (jednom uzlu) a komunikují přes sdílenou paměť. MPI implementace tento fakt detekuje a používá optimalizované přenosy – data se kopírují přímo v RAM bez síťové režie. Latence je řádově mikrosekundy.
• Cluster / Distributed memory mode: Procesy běží na různých fyzických uzlech propojených sítí (Ethernet, InfiniBand, OmniPath). Data putují přes síťové rozhraní. Latence je řádově desítky mikrosekund až milisekundy.

Validní příjem zprávy

Aby byl příjem zprávy úspěšný, musí se shodovat obálka (envelope) zprávy s parametry volání MPI_Recv. Obálku tvoří tři hodnoty:

Všechny tři musí souhlasit. Pokud ne, zpráva zůstane ve frontě a čeká na odpovídající MPI_Recv.

MPI_Status status;
MPI_Recv(buf, count, MPI_INT, MPI_ANY_SOURCE, MPI_ANY_TAG, comm, &status);
printf("Zpráva od procesu %d s tagem %d\n", status.MPI_SOURCE, status.MPI_TAG);

MPI_Status status;
MPI_Probe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
int count;
MPI_Get_count(&status, MPI_INT, &count);
int* buf = malloc(count * sizeof(int));
MPI_Recv(buf, count, MPI_INT, status.MPI_SOURCE, status.MPI_TAG, MPI_COMM_WORLD, MPI_STATUS_IGNORE);

Problém blokujících operací a jak je řešit

Blokující operace (MPI_Send, MPI_Recv) pozastaví volající proces, dokud není operace dokončena. To přináší dva zásadní problémy:

// Proces 0
MPI_Send(data_to_1, count, MPI_INT, 1, 0, comm);
MPI_Recv(data_from_1, count, MPI_INT, 1, 0, comm, &status);

// Proces 1
MPI_Send(data_to_0, count, MPI_INT, 0, 0, comm);
MPI_Recv(data_from_0, count, MPI_INT, 0, 0, comm, &status);

Řešení blokujících operací

MPI_Request request;
MPI_Isend(data, count, MPI_INT, dest, tag, comm, &request);
// Okamžitě pokračujeme – můžeme počítat další fázi
compute_next_phase();
// Po dokončení výpočtu počkáme, až se data skutečně odešlou
MPI_Wait(&request, MPI_STATUS_IGNORE);

Srovnání MPI s PVM

PVM (Parallel Virtual Machine) byl předchůdce MPI, který vznikl v 90. letech na Oak Ridge National Laboratory. Dnes je prakticky nahrazen MPI, ale historicky to byla první široce používaná knihovna pro distribuované výpočty.

Konzistenční modely

Zatímco CAP teorém říká, že si při partition musíme vybrat mezi C a A, konzistenční modely určují, jak silnou konzistenci systém garantuje za běžného provozu. Jsou to smlouvy mezi systémem a programátorem – „toto ti garantuju, zbytek si ošetři sám".

Silná konzistence (Strong Consistency)

Po potvrzení zápisu každé následující čtení (odkudkoliv) vrátí hodnotu tohoto zápisu. Systém se chová, jako by existovala jediná kopie dat – ve skutečnosti ji jen dokonale simuluje pomocí konsenzus protokolů.

• Výhody: Intuitivní – programátor nemusí řešit nekonzistence.
• Nevýhody: Vysoká latence (čekání na quorum), nízká dostupnost (při partition systém odmítá zápisy).
• Příklad: Linearizovatelné úložiště (ZooKeeper, etcd, Consul), databáze v CP režimu.

Sekvenční konzistence (Sequential Consistency)

Výsledek všech operací je stejný, jako by byly provedeny v nějakém sekvenčním pořadí – a operace z jednoho procesu se provedou v pořadí, v jakém byly zadány. Slabší než striktní (linearizovatelná), protože negarantuje, že ostatní procesy vidí zápis ihned.

• Příklad: Moderní CPU s out-of-order execution (x86 TSO model).

Příčinná konzistence (Causal Consistency)

Pokud operace A kauzálně předchází operaci B, pak všechny procesy uvidí A před B. Pokud spolu operace kauzálně nesouvisí (jsou nezávislé), jejich pořadí může být na různých replikách různé.

• Výhody: Nižší latence než sekvenční – nevyžaduje globální synchronizaci.
• Nevýhody: Složitější na pochopení – programátor musí chápat kauzální závislosti.
• Příklad: COPS databáze, Apache Cassandra s LOCAL_QUORUM a CL řádkem.

Eventualní konzistence (Eventual Consistency)

Pokud do systému nepřicházejí nové zápisy, nakonec se všechny repliky sjednotí na stejném stavu. Během replikace ale mohou různé repliky vracet různé hodnoty. Je to nejslabší model – ale nejškálovatelnější.

• Výhody: Nízká latence, vysoká dostupnost, horizontální škálování.
• Nevýhody: Klient může číst zastaralá data, konflikty (např. dva různé zápisy do stejného klíče).
• Příklad: DNS, DynamoDB (výchozí), Cassandra (výchozí), CouchDB.

Read-Your-Writes

Klient vždy uvidí své vlastní zápisy – i když jiné repliky ještě nemusí být aktuální. Typicky implementováno tak, že čtení po zápisu vždy jde na primární repliku.

Monotonic Reads

Klient nikdy neuvidí starší data, než viděl předtím. Zabraňuje „cestování v čase" – klient nemůže přejít na repliku, která ještě nemá jeho poslední čtení.

Srovnání konzistenčních modelů

Asynchronní komunikace

V distribuovaných systémech je synchronní komunikace (klient čeká na odpověď) často úzkým hrdlem a zdrojem kaskádových selhání. Asynchronní komunikace odděluje odesílatele od příjemce v čase i prostoru.

Synchronní vs. asynchronní komunikace

Fronty zpráv (Message Queues)

Fronta zpráv je mezivrstva mezi producentem a konzumentem, která umožňuje asynchronní komunikaci. Producent pošle zprávu do fronty a konzument si ji vyzvedne, až bude připraven.

Producent → [Fronta] → Konzument
              ↕
            Broker

Typy výměny zpráv

Garance doručení

Backpressure

Když konzument nestíhá zpracovávat zprávy, hrozí přetečení fronty a pád systému. Backpressure je mechanismus, kterým konzument dává najevo „zpomal, nestíhám":

• Pull model: Konzument si aktivně vyžádá dávku zpráv, až je připraven (např. consumer.poll() v Kafce).
• Flow control: Broker omezí producenta na základě stavu fronty.
• Reactive Streams: Standardizovaný protokol pro backpressure (RxJava, Project Reactor, Akka Streams).

Událostmi řízená architektura (Event-Driven Architecture)

Namísto přímého volání služeb produkují komponenty události (events), které popisují, co se stalo. Ostatní komponenty na tyto události reagují. Klíčová je inverze závislosti – producent události neví, kdo ji zpracuje.

Architektury paralelních systémů

Multiprocesorové systémy

Všechna jádra sdílejí společnou fyzickou paměť. Komunikace je rychlá, ale při větším počtu jader vzniká soupeření o paměťovou sběrnici (bus contention).

• UMA (Uniform Memory Access): Všechna jádra mají stejnou latenci přístupu do paměti. Typické pro malé SMP servery.
• NUMA (Non-Uniform Memory Access): Každý procesor má vlastní „blízkou" paměť. Přístup k cizí paměti je pomalejší. Typické pro velké servery (AMD EPYC, Intel Xeon).

Clusterové systémy

Skupina nezávislých počítačů propojených sítí, které se navenek tváří jako jeden systém.

Gridové systémy

Volně propojené, geograficky rozptýlené zdroje (počítače, úložiště), které spolupracují na řešení rozsáhlých problémů. Na rozdíl od clusteru nejsou uzly dedikované a jsou heterogenní.

• Příklad: SETI@home, Folding@home – dobrovolníci poskytují výpočetní výkon svých počítačů.

Cloudové systémy

Abstrakce nad fyzickou infrastrukturou – výpočetní zdroje jsou poskytovány jako služba přes internet.

Granularita

Granularita popisuje velikost jednotky práce, která je distribuována mezi výpočetní jednotky. Ovlivňuje režii, vyvážení zátěže a celkovou efektivitu paralelního systému. Granularitu lze posuzovat podle několika os.

Podle velikosti jednotky práce

Jemná granularita (Fine-Grained)

Úloha je rozdělena na velké množství malých jednotek. Každá jednotka práce je rychle hotová.

• Výhody: Lepší vyvážení zátěže, kratší čekání na dokončení poslední jednotky.
• Nevýhody: Vysoká režie na rozdělování, komunikaci a synchronizaci.
• Příklad: Zpracování jednotlivých pixelů obrázku, CUDA kernely nad maticí, actor model v Erlang/Elixir.

Hrubá granularita (Coarse-Grained)

Úloha je rozdělena na malé množství velkých jednotek. Každá jednotka pracuje dlouho samostatně.

• Výhody: Nízká režie, méně komunikace a synchronizace.
• Nevýhody: Horší vyvážení zátěže – jeden pomalý uzel zdrží celý výpočet, zatímco ostatní čekají.
• Příklad: Každý uzel počítá celou kapitolu simulace, MapReduce joby nad velkými bloky dat.

Smíšená granularita (Mixed / Hybrid)

Kombinuje oba přístupy – na vyšší úrovni hrubá, na nižší jemná. Například cluster uzlů počítá hrubé MapReduce joby, ale uvnitř každého jobu se použije SIMD vektorizace nebo GPU akcelerace.

Podle dynamiky rozdělení

Podle typu rozdělení

Konsenzus algoritmy

V distribuovaném systému je konsenzus dohoda mezi skupinou uzlů na jedné společné hodnotě – například: kdo je aktuální leader clusteru, zda byla transakce potvrzena, nebo která hodnota je ta správná při konfliktu replik. Konsenzus je teoreticky obtížný problém – FLP teorém (Fischer–Lynch–Paterson, 1985) dokázal, že v asynchronním systému nelze dosáhnout deterministického konsenzu, pokud může selhat byť jen jeden uzel.

Paxos

Paxos (Leslie Lamport, 1989) je rodina protokolů pro dosažení konsenzu v nespolehlivé síti. Patří k nejvlivnějším algoritmům distribuovaných systémů – prakticky všechny moderní konsenzus systémy z něj vycházejí.

Raft

Raft (Diego Ongaro, 2014) je konsenzus algoritmus navržený jako srozumitelnější alternativa k Paxosu. Jeho hlavním cílem je, aby ho šlo snadno pochopit, implementovat a ladit – proto je strukturovaný kolem tří jasně oddělených mechanismů.

Paxos vs. Raft

Distribuované transakce

Distribuovaná transakce je posloupnost operací nad více nezávislými uzly nebo databázemi, která musí splňovat ACID vlastnosti – i když každý uzel drží jinou část dat. Problém: pokud jeden uzel selže uprostřed transakce, musíme se umět vrátit zpět (rollback) napříč všemi zúčastněnými uzly.

Two-Phase Commit (2PC)

2PC je protokol, který zajišťuje, že se všechny uzly jednomyslně shodnou na potvrzení nebo zrušení transakce. Jeden uzel je určen jako koordinátor.

Saga

Saga je alternativa ke 2PC pro dlouho běžící transakce a mikroslužby. Namísto jednoho atomického commitu rozbíjí globální transakci na posloupnost lokálních transakcí – každá má vlastní commit. Pokud některá z nich selže, spustí se kompenzační transakce pro vrácení změn.

Flynnova klasifikace

Flynnova taxonomie (1966) je klasické rozdělení počítačových architektur podle toho, zda pracují nad jednou nebo více instrukcemi a nad jedním nebo více datovými proudy. Je to základní rámec pro pochopení paralelismu na úrovni hardware.

SISD (Single Instruction, Single Data)

Klasický von Neumannovský počítač. V každém hodinovém cyklu se načte a vykoná právě jedna instrukce nad právě jednou datovou položkou. Žádný paralelismus na úrovni instrukcí ani dat.

• Příklad: Původní procesory Intel 8086, jednoduché mikrokontroléry.

SIMD (Single Instruction, Multiple Data)

Jedna instrukce je aplikována na více datových prvků současně. Typicky se používá pro vektorové operace – sčítání dvou polí po prvcích, násobení matic, zpracování obrazu.

• Příklad: MMX, SSE, AVX instrukce v x86 procesorech; NEON v ARM; GPU (NVIDIA CUDA, OpenCL).

MISD (Multiple Instruction, Single Data)

Různé instrukce operují nad stejnými daty. Tato třída je spíše teoretická – v praxi nemá běžné využití.

• Příklad: Redundantní systémy s více výpočetními jednotkami, kde každá počítá jiným algoritmem a výsledky se porovnávají (např. avionika, řídící systémy raket). Pipeline procesor, kde každá fáze provádí jinou operaci nad daty procházejícími pipeline.

MIMD (Multiple Instruction, Multiple Data)

Každý procesor vykonává vlastní instrukce nad vlastními daty. To je model, který odpovídá většině dnešních paralelních a distribuovaných systémů. MIMD se dále dělí podle paměťového modelu:

Shrnutí pojmů