1. gyakorlat

Párhuzamos számítógéprendszerek

A számítástechnikában a párhuzamos programozás azt jelenti, hogy egy problémát kisebb részfeladatokra bontunk, amelyeket egyszerre, párhuzamosan hajthatunk végre több végrehajtási egység segítségével. A párhuzamosság tehát a feladatok időbeli átfedésére vagy egyidejű végrehajtására utal. A végrehajtási egység itt egy magasabb absztrakciós szintet képvisel, mivel a párhuzamosságot különböző szinteken alkalmazhatunk, például több számítógép között, egy számítógépen belül vagy akár a számítógépen futó program szintjén is. Ezért minden esetben fontos tisztázni, hogy a párhuzamosság alatt párhuzamos hardverrendszerekre vagy párhuzamos szoftverrendszerekre utalunk éppen.

Hardver

Hardverrendszerek esetén Flynn-féle osztályozás négy fő csoportot különböztet meg aszerint, hogy mennyi utasításfolyam (instruction stream), illetve adatfolyam (data stream) különíthető el a rendszerben.

• SISD (Single Instruction stream, Single Data stream): egy egyszerű, egyprocesszoros gép, ahol az összes műveletet szekvenciálisan hajtják végre

• SIMD (Single Instruction stream, Multiple Data streams): a GPU-k képesek egyszerre több adatot feldolgozni ugyanazzal az utasítással, például több pixel vagy vektor műveletet végezhetnek párhuzamosan

• MIMD (Multiple Instruction streams, Multiple Data streams): manapság elterjedten használt párhuzamos számítógépek, ahol minden processzor saját utasítássorozatot hajt végre különböző adatfolyamokon

• MISD (Multiple Instruction streams, Single Data stream): a gyakorlatban nem terjedt el, de a megbízhatóság növelésére szolgálhatnak (redundáns adatfeldolgozás)

Elosztott rendszerek (Grid Computing, Cloud Computing)

Egy hálózatba kötött, földrajzilag elosztott, heterogén számítógépek hálózata ideálisan alkalmas párhuzamos programozás alkalmazására, mivel lehetővé teszi a különböző számítási feladatok párhuzamos végrehajtását több, egymástól független végrehajtási egységen, miközben az erőforrásokat hatékonyan osztja meg a hálózaton keresztül.
Ezzel viszont ezen a gyakorlaton nem foglalkozunk. Ahogy GPU programozással sem. 🙁

Szoftver

A processzorok teljesítményének növekedését hosszú ideig a tranzisztorok egyre növekvő számával érték el. Ahogyan a tranzisztorok mérete csökkent, úgy nőtt a sebesség, ezzel párhuzamosan viszont nőtt az energiafogyasztás is. Azonban az energia hővé alakulása az áramkörök túlmelegedéshez vezethetnek, korlátozva az egyprocesszoros rendszerek fejleszthetőségét. A számítási teljesítmény növelése érdekében az ipar a párhuzamosságra váltott: több, "egyszerűbb" processzort helyeznek el az integrált áramkörre, amelyeket többmagos processzoroknak nevezünk.

Ha van már egy párhuzamos végrehajtásra alkalmas hardverünk, akkor arra több fajta párhuzamos programozási szoftvermodellt is leképezhetünk.

• Közös memóriás párhuzamos programozás: folyamatokból és egy közös memóriából áll (a közös memória azt jelenti, hogy a folyamatok ugyanazokat a változókat használhatják, akár egyidőben is megkísérelhetik őket olvasni vagy írni - ennek hatékony menedzselése jelenti a legnagyobb kihívást)

Folyamatinterakciók

A közös memóriás programozás nagy figyelmet kell fordítanunk arra, hogy hogyan biztosítjuk az erőforrásokhoz való kizárólagos hozzáférést (kölcsönös kizárás), hogyan szinkronizáljuk a szálakat és hogyan kerüljük el a programunk holtpontba jutását.

