Python-এ asyncio: async/await, TaskGroup, thread, multiprocessing ও GIL

Asyncio হলো Python-এর অন্তর্নির্মিত লাইব্রেরি অ্যাসিঙ্ক্রোনাস কোডের জন্য। async def দিয়ে ঘোষিত ফাংশন await-এ থেমে যেতে পারে, আর যতক্ষণ একটি ফাংশন নেটওয়ার্কের উত্তরের অপেক্ষায়, event loop অন্যগুলো চালায়। ফলে একই থ্রেডে শত শত অনুরোধ একসঙ্গে চলে। তবে asyncio একটিমাত্র হাতিয়ার: আছে থ্রেড (threading) ও প্রসেস (multiprocessing)-ও। দেখা যাক asyncio Python কীভাবে কাজ করে, কোন কাজে কোনটি নেবেন, এবং GIL-এর ভূমিকা কী।

অপেক্ষা করা ও হিসাব করা: দুটি ভিন্ন কাজ

এই পার্থক্যটি না বুঝলে পরের কিছুই বোঝা যাবে না। প্রোগ্রাম ধীর হওয়ার দুটি কারণ থাকে:

• I/O-bound — ইনপুট-আউটপুটের জন্য অপেক্ষা: API-র উত্তর, ডেটাবেস, ডিস্ক পড়া। এই সময় প্রসেসর অলস বসে থাকে।
• CPU-bound — হিসাব: ছবি প্রক্রিয়া, লাখ লাখ রেকর্ডের ওপর লুপ। প্রসেসর ব্যস্ত থাকে।

অপেক্ষার জন্য asyncio বা থ্রেড কাজ করে: একটি কাজ অপেক্ষা করলে অন্যটি চলে। হিসাবের জন্য দরকার আসল সমান্তরাল গণনা একাধিক কোরে — তার জন্য প্রসেস। এই দুটি গুলিয়ে ফেলাই প্রধান ভুল: সাধারণ Python-এ থ্রেড হিসাব দ্রুত করে না, আর তিনটি HTTP অনুরোধের জন্য প্রসেস বানানো মাছি মারতে কামান দাগার মতো।

asyncio কী: coroutine ও event loop

অ্যাসিঙ্ক্রোনাস কোড দুটি শব্দের ওপর দাঁড়িয়ে। async def একটি coroutine ঘোষণা করে — এমন একটি ফাংশন, যা থামতে পারে। এটি ডাকলে শরীর চলে না, বরং একটি coroutine অবজেক্ট তৈরি হয়। ভেতরের await বলে: «এখানে আমি অপেক্ষা করব, তুমি এই সময়ে অন্য কাজ করো।» পুরো ব্যবস্থার নিয়ন্ত্রণ থাকে event loop-এর, যেটি চালু হয় asyncio.run() দিয়ে।

import asyncio

async def fetch_rate(currency):
    await asyncio.sleep(1)          # API অনুরোধের অনুকরণ
    return {"USD": 92.5, "EUR": 100.1, "CNY": 12.7}[currency]

async def main():
    rate = await fetch_rate("USD")
    print(rate)

asyncio.run(main())  # 92.5

asyncio.sleep() একটি অ্যাসিঙ্ক্রোনাস বিরতি: এটি নিয়ন্ত্রণ event loop-কে ফিরিয়ে দেয়। সাধারণ time.sleep() coroutine-এর ভেতরে পুরো loop-কে আটকে দেয়, তখন বাকি সব কাজ দাঁড়িয়ে থাকে।

একের পর এক না একসঙ্গে

লুপে await লিখলে অনুরোধগুলো একের পর এক যায় — তিনটি মুদ্রার জন্য তিন সেকেন্ড। একসঙ্গে চালাতে হলে সেগুলো টাস্ক হিসেবে শুরু করতে হয়।

import asyncio, time

async def main():
    start = time.perf_counter()
    for cur in ("USD", "EUR", "CNY"):
        await fetch_rate(cur)                       # একের পর এক
    print(f"{time.perf_counter() - start:.1f} সে")   # 3.0 সে

asyncio.run(main())

