هم‌زمانی برای کارِ داده

در این فصل چه یاد می‌گیری#

تا اینجا همه‌چیز یک کار در یک زمان بود. هم‌زمانی سه ابزار دارد — نخ، فرآیند، async — و هر سه در جای اشتباه نه‌تنها کمک نمی‌کنند، بلکه ضرر می‌زنند.

این فصل یک سرویسِ واقعی روی 127.0.0.1 بالا می‌آورد و همه‌چیز را رویش می‌سنجد. چهار عدد از این فصل بیرون می‌آید و سه‌تایشان ناخوشایندند: با ۸ نخ، ۲۴ درخواستِ شبکه ۷٫۶ برابر سریع‌تر می‌شوند؛ با ۲۴ نخ فقط ۱٫۲ برابر — و ثابت می‌کنیم چرا، چون گلوگاه اصلاً سمتِ ما نبود. کارِ محاسباتی با چهار نخ ۱٫۰۴ برابر می‌شود، یعنی هیچ. و کارِ ریز با فرآیند بیش از صد برابر کندتر از حالتِ عادی می‌شود.

آخر این فصل می‌توانی:

• بگویی کدام کار با نخ سریع می‌شود و کدام با فرآیند
• GIL را با یک اندازه‌گیری توضیح بدهی، نه با تعریف
• یک کارِ شبکه‌ای را با asyncio هم‌زمان کنی
• تشخیص بدهی کِی هیچ‌کدام کمک نمی‌کند

قبل از شروع#

از فصلِ ۷: «سقفِ بهبودت را قبل از اندازه‌گیری حساب کن». اینجا هم همان: کفِ نظریِ ۲۴ درخواستِ سریالی، ۲۴ ضربدر تأخیرِ هر درخواست است.

هیچ سرویسِ بیرونی‌ای در کار نیست. سرور را با http.server روی همین ماشین بالا می‌آوریم، با یک تأخیرِ ساختگی که نقشِ پایگاه داده یا دیسک را بازی می‌کند.

📓 نوت‌بوک: نوت‌بوک این فصل را در Colab باز کن — همهٔ کدهای این فصل آماده و به‌ترتیب داخلش هست.

۱. یک سرویسِ واقعی، روی همین ماشین#

import http.server
import socketserver
import threading
import time
import urllib.request

SERVED = 0
DELAY = 0.05


class Handler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        global SERVED
        SERVED += 1
        time.sleep(DELAY)                 # کارِ واقعیِ یک سرویس: پایگاه داده، دیسک، شبکه
        body = b"ok"
        self.send_response(200)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), Handler)
server.daemon_threads = True
PORT = server.server_address[1]
threading.Thread(target=server.serve_forever, daemon=True).start()


def fetch(index):
    with urllib.request.urlopen(f"http://127.0.0.1:{PORT}/ticket/{index}") as response:
        return response.read()


print("سرور روی 127.0.0.1 بالا آمد. یک درخواست:", fetch(0))
print("تأخیرِ ساختگیِ هر درخواست:", DELAY, "ثانیه")

سرور روی 127.0.0.1 بالا آمد. یک درخواست: b'ok'
تأخیرِ ساختگیِ هر درخواست: 0.05 ثانیه

آن time.sleep داخلِ سرور، کلِ استعارهٔ این فصل است. در آن پنجاه میلی‌ثانیه، پردازنده هیچ کاری نمی‌کند — منتظر است. و هر جا منتظری، جا برای کارِ هم‌زمان هست.

N = 24


def clock(fn):
    start = time.perf_counter()
    result = fn()
    return time.perf_counter() - start, result


serial_time, serial_result = clock(lambda: [fetch(i) for i in range(N)])
print(f"{N} درخواست، یکی‌یکی: {serial_time:.2f} ثانیه")
print(f"کفِ نظری (N × تأخیر): {N * DELAY:.2f} ثانیه")

24 درخواست، یکی‌یکی: 1.25 ثانیه
کفِ نظری (N × تأخیر): 1.20 ثانیه

۱٫۲۵ در برابرِ کفِ نظریِ ۱٫۲۰ — یعنی تقریباً تمامِ این زمان، انتظار بوده. این بهترین حالتِ ممکن برای هم‌زمانی است.

۲. نخ برای کارِ شبکه#

from concurrent.futures import ThreadPoolExecutor