• Osztott memóriás párhuzamos programozás: folyamatainak olyan változói vannak, amelyeket azok kizárólagosan használhatnak, itt tehát az egyidejű hozzáférés ki van zárva (a folyamatok együttműködése üzenetküldéssel valósul meg).

Közös változók hiánya

Az osztott memóriás programozás esetén a folyamatok egymás közötti adatcseréjén, azaz az üzenetküldésen és annak szinkronizációján van a hangsúly.

Közös memóriás párhuzamos programozás esetén, amikor egy felhasználó elindít egy programot, az operációs rendszer létrehoz egy folyamatot (process), azaz egy futó példányt a programból, amely saját memóriaterülettel rendelkezik. A szálak (thread) egy adott folyamat belső végrehajtási egységei, amelyek hozzáférnek a folyamat által használt memóriaterülethez. A szálkezelés lehetőséget biztosít a programozók számára, hogy a programjaikat többé-kevésbé független feladatokra bontsák fel.

Osztott memóriás párhuzamos programok esetén pedig a végrehajtási egységnek magát a párhuzamosan futó folyamatokat tekintjük, amelyek saját memóriaterületükkel rendelkeznek, és üzenetküldés segítségével kommunikálnak egymással.

Lightweight végrehajtási egységek

A hagyományos szálakkal (közvetlenül az operációs rendszer által kezelt erőforrások) ellentétben bizonyos programozási nyelvekben elérhetőek a futtató környezet által közvetlenül kezelt végrehajtási egységek (is). Előnye a kisebb erőforrásigény és a jobb skálázhatóság.

Parallelism vs. Concurrency

Párhuzamosság esetén több feladat egyszerre, párhuzamosan fut, különböző processzorokon vagy magokon, így gyorsítva a számításokat. Ezzel szemben konkurens programozás esetén is több feladattal foglalkozunk, de nem feltétlenül párhuzamosan, hanem felváltva fut, miközben egyetlen processzor osztja meg az időt a különböző feladatok között. A két megközelítés nem zárja ki egymás jelenlétét, egy implementáció lehet egyszerre párhuzamos és konkurens is.

Párhuzamosság előnyei

• Teljesítményjavulás: Megfelelő hardveres támogatás esetén a párhuzamos programozás gyorsabb és hatékonyabb az erőforrás-kihasználása, mivel több végrehajtási egység dolgozik egyszerre. Ez különösen fontos nagy számításigényű alkalmazásoknál, például tudományos számításoknál, szimulációknál vagy mesterséges intelligencia algoritmusoknál.

• Természetesebb problémaleírás: A párhuzamosság lehetővé teszi, hogy a problémát az összetevőire bontva, természetesebb módon közelítsük meg. A párhuzamos programozási nyelvek általában tartalmaznak szekvenciális utasításokat, amelyeket kiegészítenek párhuzamos konstrukciókkal, ezáltal egyszerűsítve a komplexebb problémák implementációját.

Például egy nagy adatbázisban végzett keresési feladat esetén a párhuzamosság azt jelenti, hogy az adatbázis különböző részeit egyidejűleg több szál dolgozza fel, így csökkenhet a végrehajtási idő. A probléma (egyszerűsített) reprezentációja:

Szekvenciális keresés: 23 kocka | 2 szálas keresés: 14 kocka | 6 szálas keresés: 3 kocka

Nem mindig számíthatunk ekkora javulásra!

Például 2 szál nem kétszer, 3 szál nem háromszor, és 6 szál sem hatszor gyorsítja a végrehajtást. A párhuzamosság hatékonyságát korlátozhatja a szekvenciális részek aránya, a szinkronizációs költségek és az erőforrások korlátai.

Párhuzamosság Java-ban

A JDK 21 hivatalos Javadoc-ja itt érthető el.

Thread osztály, Runnable interfész

Most, hogy megértettük, mi a párhuzamosság és milyen előnyökkel jár, nézzük meg, hogyan valósítható meg mindez a Java programozási nyelvben. A nyelv már az első verziójától kezdve támogatja a párhuzamosságot, és számos eszközt biztosít arra, hogy a programok egyszerre több szálon fussanak.

