Parallellik: thread, process va asyncio

"""CPU-bog'liq ish. ProcessPoolExecutor uchun alohida modulda bo'lishi SHART."""


def ogir_hisob(n: int) -> int:
    """Sof hisob: GIL ni ushlab turadi."""
    s = 0
    for i in range(n):
        s += i * i
    return s

import sys, time
sys.path.insert(0, str(ISH))
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from ogir_ish import ogir_hisob

N = 3_000_000
ISH_SONI = 4

t0 = time.perf_counter()
[ogir_hisob(N) for _ in range(ISH_SONI)]
ketma_ket = time.perf_counter() - t0

t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as p:
    list(p.map(ogir_hisob, [N] * ISH_SONI))
threadlar = time.perf_counter() - t0

t0 = time.perf_counter()
with ProcessPoolExecutor(max_workers=4) as p:
    list(p.map(ogir_hisob, [N] * ISH_SONI))
jarayonlar = time.perf_counter() - t0

import pandas as pd
show(pd.DataFrame([
    {"usul": "ketma-ket", "vaqt, s": round(ketma_ket, 2), "tezlanish": "1.00x"},
    {"usul": "4 thread", "vaqt, s": round(threadlar, 2), "tezlanish": f"{ketma_ket / threadlar:.2f}x"},
    {"usul": "4 jarayon", "vaqt, s": round(jarayonlar, 2), "tezlanish": f"{ketma_ket / jarayonlar:.2f}x"},
]))
import os
print("mavjud yadrolar:", os.cpu_count())

import sys, time
sys.path.insert(0, str(ISH))
from concurrent.futures import ThreadPoolExecutor
import urllib.request
from sinov_server import ishga_tushir

server2, manzil2 = ishga_tushir()

def sekin_sorov(_):
    """Har so'rov taxminan 0.2 s kutadi (server tomonida)."""
    with urllib.request.urlopen(manzil2 + "/ok", timeout=10) as j:
        time.sleep(0.2)                  # tarmoq kechikishini taqlid qilamiz
        return j.status

SORO_SONI = 12
t0 = time.perf_counter()
[sekin_sorov(i) for i in range(SORO_SONI)]
ketma_ket_io = time.perf_counter() - t0

natijalar = {}
for ishchilar in (2, 4, 12):
    t0 = time.perf_counter()
    with ThreadPoolExecutor(max_workers=ishchilar) as p:
        list(p.map(sekin_sorov, range(SORO_SONI)))
    natijalar[ishchilar] = time.perf_counter() - t0

import pandas as pd
qatorlar = [{"usul": "ketma-ket", "vaqt, s": round(ketma_ket_io, 2), "tezlanish": "1.00x"}]
for n, v in natijalar.items():
    qatorlar.append({"usul": f"{n} thread", "vaqt, s": round(v, 2),
                     "tezlanish": f"{ketma_ket_io / v:.2f}x"})
show(pd.DataFrame(qatorlar))

import asyncio, time

async def sekin_ish(nom: str, kutish: float) -> str:
    await asyncio.sleep(kutish)          # time.sleep EMAS!
    return f"{nom} tayyor"

async def ketma_ket_async():
    t0 = time.perf_counter()
    for i in range(5):
        await sekin_ish(f"ish{i}", 0.2)
    return time.perf_counter() - t0

async def parallel_async():
    t0 = time.perf_counter()
    await asyncio.gather(*(sekin_ish(f"ish{i}", 0.2) for i in range(5)))
    return time.perf_counter() - t0

k = asyncio.run(ketma_ket_async())
p = asyncio.run(parallel_async())
print(f"await ketma-ket : {k:.2f} s")
print(f"asyncio.gather  : {p:.2f} s  ({k / p:.1f}x tezroq)")

import asyncio, time

async def togri(_):
    await asyncio.sleep(0.15)

async def notogri(_):
    time.sleep(0.15)                     # BLOKLAYDI