for workers in (4, 8, 24):
    spent, result = clock(
        lambda: list(ThreadPoolExecutor(max_workers=workers).map(fetch, range(N))))
    print(f"  {workers:>2} نخ → {spent:.2f} ثانیه   ({serial_time / spent:.1f} برابر)"
          f"   همان جواب؟ {result == serial_result}")
    if workers == 24:
        thread_time = spent

   4 نخ → 0.32 ثانیه   (3.8 برابر)   همان جواب؟ True
   8 نخ → 0.17 ثانیه   (7.6 برابر)   همان جواب؟ True
  24 نخ → 1.07 ثانیه   (1.2 برابر)   همان جواب؟ True

چهار نخ ۳٫۸ برابر، هشت نخ ۷٫۶ برابر — تا اینجا دقیقاً همان‌طور که انتظار داری، تقریباً خطی.

و بعد بیست‌وچهار نخ همه‌چیز را خراب می‌کند: ۱٫۲ برابر، یعنی تقریباً به سرعتِ حالتِ یکی‌یکی برگشتیم.

جوابِ رایج به این پدیده «سربارِ نخ‌ها» است — و در این مورد غلط است. بخشِ بعد جوابِ درست را با آزمایش نشان می‌دهد.

۳. گلوگاه سمتِ ما نبود#

فرضیه: مشکل از نخ‌های ما نیست، از سرور است. هر سرورِ TCP یک صفِ گوش‌دادن دارد: تعدادِ اتصال‌هایی که می‌تواند پشتِ در نگه دارد تا نوبتشان برسد. پیش‌فرضِ این کتابخانه پنج است.

و این صف جای دقیقی دارد که باید بدانی: داخلِ خودِ سیستم‌عامل است، نه داخلِ برنامهٔ تو. وقتی برنامه‌ات یک پورت را باز می‌کند، به سیستم‌عامل می‌گوید «اتصال‌های تازه را برایم نگه دار»؛ هر کلاینتی که وصل می‌شود در همان صف می‌نشیند تا برنامه‌ات وقت کند برش دارد. اگر صف پر باشد، سیستم‌عامل اتصالِ بعدی را رد می‌کند و کلاینت ConnectionRefusedError می‌گیرد — بی‌آنکه حتی یک خط از کدِ تو اجرا شده باشد. پس عددِ request_queue_size نمی‌گوید سرورت چقدر سریع است؛ می‌گوید چند نفر می‌توانند هم‌زمان پشتِ در منتظر بمانند.

آزمونش ساده است: همان کد، سرورِ دوم، فقط با صفِ بزرگ‌تر.

class RoomyServer(socketserver.ThreadingTCPServer):
    request_queue_size = 64               # پیش‌فرضِ کتابخانه ۵ است
    daemon_threads = True
    allow_reuse_address = True


roomy = RoomyServer(("127.0.0.1", 0), Handler)
ROOMY_PORT = roomy.server_address[1]
threading.Thread(target=roomy.serve_forever, daemon=True).start()


def fetch_roomy(index):
    with urllib.request.urlopen(f"http://127.0.0.1:{ROOMY_PORT}/ticket/{index}") as response:
        return response.read()


print("صفِ گوش‌دادنِ سرورِ اول:", socketserver.TCPServer.request_queue_size)
print("صفِ گوش‌دادنِ سرورِ دوم:", RoomyServer.request_queue_size)
for workers in (8, 24):
    spent, _ = clock(lambda: list(ThreadPoolExecutor(max_workers=workers).map(fetch_roomy, range(N))))
    print(f"  {workers:>2} نخ روی سرورِ دوم → {spent:.2f} ثانیه   ({serial_time / spent:.1f} برابر)")

صفِ گوش‌دادنِ سرورِ اول: 5
صفِ گوش‌دادنِ سرورِ دوم: 64
   8 نخ روی سرورِ دوم → 0.17 ثانیه   (7.5 برابر)
  24 نخ روی سرورِ دوم → 0.06 ثانیه   (19.2 برابر)

همان بیست‌وچهار نخ، همان کد، همان ماشین: از ۱٫۲ برابر به ۱۹٫۲ برابر. فقط یک عددِ سمتِ سرور عوض شد.

این مهم‌ترین درسِ این فصل است و ربطی به پایتون ندارد: وقتی هم‌زمانی را زیاد می‌کنی و سود نمی‌گیری، قبل از اینکه کدِ خودت را متهم کنی، بپرس گلوگاه کجاست. ما اگر آن آزمایش را نمی‌کردیم، الان یک باورِ غلط داشتیم («بیشتر از هشت نخ در پایتون فایده ندارد») که در همین صفحه ابطال شد.

