همروندی: نخ‌ها و asyncio

۱. سه ابزار، یک جدول

اگر فقط یک چیز از این فصل یادتان بماند، همین جدول باشد:

۲. اولین نخ

نخ یک خط اجرای دوم در همان برنامه است، پس دو تکه کد می‌توانند با هم پیش بروند.

import threading
import time


def worker() -> None:
    '''This function will run in a thread of its own.'''
    name = threading.current_thread().name
    for step in range(3):
        print(f'[{name}] step {step}')
        time.sleep(1)
    print(f'[{name}] finished')


def main() -> None:
    print(f'[{threading.current_thread().name}] creating the thread')

    thread = threading.Thread(target=worker, name='worker')
    thread.start()                 # start() -- never run()

    # The main thread is free while the worker sleeps.
    print('[MainThread] the thread is running; I am not blocked')
    print(f'[MainThread] is it alive? {thread.is_alive()}')

    thread.join()                  # wait for it
    print('[MainThread] done')

اولین نخ: نمونه 01_first_thread/app.py

سه قاعده از همان روز اول

• start() تابع را در نخ تازه اجرا می‌کند. اگر اشتباهی thread.run() بنویسید، تابع در نخ فعلی اجرا می‌شود: هیچ همروندی‌ای در کار نیست و هیچ چیز هم به شما هشدار نمی‌دهد.
• join() هیچ چیز را متوقف یا نابود نمی‌کند؛ فقط منتظر می‌ماند.
• نخ‌های یک برنامه متغیرهای مشترک دارند. این هم قدرتشان است و هم خطرشان، که در بخش چهارم می‌بینید.

نکته‌ای که همه یک بار گیرش می‌افتند: args=('A',) با ویرگول انتهایی. بدون آن یک رشته است نه تاپل، و پایتون کاراکترهایش را به عنوان آرگومان‌های جدا می‌فرستد.

۳. نخ برای کار انتظاری

نخ کی برنامه را واقعاً سریع‌تر می‌کند؟ وقتی برنامه منتظر است. هنگام دانلود یک فایل یا پرس‌وجو از پایگاه داده، پردازنده بیکار است و نخ دیگری می‌تواند از آن وقت استفاده کند.

SITES = ['kadoos.ir', 'python.org', 'sqlite.org', 'qt.io']


def fake_download(site: str, seconds: float = 1.0) -> None:
    '''Pretend to download a page. `sleep` = waiting for the network.'''
    time.sleep(seconds)
    print(f'  downloaded {site}')


def sequential() -> float:
    start = time.perf_counter()
    for site in SITES:
        fake_download(site)
    return time.perf_counter() - start


def threaded() -> float:
    start = time.perf_counter()
    threads = [threading.Thread(target=fake_download, args=(site,))
               for site in SITES]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    return time.perf_counter() - start

چهار ثانیه به یک ثانیه: نمونه 03_threads_for_io/app.py

one after the other (4 sites):
  -> 4.01 seconds

with 4 threads:
  -> 1.00 seconds

4.0 times faster

دقت کنید الگو همیشه یکی است: اول همه نخ‌ها را start کنید، بعد همه را join. اگر هر نخ را بلافاصله بعد از راه‌اندازی join کنید، دوباره پشت سر هم اجرا می‌شوند و هیچ سودی نمی‌برید.

برای اندازه‌گیری زمان از time.perf_counter() استفاده کنید، نه time.time(): فقط جلو می‌رود و دقت بالایی دارد.

۴. شرایط رقابتی

حالا خطرناک‌ترین باگ دنیای همروندی: باگی که فقط گاهی خودش را نشان می‌دهد.

counter = 0                 # shared between every thread


def worker(rounds: int) -> None:
    global counter
    for _ in range(rounds):
        # This is `counter += 1` written out in slow motion.
        temp = counter          # 1. read
        time.sleep(0.000_01)    # <- the interruption, made visible
        temp += 1               # 2. add
        counter = temp          # 3. write