async def olch(f):
    t0 = time.perf_counter()
    await asyncio.gather(*(f(i) for i in range(6)))
    return time.perf_counter() - t0

print(f"await asyncio.sleep : {asyncio.run(olch(togri)):.2f} s  (parallel)")
print(f"time.sleep          : {asyncio.run(olch(notogri)):.2f} s  (ketma-ket, 6 x 0.15)")

import threading, time
from concurrent.futures import ThreadPoolExecutor

class Oddiy:
    """+= 1 — bir qarashda atomar ko'rinadi."""
    def __init__(self):
        self.qiymat = 0
    def oshir(self):
        self.qiymat += 1

class Ochiq:
    """O'sha amal, lekin uch qadami ochiq yozilgan."""
    def __init__(self):
        self.qiymat = 0
    def oshir(self):
        tmp = self.qiymat          # 1) o'qish
        time.sleep(0)              # 2) shu yerda boshqa thread ishga tushishi mumkin
        self.qiymat = tmp + 1      # 3) yozish

class Qulfli:
    def __init__(self):
        self.qiymat = 0
        self._qulf = threading.Lock()
    def oshir(self):
        with self._qulf:
            tmp = self.qiymat
            time.sleep(0)
            self.qiymat = tmp + 1

def sinov(sanagich, threadlar=8, har_biri=20_000):
    def ishla():
        for _ in range(har_biri):
            sanagich.oshir()
    with ThreadPoolExecutor(max_workers=threadlar) as p:
        list(p.map(lambda _: ishla(), range(threadlar)))
    return sanagich.qiymat, threadlar * har_biri

import pandas as pd
qatorlar = []
for nom, sinf in [("`+= 1`", Oddiy), ("o'qish/yozish ochiq", Ochiq), ("qulf bilan", Qulfli)]:
    olingan, kutilgan = sinov(sinf())
    qatorlar.append({"usul": nom, "kutilgan": kutilgan, "olingan": olingan,
                     "yo'qolgan": kutilgan - olingan,
                     "yo'qotish %": round(100 * (kutilgan - olingan) / kutilgan, 1)})
show(pd.DataFrame(qatorlar))

from concurrent.futures import ThreadPoolExecutor

def mustaqil_ish(n):
    """Umumiy holat yo'q — natijani QAYTARADI."""
    return sum(1 for _ in range(n))

with ThreadPoolExecutor(max_workers=8) as p:
    natijalar = list(p.map(mustaqil_ish, [50_000] * 8))
print("har ishchi qaytardi:", natijalar)
print("yig'indi:", sum(natijalar), "| kutilgan:", 8 * 50_000, "| qulf kerak bo'lmadi")

"""Bir nechta CSV faylni parallel qayta ishlash."""
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from .modellar import Natija
from .oqish import oqi, tozala
from .xatolar import HisobotXato

log = logging.getLogger(__name__)


def bitta(yol: Path) -> tuple[Path, Natija | None, str | None]:
    """Bitta faylni qayta ishlaydi. Xato otmaydi — uni qaytaradi."""
    try:
        return yol, tozala(oqi(yol)), None
    except HisobotXato as e:
        return yol, None, str(e)


def hammasi(yollar: list[Path], ishchilar: int = 4) -> dict[str, Natija | str]:
    """Fayllarni parallel qayta ishlaydi. Bitta faylning xatosi boshqalarni to'xtatmaydi."""
    natijalar: dict[str, Natija | str] = {}
    with ThreadPoolExecutor(max_workers=ishchilar) as p:
        kelajaklar = {p.submit(bitta, y): y for y in yollar}
        for k in as_completed(kelajaklar):
            yol, natija, xato = k.result()
            natijalar[yol.name] = natija if natija is not None else f"XATO: {xato}"
            log.debug("tugadi: %s", yol.name)
    return natijalar