📏 اندازه بگیر: با چه چیزی مقایسه شد؟ با همان تعدادِ نخ روی همان کد، فقط با یک تنظیمِ متفاوتِ سرور — یعنی یک متغیر عوض شد، نه دوتا. روی کدام داده؟ همان ۲۴ درخواست با همان تأخیر. با چند seed؟ بی‌ربط، ولی هر عدد دو بار با ترتیبِ یکسان تکرار شد و جهتِ نتیجه هر بار همان بود. در اندازه‌گیریِ زمان، تکرار جای seed را می‌گیرد.

۴. asyncio: همان سود، بدونِ نخ#

نخ گران است: هر نخ حافظه و زمانِ راه‌اندازی می‌خواهد. asyncio همان کار را با یک نخ انجام می‌دهد: به‌جای اینکه چند نخ هرکدام منتظر بمانند، یک حلقه همهٔ انتظارها را با هم مدیریت می‌کند.

اینجا حتی کتابخانهٔ HTTP هم لازم نداریم؛ خودمان درخواست را می‌نویسیم:

import asyncio


async def fetch_async(index):
    reader, writer = await asyncio.open_connection("127.0.0.1", PORT)
    writer.write(f"GET /ticket/{index} HTTP/1.0\r\nHost: 127.0.0.1\r\n\r\n".encode())
    await writer.drain()
    data = await reader.read()
    writer.close()
    await writer.wait_closed()
    return data.split(b"\r\n\r\n", 1)[1]


async def fetch_all(n):
    return await asyncio.gather(*(fetch_async(i) for i in range(n)))


def run_async(coro):
    # در نوت‌بوک یک حلقهٔ رویداد از قبل در حال اجراست، پس asyncio.run را
    # در یک نخِ جدا می‌زنیم تا هم اینجا کار کند هم در یک اسکریپتِ ساده.
    box = {}
    thread = threading.Thread(target=lambda: box.update(result=asyncio.run(coro)))
    thread.start()
    thread.join()
    return box["result"]


async_time, async_result = clock(lambda: run_async(fetch_all(N)))
print(f"{N} درخواست با asyncio: {async_time:.2f} ثانیه")
print(f"بهبود: {serial_time / async_time:.1f} برابر")
print("همان جواب؟", async_result == serial_result)
print("نخ‌های زنده در این لحظه:", threading.active_count())

24 درخواست با asyncio: 0.07 ثانیه
بهبود: 17.8 برابر
همان جواب؟ True
نخ‌های زنده در این لحظه: 3

۱۷٫۸ برابر — و آن سه نخ، نخِ اصلی به‌علاوهٔ دو نخِ خودِ سرورهاست. یعنی asyncio برای بیست‌وچهار درخواستِ هم‌زمان هیچ نخِ تازه‌ای نساخت، در حالی که استخرِ نخ بیست‌وچهارتا ساخته بود.

دو نکتهٔ عملی که نباید فراموش شوند:

• await فقط جایی کار می‌کند که کتابخانه‌اش هم async باشد. اگر داخلِ یک coroutine یک تابعِ معمولیِ کند صدا بزنی، کلِ حلقه می‌ایستد و هیچ‌کدام از این سود عاید نمی‌شود.
• asyncio.run را در نوت‌بوک مستقیم نمی‌شود صدا زد، چون آنجا از قبل یک حلقهٔ رویداد در حال اجراست. راهِ ما یک نخِ کمکی بود؛ راهِ دیگر await مستقیم در سلول است که فقط در نوت‌بوک کار می‌کند و در فایلِ .py خطای نحوی می‌دهد.

🔧 اگر کار نکرد: اگر RuntimeError: asyncio.run() cannot be called from a running event loop گرفتی، asyncio.run را مستقیم در سلول زده‌ای. همان run_async بالا علاجش است. و اگر IndexError روی data.split(...) گرفتی، سرور پاسخِ خالی داده — معمولاً یعنی سرور را قبلاً بسته‌ای و باید سلولِ اول را دوباره اجرا کنی.

۵. GIL: نخ برای کارِ محاسباتی#

حالا همان نخ‌ها، این بار روی کاری که منتظرِ هیچ‌چیز نیست:

