التوازي والتزامن: threads و processes و asyncio — Python المتقدم

التوازي والتزامن: أيّها تختار؟

تعلمت async/await. لكن Python توفر ثلاث طرق لتنفيذ أكثر من شيء في نفس الوقت، ولكل منها مجال يناسبه. الاختيار الخاطئ يعطي برنامجاً أعقد دون أي تسريع.

نوعان من العمل

| النوع | المعنى | أمثلة |

|---|---|---|

| مقيد بالانتظار (I/O-bound) | معظم الوقت انتظار لشيء خارجي | طلبات شبكة، قراءة ملفات، قواعد بيانات |

| مقيد بالمعالج (CPU-bound) | المعالج يعمل باستمرار | معالجة صور، حسابات رياضية، ضغط بيانات |

الخيارات الثلاثة

| الطريقة | آلية العمل | الأنسب لـ |

|---|---|---|

| asyncio | خيط واحد يتنقل بين المهام عند كل await | آلاف عمليات I/O المتزامنة بكفاءة |

| threading | عدة خيوط في نفس العملية | I/O مع مكتبات متزامنة (requests مثلاً) |

| multiprocessing | عدة عمليات منفصلة، لكل منها مفسر وذاكرة | العمل الحسابي الثقيل (CPU-bound) |

GIL: لماذا لا تسرّع الخيوط الحسابات؟

في Python القياسية (CPython) يوجد قفل المفسر العام (GIL): خيط واحد فقط ينفذ كود Python في كل لحظة. أثناء انتظار I/O يُحرَّر القفل فتستفيد الخيوط، أما في الحسابات الصافية فتتناوب الخيوط على القفل دون تسريع حقيقي. لذلك للحسابات الثقيلة نستخدم العمليات، لأن لكل عملية مفسرها وقفلها.

ملاحظة: Python 3.13+ تقدم بناءً تجريبياً بدون GIL (free-threaded)، لكنه ليس الافتراضي بعد.

concurrent.futures: واجهة موحدة وبسيطة

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

def download(url):            # انتظار شبكة (محاكاة)
    time.sleep(0.1)
    return url

def heavy_compute(n):         # حساب
    return sum(i * i for i in range(n))

urls = [f"https://example.com/{i}" for i in range(8)]
numbers = [200_000] * 4


if __name__ == "__main__":          # ضروري مع العمليات (Windows و macOS)
    with ThreadPoolExecutor(max_workers=8) as pool:      # للانتظار
        pages = list(pool.map(download, urls))

    with ProcessPoolExecutor() as pool:                   # للحساب
        results = list(pool.map(heavy_compute, numbers))
    print(len(pages), len(results))

نفس الشكل للطريقتين، فيسهل التبديل والقياس. map ترجع النتائج بترتيب المدخلات.

تحذيرات مهمة

- مع العمليات: ضع كود التشغيل تحت if __name__ == "__main__":، وإلا ستعيد كل عملية فرعية تنفيذ الملف (مطلوب في Windows و macOS). الدوال والبيانات المرسلة للعمليات يجب أن تكون قابلة للتسلسل (pickle)، فلا تصلح دوال lambda.

- مع الخيوط: البيانات المشتركة التي تُعدَّل من عدة خيوط تحتاج قفلاً (threading.Lock) وإلا تحدث حالات تسابق (race conditions) تعطي نتائج خاطئة أحياناً فقط، وهي أصعب الأخطاء اكتشافاً.

- إنشاء عملية أبطأ بكثير من إنشاء خيط؛ للمهام الصغيرة جداً قد تكون الطريقة المتتابعة أسرع.

القاعدة العملية

1. اكتب الحل المتتابع البسيط أولاً، وقِس الوقت.

2. إن كان البطء من الانتظار ← asyncio (أو الخيوط إن كانت مكتباتك متزامنة).

3. إن كان من الحساب ← ProcessPoolExecutor.

4. قِس مرة أخرى: هل تحسّن فعلاً؟

تمرين

شغّل المثال وقارن الأزمنة الثلاثة للعمل الحسابي. ثم غيّر الدالة لتكون انتظاراً (time.sleep(0.2)) وأعد المقارنة، ولاحظ أن الخيوط أصبحت الأسرع.

import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

def cpu_task(n: int) -> int:
    """عمل حسابي صافٍ"""
    total = 0
    for i in range(n):
        total += i * i % 7
    return total

def io_task(seconds: float) -> float:
    """محاكاة انتظار شبكة"""
    time.sleep(seconds)
    return seconds

def measure(label, func):
    start = time.perf_counter()
    func()
    print(f"{label:<28} {time.perf_counter() - start:.2f} ث")

if __name__ == "__main__":   # ضروري مع العمليات
    jobs = [600_000] * 4
    measure("حساب: متتابع", lambda: [cpu_task(n) for n in jobs])
    with ThreadPoolExecutor(4) as pool:
        measure("حساب: خيوط (GIL)", lambda: list(pool.map(cpu_task, jobs)))
    with ProcessPoolExecutor(4) as pool:
        measure("حساب: عمليات", lambda: list(pool.map(cpu_task, jobs)))

    waits = [0.3] * 4
    measure("انتظار: متتابع", lambda: [io_task(s) for s in waits])
    with ThreadPoolExecutor(4) as pool:
        measure("انتظار: خيوط", lambda: list(pool.map(io_task, waits)))