TaskGroup: একসঙ্গে শুরু, সবার জন্য অপেক্ষা

আধুনিক উপায় (Python 3.11+) হলো টাস্ক-গ্রুপ। async with ব্লকের ভেতরে প্রতিটি টাস্ক tg.create_task() দিয়ে সঙ্গে সঙ্গে শুরু হয়, আর ব্লক থেকে বের হওয়া হয় সব শেষ হলে।

async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = {cur: tg.create_task(fetch_rate(cur)) for cur in ("USD", "EUR", "CNY")}
    rates = {cur: t.result() for cur, t in tasks.items()}
    print(rates, f"{time.perf_counter() - start:.1f} সে")   # ... 1.0 সে

asyncio.run(main())

TaskGroup কঠোরভাবে কাজ করে: একটি টাস্ক ব্যর্থ হলে বাকিগুলো বাতিল হয়, আর সব ত্রুটি জমা হয় ExceptionGroup-এ। ধরা হয় except* দিয়ে:

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_rate("USD"))
            tg.create_task(fetch_rate("XXX"))   # KeyError
    except* KeyError as eg:
        print("অজানা মুদ্রা:", eg.exceptions)

asyncio.gather ও create_task

TaskGroup-এর আগে মানদণ্ড ছিল asyncio.gather(): এটি coroutine-গুলো একসঙ্গে চালায় এবং ফল দেয় একই ক্রমে। পুরনো কোডে এটি এখনো অনেক।

async def main():
    usd, eur, cny = await asyncio.gather(
        fetch_rate("USD"), fetch_rate("EUR"), fetch_rate("CNY")
    )
    # return_exceptions=True হলে ত্রুটিগুলো তালিকায় ফিরে আসে, সব থামিয়ে দেয় না

asyncio.create_task() একটি coroutine-কে গ্রুপ ছাড়াই ব্যাকগ্রাউন্ডে চালায়। টাস্কের রেফারেন্স রাখতে হয়, নইলে garbage collector সেটি মুছে দিতে পারে এবং ত্রুটি কেউ দেখবে না। তাই নতুন প্রজেক্টে TaskGroup নিরাপদতর।

টাস্ক বাতিল ও সময়সীমা

async def main():
    task = asyncio.create_task(fetch_rate("USD"))
    task.cancel()                     # টাস্কের ভেতরে CancelledError উঠবে
    try:
        await task
    except asyncio.CancelledError:
        print("বাতিল")

    try:
        async with asyncio.timeout(0.5):   # Python 3.11+
            await fetch_rate("EUR")
    except TimeoutError:
        print("অনেক সময় লাগছে")

পুরনো সংস্করণে asyncio.timeout()-এর বদলে ব্যবহৃত হয় asyncio.wait_for(coro, timeout=0.5)। আরও আছে asyncio.wait() — এটি সমাপ্ত ও অসমাপ্ত টাস্কের সেট ফেরত দেয় এবং প্রথমটির জন্য অপেক্ষা করতে পারে (return_when=asyncio.FIRST_COMPLETED)।

কোর্সের বিবরণ অনুযায়ী Python 3.15-এ TaskGroup-এ যুক্ত হয়েছে tg.cancel() মেথড: এটি গ্রুপের সব টাস্ক ব্যতিক্রম ছাড়াই বাতিল করে। এতে «কে আগে উত্তর দেয়» পরিস্থিতি সহজ হয়: তিনটি মিরর সার্ভিসকে প্রশ্ন করা, প্রথম উত্তরটি নেওয়া, বাকিগুলো বাতিল করা। আগে এর জন্য নিজের সংকেত-ব্যতিক্রম ছুড়ে দিতে হতো বা টাস্কগুলো হাতে বাতিল করতে হতো।

সিঙ্ক্রোনাইজেশন: Semaphore, Lock, Event, Queue

asyncio-র নিজস্ব সিঙ্ক্রোনাইজেশন উপাদান আছে, যেগুলো await-এর সঙ্গে কাজ করে। সবচেয়ে কাজের হলো Semaphore: এটি একসঙ্গে কতগুলো কাজ চলবে সীমিত করে, যাতে অন্যের API-কে হাজার অনুরোধে ভাসিয়ে না দেওয়া হয়।