A Java-ban a végrehajtási egységet szálnak nevezzük. Szekvenciális programok esetében az összes utasítás a main metódusban fut le, amelyet a main thread hajt végre. A fő szál által létrehozott újabb szálak használatával lehetőség nyílik arra, hogy a program különböző részei párhuzamosan hajtsanak végre műveleteket, ezzel felgyorsítva a feldolgozást és növelve a teljesítményt.

A Java-ban a párhuzamosságot a Thread osztály és a Runnable interfész biztosítja, amelyeket felhasználva könnyedén hozhatunk létre saját szálakat, és meghatározhatjuk, hogy ezek milyen feladatokat végezzenek. A szálak működési logikáját a run() metódus felülírásával tudjuk változtatni. Miután meghatároztuk a működést, a szálakat a start() metódussal fogjuk tudni elindítani.

Tekintsük a HelloWorld problémának a megoldását párhuzamos Java-ban!

public class HelloWorld {

    public static void main(String[] args) {
        Thread thread = new ExampleThread();
        Thread runnable = new Thread(new ExampleRunnable());

        thread.start();
        runnable.start();
    }
}

class ExampleThread extends Thread {
    @Override
    public void run() {
        System.out.println("Hello World - Thread");
    }
}

class ExampleRunnable implements Runnable {
    @Override
    public void run() {
        System.out.println("Hello World - Runnable");
    }
}

Háttérben futó szálak

A main thread által végreahajtott szekvenciális programokat tekinthetjük egyszálú vezérlésnek, de ez nem jelenti azt, hogy nincsenek további, a háttérben futó szálak. Például a Java rendszer által kezelt daemon szálak olyan háttérfolyamatokat futtatnak, mint a garbage collector vagy a finalizer, és automatikusan megszűnnek, amikor az összes non-daemon szál befejeződött.

A main thread terminálása nem jelenti a teljes program végét, ha van még futó, nem-démon szál. A JVM aktívan tartja az alkalmazást mindaddig, amíg legalább egy ilyen szál fut.

A Thread osztály számos hasznos metódust biztosít, amelyek segítségével kezelhetjük és monitorozhatjuk a szálak működését, állapotát. Az alábbiakban néhány fontosabb metódust emelünk csak ki:

• Thread.currentThread(): Az aktuálisan futó szálra mutató referencia lekérése.

• Thread.sleep(n): A szál futását egy megadott időre (n: milliszekundum) felfüggeszti.

• threadId(): A szál egyedi azonosítóját adja vissza. A getId() 19-es verzió óta deprecated, használata hosszútávon váratlan viselkedést eredményezhet!

• getName(): A szál nevét adja vissza. Ez a szálnak adott tetszőleges név, nem egyedi.

• isAlive(): Ellenőrzi, hogy az adott szál éppen végrehajtás alatt van-e.

• isDaemon(): Ellenőrzi, hogy az adott szál daemon szál-e.

• getPriority() és setPriority(): A szál prioritásának lekérdezése vagy beállítása. A nagyobb prioritású szál nagyobb eséllyel kap processzoridőt.

• yield(): Egy jelzés az ütemezőnek, hogy a jelenlegi szál hajlandó átadni a vezérlést más szálaknak, de nem garantálja, hogy ez a váltás valóban meg is történik.

• getState(): Lekérdezi a szál aktuális állapotát.

• join(): Biztosítja, hogy egy szál befejezze a futását, mielőtt a program továbblépne a következő utasításra.

InterruptedException

Van arra is lehetőség, hogy egy másik szál megszakítson egy szálat, jelezve neki, hogy le kell állnia a végrehajtással, amint lehetséges. Erre akkor lehet szükség, ha egy szál blokkoló műveletet végez, például túl hosszú ideig várakozik (sleep(n)). Ahhoz, hogy a szál megfelelően kezelhesse a megszakítást, bizonyos metódusok InterruptedException kivételt dobnak.

