threading, concurrent.futures ve GIL

Üç araç, üç farklı iş

Anadolu Market'in gece işi: 200 tedarikçi dosyasını indir (ağ beklemesi), her birini ayrıştır (hafif), sonra 2 milyon satırda fiyat optimizasyonu hesapla (ağır hesap). Üçü de "aynı anda daha çok iş" ister ama farklı araçlarla: ağ için thread ya da asyncio, ağır hesap için süreç (process). Bu dersin amacı doğru aracı seçmeyi ve paylaşılan veriyi güvenle korumayı öğretmek.

İş parçacığı (thread) ve havuz

Thread, aynı süreç içinde paralel akan bir çalışma hattıdır; belleği paylaşır. Elle threading.Thread başlatabilirsin ama modern kalıp concurrent.futures.ThreadPoolExecutor: sabit sayıda işçi, submit ile iş ver, Future ile sonucu al; ya da map ile liste gibi kullan.

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def indir(tedarikci: str) -> str:
    time.sleep(0.1)                       # sahte ağ beklemesi
    return f"{tedarikci}.csv"

tedarikciler = ["Ege", "Marmara", "Karadeniz", "Akdeniz", "Anadolu", "Trakya"]

t0 = time.perf_counter()
sirali = [indir(t) for t in tedarikciler]
print(f"sıralı: {time.perf_counter() - t0:.2f} s")

t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=3) as havuz:
    sonuc = list(havuz.map(indir, tedarikciler))      # sıra korunur
print(sonuc, f"havuz(3): {time.perf_counter() - t0:.2f} s")

with ThreadPoolExecutor(max_workers=6) as havuz:
    gelecekler = {havuz.submit(indir, t): t for t in tedarikciler}
    for f in as_completed(gelecekler):                  # bitiş sırasıyla
        print("bitti:", f.result())

Kendi bilgisayarında beklenen çıktı (süre değişebilir, "bitti" sırası her seferinde farklı olabilir):
sıralı: 0.60 s
['Ege.csv', 'Marmara.csv', 'Karadeniz.csv', 'Akdeniz.csv', 'Anadolu.csv', 'Trakya.csv'] havuz(3): 0.20 s
bitti: Marmara.csv
bitti: Ege.csv
…
Tarayıcıda iki süre de 0.60 s görünür, çünkü thread'ler sırayla çalışır.

GIL: tek anda tek Python

CPython'da GIL (Global Interpreter Lock) aynı anda yalnızca bir thread'in Python bytecode çalıştırmasına izin verir. Thread'ler beklerken (ağ, disk, sleep) GIL'i bırakır; o yüzden I/O işinde havuz gerçek hız sağlar. Ama saf Python hesabında iki thread sırayla çalışır, toplam süre düşmez, hatta kilit alışverişi yüzünden artabilir.

import time
from concurrent.futures import ThreadPoolExecutor

def hesap(n: int) -> int:
    return sum(i * i for i in range(n))         # saf Python, CPU işi

N = 300_000
t0 = time.perf_counter()
[hesap(N) for _ in range(4)]
tek = time.perf_counter() - t0

t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as havuz:
    list(havuz.map(hesap, [N] * 4))
coklu = time.perf_counter() - t0
print(f"sıralı {tek:.2f} s, 4 thread {coklu:.2f} s -> GIL yüzünden kazanç yok")

Yarış durumu ve Lock

İki thread aynı sayacı artırırsa ne olur? sayac += 1 aslında oku-artır-yaz üç adımdır; GIL adımların arasında thread değiştirebilir ve bir artış kaybolur. Buna yarış durumu (race condition) denir. Çözüm: paylaşılan veriye erişimi threading.Lock ile tek seferde bir thread'e vermek.

import threading, time
from concurrent.futures import ThreadPoolExecutor

bakiye = 0
kilit = threading.Lock()

def yatir_guvensiz(n):
    global bakiye
    for _ in range(n):
        tmp = bakiye            # oku
        time.sleep(0)           # başka thread'e sıra ver: yarışı görünür kılar
        bakiye = tmp + 1        # yaz

def yatir_guvenli(n):
    global bakiye
    for _ in range(n):
        with kilit:             # kritik bölge
            bakiye += 1

for fn in (yatir_guvensiz, yatir_guvenli):
    bakiye = 0
    with ThreadPoolExecutor(max_workers=4) as havuz:
        list(havuz.map(fn, [2_000] * 4))
    print(fn.__name__, "beklenen 8000, gelen", bakiye)