def cpu_work(rounds):
    total = 0
    for i in range(rounds):
        total += (i * i) % 7
    return total


ROUNDS = 3_000_000
CHUNKS = 4

cpu_serial, serial_sums = clock(lambda: [cpu_work(ROUNDS) for _ in range(CHUNKS)])
cpu_threads, thread_sums = clock(
    lambda: list(ThreadPoolExecutor(max_workers=CHUNKS).map(cpu_work, [ROUNDS] * CHUNKS)))
print(f"{CHUNKS} تکه کارِ محاسباتی، یکی‌یکی: {cpu_serial:.2f} ثانیه")
print(f"{CHUNKS} تکه با {CHUNKS} نخ          : {cpu_threads:.2f} ثانیه")
print(f"بهبود: {cpu_serial / cpu_threads:.2f} برابر")
print("همان جواب؟", thread_sums == serial_sums)

4 تکه کارِ محاسباتی، یکی‌یکی: 0.98 ثانیه
4 تکه با 4 نخ          : 0.94 ثانیه
بهبود: 1.04 برابر
همان جواب؟ True

چهار نخ، چهار برابر کار، همان زمان. یعنی هیچ.

دلیلش GIL است: قفلی که اجازه می‌دهد در هر لحظه فقط یک نخ کدِ پایتون اجرا کند. نخ‌ها واقعاً موازی‌اند وقتی منتظر چیزی هستند — شبکه، دیسک، یا کدِ C که قفل را آزاد می‌کند — ولی برای اجرای خودِ دستورهای پایتون نوبتی می‌شوند.

و همین یک جمله کلِ فصل را دو نیم می‌کند:

• کارِ انتظاری (شبکه، فایل، پایگاه داده) → نخ یا async. بخشِ ۲ و ۴.
• کارِ محاسباتیِ خالصِ پایتونی → نخ بی‌فایده است. باید فرآیندِ جدا بسازی.

نکتهٔ مهم برای کارِ داده: حلقه‌های numpy و بخشِ سنگینِ scikit-learn در C اجرا می‌شوند و قفل را آزاد می‌کنند، پس آن‌ها از نخ سود می‌برند. قاعده این نیست که «نخ برای محاسبه بی‌فایده است»، قاعده این است که «نخ برای کدِ پایتونیِ محاسباتی بی‌فایده است».

۶. فرآیند: سودِ واقعی، با هزینهٔ ثابت#

هر فرآیند مفسرِ خودش و GILِ خودش را دارد، پس واقعاً هم‌زمان می‌شود. قیمتش این است که راه‌اندازیِ هر مفسر وقت می‌برد — و آن وقت را جدا می‌سنجیم:

import subprocess
import sys
from pathlib import Path

Path("worker.py").write_text('''import sys


def cpu_work(rounds):
    total = 0
    for i in range(rounds):
        total += (i * i) % 7
    return total


print(cpu_work(int(sys.argv[1])))
''', encoding="utf-8")

Path("nothing.py").write_text("pass\n", encoding="utf-8")


def run_processes(count, rounds):
    procs = [subprocess.Popen([sys.executable, "worker.py", str(rounds)],
                              stdout=subprocess.PIPE, text=True) for _ in range(count)]
    return [int(p.communicate()[0].strip()) for p in procs]


startup, _ = clock(lambda: [subprocess.run([sys.executable, "nothing.py"]) for _ in range(CHUNKS)])
cpu_procs, proc_sums = clock(lambda: run_processes(CHUNKS, ROUNDS))
print(f"{CHUNKS} فرآیندِ جدا: {cpu_procs:.2f} ثانیه")
print(f"بهبود در برابرِ یکی‌یکی: {cpu_serial / cpu_procs:.2f} برابر")
print(f"از این زمان، {startup:.2f} ثانیه فقط بالا آمدنِ {CHUNKS} مفسر بود")
print("همان جواب؟", proc_sums == serial_sums)

4 فرآیندِ جدا: 0.35 ثانیه
بهبود در برابرِ یکی‌یکی: 2.80 برابر
از این زمان، 0.17 ثانیه فقط بالا آمدنِ 4 مفسر بود
همان جواب؟ True

۲٫۸ برابر — و تقریباً نصفِ همین زمانِ چهارفرآیندی، فقط بالا آمدنِ چهار مفسر بود.