sem = asyncio.Semaphore(10)

async def limited(url):
    async with sem:              # একসঙ্গে সর্বোচ্চ 10টি
        await asyncio.sleep(0.1) # এখানে অনুরোধ হতো
        return url

• asyncio.Lock — critical section-এ একবারে একটি coroutine।
• asyncio.Event — একটি coroutine await event.wait()-এ অপেক্ষা করে, অন্যটি event.set() দিয়ে সংকেত দেয়।
• asyncio.Queue — «উৎপাদক — ভোক্তা» সারি: await q.put(), await q.get(), q.task_done()।

ফাঁদ: await ভুলে যাওয়া

সবচেয়ে সাধারণ ভুল — await ছাড়া coroutine ডাকা:

async def main():
    rate = fetch_rate("USD")    # await ভুলে গেছি
    print(rate * 2)             # TypeError
# RuntimeWarning: coroutine 'fetch_rate' was never awaited

Python ফাংশনটি চালায়নি, শুধু একটি coroutine অবজেক্ট বানিয়েছে। লগে «was never awaited» দেখলে বুঝবেন await বাদ পড়েছে। দ্বিতীয় সাধারণ ভুল — coroutine-এর ভেতরে ব্লকিং ডাক (time.sleep, requests.get)। HTTP-এর জন্য অ্যাসিঙ্ক্রোনাস library নিন (httpx, aiohttp), আর অনিবার্য সিঙ্ক্রোনাস কোড থ্রেডে পাঠান await asyncio.to_thread(func, arg) দিয়ে।

Python-এ থ্রেড: threading মডিউল

Thread হলো একই প্রসেসের ভেতরে একটি স্বাধীন কাজের ধারা; থ্রেডগুলো মেমোরি ভাগ করে। নেটওয়ার্কের অপেক্ষায় এরা দারুণ কাজ করে এবং কোড async-এ নতুন করে লেখার প্রয়োজন হয় না।

import threading, time

def download(name):
    time.sleep(1)               # নেটওয়ার্ক অপেক্ষা
    print(name, "সম্পন্ন")

threads = [threading.Thread(target=download, args=(n,)) for n in ("a", "b", "c")]
for t in threads:
    t.start()
for t in threads:
    t.join()                    # সব মিলিয়ে প্রায় 1 সেকেন্ড

ভাগ করা মেমোরি সুবিধাও, উৎসও ভুলের: দুটি থ্রেড একই ভেরিয়েবল বদলালে threading.Lock লাগে। সাধারণত থ্রেড হাতে না বানিয়ে একটি pool নেওয়া সহজ।

ThreadPoolExecutor

concurrent.futures.ThreadPoolExecutor একটি থ্রেড-পুল বানায় এবং তাতে কাজ বিলি করে। map ফল দেয় মূল ক্রমে।

from concurrent.futures import ThreadPoolExecutor
import time

def fetch(url):
    time.sleep(1)
    return f"{url}: 200"

urls = [f"https://example.com/{i}" for i in range(5)]
with ThreadPoolExecutor(max_workers=5) as pool:
    for result in pool.map(fetch, urls):
        print(result)           # সব মিলিয়ে প্রায় 1 সেকেন্ড

GIL কী, এবং থ্রেড হিসাব কেন দ্রুত করে না

GIL (Global Interpreter Lock) হলো CPython-এর বৈশ্বিক লক। যেকোনো মুহূর্তে Python বাইটকোড একটিমাত্র থ্রেড চালায়। থ্রেড ইনপুট-আউটপুটের জন্য অপেক্ষা করলে সে লক ছেড়ে দেয়, তাই নেটওয়ার্কে থ্রেড কাজে আসে। কিন্তু বিশুদ্ধ হিসাবে চারটি থ্রেড একটির চেয়ে দ্রুত হয় না: তারা পালা করে লক হাত বদলায়।

from concurrent.futures import ThreadPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

with ThreadPoolExecutor(4) as pool:
    list(pool.map(heavy, [5_000_000] * 4))   # প্রায় একের পর এক চালানোর মতোই সময়