def main() -> None:
    threads = [threading.Thread(target=worker, args=(1000,))
               for _ in range(10)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    print(f'expected: 10000')
    print(f'obtained: {counter}')
    print(f'lost:     {10000 - counter} increments')

شرایط رقابتی: نمونه 04_race_condition/app.py

expected: 10000
obtained: 1247
lost:     8753 increments

شرح مثال

• ده نخ، هر کدام هزار بار یکی اضافه می‌کنند، پس پاسخ باید ده هزار باشد. اجرایش کنید: تقریباً هیچ وقت ده هزار نمی‌گیرید.
• علت این است که counter += 1 یک قدم نیست، بلکه سه قدم است: خواندن، اضافه کردن، نوشتن. اگر نخ A بین قدم یک و سه قطع شود، نخ B همان مقدار قدیمی را می‌خواند و هر دو یک عدد را برمی‌گردانند: یک افزایش گم می‌شود.
• آن sleep کوچک فقط پنجره را آنقدر پهن می‌کند که هر بار ببینیدش. باگ بدون sleep هم هست؛ فقط یک بار در چند هزار اجرا ظاهر می‌شود: روی کامپیوتر مشتری، عصر پنجشنبه. و همین است که خطرناکش می‌کند.
• قاعده: لحظه‌ای که دو نخ روی یک متغیر می‌نویسند، به قفل نیاز دارید.

۵. قفل و رویداد

class BankAccount:
    '''A shared account. Every change to `balance` goes through the lock.'''

    def __init__(self, balance: int = 0) -> None:
        self.balance = balance
        self.lock = threading.Lock()

    def deposit(self, amount: int) -> None:
        with self.lock:
            # Slow motion again, to prove the lock really protects us.
            new_balance = self.balance + amount
            time.sleep(0.000_01)
            self.balance = new_balance

    def withdraw(self, amount: int) -> bool:
        with self.lock:
            # "Check, then act" MUST be inside the same lock, otherwise the
            # balance can change between the `if` and the subtraction.
            if amount > self.balance:
                return False
            self.balance -= amount
            return True

قفل: نمونه 05_locks/app.py

شرح مثال

• تا وقتی یک نخ قفل را در دست دارد، هر نخ دیگری که به with lock: برسد منتظر می‌ماند. سه قدم افزایش دیگر نصف نمی‌شوند.
• with lock: قفل را حتی اگر استثنا رخ دهد آزاد می‌کند. هرگز دستی acquire و release ننویسید.
• بخش بحرانی را تا می‌توانید کوتاه نگه دارید. هر چه داخل قفل است دوباره ترتیبی اجرا می‌شود، پس قفل دور یک عملیات کند یعنی دور ریختن همه همروندی.
• الگوی «بررسی کن، بعد عمل کن» باید داخل یک قفل باشد، وگرنه موجودی می‌تواند بین if و تفریق عوض شود.

    stop = threading.Event()

    def blinker() -> None:
        while not stop.is_set():
            print('  ...working')
            # wait() returns as soon as the event is set: much better than
            # sleep(), which would always run to the end.
            stop.wait(timeout=0.3)
        print('  worker stopped cleanly')

    worker = threading.Thread(target=blinker)
    worker.start()
    time.sleep(1)
    stop.set()                  # ask it to stop
    worker.join()

رویداد: توقف تمیز یک نخ

۶. صف و کارگرها

queue.Queue فهرستی است که خودش از قبل امن است: put و get قفل را برایتان می‌گیرند. برای همین است که توصیه اول هر برنامه‌نویس باتجربه این است: «داده را از راه صف رد و بدل کنید، نه از راه متغیر مشترک».

def worker(tasks: queue.Queue, results: queue.Queue) -> None:
    '''Take numbers from the queue until the sentinel arrives.'''
    name = threading.current_thread().name
    while True:
        number = tasks.get()
        try:
            if number is None:          # the sentinel: "no more work"
                return
            time.sleep(0.2)             # pretend the job takes time
            results.put((name, number, number * number))
        finally:
            # In a `finally`, so an exception cannot make queue.join() hang
            # for ever waiting for a task_done() that never comes.
            tasks.task_done()


def main() -> None:
    tasks: queue.Queue = queue.Queue()
    results: queue.Queue = queue.Queue()

    workers = [threading.Thread(target=worker, args=(tasks, results),
                                name=f'worker-{index}', daemon=True)
               for index in range(3)]
    for thread in workers:
        thread.start()

    for number in range(1, 11):
        tasks.put(number)

    # One sentinel per worker, so each of them sees exactly one.
    for _ in workers:
        tasks.put(None)

    tasks.join()                    # wait until everything is handled
    for thread in workers:
        thread.join()

صف و کارگرها: نمونه 06_queue_and_workers/app.py

شرح مثال

• get() یک آیتم برمی‌دارد و اگر صف خالی باشد منتظر می‌ماند. پس کارگر حلقه شلوغ نمی‌زند.
• نگهبان (None) به کارگرها می‌گوید کار تمام شد. به تعداد کارگرها نگهبان بگذارید تا هر کدام دقیقاً یکی ببیند.
• task_done() باید در finally باشد، وگرنه یک استثنا باعث می‌شود queue.join() تا ابد منتظر بماند.
• daemon=True یعنی «این نخ مهم نیست»؛ پایتون هنگام پایان برنامه منتظرش نمی‌ماند. برای کارگری که تا ابد حلقه می‌زند عالی است، ولی برای نخی که فایل می‌نویسد خطرناک، چون وسط کار قطع می‌شود.

۷. استخر نخ

concurrent.futures راه کوتاه و امروزی همه کارهای بالاست: استخری از نخ‌های آماده که راه‌اندازی و انتظار و جمع‌آوری نتیجه را خودش انجام می‌دهد.

from concurrent.futures import ThreadPoolExecutor, as_completed

    # -- map: simple, ordered, and it stops at the first exception ---------
    with ThreadPoolExecutor(max_workers=4) as pool:
        good_sites = [site for site in SITES if site != 'flask.dev']
        for site, size in zip(good_sites, pool.map(fetch, good_sites)):
            print(f'  {site}: {size} bytes')

    # -- submit + as_completed: answers as soon as they are ready ----------
    with ThreadPoolExecutor(max_workers=4) as pool:
        # A dict Future -> site, to know which job an answer belongs to.
        futures = {pool.submit(fetch, site): site for site in SITES}

        for future in as_completed(futures):
            site = futures[future]
            try:
                size = future.result()      # re-raises the worker's exception
            except ConnectionError as error:
                print(f'  {site}: FAILED ({error})')
            else:
                print(f'  {site}: {size} bytes')

استخر نخ: نمونه 07_thread_pool/app.py

بلوک with پیش از پایان، منتظر همه کارها می‌ماند. پس دیگر حلقه join ضرورتی ندارد.

۸. نخ و رابط گرافیکی

اینجا دلیل واقعی نیاز شما به نخ است. فصل سی و پنجم گفت کار طولانی داخل تابع پاسخ، پنجره را منجمد می‌کند. after() برای یک زمان‌سنج جواب داد، ولی برای کار واقعی نه: یک دانلود را نمی‌شود به تکه‌های صدمیلی‌ثانیه‌ای برید.

الگوی امن این است:

worker thread  --put-->  queue.Queue  --get-->  main thread (after())

    # -- the worker thread -------------------------------------------------
    def download(self) -> None:
        '''Runs in a THREAD. It must not touch a single widget.'''
        for percent in range(0, 101, 5):
            if self.stop_event.is_set():
                self.messages.put(('done', 'cancelled'))
                return
            time.sleep(0.2)                     # the waiting work
            self.messages.put(('progress', percent))
        self.messages.put(('done', 'finished'))

    # -- the GUI thread ----------------------------------------------------
    def poll_queue(self) -> None:
        '''Runs in the MAIN thread: here, and only here, we touch widgets.'''
        try:
            while True:
                # get_nowait raises Empty instead of blocking the interface.
                kind, value = self.messages.get_nowait()
                if kind == 'progress':
                    self.bar['value'] = value
                    self.status.config(text=f'downloading... {value} %')
                elif kind == 'done':
                    self.status.config(text=str(value))
                    self.start_button.config(state='normal')
        except queue.Empty:
            pass
        # Ask to be called again. This line must never be skipped.
        self.root.after(self.POLL_MS, self.poll_queue)

    def cancel(self) -> None:
        '''You cannot kill a thread; you ASK it to stop.'''
        self.stop_event.set()
        self.status.config(text='cancelling...')

نخ و tkinter: نمونه 08_thread_with_gui/app.py

شرح مثال

• کارگر فقط پیام در صف می‌گذارد. نخ اصلی هر صد میلی‌ثانیه با after() صف را می‌خواند و تنها او به ویجت‌ها دست می‌زند.
• get_nowait() اگر صف خالی باشد Empty پرتاب می‌کند به جای اینکه رابط کاربری را قفل کند.
• نخ را نمی‌شود کشت. فقط می‌شود از او خواهش کرد که تمام کند، با یک Event که خودش مرتب بررسی‌اش می‌کند.
• نسخه Qt همین ایده: کارگر یک سیگنال منتشر می‌کند و Qt خودش آن را به نخ رابط کاربری تحویل می‌دهد.

۹. قفل سراسری و پردازه‌ها

GIL یا قفل سراسری مفسر، قفلی است داخل خود CPython: در هر لحظه فقط یک نخ بایت‌کد پایتون اجرا می‌کند. نتیجه‌اش همان جدول بخش اول است.

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor


def count_primes(limit: int) -> int:
    '''Pure calculation, no waiting at all: the GIL's worst case.'''
    total = 0
    for number in range(2, limit):
        for divisor in range(2, int(number ** 0.5) + 1):
            if number % divisor == 0:
                break
        else:
            total += 1
    return total

محاسبه خالص: نمونه 09_gil_and_processes/app.py

این برنامه را اجرا کنید: نسخه نخ‌دار یک محاسبه خالص، به همان کندی نسخه ترتیبی است و گاهی کندتر، به خاطر هزینه جابه‌جایی بین نخ‌ها. پاسخ درست برای محاسبه، ProcessPoolExecutor است: چند پردازه واقعی سیستم‌عامل، هر کدام با مفسر و قفل خودش، روی چند هسته.

بهای یک پردازه

• آرگومان‌ها و نتیجه‌ها pickle و کپی می‌شوند، پس فرستادن شیء بزرگ گران است. یک lambda یا تابع محلی اصلاً قابل pickle نیست.
• دیگر هیچ متغیر مشترکی وجود ندارد. این خبر بد و خبر خوب است: دیگر شرایط رقابتی هم ندارید.
• روی ویندوز هر پردازه ماژول شما را دوباره وارد می‌کند، و برای همین if __name__ == '__main__': اینجا اختیاری نیست. بدون آن، برنامه بی‌نهایت بار خودش را راه می‌اندازد.

۱۰. اولین کوروتین

یک نخ گران است: سیستم‌عامل می‌سازدش، به آن سوئیچ می‌کند، و هر کدام حافظه می‌خورند. هزار نخ یک مشکل است؛ هزار کوروتین هیچ چیز نیست. asyncio همروندی با یک نخ است.

import asyncio
import time


async def worker(name: str, seconds: float) -> str:
    '''A coroutine: it can pause at every `await`.'''
    print(f'{name}: starting')
    await asyncio.sleep(seconds)        # NOT time.sleep()
    print(f'{name}: finished after {seconds} s')
    return f'{name} is done'


async def main_async() -> None:
    # `await` one after the other = sequential. Two seconds in total.
    first = await worker('A', 1)
    second = await worker('B', 1)

    # What happens if you forget the await:
    coroutine = worker('C', 1)          # nothing runs here
    print(f'without await, worker() gives: {type(coroutine).__name__}')
    print(await coroutine)              # NOW it runs


def main() -> None:
    # asyncio.run creates the event loop, runs the coroutine, closes the loop.
    # Call it ONCE, at the very top of your program.
    asyncio.run(main_async())

اولین کوروتین: نمونه 10_first_coroutine/app.py

دو قاعده‌ای که همه رویشان می‌لغزند

• صدا زدن یک کوروتین اجرایش نمی‌کند. worker() فقط یک شیء کوروتین می‌سازد؛ await worker() اجرایش می‌کند. پایتون هشدار می‌دهد: coroutine 'worker' was never awaited.
• await فقط داخل async def مجاز است.

۱۱. اجرای همزمان با gather

async def main_async() -> None:
    sites = [('kadoos.ir', 1.0), ('python.org', 0.4),
             ('sqlite.org', 0.7), ('qt.io', 0.2)]

    print('sequential:')
    for site, seconds in sites:
        await fetch(site, seconds)

    print('with gather:')
    sizes = await asyncio.gather(*(fetch(site, seconds)
                                   for site, seconds in sites))
    print(f'   results in the ORDER OF THE ARGUMENTS: {sizes}')

    print('with a failure and return_exceptions=True:')
    results = await asyncio.gather(
        fetch('kadoos.ir', 0.3),
        fetch('broken.ir', 0.1),
        fetch('qt.io', 0.2),
        return_exceptions=True,
    )
    for result in results:
        if isinstance(result, Exception):
            print(f'   FAILED: {result}')
        else:
            print(f'   ok: {result} bytes')

gather: نمونه 11_gather/app.py

شرح مثال

• await one(); await two() ترتیبی است. برای اجرای همزمان باید هر دو را به gather بدهید.
• gather نتیجه‌ها را به ترتیبی که پاس داده‌اید برمی‌گرداند، نه به ترتیب تمام شدن.
• به طور پیش‌فرض اولین استثنا بلافاصله پرتاب می‌شود. با return_exceptions=True به جایش شیء استثنا داخل فهرست نتیجه می‌آید، و این دقیقاً چیزی است که می‌خواهید وقتی یک دانلود خراب نباید چهار دانلود موفق را از بین ببرد.
• جایگزین امروزی‌تر (پایتون ۳.۱۱ به بعد) asyncio.TaskGroup است که وقتی یکی شکست بخورد بقیه را لغو می‌کند؛ معمولاً همان چیزی است که در یک سرور می‌خواهید.

۱۲. تسک، لغو و مهلت

async def long_job(name: str, seconds: float) -> str:
    '''A job that cleans up properly when it is cancelled.'''
    try:
        for second in range(int(seconds)):
            print(f'  {name}: working ({second + 1}/{int(seconds)})')
            await asyncio.sleep(1)
        return f'{name} finished'
    except asyncio.CancelledError:
        print(f'  {name}: cancelled -- closing the file, the socket...')
        raise            # ALWAYS re-raise: the caller must know


async def main_async() -> None:
    # -- create_task: the work starts, we do something else meanwhile -------
    task = asyncio.create_task(long_job('background', 3))
    print('  the main coroutine is free')
    await asyncio.sleep(1)
    print(f'  is it done? {task.done()}')
    print(f'  result: {await task}')

    # -- cancelling --------------------------------------------------------
    task = asyncio.create_task(long_job('endless', 10))
    await asyncio.sleep(2.5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print('  the caller sees CancelledError')

    # -- timeout (Python 3.11+) --------------------------------------------
    try:
        async with asyncio.timeout(2):
            await long_job('slow', 10)
    except TimeoutError:
        print('  too slow: the job was cancelled for us')

تسک‌ها: نمونه 12_tasks_cancel_timeout/app.py

شرح مثال

• create_task کوروتین را در همین مرحله در پس‌زمینه شروع می‌کند و کد شما ادامه می‌یابد. بعداً با await task پاسخ را جمع می‌کنید.
• یک ارجاع به تسک نگه دارید. حلقه رویداد فقط ارجاع ضعیف نگه می‌دارد، پس تسکی که کسی نگهش نداشته می‌تواند وسط کار زباله‌روبی شود.
• task.cancel() باعث می‌شود await داخل کوروتین CancelledError پرتاب کند. آن را بگیرید تا تمیزکاری کنید، و بعد دوباره پرتابش کنید. قورت دادنش یعنی تسک از لغو شدن سر باز می‌زند، و این‌طور است که برنامه به Ctrl-C جواب نمی‌دهد.
• asyncio.timeout() کوروتین را برایتان لغو می‌کند و TimeoutError می‌دهد. شکل قدیمی‌تر که همه جا کار می‌کند asyncio.wait_for() است.

۱۳. صف ناهمگام

همان ایده بخش ششم، ولی با کوروتین به جای نخ. و چون همه چیز در یک نخ اجرا می‌شود، اصلاً شرایط رقابتی‌ای وجود ندارد که از آن محافظت کنیم.

async def producer(queue: asyncio.Queue, count: int) -> None:
    '''Puts `count` jobs into the queue.'''
    for number in range(1, count + 1):
        await asyncio.sleep(random.uniform(0.05, 0.15))
        await queue.put(number)
        print(f'produced  {number:>2}   (queue: {queue.qsize()})')
    print('the producer has finished')


async def consumer(name: str, queue: asyncio.Queue) -> None:
    '''Takes jobs for ever, until it is cancelled.'''
    while True:
        number = await queue.get()
        try:
            await asyncio.sleep(random.uniform(0.1, 0.3))
            print(f'  {name} handled {number:>2} -> {number ** 2}')
        finally:
            # In a `finally`, so a failure cannot make join() wait for ever.
            queue.task_done()

صف ناهمگام: نمونه 14_async_queue/app.py

maxsize به شما فشار برگشتی می‌دهد: وقتی مصرف‌کننده‌ها عقب می‌مانند، صف پر می‌شود و put شروع به انتظار می‌کند، پس تولیدکننده کند می‌شود به جای اینکه همه حافظه را بخورد. تمام ترفند یک خط لوله سالم همین است.

۱۴. پروژه: ربات پیام‌رسان

ربات طبیعی‌ترین پروژه asyncio است: تقریباً همیشه منتظر شبکه است و همزمان به آدم‌های زیادی خدمت می‌کند؛ دقیقاً همان شکلی که asyncio برایش ساخته شد. کتابخانه python-telegram-bot از بالا تا پایین روی asyncio ساخته شده و برای همین هر پاسخگر یک async def است.

async def hello(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    '''Answers /hello.'''
    # update.message CAN be None (an edited message, a channel post...), so
    # check instead of silencing the type checker with a `# type: ignore`.
    if update.message is None or update.effective_user is None:
        return
    name = update.effective_user.first_name
    await update.message.reply_text(f'Hello {name}! Welcome to Kadoos.')


def main() -> None:
    # `.base_url(...)` is what makes the same library talk to Bale.
    application = (Application.builder()
                   .token(get_token())
                   .base_url(BASE_URL)
                   .build())

    application.add_handler(CommandHandler('hello', hello))
    application.add_handler(CommandHandler('dice', dice))
    application.add_handler(CommandHandler('help', help_command))

    print('The bot is running. Press Ctrl-C to stop it.')
    application.run_polling()


# The main guard matters here: without it, importing this file would start the
# bot. The first version of this course built the application at module level.
if __name__ == '__main__':
    main()

ربات: نمونه 15_messenger_bot/01_hello_bot.py

def get_token() -> str:
    '''Read the token, or explain clearly what is missing.'''
    token = os.environ.get('BOT_TOKEN')
    if not token:
        raise SystemExit(
            'BOT_TOKEN is not set.\n\n'
            '  Windows : set BOT_TOKEN=your-token-here\n'
            '  Linux   : export BOT_TOKEN="your-token-here"\n\n'
            'Get a token from @BotFather (Telegram) or @BotFather (Bale).')
    return token

خواندن توکن از محیط: نمونه 15_messenger_bot/config.py

نکته امنیتی دوم در بخش دریافت فایل است. نام فایل را فرستنده انتخاب می‌کند، پس نامی مثل ../../passwords.txt بیرون از پوشه دانلود شما نوشته می‌شود. به این حمله پیمایش مسیر می‌گویند و دفاعش این است که فقط آخرین بخش نام را نگه دارید و بعد بررسی کنید مسیر نهایی واقعاً داخل پوشه مورد نظر است.

def safe_filename(name: str) -> str:
    '''Turn a name chosen by a stranger into a name we can safely write.'''
    # Path(...).name drops every folder: '../../etc/passwd' -> 'passwd'
    name = Path(name).name
    # Then keep only harmless characters.
    name = re.sub(r'[^A-Za-z0-9._-]', '_', name)
    # '.' and '..' survive both steps above and are not file names at all.
    if name in {'', '.', '..'}:
        return 'file.bin'
    return name

نام فایل امن: نمونه 15_messenger_bot/04_files_bot.py

خلاصه فصل