آن startup را جدا سنجیدیم و این کارِ اصلیِ بخش است. بدونش می‌گفتیم «چهار فرآیند فقط ۲٫۸ برابر شد، پس موازی‌سازی خوب کار نمی‌کند». با آن می‌دانیم که خودِ محاسبه تقریباً کاملاً موازی شد و هزینهٔ ثابت، سود را خورد.

و همین عدد قاعدهٔ تصمیم را می‌سازد: فرآیند وقتی می‌ارزد که کارِ هر تکه از زمانِ راه‌اندازیِ یک مفسر خیلی بیشتر باشد. بخشِ بعد نشان می‌دهد وقتی این شرط برقرار نباشد چه فاجعه‌ای می‌شود.

۷. کِی هیچ‌کدام کمک نمی‌کنند#

TINY = 2_000

tiny_serial, tiny_a = clock(lambda: [cpu_work(TINY) for _ in range(CHUNKS)])
tiny_threads, tiny_b = clock(
    lambda: list(ThreadPoolExecutor(max_workers=CHUNKS).map(cpu_work, [TINY] * CHUNKS)))
tiny_procs, tiny_c = clock(lambda: run_processes(CHUNKS, TINY))

print(f"کارِ ریز، یکی‌یکی : {tiny_serial * 1000:>8.1f} میلی‌ثانیه")
print(f"کارِ ریز، با نخ   : {tiny_threads * 1000:>8.1f} میلی‌ثانیه  ({tiny_serial / tiny_threads:.2f} برابر)")
print(f"کارِ ریز، با فرآیند: {tiny_procs * 1000:>8.1f} میلی‌ثانیه  ({tiny_serial / tiny_procs:.2f} برابر)")

کارِ ریز، یکی‌یکی :      0.5 میلی‌ثانیه
کارِ ریز، با نخ   :      1.0 میلی‌ثانیه  (0.49 برابر)
کارِ ریز، با فرآیند:     78.3 میلی‌ثانیه  (0.01 برابر)

نخ حدودِ دو برابر کندتر، فرآیند بیش از صد برابر کندتر.

خودِ کار نیم میلی‌ثانیه بود. ساختنِ نخ‌ها بیشتر از خودِ کار طول کشید، و ساختنِ چهار مفسرِ پایتون ده‌ها برابرِ کاری بود که قرار بود انجام بدهند. نسبتِ دقیقش روی ماشینِ تو فرق می‌کند و مهم هم نیست؛ مرتبهٔ بزرگی‌اش همیشه همین است.

این جدولِ نهایی، خلاصهٔ تصمیم است:

print("درخواست‌هایی که دو سرور جواب دادند:", SERVED)
for running in (server, roomy):
    running.shutdown()
    running.server_close()
print("سرورها بسته شدند.")

درخواست‌هایی که دو سرور جواب دادند: 169
سرورها بسته شدند.

✅ چک کن: آن عددِ ۱۶۹ باید دقیقاً همین باشد: یک درخواستِ آزمایشی، ۲۴ سریالی، سه دورِ ۲۴تایی با نخ، دو دورِ ۲۴تایی روی سرورِ دوم، و ۲۴ با asyncio. اگر عددت فرق دارد یعنی سلولی را دو بار اجرا کرده‌ای — و همین شمارنده، ساده‌ترین شکلِ چیزی است که در ترمِ ۵ اسمش «پایش» می‌شود.

⚠️ مواظب باش: سرور را در همان نوت‌بوک با daemon=True بالا آوردیم تا با بسته شدنِ runtime خودش برود. اگر سلولِ اول را دو بار بزنی، دو سرور روی دو پورت بالا می‌آید و اولی تا آخرِ نشست زنده می‌ماند. بستنِ صریح — همان shutdown() بالا — عادتِ درستی است که در ترمِ ۴ به یک قاعده تبدیل می‌شود.

🤖 از دستیارت بپرس: «تفاوتِ concurrency و parallelism چیست؟» جوابِ درست: هم‌زمانی یعنی چند کار در جریان باشند، موازی‌سازی یعنی چند کار در همان لحظه اجرا شوند — و asyncio اولی را بدونِ دومی می‌دهد. بعد خودت این را امتحان کن: DELAY را صفر کن و کلِ بخشِ ۲ را دوباره بزن. حالا نخ‌ها چقدر کمک می‌کنند؟ جوابت باید دقیقاً با جدولِ سطرِ آخر بخواند.

واژه‌های تازهٔ این فصل#

در فصل بعد#

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