33.2 Thread, process és könyvtári szálak

33.2.1 Cél és konkrét helyzet

Ebben az alfejezetben egy kutatói gépen szeretnénk felgyorsítani két, eltérő jellegű munkafázist: több száz kis CSV-fájl kiírását és visszaolvasását (I/O-kötött feladat), valamint 12 darab $500 \times 500$-as valós mátrix szorzását (CPU-kötött NumPy-feladat). A megfigyelési egység egy feladatfutás, a mért eredmény a falidő mediánja öt ismétlésből. A cél annak eldöntése, hogy ThreadPoolExecutor vagy ProcessPoolExecutor illik-e az adott munkaterheléshez, és hogy a BLAS-könyvtár saját szálait hogyan kell kézben tartani. Előzetes becslés: az I/O feladatnál a szálak használnak, mert a Python a fájlműveletek alatt várakozik; a tiszta NumPy szorzásnál a natív kód felszabadítja a GIL-t, így a szálak is gyorsíthatnak, de csak akkor, ha a BLAS szálainak száma 1-re van korlátozva. A konkrét labkimenetet itt nem futtattuk, ezért a számokat „szemléltető” becslésként adjuk meg; a valódi méréshez a kódot a saját gépen kell lefuttatni.

33.2.2 Thread, process, GIL és natív kód

A CPython implementációban egyszerre csak egy szál hajthat végre Python-bájtkódot: ez a Global Interpreter Lock (GIL). A GIL nem általános Python-tulajdonság, hanem a CPython build sajátossága; a 3.13-tól létezik kísérleti free-threaded build, a PyPy és más implementációk pedig másként viselkednek. A tanulási főút CPython 3.12-t használ, ezért itt GIL-es modellt feltételezünk.

A GIL hatása két fázisban enyhül:

• I/O-várakozás: amikor a szál fájlműveletre, hálózatra vagy alvásra vár, a CPython elengedi a GIL-t, így egy másik szál futhat. Ezért a sok kis fájlt író-olvasó feladatra a threading vagy a ThreadPoolExecutor természetes választás.
• Natív kód: a NumPy nagy tömbműveletei C-ben futnak, és a CPython a hosszú számítások alatt szintén elengedi a GIL-t. Így egy A @ B mátrixszorzás közben más Python-szálak is dolgozhatnak. A Python-szintű ciklus viszont továbbra is a GIL alatt fut, tehát a tiszta Python számításokat a szálak nem párhuzamosítják.

A process ezzel szemben külön interpretert indít minden workerhez, saját memóriával és saját GIL-lel. A ProcessPoolExecutor vagy a multiprocessing ezért CPU-kötött Python-kódhoz is valódi párhuzamosságot ad. Az ára a serialization: a munkafüggvényt, az argumentumokat és a visszatérési értéket a pickle segítségével sorosítani kell a folyamatok között. Kis egész számoknál ez elhanyagolható; egy több száz megabájtos pandas DataFrame esetén viszont maga a másolás lehet a szűk keresztmetszet.

A NumPy, a SciPy és a scikit-learn nagy lineáris algebrai műveletei gyakran egy BLAS-könyvtárat használnak (OpenBLAS, MKL, Accelerate). A BLAS a saját, natív szintű szálait is elindíthatja; a szálak számát az OPENBLAS_NUM_THREADS, MKL_NUM_THREADS, OMP_NUM_THREADS környezeti változók szabályozzák, és ezeket az értékeket a NumPy importja előtt kell beállítani. Ha például 4 Python-worker mindegyike 8 BLAS-szálat indít egy 8 logikai magos gépen, akkor 32 szál verseng ugyanazért a CPU-ért: ez az oversubscription. A kontextusváltás és a gyorsítótár-versengés ilyenkor gyakran lassabb futást ad, mint a szekvenciális változat.

A joblib a lokális párhuzamosítás kényelmes rétege: joblib.Parallel(n_jobs=4, prefer="threads") vagy prefer="processes" mögött ugyanez a mechanizmus áll, és a scikit-learn n_jobs paraméterei is erre épülnek. A cProfile CPU-időt mér, de szálas programnál a profilozó overhead torzíthat; a párhuzamosítási döntéshez falidőt és CPU-időt együtt érdemes mérni. A memory_profiler a processz memóriaforgalmát figyeli, ami a process-alapú megoldásnál fontos, mert minden worker külön memóriát foglal.

33.2.3 Adatmozgatás és műveletszám becslése

Egy $n \times n$-es, float64 típusú mátrix szorzásánál a műveletszám közelítőleg

$$M_{\text{comp}} \approx 2n^3 \; \text{flop},$$