Kendi bilgisayarında beklenen çıktı (ilk sayı her çalıştırmada farklı olur, 8000'den küçüktür):
yatir_guvensiz beklenen 8000, gelen 2004
yatir_guvenli beklenen 8000, gelen 8000
Tarayıcıda ikisi de 8000 görünür, çünkü thread'ler sırayla çalışır: yarış burada oluşmaz.

queue ile üretici-tüketici

queue.Queue thread-güvenli bir kuyruktur: üreticiler put, tüketiciler get yapar, kilitle uğraşmazsın. None gibi bir "zehirli hap" (sentinel) ile tüketicilere bitişi bildir; task_done/join ile tüm işlerin bittiğini bekle.

import queue, threading, time

kuyruk: queue.Queue = queue.Queue()
islenen = []
kilit = threading.Lock()

def tuketici(ad):
    while True:
        siparis = kuyruk.get()
        if siparis is None:          # bitiş sinyali
            kuyruk.task_done()
            break
        time.sleep(0.01)             # sahte işleme
        with kilit:
            islenen.append((ad, siparis))
        kuyruk.task_done()

isciler = [threading.Thread(target=tuketici, args=(f"isci{i}",)) for i in range(3)]
for t in isciler:
    t.start()
for sid in range(1001, 1011):
    kuyruk.put(sid)
for _ in isciler:
    kuyruk.put(None)
kuyruk.join()
print("işlenen sipariş:", sorted(s for _, s in islenen))
print("işçi başına:", {ad: sum(1 for a, _ in islenen if a == ad) for ad in ("isci0", "isci1", "isci2")})

Kendi bilgisayarında beklenen çıktı (dağılım her seferinde değişir):
işlenen sipariş: [1001, 1002, 1003, 1004, 1005, 1006, 1007, 1008, 1009, 1010]
işçi başına: {'isci0': 3, 'isci1': 3, 'isci2': 4}
Tarayıcıda işin tamamını ilk işçi yapar: {'isci0': 10, 'isci1': 0, 'isci2': 0}.

Hangi iş için hangisi?

# fiyat_opt.py — kendi bilgisayarında: python fiyat_opt.py
import os
from concurrent.futures import ProcessPoolExecutor

def optimize(urun_id: int) -> tuple[int, float]:      # modül düzeyinde (pickle edilebilir)
    toplam = sum((urun_id * i) % 97 for i in range(200_000))
    return urun_id, toplam / 200_000

def main():
    urunler = range(1, 401)
    with ProcessPoolExecutor(max_workers=os.cpu_count()) as havuz:
        sonuc = dict(havuz.map(optimize, urunler, chunksize=25))
    print(len(sonuc), "ürün hesaplandı")

if __name__ == "__main__":       # Windows ve macOS'ta ZORUNLU
    main()

def hizlanma(p: float, n: int) -> float:
    return 1 / ((1 - p) + p / n)

for n in (1, 2, 4, 8, 16, 64):
    print(f"{n:>3} işçi: %90 paralel -> {hizlanma(0.90, n):5.2f}x   %50 paralel -> {hizlanma(0.50, n):4.2f}x")
print("üst sınır (%90):", round(1 / (1 - 0.90), 1))

from collections import Counter

siparisler = [("Ankara", 120.0), ("İzmir", 80.0), ("Ankara", 45.5), ("Bursa", 300.0),
              ("İzmir", 19.9), ("Ankara", 10.0), ("Bursa", 5.0)]

def parcala(veri, n):
    boyut, artan = divmod(len(veri), n)
    parcalar, bas = [], 0
    for i in range(n):
        son = bas + boyut + (1 if i < artan else 0)
        parcalar.append(veri[bas:son])
        bas = son
    return parcalar

def kismi_ciro(parca):
    c = Counter()
    for sehir, tutar in parca:
        c[sehir] += tutar
    return c

parcalar = parcala(siparisler, 3)
print([len(p) for p in parcalar])
toplam = sum(map(kismi_ciro, parcalar), Counter())    # havuz.map ile aynı arayüz
print(dict(sorted(toplam.items())))

import threading

kilit = threading.Lock()
kilit.acquire()
print("Lock ikinci kez alınabilir mi?", kilit.acquire(blocking=False))   # kendini kilitler
kilit.release()

rk = threading.RLock()
with rk:
    with rk:                       # aynı thread: sorun yok
        print("RLock iç içe alındı")

dur = threading.Event()
print("dur?", dur.is_set())
dur.set()                          # işçi döngüleri bunu görüp çıkar
print("dur?", dur.is_set(), "| bekleme sonucu:", dur.wait(timeout=0.01))

from concurrent.futures import Future

f = Future()                      # havuzun içeride yaptığını elle yapıyoruz
f.add_done_callback(lambda g: print("geri çağrı: bitti =", g.done()))
f.set_result({"USD": 32.5})
print(f.result())

hata = Future()
hata.set_exception(TimeoutError("kur servisi cevap vermedi"))
try:
    hata.result()
except TimeoutError as e:
    print("result() istisnayı yeniden fırlattı:", e)

bekleyen = Future()
print("iptal edildi mi?", bekleyen.cancel(), bekleyen.cancelled())