المعالجة المتعددة

عمليات منفصلة وذاكرات منفصلة

لا تستطيع الخيوط تسريع بايثون المقيَّدة بالمعالج بسبب GIL، لكن العمليات تستطيع: فكل واحدة مفسّر بايثون كامل بقفل GIL خاص به، ويشغّلها نظام التشغيل على أنوية مختلفة. وتبدأ وحدة multiprocessing والأسهل concurrent.futures.ProcessPoolExecutor مجموعة من عمليات العمل وتعطي كلًّا حصة من العمل. واختبار رجعي على ألف مجموعة معاملات أو مونت كارلو بملايين المسارات يستطيع استعمال الأنوية الثمانية في حاسوب محمول وينتهي في نحو ثُمن الوقت.

# A pool of processes: each gets a slice of the work, computed on its own core
from concurrent.futures import ProcessPoolExecutor

def sum_of_squares(bounds):
    start, stop = bounds
    return sum(i * i for i in range(start, stop))

if __name__ == "__main__":                           # required on Windows and macOS, where workers re-import this file
    bounds = [(0, 250_000), (250_000, 500_000), (500_000, 750_000), (750_000, 1_000_000)]
    with ProcessPoolExecutor(max_workers=4) as pool:
        partials = list(pool.map(sum_of_squares, bounds))
    print(sum(partials))                             # 333332833333500000
    print(sum(partials) == sum(i * i for i in range(1_000_000)))        # True  the same answer as one process

ما يعبر الحدود: النسخ وpickle والأجزاء

لا تتشارك العمليات المتغيرات. وكل ما ترسله إلى عامل، الدالة ووسائطها، يُسلسَل بـ pickle (إلى بايتات) ويُنسخ ويُعاد بناؤه، وتعود النتيجة بالطريقة نفسها. فلا يعبر إلا ما يمكن تسلسله: الدوال على مستوى الوحدة والبيانات البسيطة نعم، أما lambda والدوال المتداخلة والأقفال والمولدات فلا وتفشل بـ PicklingError أو AttributeError. والنسخ له كلفة: أرسل القليل واحسب الكثير. أعطِ كل عامل زوج فهارس (start, stop) بدل نسخة من البيانات حين يستطيع تحميلها أو توليدها بنفسه، ووزّع بضعة أجزاء كبيرة بدل مليون مهمة صغيرة، وأرجع نتائج صغيرة كمجموع لا مصفوفة كاملة.

# Index pairs instead of data, and a check that something can be sent to another process
import pickle

def chunk_bounds(n, parts):
    size, extra = divmod(n, parts)
    bounds, start = [], 0
    for index in range(parts):
        stop = start + size + (1 if index < extra else 0)
        bounds.append((start, stop))
        start = stop
    return bounds

def is_picklable(obj):
    try:
        pickle.dumps(obj)
    except (pickle.PicklingError, AttributeError, TypeError):
        return False
    return True

print(chunk_bounds(1_000_000, 4))                    # [(0, 250000), (250000, 500000), (500000, 750000), (750000, 1000000)]
print(is_picklable({"AAPL": 184.25}), is_picklable(lambda x: x))      # True False

تدفقات عشوائية مستقلة ودمج النتائج الجزئية

تحدد تفصيلتان صحة مونت كارلو المتوازي. الأولى البذور العشوائية: عامل يبدأ بالبذرة نفسها التي يبدأ بها عامل آخر ينتج الأرقام «العشوائية» نفسها، وإعطاء العمال seed + 0 وseed + 1 ... يصنع تدفقات قد تتداخل أو ترتبط. وتشتق np.random.SeedSequence(seed).spawn(n) عدد n من البذور الفرعية المستقلة إحصائيًا من أب واحد فيبقى التشغيل كله قابلًا للتكرار من رقم واحد. والثانية الدمج: يرجع كل عامل ملخصًا صغيرًا وتدمجها. تُجمع الأعداد، وتُدمج المتوسطات متوسطًا مرجّحًا بالعدد، ويجب أن يشمل تباين المجموع التشتت بين متوسطات العمال، فلا يكفي متوسط التباينات.

# Independent seeds for the workers, and merging their partial statistics exactly
import numpy as np

def spawn_seeds(seed, n):
    return [int(child.generate_state(1)[0]) for child in np.random.SeedSequence(seed).spawn(n)]

def combine_mean_var(partials):
    count = sum(n for n, _, _ in partials)
    mean = sum(n * m for n, m, _ in partials) / count
    variance = sum(n * (v + (m - mean) ** 2) for n, m, v in partials) / count
    return count, mean, variance

seeds = spawn_seeds(7, 4)
print(len(set(seeds)), spawn_seeds(7, 4) == seeds)   # 4 True   four different seeds, reproducible from one number

data = np.random.default_rng(seeds[0]).normal(0.0, 1.0, size=1_000)
partials = [(len(p), p.mean(), p.var()) for p in np.array_split(data, 4)]
count, mean, variance = combine_mean_var(partials)
print(count, np.isclose(mean, data.mean()), np.isclose(variance, data.var()))      # 1000 True True

# A parallel Monte Carlo: every worker has its own seed, and the run is still exactly reproducible
from concurrent.futures import ProcessPoolExecutor

import numpy as np

def simulate(job):
    seed, paths = job
    z = np.random.default_rng(seed).standard_normal(paths)
    payoff = np.maximum(100.0 * np.exp(-0.5 * 0.2 ** 2 + 0.2 * z) - 100.0, 0.0)   # at-the-money call, 20% volatility
    return paths, payoff.mean(), payoff.var()

if __name__ == "__main__":
    seeds = [int(child.generate_state(1)[0]) for child in np.random.SeedSequence(2024).spawn(4)]
    jobs = [(seed, 50_000) for seed in seeds]
    with ProcessPoolExecutor(max_workers=4) as pool:
        partials = list(pool.map(simulate, jobs))
    count = sum(n for n, _, _ in partials)
    price = sum(n * mean for n, mean, _ in partials) / count
    print(count, round(price, 1))                    # 200000 7.9   close to the exact Black-Scholes value of about 7.97
    print(partials == [simulate(job) for job in jobs])      # True  the same seeds give the same numbers in one process

• لكل عملية GIL وذاكرتها فيمكن لبايثون المقيَّدة بالمعالج استعمال كل الأنوية؛ ومجموعة (ProcessPoolExecutor) توزع العمل.
• تُسلسَل الوسائط والنتائج وتُنسخ: أرسل أزواج فهارس ونتائج صغيرة في أجزاء كبيرة قليلة؛ ولا يمكن إرسال lambda والدوال المتداخلة.
• امنح كل عامل بذرة مستقلة بـ SeedSequence.spawn وادمج الإحصاءات الجزئية بدقة وأحِط نقطة الدخول وقِس التسريع الحقيقي.

التالي: قواعد بيانات SQL، تخزين بيانات السوق والاستعلام عنها بالجداول والاستعلامات.

المختبر: ما الذي يُقيَّم

• قطّع المجال إلى شرائح باستخدام chunk_bounds25 نقطة
• ادمج إحصاءات جزئية باستخدام combine_mean_var25 نقطة
• امنح كل عامل بذرته باستخدام spawn_seeds25 نقطة
• احسب كلفة التوازي باستخدام parallel_speedup25 نقطة