mert minden kimeneti elemhez $n$ szorzás és $n$ összeadás tartozik. A bemenő $A$, $B$ és a kimenő $C$ mátrix egyenként $8n^2$ bájtot foglal, így a minimális tárolási méret (nem a tényleges RAM-forgalom, amely cache- és implementációfüggő)

$$M_{\text{data,min}} \approx 3 \cdot 8n^2 \; \text{bájt}.$$

Ha a ténylegesen elérhető lebegőpontos teljesítmény $R_{\text{flop}}$ (GFLOP/s), és a RAM-sávszélesség $B_{\text{RAM}}$ (GB/s), akkor a két idő hányadosa

$$\frac{t_{\text{comp}}}{t_{\text{mem}}} \approx \frac{2n^3 / R_{\text{flop}}}{24n^2 / B_{\text{RAM}}} = \frac{n B_{\text{RAM}}}{12 R_{\text{flop}}}.$$

Szemléltető értékekkel: $n = 500$, $R_{\text{flop}} = 2$ GFLOP/s, $B_{\text{RAM}} = 20$ GB/s esetén $t_{\text{comp}} \approx 250\,000\,000 / 2 \cdot 10^9 = 0{,}125$ s feladatonként, míg $t_{\text{mem}} \approx 24 \cdot 250\,000 / 20 \cdot 10^9 \approx 0{,}3$ ms, tehát a hányados körülbelül 400. A feladat erősen CPU-kötött, ezért a szálak vagy processzek közötti adatmegosztás kevésbé számít, a CPU-versengés viszont annál inkább.

Az I/O-feladatnál a becslés más: $12$ worker $\times 25$ fájl $= 300$ kis fájl, és $12 \times 25 \times 2000 = 600\,000$ sor. A fájlonkénti nyitási/zárási késleltetés és a soronkénti írási költség dominál; a szálak akkor használnak, ha a Python a fájlműveletek alatt más workernek adja át a vezérlést.

33.2.4 Teljes minimális munkapélda

A következő script a concurrent.futures szabványos modullal hasonlítja össze a szekvenciális, szálas és processzes futást. A platform itt Linux, CPython 3.12, 4 fizikai mag; más platformon a ProcessPoolExecutor spawn viselkedése eltérhet. A scriptet a s130_workload.py fájlba mentjük.

import os

# A BLAS szálkeretét a NumPy importja előtt rögzítjük,
# különben a belső párhuzamosság oversubscriptiont okoz.
os.environ["OPENBLAS_NUM_THREADS"] = "1"
os.environ["MKL_NUM_THREADS"] = "1"
os.environ["OMP_NUM_THREADS"] = "1"

import platform
import statistics
import tempfile
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

import numpy as np

TASK_COUNT = 12
WORKERS = 4
REPEATS = 5


def io_task(worker_id: int) -> int:
    """Egy worker 25 kis CSV-fájlt ír és olvas vissza."""
    tmpdir = tempfile.gettempdir()
    seen = []
    for i in range(25):
        path = os.path.join(tmpdir, f"s130_{worker_id}_{i}.csv")
        with open(path, "w", encoding="utf-8") as f:
            for j in range(2000):
                f.write(f"{j},{worker_id},{j % 7}\n")
        with open(path, "r", encoding="utf-8") as f:
            n = sum(1 for _ in f)
        assert n == 2000
        seen.append(n)
        os.remove(path)
    return sum(seen)


def np_task(seed: int) -> float:
    """Egy 500x500-as mátrixpár szorzása determinisztikus seeddel."""
    rng = np.random.default_rng(seed)
    A = rng.normal(size=(500, 500))
    B = rng.normal(size=(500, 500))
    C = A @ B
    return float(C[0, 0])


def run_sequential(fn, n):
    t0 = time.perf_counter()
    results = [fn(i) for i in range(n)]
    return time.perf_counter() - t0, results


def run_threads(fn, n, workers):
    t0 = time.perf_counter()
    with ThreadPoolExecutor(max_workers=workers) as ex:
        results = list(ex.map(fn, range(n)))
    return time.perf_counter() - t0, results


def run_processes(fn, n, workers):
    t0 = time.perf_counter()
    with ProcessPoolExecutor(max_workers=workers) as ex:
        results = list(ex.map(fn, range(n)))
    return time.perf_counter() - t0, results


def measure(runner, fn, n, workers, repeats):
    times = []
    results = None
    for _ in range(repeats):
        elapsed, results = runner(fn, n, workers)
        times.append(elapsed)
    return statistics.median(times), results


