Praktické cvičenie: Paralelné počítanie prvočísel

Cieľ

Implementujte výpočet počtu prvočísel v intervale najprv sekvenčne a potom paralelne pomocou viacerých procesov.

Študent si precvičí:

• rozdelenie CPU-bound úlohy na nezávislé časti,
• rozdiel medzi sekvenčným a paralelným riešením,
• čisté funkcie vhodné pre multiprocessing,
• použitie ProcessPoolExecutor,
• skladanie čiastkových výsledkov,
• testovateľný návrh paralelného kódu bez krehkého merania času.

Zadanie

Vytvorte súbor solution.py a implementujte funkcie:

from concurrent.futures import ProcessPoolExecutor

def is_prime(number):
    pass

def count_primes_in_range(chunk):
    pass

def split_range(start, end, parts):
    pass

def count_primes_sequential(start, end):
    pass

def count_primes_parallel(start, end, workers=4, executor_class=ProcessPoolExecutor):
    pass

Intervaly v tomto zadaní sú polootvorené:

[start, end)

To znamená, že start do intervalu patrí a end už nie.

1. is_prime(number)

Funkcia vráti True, ak je number prvočíslo, inak False.

Príklady:

is_prime(2)
# True

is_prime(17)
# True

is_prime(1)
# False

is_prime(21)
# False

Požiadavky:

• čísla menšie ako 2 nie sú prvočísla,
• netestujte deliteľov väčších ako druhá odmocnina čísla,
• funkcia nemá meniť globálny stav.

2. count_primes_in_range(chunk)

Funkcia dostane dvojicu (start, end) a vráti počet prvočísel v intervale [start, end).

Príklad:

count_primes_in_range((0, 10))
# 4

Prvočísla v intervale [0, 10) sú 2, 3, 5, 7.

Táto funkcia je pracovná jednotka vhodná pre executor.map().

3. split_range(start, end, parts)

Funkcia rozdelí interval [start, end) na najviac parts neprázdnych častí.

Príklad:

split_range(0, 10, 3)
# [(0, 4), (4, 7), (7, 10)]

Požiadavky:

• časti musia pokryť celý interval bez medzier a prekryvov,
• dĺžky častí sa môžu líšiť najviac o 1,
• ak je interval prázdny, vráťte prázdny zoznam,
• ak parts <= 0, vyhoďte ValueError,
• ak je parts väčšie než počet hodnôt v intervale, vráťte iba neprázdne časti.

Príklady:

split_range(5, 5, 4)
# []

split_range(0, 3, 10)
# [(0, 1), (1, 2), (2, 3)]

4. count_primes_sequential(start, end)

Funkcia vráti počet prvočísel v intervale [start, end) sekvenčne v jednom procese.

Príklad:

count_primes_sequential(0, 20)
# 8

Prvočísla menšie ako 20 sú:

2, 3, 5, 7, 11, 13, 17, 19

5. count_primes_parallel(start, end, workers=4, executor_class=ProcessPoolExecutor)

Funkcia vráti počet prvočísel v intervale [start, end) paralelne.

Postup:

1. Rozdeľte interval pomocou split_range(start, end, workers).
2. Spustite count_primes_in_range nad jednotlivými časťami cez executor.map(...).
3. Spočítajte čiastkové výsledky.

Odporúčaná štruktúra:

chunks = split_range(start, end, workers)

with executor_class(max_workers=workers) as executor:
    partial_counts = executor.map(count_primes_in_range, chunks)

return sum(partial_counts)

Parameter executor_class nechajte v rozhraní funkcie. Automatické testy vďaka nemu vedia použiť falošný executor a overiť rozdelenie práce bez spúšťania skutočných procesov.

Požiadavky:

• pre prázdny interval vráťte 0,
• pre workers <= 0 vyhoďte ValueError,
• výsledok musí byť rovnaký ako pri count_primes_sequential,
• pracovná funkcia musí byť definovaná na najvyššej úrovni modulu, aby bola použiteľná v multiprocessing.

Automatické hodnotenie

Automatické testy nemajú merať čas. Meranie výkonu je vhodné na demonštráciu, ale v hodnotiacom systéme býva nestabilné. Namiesto toho testy overujú:

• správnosť výpočtu,
• správne rozdelenie intervalov,
• to, že paralelná funkcia používa executor a posiela mu časti práce.

Testy

import pytest

from solution import (
    is_prime,
    count_primes_in_range,
    split_range,
    count_primes_sequential,
    count_primes_parallel,
)

class FakeExecutor:
    instances = []

    def __init__(self, max_workers=None):
        self.max_workers = max_workers
        self.mapped_function = None
        self.mapped_items = None
        FakeExecutor.instances.append(self)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, traceback):
        return False

    def map(self, function, items):
        self.mapped_function = function
        self.mapped_items = list(items)
        return [function(item) for item in self.mapped_items]

