Parallellik va asyncio: bir vaqtda ko'p ish

import time
from concurrent.futures import ThreadPoolExecutor
 
 
def fetch(i: int) -> int:
    time.sleep(0.5)            # tarmoq so'rovi o'rnida
    return i * i
 
 
start = time.perf_counter()
results = [fetch(i) for i in range(10)]
print(f"ketma-ket: {time.perf_counter() - start:.2f}s")     # ~5.0s
 
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=10) as ex:
    results = list(ex.map(fetch, range(10)))                # tartib saqlanadi
print(f"threadlar: {time.perf_counter() - start:.2f}s")     # ~0.5s

from concurrent.futures import ThreadPoolExecutor, as_completed
 
with ThreadPoolExecutor(max_workers=8) as ex:
    futures = {ex.submit(process_document, doc): doc.id for doc in docs}
    for fut in as_completed(futures):           # qaysi biri birinchi tugasa
        doc_id = futures[fut]
        try:
            result = fut.result()               # vazifadagi xato shu yerda ko'tariladi
        except Exception as e:
            log.warning("%s: %r", doc_id, e)
            continue
        save(doc_id, result)

from concurrent.futures import ProcessPoolExecutor
 
 
def tokenize_file(path: str) -> int:
    with open(path, encoding="utf-8") as f:
        return sum(len(line.split()) for line in f)    # sof Python — CPU ishi
 
 
if __name__ == "__main__":                              # ← process'lar uchun SHART
    with ProcessPoolExecutor() as ex:                   # standart: yadrolar soni
        counts = list(ex.map(tokenize_file, paths, chunksize=16))

import threading
 
counter = 0
 
def increment(n: int) -> None:
    global counter
    for _ in range(n):
        counter += 1          # o'qish → qo'shish → yozish: atomik EMAS
 
threads = [threading.Thread(target=increment, args=(100_000,)) for _ in range(8)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)                # 800000 bo'lishi kerak — ba'zan kamroq chiqishi mumkin

lock = threading.Lock()
 
def increment(n: int) -> None:
    global counter
    for _ in range(n):
        with lock:            # bir vaqtda faqat bitta thread kiradi
            counter += 1

import asyncio
 
 
async def fetch(i: int) -> int:          # korutina funksiyasi
    await asyncio.sleep(0.5)             # kutish — boshqaruv loop'ga qaytadi
    return i * i
 
 
async def main() -> None:
    result = await fetch(3)              # bitta korutinani kutish
    print(result)
 
 
asyncio.run(main())                      # event loop'ni ishga tushiradi — dastur kirish nuqtasi

async def main() -> None:
    results = await asyncio.gather(*(fetch(i) for i in range(10)))
    print(results)           # [0, 1, 4, ...] — tartib saqlanadi; jami ~0.5s, 5s emas

results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        log.warning("vazifa yiqildi: %r", r)

async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch(1))
        t2 = tg.create_task(fetch(2))
    # blokdan chiqqanda hammasi tugagan; biri yiqilsa — qolganlari bekor qilinadi
    print(t1.result(), t2.result())

async def main() -> None:
    try:
        async with asyncio.timeout(10):          # 3.11+
            answer = await call_llm(prompt)
    except TimeoutError:
        answer = None

import asyncio
 
 
async def limited_fetch(sem: asyncio.Semaphore, i: int) -> int:
    async with sem:                      # bo'sh "o'rin" bo'lguncha kutadi
        await asyncio.sleep(0.5)
        return i
 
 
async def main() -> None:
    sem = asyncio.Semaphore(3)           # bir vaqtda 3 ta
    await asyncio.gather(*(limited_fetch(sem, i) for i in range(9)))   # ~1.5s: 3 to'lqin
 
 
asyncio.run(main())

import asyncio
import json
import logging
from pathlib import Path
 
import anthropic
from anthropic import AsyncAnthropic
 
log = logging.getLogger(__name__)
 
LABELS = {"positive", "negative", "neutral"}
PROMPT = ("Quyidagi sharhning kayfiyatini aniqla. Faqat bitta so'z bilan javob ber: "
          "positive, negative yoki neutral.\n\nSharh: {text}")
 
 
async def classify(client: AsyncAnthropic, sem: asyncio.Semaphore, item: dict) -> dict:
    async with sem:
        try:
            msg = await client.messages.create(
                model="claude-opus-5-5",
                max_tokens=16,
                messages=[{"role": "user", "content": PROMPT.format(text=item["text"])}],
            )
        except anthropic.APIError as e:
            log.warning("id=%s: %r", item["id"], e)
            return {**item, "label": None, "error": type(e).__name__}
 
    text = "".join(b.text for b in msg.content if b.type == "text").strip().lower()
    return {**item, "label": text if text in LABELS else "neutral"}
 
 
async def classify_all(items: list[dict], concurrency: int = 10) -> list[dict]:
    sem = asyncio.Semaphore(concurrency)
    # SDK o'zi 429/5xx xatolarda backoff bilan qayta urinadi (max_retries)
    async with AsyncAnthropic(max_retries=4, timeout=60) as client:
        tasks = [classify(client, sem, it) for it in items]
        results = []
        for i, coro in enumerate(asyncio.as_completed(tasks), 1):   # tayyor bo'lishi bilan
            results.append(await coro)
            if i % 100 == 0:
                log.info("%d/%d tayyor", i, len(items))
    return results
 
 
def main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
    items = [json.loads(line) for line in Path("reviews.jsonl").read_text(encoding="utf-8").splitlines()]
    results = asyncio.run(classify_all(items, concurrency=10))
    failed = [r for r in results if r.get("error")]
    Path("labeled.jsonl").write_text(
        "".join(json.dumps(r, ensure_ascii=False) + "\n" for r in results), encoding="utf-8"
    )
    log.info("Tayyor: %d ta, %d ta xato", len(results), len(failed))
 
 
if __name__ == "__main__":
    main()

# Async ichidan sinxron (bloklovchi) funksiya — threadda
result = await asyncio.to_thread(heavy_sync_function, arg)
 
# Async ichidan CPU-og'ir ish — process'da
loop = asyncio.get_running_loop()
with ProcessPoolExecutor() as pool:
    result = await loop.run_in_executor(pool, cpu_heavy, data)
 
# Ishlab chiqaruvchi–iste'molchi: asyncio.Queue
queue: asyncio.Queue[str] = asyncio.Queue(maxsize=100)