mkdir -p ko_p_sinov
for i in 1 2 3 4 5 6; do
  printf 'ism,guruh,baho,sana\nT%s,A,8%s,2026-09-0%s\nX%s,B,9%s,2026-09-0%s\n' $i $i 1 $i $i 2 > ko_p_sinov/f$i.csv
done
printf 'ism,guruh\nBuzuq,A\n' > ko_p_sinov/buzuq.csv
ls ko_p_sinov/

import sys, time, importlib
from pathlib import Path
sys.path.insert(0, str(ISH / "src"))
import hisobot.ko_p as ko_p
importlib.reload(ko_p)

yollar = sorted((ISH / "ko_p_sinov").glob("*.csv"))
print("fayllar:", [y.name for y in yollar])

t0 = time.perf_counter()
ketma = {y.name: ko_p.bitta(y)[1] for y in yollar}
ketma_s = time.perf_counter() - t0

t0 = time.perf_counter()
parallel = ko_p.hammasi(yollar, ishchilar=4)
parallel_s = time.perf_counter() - t0

print(f"\nketma-ket: {ketma_s * 1000:.0f} ms | parallel: {parallel_s * 1000:.0f} ms")
for nom, natija in sorted(parallel.items()):
    holat = natija if isinstance(natija, str) else f"{len(natija.talabalar)} to'g'ri"
    print(f"  {nom:12} {holat}")

import pandas as pd
show(pd.DataFrame([
    {"holat": "Bir nechta faylni o'qish", "tanlov": "ThreadPoolExecutor", "sabab": "I/O kutish"},
    {"holat": "100 ta HTTP so'rov", "tanlov": "ThreadPoolExecutor yoki asyncio", "sabab": "I/O kutish"},
    {"holat": "10 000 ta HTTP so'rov", "tanlov": "asyncio", "sabab": "Thread xotira yeydi"},
    {"holat": "Rasmlarni qayta ishlash", "tanlov": "ProcessPoolExecutor", "sabab": "CPU ish, GIL to'siq"},
    {"holat": "Katta matritsani hisoblash", "tanlov": "numpy (GIL ni bo'shatadi)", "sabab": "C darajasida parallel"},
    {"holat": "Kichik, tez ish", "tanlov": "Oddiy sikl", "sabab": "Parallellik xarajati ishdan ko'p"},
]))

from hisobot.ko_p import bitta, hammasi
from hisobot.modellar import Natija


def _fayl(papka, nom, matn):
    y = papka / nom
    y.write_text(matn, encoding="utf-8")
    return y


TOGRI = "ism,guruh,baho,sana\nA,X,85,2026-01-01\n"
BUZUQ = "ism,guruh\nA,X\n"


def test_bitta_togri(tmp_path):
    _yol, natija, xato = bitta(_fayl(tmp_path, "a.csv", TOGRI))
    assert xato is None
    assert natija is not None
    assert len(natija.talabalar) == 1


def test_bitta_buzuq(tmp_path):
    _yol, natija, xato = bitta(_fayl(tmp_path, "b.csv", BUZUQ))
    assert natija is None
    assert "ustunlar yetishmaydi" in xato


def test_hammasi_xato_boshqalarni_toxtatmaydi(tmp_path):
    yollar = [
        _fayl(tmp_path, "a.csv", TOGRI),
        _fayl(tmp_path, "buzuq.csv", BUZUQ),
        _fayl(tmp_path, "c.csv", TOGRI),
    ]
    n = hammasi(yollar, ishchilar=3)
    assert len(n) == 3
    assert isinstance(n["a.csv"], Natija)
    assert isinstance(n["c.csv"], Natija)
    assert str(n["buzuq.csv"]).startswith("XATO")


def test_hammasi_bosh_royxat():
    assert hammasi([]) == {}

.venv/bin/python3 -m pytest tests/test_ko_p.py -q
.venv/bin/python3 -m pytest -q
python3 -m mypy