def test_is_prime():
    primes = [2, 3, 5, 7, 11, 13, 17, 19, 97]
    non_primes = [-5, -1, 0, 1, 4, 6, 9, 21, 100]

    for number in primes:
        assert is_prime(number) is True

    for number in non_primes:
        assert is_prime(number) is False

def test_count_primes_in_range():
    assert count_primes_in_range((0, 10)) == 4
    assert count_primes_in_range((10, 20)) == 4
    assert count_primes_in_range((20, 30)) == 2
    assert count_primes_in_range((30, 30)) == 0

def test_split_range_balanced_chunks():
    chunks = split_range(0, 10, 3)

    assert chunks == [(0, 4), (4, 7), (7, 10)]

    lengths = [end - start for start, end in chunks]
    assert max(lengths) - min(lengths) <= 1

def test_split_range_more_parts_than_numbers():
    assert split_range(0, 3, 10) == [(0, 1), (1, 2), (2, 3)]

def test_split_range_empty_interval():
    assert split_range(5, 5, 4) == []
    assert split_range(10, 5, 4) == []

def test_split_range_invalid_parts():
    with pytest.raises(ValueError):
        split_range(0, 10, 0)

def test_count_primes_sequential():
    assert count_primes_sequential(0, 20) == 8
    assert count_primes_sequential(100, 130) == 6
    assert count_primes_sequential(10, 10) == 0

def test_parallel_matches_sequential_with_fake_executor():
    FakeExecutor.instances.clear()

    parallel = count_primes_parallel(
        0,
        100,
        workers=4,
        executor_class=FakeExecutor,
    )
    sequential = count_primes_sequential(0, 100)

    assert parallel == sequential == 25

    executor = FakeExecutor.instances[-1]
    assert executor.max_workers == 4
    assert executor.mapped_function is count_primes_in_range
    assert executor.mapped_items == split_range(0, 100, 4)

def test_parallel_empty_interval_does_not_create_work():
    FakeExecutor.instances.clear()

    assert count_primes_parallel(10, 10, workers=4, executor_class=FakeExecutor) == 0
    assert FakeExecutor.instances == []

def test_parallel_invalid_workers():
    with pytest.raises(ValueError):
        count_primes_parallel(0, 100, workers=0, executor_class=FakeExecutor)

Skryté testy

Skryté testy môžu overovať:

• väčšie intervaly,
• intervaly začínajúce záporným číslom,
• rôzne počty workerov,
• workers väčšie než dĺžka intervalu,
• to, že paralelné a sekvenčné riešenie dávajú rovnaký výsledok,
• to, že count_primes_parallel rešpektuje parameter executor_class.

Odporúčaná stratégia riešenia

• is_prime: špeciálne ošetrite čísla menšie ako 2, párne čísla a potom skúšajte nepárnych deliteľov po druhú odmocninu.
• split_range: vypočítajte základnú veľkosť časti cez delenie a zvyšok rozdeľte medzi prvé časti.
• count_primes_in_range: použite sum(1 for number in range(start, end) if is_prime(number)).
• count_primes_parallel: neukladajte výsledky do globálnej premennej; každá práca má vrátiť svoje číslo a hlavný proces ich spočíta.

Hodnotenie

Časť	Body
is_prime	3
count_primes_in_range	2
split_range	4
count_primes_sequential	2
count_primes_parallel	5
Ošetrenie prázdnych intervalov a neplatného počtu workerov	2
Čitateľnosť kódu a čisté funkcie bez zdieľaného stavu	2

Spolu: 20 bodov.

Meranie výkonu na cvičení

Toto nie je vhodné ako automatický test, ale je vhodné na diskusiu počas cvičenia:

import time

start = time.perf_counter()
print(count_primes_sequential(0, 500_000))
print("sequential:", time.perf_counter() - start)

start = time.perf_counter()
print(count_primes_parallel(0, 500_000, workers=4))
print("parallel:", time.perf_counter() - start)

Výsledok závisí od počtu jadier, operačného systému a aktuálneho zaťaženia počítača. Preto sa výkon nemá hodnotiť pevným časovým limitom.

Pri spúšťaní benchmarku zo súboru dajte merací kód pod podmienku:

if __name__ == "__main__":
    ...

Je to dôležité najmä pri multiprocessing na systémoch, ktoré nové procesy štartujú cez import hlavného modulu.

Bonus: Race condition a Lock

Pridajte funkciu:

def locked_counter(workers, increments_per_worker):
    pass

Funkcia vytvorí viac vlákien. Každé vlákno zvýši spoločný čítač increments_per_worker krát. Použite threading.Lock, aby bol výsledok deterministický.

Očakávaný výsledok:

locked_counter(4, 1000)
# 4000

Túto časť je vhodné ukázať aj bez zámku a porovnať správanie, ale do automatického hodnotenia zaraďte iba verziu so zámkom.

Previous Post Next Post