def main():
    print("platform:", platform.platform())
    print("logical CPUs:", os.cpu_count())
    print("task count:", TASK_COUNT)
    print("workers:", WORKERS)
    print("repeats:", REPEATS)

    for label, fn in [("io", io_task), ("np", np_task)]:
        seq_t, seq_res = measure(run_sequential, fn, TASK_COUNT, 1, REPEATS)
        thr_t, thr_res = measure(run_threads, fn, TASK_COUNT, WORKERS, REPEATS)
        prc_t, prc_res = measure(run_processes, fn, TASK_COUNT, WORKERS, REPEATS)
        print(f"{label}: seq={seq_t:.3f}s thread={thr_t:.3f}s "
              f"process={prc_t:.3f}s")
        if label == "io":
            print("  io invariant sum:", sum(seq_res), sum(thr_res), sum(prc_res))
        else:
            print("  np first element:", seq_res[0], thr_res[0], prc_res[0])


if __name__ == "__main__":
    main()

A futtatási parancs Linux terminálból, a projektgyökérből:

python s130_workload.py

Szemléltető, nem mért kimenet 4 fizikai magos gépre, a fenti méretekkel:

platform: Linux-6.8.0-45-generic-x86_64-with-glibc2.39
logical CPUs: 8
task count: 12
workers: 4
repeats: 5
io: seq=2.100s thread=0.700s process=1.900s
  io invariant sum: 600000 600000 600000
np: seq=1.500s thread=0.500s process=0.600s
  np first element: -17.4213 -17.4213 -17.4213

Ezek a számok a szakasz eleji becslésből származó várt nagyságrendek, nem helyi futtatás eredményei.

33.2.5 Eredmény értelmezése

Az I/O feladatnál a szálas változat a szekvenciális idő nagyjából harmadára csökkenhet, mert a fájlműveletek alatt a GIL felszabadul. A processzes változat keveset nyer, mert a pickle-sorosítás és a processzindítás költsége hozzáadódik. A NumPy feladatnál a szálak is gyorsítanak, mert a natív mátrixszorzás közben a GIL elengedett; a processzes változat hasonló időt ad, de a processzenkénti NumPy-import és a sorosítás miatt általában valamivel lassabb. Az invariánsok: az I/O összeg minden módnál 600 000 sor, a NumPy első kimeneti elem seedenként ugyanaz, de a lebegőpontos összegzés sorrendje miatt a BLAS-szálak számától függően kis eltérés előfordulhat, ezért bitazonosságot nem követelünk, csak relatív toleranciát, például np.allclose alapján.

33.2.6 Feladatok

1. Értelmezés. Miért lehetséges, hogy a np_task szálas változata gyorsabb a szekvenciálisnál annak ellenére, hogy a GIL a Python-bájtkódot sorossá teszi? Nevezd meg a mechanizmust, és mondd meg, mi történne, ha a mátrixszorzást tiszta Python for ciklussal írnánk meg.

2. Megvalósítás. Módosítsd a scriptet úgy, hogy a WORKERS értékét 2, 4 és 8 között is mérje, és minden futásról készítsen egy manifest.json-t a Konfiguráció és futási napló mintájára: időbélyeg, platform, CPU-k száma, feladatszám, worker-szám, ismétlésszám és a mért mediánok. A kimeneti fájl ne legyen felülírható.

3. Hibadiagnózis. Egy kolléga elhagyja a script elején az OPENBLAS_NUM_THREADS=1 beállítást, és azt tapasztalja, hogy a np_task szálas változata lassabb, mint a szekvenciális. Fogalmazd meg, mit ellenőriznél először, és mi a valószínű magyarázat.

4. Megváltozott feltétel. A feladat mostantól nem 12 kis mátrixot, hanem 4 darab 200 MB-os pandas DataFrame-et kap, amelyeket a processzeknek kell feldolgozniuk. Miért nőhet meg drasztikusan a falidő, és milyen adatátadási stratégia csökkentheti a sorosítási költséget?

5. Lebegőpontos transzfer. Ugyanaz a seed és ugyanaz a gép mellett a np_task C[0, 0] értéke a szekvenciális és a szálas futás között a tizedesjegyekben eltérhet. Miért nem sérti ez a numerikus reprodukálhatóságot, és milyen ellenőrzést használj a == helyett?

33.2.7 Megoldások és indoklás

1. A NumPy @ művelete natív C kódot hív, és a CPython a hosszú C-számítás alatt elengedi a GIL-t, így több szál párhuzamosan végezhet mátrixszorzást. Ha a szorzást Python for ciklusokkal írnánk, a ciklus bájtkódja a GIL alatt futna, és a szálas változat nem gyorsítana, sőt a szálkezelés overheadje miatt még lassulhatna is. A GIL tehát a Python-szintű számítást korlátozza, nem a natív kiterjesztéseket.

2. A módosítás pszeudokódja:

import json
import datetime as dt
from pathlib import Path

