آموزش ماژول threading – کلاس Barrier

مقدمه

فرض کنید چند ترد دارید که هر کدام یک بخش از یک محاسبه را در «فاز اول» انجام می‌دهند، و فاز دوم باید تنها زمانی شروع شود که همه‌ی تردها فاز اول را تمام کرده باشند – مثل یک مسابقه‌ی دو امدادی که نفر بعدی نباید زودتر از موعد شروع کند. کلاس Barrier دقیقاً برای همین نوع همگام‌سازی طراحی شده است: تعدادی ترد را در یک نقطه نگه می‌دارد تا همه به آن نقطه برسند و سپس همه را همزمان آزاد می‌کند.

ایجاد یک Barrier

import threading

barrier = threading.Barrier(3)

پارامتر اول، parties نام دارد و تعداد تردهایی است که باید wait را صدا بزنند تا همه آزاد شوند. یک پارامتر اختیاری به نام action هم وجود دارد: یک تابع بدون آرگومان که دقیقاً یک بار، توسط یکی از تردها، در لحظه‌ی آزاد شدن همه فراخوانی می‌شود؛ مناسب برای کارهایی مثل چاپ گزارش بین فازها. پارامتر سوم هم timeout نام دارد و مقدار پیش‌فرض زمان انتظار را تعیین می‌کند.

متد wait

هر ترد وقتی به نقطه‌ی همگام‌سازی می‌رسد، متد wait روی شیء barrier را صدا می‌زند. این متد تا زمانی که به تعداد parties ترد آن را صدا نزده باشند، مسدود می‌ماند. به محض رسیدن آخرین ترد، همه‌ی تردهای منتظر همزمان آزاد می‌شوند.

import threading

def worker(barrier, name):
    print(f'{name}: phase 1 started')
    barrier.wait()
    print(f'{name}: phase 2 started')

barrier = threading.Barrier(3)
threads = [threading.Thread(target=worker, args=(barrier, f'worker-{i}')) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

خروجی متد wait یک عدد صحیح بین صفر تا parties منهای یک است؛ می‌توان از آن برای انتخاب یکی از تردها (مثلاً همان ترد که این عدد برایش صفر شد) جهت انجام یک کار ویژه پس از رسیدن همه استفاده کرد. دقت کنید که اگر تعداد تردهایی که wait را صدا می‌زنند کمتر یا بیشتر از parties باشد، یا یکی از تردها هرگز به آن نقطه نرسد، بقیه‌ی تردها برای همیشه یا تا پایان timeout منتظر می‌مانند.

وضعیت خراب Barrier و استثنای مربوط به آن

اگر یکی از تردها هنگام انتظار به timeout بخورد، یا متد abort فراخوانی شود، Barrier وارد حالت «خراب» می‌شود. در این حالت، تمام تردهای دیگری که در همان لحظه منتظر بودند یا بعداً وارد می‌شوند، با استثنای threading.BrokenBarrierError مواجه می‌شوند. این رفتار عمدی است: چون قرارداد Barrier این است که «یا همه با هم رد می‌شوند، یا هیچ‌کس»، اگر یک ترد نتواند برسد، ادامه دادن برای بقیه هم منطقی نیست.

import threading
import time

def worker(barrier, name, delay):
    try:
        time.sleep(delay)
        barrier.wait(timeout=1)
        print(f'{name}: passed')
    except threading.BrokenBarrierError:
        print(f'{name}: broken')

barrier = threading.Barrier(3)
delays = [0, 0, 2]
threads = [threading.Thread(target=worker, args=(barrier, f'worker-{i}', delays[i])) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

در این مثال، ترد سوم با تاخیر دو ثانیه‌ای، از timeout یک ثانیه‌ای عبور می‌کند و باعث می‌شود Barrier بشکند؛ به همین دلیل دو ترد دیگر هم استثنا می‌گیرند، نه فقط همان تردی که دیر کرده. برای بازگرداندن Barrier به حالت سالم پس از شکستن، می‌توانید متد reset را روی آن فراخوانی کنید.

ویژگی‌های کاربردی

• parties: تعداد تردهای لازم برای آزاد شدن.
• n_waiting: تعداد تردهایی که در حال حاضر منتظرند.
• broken: مقدار بولی که نشان می‌دهد Barrier در حالت خراب است یا نه.

جمع‌بندی

Barrier برای زمانی مناسب است که چند ترد باید مرحله به مرحله با هم پیش بروند و هیچ تردی نباید زودتر از بقیه وارد مرحله‌ی بعد شود؛ مثلاً در شبیه‌سازی‌های گام‌به‌گام یا الگوریتم‌های موازی که هر مرحله به نتیجه‌ی کامل مرحله‌ی قبل نیاز دارد. برخلاف Lock و Semaphore که برای محافظت از منابع استفاده می‌شوند، Barrier ابزاری برای «همگام‌سازی زمانی» بین تردهاست.