Példák a felsorolt metódusokra:

public class ThreadMethods {

    public static void main(String[] args) {
        Thread thread = new MethodThread("Snicker");
        thread.start();
    }
}

class MethodThread extends Thread {

    MethodThread(String name){
        super(name);
    }

    @Override
    public void run() {
        System.out.println("Current thread: " + this);  // megegyezik a Thread.currentThread() metódussal
        System.out.println("Id: " + this.threadId());
        System.out.println("Name: " + this.getName());
        System.out.println("Alive: " + this.isAlive());
        System.out.println("Daemon: " + this.isDaemon());
        System.out.println("Priority: " + this.getPriority());
    }
}

Szálállapotok

Egy szál életciklusa során különböző állapotokat vehet fel, amit a fent említett Thread.getState() metódussal hívhatunk le.

• NEW: A szálat létrehozták, de még nem indították el a start() metódussal. Ebben az állapotban a szál nem hajt végre semmilyen műveletet.

• RUNNABLE: A szál készen áll a futásra, és a JVM megpróbálja ütemezni a processzoron. Bár "futásra kész", nem feltétlenül fut azonnal, mivel más szálak is versenyezhetnek az erőforrásokért.

• TERMINATED: A szál a run() metódus végrehajtásával kerül ebbe az állapotba. Végállapot, a szál nem indítható el ismét.

• TIMED_WAITING: A szál egy meghatározott ideig vár, például a sleep() metódus hívásakor. Az idő lejárta után automatikusan visszakerülhet a RUNNABLE állapotba.

• WAITING: A szál határozatlan ideig vár egy másik szál műveletére vár, hogy folytathassa a végrehajtást. Tipikusan valamilyen feltétel teljesülésére van szükség a folytatáshoz, amihez explicit értesítés szükséges.

• BLOCKED: Amikor egy másik szál jelenleg használja a szükséges erőforrást, és a várakozó szál nem tudja folytatni a végrehajtást, amíg az erőforrás zárolása fel nem oldódik.

WAITING & BLOCKED állapot

Ezeket az állapotokat egy későbbi órán tárgyaljuk részletesebben, mivel a teljes megértésükhöz a monitor koncepció és a szálak közötti szinkronizációs mechanizmusok ismerete szükséges.

public class ThreadStates {

    public static void main(String[] args) {
        Thread thread1 = new Thread();
        System.out.println(thread1.getState()); // NEW

        Thread thread2 = new RunnableThread();
        thread2.start();

        Thread thread3 = new TimedWaitingThread();
        thread3.start();

        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println(thread3.getState()); // TIMED_WAITING

        System.out.println(thread2.getState()); // TERMINATED
    }
}

class RunnableThread extends Thread {
    @Override
    public void run() {
        System.out.println(this.getState()); // RUNNABLE
    }
}