def save_manifest(run_id, records):
    out = Path("artifacts") / run_id
    out.mkdir(parents=True, exist_ok=False)  # immutable output
    manifest = {
        "created_at": dt.datetime.now(dt.timezone.utc).isoformat(),
        "platform": platform.platform(),
        "logical_cpus": os.cpu_count(),
        "task_count": TASK_COUNT,
        "repeats": REPEATS,
        "results": records,
    }
    (out / "manifest.json").write_text(json.dumps(manifest, indent=2))

A measure hívását minden worker-számra el kell végezni, a rekordokat listába gyűjteni, majd a lista alapján manifestet írni. Az exist_ok=False garantálja, hogy véletlenül sem írunk felül korábbi futást. A manifest tehát a futás környezeti adatait és a mért mediánokat is rögzíti, ahogy a Konfiguráció és futási napló konfigurációs és futási napló elve megköveteli.

3. Először azt ellenőrizném, hogy a BLAS ténylegesen hány szálat indít: threadpoolctl.threadpool_info() vagy az operációs rendszer folyamatnézetében a szálak száma. Ha a környezeti változó hiányzik, a BLAS alapértelmezés szerint akár az összes logikai magot használhatja. Ekkor a 4 Python-worker mindegyike 8 BLAS-szálat indít, összesen 32 szálat: ez oversubscription, a kontextusváltás és a cache-versengés miatt a falidő nagyobb, mint a szekvenciális 8 szálas BLAS-futásé. A javítás: a környezeti változót a NumPy importja elé tenni, vagy a threadpoolctl.threadpool_limits(limits=1) hívást használni.

4. A 200 MB-os DataFrame-et a ProcessPoolExecutor minden feladatindításkor pickle-lel sorosítja, a célprocessz memóriájába másolja, majd a visszatérési értéket is sorosítja. Ez négy feladatnál több száz MB adatmozgatást és jelentős CPU-időt adhat. A költség csökkenthető, ha a munkafüggvény maga olvassa be a saját partícióját egy Parquet-fájlból; így csak a fájlnév megy át pickle-ként. Alternatíva a joblib loky backend memmap-támogatása vagy multiprocessing.shared_memory, de ezek bonyolultabb élettartam-kezelést igényelnek, ezért a fájlalapú partíció a kezdő tanulási úton a tisztább választás.

5. A BLAS a mátrixszorzást különböző szálkiosztás esetén más sorrendben összegezheti, és a lebegőpontos összeadás nem asszociatív, ezért az utolsó bitek eltérhetnek. Ez nem sérti a numerikus reprodukálhatóságot, mert a seed rögzítése a bemeneti mátrixokat reprodukálja, és az eltérés a lebegőpontos kerekítésből fakad. Az ellenőrzésre np.allclose(a, b, rtol=1e-12, atol=1e-12) vagy egy explicit relatív hibakorlát használandó, nem a ==.

33.2.8 Gyakori tévedés ellenpéldával

A leggyakoribb tévedés az, hogy „a GIL miatt a szálak soha nem gyorsítanak”. Ellenpélda a fenti I/O-feladat: a szekvenciális 2,1 s-hoz képest a szálas változat 0,7 s-ot adhat, mert a fájlvárakozás alatt más szál fut. A második tévedés az, hogy „több worker mindig jobb”. Ha 8 logikai magos gépen 4 worker egyenként 8 BLAS-szálat indít, a 32 szál túlterheli a CPU-t, és a falidő meghaladhatja a szekvenciális futást. A helyes döntés a mérésből következik: a Szűk keresztmetszet és mérési terv szűk keresztmetszet-elemzés után először a BLAS belső szálait korlátozzuk, majd a külső worker-számot növeljük, és a falidőt ismételt méréssel, mediánnal hasonlítjuk össze.

33.2.9 Rövid összefoglaló

A CPython GIL-je a Python-bájtkódot sorossá teszi, de az I/O-várakozás és a natív NumPy/BLAS-kód alatt felszabadul. A ThreadPoolExecutor ezért kis fájlok tömeges írására-olvasására és natív tömbműveletekre jó; a ProcessPoolExecutor akkor indokolt, ha tiszta Python CPU-kódot kell párhuzamosítani, de a pickle-sorosítás költségét mérlegelni kell. A BLAS saját szálait a NumPy importja előtt explicit módon 1-re kell állítani, különben a külső workerek és a belső szálak oversubscriptiont okoznak. A mérési jegyzék platformot, magszámot, feladatszámot, ismétlést és mediánt rögzít. A következő lépésben ezt a szálkeretet párhuzamos modellkeresésre alkalmazzuk az erőforráskeretes Párhuzamos modellkeresés erőforráskerettel alfejezetben.