C-তে লেখা library (NumPy ও এ-জাতীয়) ভারী অপারেশনের সময় প্রায়ই GIL ছেড়ে দেয়, তাই সেখানে থ্রেডে গতি বাড়ে।

multiprocessing: আসল সমান্তরাল কাজ

প্রতিটি প্রসেসের নিজস্ব ইন্টারপ্রেটার ও নিজস্ব GIL থাকে, তাই প্রসেসগুলো বিভিন্ন কোরে একসঙ্গে হিসাব করে। দাম — আলাদা মেমোরি: প্রসেসগুলোর মধ্যে ডেটা যায় সিরিয়ালাইজেশনে (pickle), আর প্রসেস চালু করা থ্রেডের চেয়ে ব্যয়বহুল।

from concurrent.futures import ProcessPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

if __name__ == "__main__":      # Windows ও macOS-এ বাধ্যতামূলক
    with ProcessPoolExecutor() as pool:
        print(list(pool.map(heavy, [5_000_000] * 4)))

if __name__ == "__main__" পরীক্ষা না থাকলে শিশু-প্রসেসগুলো মডিউল ইমপোর্টের সময় নতুন প্রসেস বানাতে শুরু করে। নিম্নস্তরের multiprocessing মডিউলে Process, Pool, Queue আছে, কিন্তু ProcessPoolExecutor সহজ এবং থ্রেড-পুলের মতো একই ইন্টারফেস দেয়।

কোনটি বেছে নেবেন: asyncio, threading বা multiprocessing

GIL-মুক্ত Python ও সাব-ইন্টারপ্রেটার: নতুন কী

Python 3.13 থেকে আলাদা একটি বিল্ড আছে যেখানে GIL নেই (free-threading, PEP 703) — তার নামে একটি t থাকে, যেমন python3.15t। 3.13-এ এটি পরীক্ষামূলক ছিল, 3.14 থেকে (PEP 779) আনুষ্ঠানিকভাবে সমর্থিত। সেখানে থ্রেড বিভিন্ন কোরে একসঙ্গে হিসাব করে। তবে ডিফল্টভাবে আপনি এখনো সাধারণ GIL-সহ Python পান: free-threaded বিল্ডে শুধু এক থ্রেডের কোড কিছুটা ধীর, এবং সব library এখনো তার জন্য প্রস্তুত নয়।

import sys
print(sys._is_gil_enabled())   # free-threaded বিল্ডে False

সমান্তরাল হিসাবের দ্বিতীয় পথটি এসেছে 3.14-এ (PEP 734) — সাব-ইন্টারপ্রেটার, concurrent.interpreters মডিউলে: একই প্রসেসে কয়েকটি আলাদা ইন্টারপ্রেটার, প্রতিটির নিজস্ব GIL। এগুলো সবচেয়ে সুবিধাজনকভাবে ব্যবহার হয় InterpreterPoolExecutor-এর মাধ্যমে, যার চেহারা থ্রেড-পুলের মতো। কোর্সের পরিমাপে চারটি ভারী কাজ একটি থ্রেডে লেগেছে 1.03 সে, আর এই পদ্ধতিতে 0.39 সে।

সারসংক্ষেপ

Asyncio Python-এ অপেক্ষার কাজের জন্য: async def-এর coroutine, await-এ বিরতি, TaskGroup বা gather-এ একসঙ্গে চালানো। থ্রেড একই কাজ করে কোড async-এ না লিখেই, আর ভারী হিসাবের জন্য দরকার প্রসেস, কারণ সাধারণ CPython-এর GIL থ্রেডগুলোকে একসঙ্গে গুনতে দেয় না। Free-threading ও সাব-ইন্টারপ্রেটার এই ছবিটি ধীরে বদলাচ্ছে, তবে মূল নিয়মটি এখনো কাজ করে: অপেক্ষা করলে asyncio বা থ্রেড, হিসাব করলে প্রসেস। কোর্সে এসব বাস্তবে শেখানো হয়েছে: খরচের ট্র্যাকার মুদ্রার বিনিময় হার এক সেকেন্ডে সমান্তরালে আনে, তিন সেকেন্ডের বদলে।