class TimedWaitingThread extends Thread {
    @Override
    public void run() {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

Párhuzamosság teljesítmény javítása

Nézzünk meg egy gyakorlati példát arra, hogyan javíthat a párhuzamosság egy szekvenciális program futási idején. Ebben a példában egy könyv szavainak előfordulási gyakoriságát szeretnénk megszámlálni. A választott könyv a Biblia, amit erről a linkről le lehet tölteni.

A megoldás során megvizsgálhatjuk, hogy a szekvenciális (1 szálú) megközelítést újabb szálak hozzáadásával párhuzamos feldolgozássá alakítva milyen mértékű javulást érünk el a futási idő tekintetében. A könyv soronkénti beolvasását követően felbontjuk egyenlő részekre, amelyeket aztán az egyes szálak párhuzamosan feldolgoznak (azaz összeszámolják, hogy a paraméterben kapott részhalmazban melyik szóból mennyi található). Az összegzéshez a szálak egy HashMap-et használnak, amelyet aztán egy másik, globális kulcs-érték párokat tartalmazó kollekcióval egyesítünk.

Folyamatinterakciók: szinkronizáció

A szinkronizációban egy küldő és egy fogadó folyamat vesz részt, mindkettőnek van egy szinkronizációs pontja. A fogadó nem lépheti át a saját szinkronizációs pontját, amíg a küldő el nem érte a sajátját, mivel a fogadónak szüksége van a küldő által előállított adatokra. Ha a fogadó ér hamarabb a szinkronizációs ponthoz, várakozik a küldőre. Ha a küldő ér oda előbb, kétféle viselkedés lehetséges: vagy bevárja a fogadót (szimmetrikus), vagy mindkét folyamat áthaladhat várakozás nélkül (asszimmetrikus).
Ez a mechanizmus kulcsfontosságú a párhuzamos programozásban, mivel garantálja a kívánt végrehajtási sorrendet és az adatok konzisztenciáját.

join()

Annak érdekében, hogy megelőzzük azt, hogy az egyik szál felülírja egy másik szál módosítását több megközelítéssel dolgozhatunk. Ebben a példában a join() metódust alkalmazzuk, amellyel egy speciális szinkronizációt valósítunk meg: az utasítást hívó szál (esetünkben ez a main lesz) addig blokkolásra kerül, amíg a metódust végrehajtó szál nem terminál.
A metódus esetén szintén gondoskodnunk kell az InterruptedException kivétel kezeléséről.
Egy időkorlát is megadható paraméterként, amely meghatározza, hogy mennyi ideig várakozzon a hívó szál a befejezére: join(2000)
A join használata kiváltható lenne más megoldással is, amely biztosítja a szálbiztos hozzáférést (a következő órákon ezekkel fogunk foglalkozni).

Ez azt jelenti, hogy szál által végzett összegzés rögzítése a globális adatstruktúrába a megadott sorrendben (a szálakat tartalmazó List iterálásával)történik - szekvenciálisan. Amennyiben a szálak által végzett "munka" különböző ideig tart, akkor egy hosszú ideig futó szálra történő várakozás plusz időt jelentene az összegzésnél. Viszont biztosak lehetünk benne, az eredmények a várt időben állnak rendelkezésre.

import java.io.*;
import java.util.*;
import java.util.concurrent.*;

public class WordCountParallel {

    public static void main(String[] args) throws InterruptedException, IOException {
        int numberOfThreads = 1;

        String fileName = "bible.txt";

        HashMap<String, Integer> totalWordCount = new HashMap<>();
        List<String> lines = new ArrayList<>();
        List<WordCounterThread> threads = new ArrayList<>();

        long startTime = System.nanoTime(); // időmérés kezdete

        // sorok beolvasása
        BufferedReader reader = new BufferedReader(new FileReader(fileName));
        String line;
        while ((line = reader.readLine()) != null) {
            lines.add(line);
        }
        reader.close();

        // szálak létrehozása
        int chunkSize = lines.size() / numberOfThreads;
        for (int i = 0; i < numberOfThreads; i++) {
            int start = i * chunkSize;
            int end = (i == numberOfThreads - 1) ? lines.size() : (i + 1) * chunkSize;

            List<String> part = lines.subList(start, end);
            WordCounterThread thread = new WordCounterThread(part, new HashMap<>());
            thread.start();
            threads.add(thread);
        }

        // a szálak eredményeinek rögzítése
        for (WordCounterThread thread : threads) {
            thread.join();
            for (Map.Entry<String, Integer> entry : thread.wordCount.entrySet()) {
                totalWordCount.merge(entry.getKey(), entry.getValue(), Integer::sum);
            }
        }

        long endTime = System.nanoTime(); // időmérés vége

        List<Map.Entry<String, Integer>> sortedWords = new ArrayList<>(totalWordCount.entrySet());
        sortedWords.sort(
                (entry1, entry2) -> entry2.getValue().compareTo(entry1.getValue())
        );

        sortedWords.stream()
                .limit(10)
                .forEach(entry -> System.out.println(entry.getKey() + ": " + entry.getValue()));
        System.out.println("Runtime: " + (endTime - startTime) / 1_000_000 + " ms");
    }
}

class WordCounterThread extends Thread {
    List<String> lines;
    HashMap<String, Integer> wordCount;

    public WordCounterThread(List<String> lines, HashMap<String, Integer> wordCount) {
        this.lines = lines;
        this.wordCount = wordCount;
    }

    @Override
    public void run() {
        for (String line : lines) {
            String[] words = line.toLowerCase().split("[^a-z]+");
            for (String word : words) {
                if (!word.isEmpty()) {
                    wordCount.put(word, wordCount.getOrDefault(word, 0) + 1);
                }
            }
        }
    }
}

Vizsgáljuk meg, hogy más input esetén mennyi szál esetén érünk el javulást a futási idő tekintetében. Minden inputra érdemes több szálat létrehozni? Használjuk a Runtime.getRuntime().availableProcessors() utasítást, amely visszadja a futtatókörnyezet számára elérhető CPU-magok számát.

Megjegyzés

Attól függetlenül, hogy a Biblia körülbelül 800.000 szót tartalmaz, nem minősül egy túl nagy inputnak. Bizonyos területeken, például nagyszabású szövegbányászati vagy genomikai elemzéseknél, a párhuzamosítás által nyújtott gyorsulás kritikus fontosságú lehet. Ezekben az esetekben a különbség a szekvenciális és párhuzamos feldolgozás között több nagyságrendű lehet.

Gyakorló feladatok

Feladat

Mi történik az alábbi programrészletben?

public static void main(String[] args) throws InterruptedException {
    System.out.println("Hello");

    Thread.currentThread().join();

    System.out.println("Word!");
}

Feladat

Hozz létre egy MyRunnable nevű osztályt, amely megvalósítja a Runnable interfészt, és implementálja a run() metódust. Az osztály konstruktora egy egész szám típusú paramétert vár, amelyet a count nevű változóban tárol.

A run() metódus feladata, hogy a count változó értékétől visszafelé számoljon nulláig (for ciklus segítségével), és minden lépésben kiírja a szám értékét a standard outputra sortöréssel.

Hozz létre egy MyThread nevű osztályt, amely a Thread osztályból származik, és szintén implementálja a run() metódust. Az osztály konstruktora egy MyRunnable típusú objektumot vár paraméterként, amelyet eltárol a myRunnable változóban. A run() metódus indítsa el a konstruktorban kapott objektumot szálként.

Feladat

Hozz létre egy Resource nevű osztályt. Az osztály konstruktorának egyetlen egész szám típusú paramétere van, amely a szálak számát jelöli. A konstruktor feladata, hogy inicializálja és feltöltse az array elnevezésű adattagot, amely egy egész számokat tartalmazó tömb. A tömb mérete a megadott szálak számának százszorosával egyenlő (szálak száma × 100), a szálak számát paraméterként kapja meg a konstruktor.

Ezután hozz létre egy Calculator osztályt, amely a Thread osztályból származik és implementálja a run() metódust. A Calculator osztály konstruktora három paramétert fogad:

• egy Resource típusú objektum referenciáját,

• egy egész számot, amely egy intervallum kezdőértékét jelöli,

• egy másik egész számot, amely az intervallum végértékét adja meg.

A run() metódus feladata, hogy végigiteráljon a megadott intervallumon belül található számokon az array tömbben, és az összeget eltárolja a Calculator osztály sum elnevezésű adattagjában.

Teszteld a programot a main metódusban. A metódus szeletelje fel a szálak számával megegyező intervallumra a tömböt és minden intervallumhoz példányosítson egy új Calculator szálat az összegzés elvégzésére. A szálak elindítása után a join() metódus segítségével biztosítani kell, hogy a főprogram csak akkor folytatódjon, ha minden szál befejezte a számításait.

---

Utolsó frissítés: 2025-03-09 09:46:28