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

پیش‌نیازها

برای درک بهتر مطالب این بخش، لازم است ابتدا با مفاهیم قفل‌گذاری در مولتی‌تردینگ آشنا باشید. در آنجا دیدیم که چطور با Lock از دسترسی هم‌زمانِ چند ترد به یک منبع مشترک جلوگیری می‌کنیم. در این مقاله یک قدم جلوتر می‌رویم و می‌بینیم چطور می‌توان به‌جای «فقط یک ترد»، به «چند ترد به‌صورت کنترل‌شده» اجازه‌ی دسترسی هم‌زمان داد.

کلاس Semaphore

برخلاف کلاس Lock که فقط به یک ترد در هر لحظه اجازه‌ی دسترسی به منابع مشترک را می‌دهد، کلاس Semaphore این امکان را فراهم می‌کند که چند ترد به‌صورت هم‌زمان — اما نه بیشتر از یک تعداد مشخص — به منبع مشترک دسترسی داشته باشند.

یک قیاس ساده: Semaphore مثل یک پارکینگ با تعداد محدودی جای پارک است. اگر پارکینگ ۲ جا داشته باشد، هر لحظه فقط ۲ خودرو می‌توانند داخل باشند. خودروی سوم باید پشت در منتظر بماند تا یکی از دو خودروی داخل، خارج شود و جا خالی کند. اینجا «خودرو» همان ترد است، «جای پارک» همان ظرفیت سمافور، ورود به پارکینگ acquire و خروج از آن release است.

تعداد تردهایی که هم‌زمان می‌توانند به منبع اشتراکی دسترسی داشته باشند با یک شمارنده‌ی داخلی مدیریت می‌شود:

• ابتدا یک سمافور با یک مقدار اولیه ساخته می‌شود؛ مثلاً Semaphore(2) یعنی شمارنده از عدد ۲ شروع می‌شود.
• هر بار که acquire فراخوانی شود، یک واحد از شمارنده کم می‌شود. اگر شمارنده به صفر برسد، ترد بعدی که acquire کند تا آزادشدن یک جا متوقف می‌ماند.
• هر بار که release فراخوانی شود، یک واحد به شمارنده اضافه می‌شود و یکی از تردهای منتظر آزاد می‌شود.

فرض کنید بخواهیم در هر لحظه حداکثر دو ترد به‌صورت هم‌زمان به منبع مشترک دسترسی داشته باشند. برای این کار کلاس Semaphore را به شکل زیر به کار می‌بریم.

from time import sleep
from threading import Thread, Semaphore, current_thread

lock = Semaphore(2)          # Maximum 2 threads at the same time

def download():
    lock.acquire()           # Take a spot (if no spot is available, it waits)
    # Access to shared resources
    print('Downloading image by', current_thread().name)
    sleep(1)
    lock.release()           # Free up a spot for the next thread

threads = [Thread(target=download) for _ in range(5)]

for t in threads:
    t.start()

from time import sleep
from threading import Thread, Semaphore, current_thread

lock = Semaphore(2)          # Maximum 2 threads at the same time

def download():
    lock.acquire()           # Take a spot (if no spot is available, it waits)
    # Access to shared resources
    print('Downloading image by', current_thread().name)
    sleep(1)
    lock.release()           # Free up a spot for the next thread

threads = [Thread(target=download) for _ in range(5)]

for t in threads:
    t.start()

در این برنامه ۵ ترد ساخته می‌شود، اما چون ظرفیت سمافور ۲ است، خروجی به‌صورت دسته‌های دوتایی ظاهر می‌شود: ابتدا دو ترد پیام «Downloading» را چاپ می‌کنند و یک ثانیه می‌خوابند؛ سپس دو ترد بعدی و در نهایت ترد پنجم. یعنی هیچ‌گاه بیش از دو ترد هم‌زمان وارد بخش بحرانی نمی‌شوند.

نکته: Semaphore(1) عملاً معادل یک Lock ساده رفتار می‌کند، چون فقط به یک ترد در هر لحظه اجازه‌ی ورود می‌دهد.

روش تمیزتر: استفاده از with

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

from time import sleep
from threading import Thread, Semaphore, current_thread

lock = Semaphore(2)

def download():
    with lock:  # acquire on entry, automatic release on exit
        print('Downloading image by', current_thread().name)
        sleep(1)

threads = [Thread(target=download) for _ in range(5)]

for t in threads:
    t.start()

from time import sleep
from threading import Thread, Semaphore, current_thread

lock = Semaphore(2)

def download():
    with lock:  # acquire on entry, automatic release on exit
        print('Downloading image by', current_thread().name)
        sleep(1)

threads = [Thread(target=download) for _ in range(5)]

for t in threads:
    t.start()

این نسخه دقیقاً همان کار قبلی را انجام می‌دهد، اما امن‌تر و خواناتر است. تا جای ممکن همین الگوی with را استفاده کنید.

صبرنکردن تا ابد: acquire با timeout

گاهی نمی‌خواهیم یک ترد بی‌نهایت پشت در منتظر بماند. متد acquire دو پارامتر مفید دارد: با blocking=False اصلاً منتظر نمی‌ماند و اگر جا نباشد فوراً False برمی‌گرداند؛ و با timeout حداکثر همان تعداد ثانیه صبر می‌کند و اگر جا باز نشد False برمی‌گرداند. مقدار بازگشتی True یعنی جا گرفته شد.

from time import sleep
from threading import Thread, Semaphore, current_thread

lock = Semaphore(2)

def worker():
    acquired = lock.acquire(timeout=1)   # Wait for a maximum of 1 second
    if not acquired:
        print('Skipped:', current_thread().name)   # No spot opened up; we skip
        return
    try:
        print('Working:', current_thread().name)
        sleep(2)
    finally:
        lock.release()

threads = [Thread(target=worker) for _ in range(5)]

for t in threads:
    t.start()

from time import sleep
from threading import Thread, Semaphore, current_thread

lock = Semaphore(2)

def worker():
    acquired = lock.acquire(timeout=1)   # Wait for a maximum of 1 second
    if not acquired:
        print('Skipped:', current_thread().name)   # No spot opened up; we skip
        return
    try:
        print('Working:', current_thread().name)
        sleep(2)
    finally:
        lock.release()

threads = [Thread(target=worker) for _ in range(5)]

for t in threads:
    t.start()

چون دو ترد اول جا را می‌گیرند و ۲ ثانیه نگه می‌دارند، سه ترد دیگر بعد از ۱ ثانیه انتظار ناموفق، پیام «Skipped» را چاپ می‌کنند و بی‌آنکه بلاک شوند از کار خارج می‌شوند. این الگو برای جلوگیری از انتظار بی‌پایان بسیار کاربردی است.

یک کاربرد واقعی: محدودکردن دانلودهای هم‌زمان

رایج‌ترین کاربرد Semaphore در دنیای واقعی، «محدودکردن نرخ هم‌زمانی» است. تصور کنید می‌خواهید ۲۰ فایل را دانلود کنید، اما نمی‌خواهید بیش از ۳ دانلود هم‌زمان انجام شود (مثلاً برای اینکه پهنای باند یا سرور مقصد را زیر فشار نگذارید). با یک سمافور به ظرفیت ۳ به‌راحتی این محدودیت را اعمال می‌کنیم:

from time import sleep
from threading import Thread, Semaphore, current_thread

MAX_CONCURRENT = 3
semaphore = Semaphore(MAX_CONCURRENT)

def download(url):
    with semaphore:                       # Maximum 3 concurrent downloads
        print('Start:', url, '-', current_thread().name)
        sleep(2)                          # Simulate download time
        print('Done :', url)

urls = [f'file-{i}.zip' for i in range(20)]
threads = [Thread(target=download, args=(u,)) for u in urls]

for t in threads:
    t.start()
for t in threads:
    t.join()

print('All downloads finished')

from time import sleep
from threading import Thread, Semaphore, current_thread

MAX_CONCURRENT = 3
semaphore = Semaphore(MAX_CONCURRENT)

def download(url):
    with semaphore:                       # Maximum 3 concurrent downloads
        print('Start:', url, '-', current_thread().name)
        sleep(2)                          # Simulate download time
        print('Done :', url)

urls = [f'file-{i}.zip' for i in range(20)]
threads = [Thread(target=download, args=(u,)) for u in urls]

for t in threads:
    t.start()
for t in threads:
    t.join()

print('All downloads finished')

هرچند ۲۰ ترد ساخته و استارت می‌شوند، سمافور تضمین می‌کند که در هر لحظه فقط ۳ دانلود واقعاً در حال اجرا باشند و بقیه در صف بمانند. همین الگو برای محدودکردن تعداد اتصال‌های هم‌زمان به یک پایگاه‌داده یا API نیز به کار می‌رود.

مشکل عملکرد Semaphore

برنامه‌هایی که در دنیای واقعی نوشته می‌شوند همیشه به سادگی مثال‌های دوره‌های آموزشی نیستند. گاهی کدها پیچیده می‌شوند و ممکن است ناخواسته تعداد فراخوانی‌های release بیشتر از acquire شود. در این حالت عملکرد Semaphore خراب می‌شود، چون مقدار شمارنده در طول اجرای برنامه از چیزی که در ابتدا مشخص کرده بودیم بیشتر و بیشتر می‌شود و در نتیجه ظرفیت واقعی از حدِ موردنظر ما فراتر می‌رود.

برای مثال در کد زیر، به‌ازای هر بار acquire دو بار release فراخوانی می‌شود:

from time import sleep
from threading import Thread, Semaphore, current_thread

lock = Semaphore(2)

def add():
    lock.acquire()
    print('Downloading image by', current_thread().name)
    sleep(2)
    lock.release()
    lock.release()          # One extra incorrect release!

threads = [Thread(target=add) for _ in range(13)]

for t in threads:
    t.start()

from time import sleep
from threading import Thread, Semaphore, current_thread

lock = Semaphore(2)

def add():
    lock.acquire()
    print('Downloading image by', current_thread().name)
    sleep(2)
    lock.release()
    lock.release()          # One extra incorrect release!

threads = [Thread(target=add) for _ in range(13)]

for t in threads:
    t.start()

مشکل کد بالا این است که هر بار یک واحد اضافه به شمارنده تزریق می‌شود؛ بنابراین به‌مرور زمان تعداد تردهایی که هم‌زمان به منابع مشترک دسترسی پیدا می‌کنند بیشتر و بیشتر می‌شود و محدودیت «حداکثر ۲» عملاً از بین می‌رود. بدتر اینکه Semaphore این اشتباه را گزارش نمی‌کند و برنامه بدون هیچ خطایی به‌اشتباه کار می‌کند — که پیداکردن چنین باگی را بسیار سخت می‌کند.

کلاس BoundedSemaphore

برای جلوگیری از مشکلی که برای Semaphore مطرح شد، می‌توان از کلاس BoundedSemaphore استفاده کرد. این کلاس دقیقاً مثل Semaphore عمل می‌کند، با یک تفاوت مهم: مقدار شمارنده هرگز اجازه ندارد از مقدار اولیه‌اش بیشتر شود.

هر بار که release باعث افزایش شمارنده شود، اگر بخواهد از مقدار اولیه فراتر برود، بلافاصله یک ValueError ایجاد می‌شود. این خطا در واقع یک «زنگ خطر» است که به شما می‌گوید جایی از منطق برنامه، تعادل acquire و release به هم خورده است.

from time import sleep
from threading import Thread, BoundedSemaphore, current_thread

lock = BoundedSemaphore(2)

def add():
    try:
        lock.acquire()
        print('Downloading image by', current_thread().name)
        sleep(2)
        lock.release()
        lock.release()                       # Extra release => ValueError
    except ValueError:
        print('Can not release', current_thread().name)

threads = [Thread(target=add) for _ in range(13)]

for t in threads:
    t.start()

from time import sleep
from threading import Thread, BoundedSemaphore, current_thread

lock = BoundedSemaphore(2)

def add():
    try:
        lock.acquire()
        print('Downloading image by', current_thread().name)
        sleep(2)
        lock.release()
        lock.release()                       # Extra release => ValueError
    except ValueError:
        print('Can not release', current_thread().name)

threads = [Thread(target=add) for _ in range(13)]

for t in threads:
    t.start()

حالا به‌جای اینکه محدودیت بی‌سروصدا از بین برود، به‌محض اولین release اضافی یک ValueError پرتاب می‌شود و ما می‌توانیم آن را بگیریم و مدیریت کنیم. توجه کنید که این خطا تردِ خطاکار را از کار نمی‌اندازد؛ صرفاً به ما هشدار می‌دهد. یک نمونه‌ی کوچک‌تر برای دیدن دقیق رفتار خطا:

from threading import BoundedSemaphore

lock = BoundedSemaphore(2)
lock.acquire()
lock.release()
lock.release()      # ValueError: Semaphore released too many times

from threading import BoundedSemaphore

lock = BoundedSemaphore(2)
lock.acquire()
lock.release()
lock.release()      # ValueError: Semaphore released too many times

مقایسه Semaphore و BoundedSemaphore

از آنجا که BoundedSemaphore علاوه بر داشتن همه‌ی امکانات Semaphore، مشکل آن را هم ندارد، احتمالاً این سؤال برایتان پیش می‌آید که پس Semaphore اصلاً چه کاربردی دارد؟

کلاس BoundedSemaphore بعد از Semaphore به زبان اضافه شده است. به همین دلیل ممکن است کدهای قدیمی زیادی وجود داشته باشند که با Semaphore نوشته شده‌اند. نگه‌داشتن Semaphore در پایتون فقط این کمک را می‌کند که آن کدهای قدیمی کمتر نیاز به بازنویسی (refactor) و هماهنگی با نسخه‌های جدید داشته باشند. پس می‌توان گفت تنها دلیل حذف‌نشدن Semaphore، در واقع سازگاری به‌عقب (Backward Compatibility) است.

Semaphore در مقابل Lock

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

خلاصه

• Semaphore با یک شمارنده، به چند ترد به‌صورت هم‌زمان (تا سقف مشخص) اجازه‌ی دسترسی می‌دهد؛ acquire شمارنده را کم و release آن را زیاد می‌کند.
• تا جای ممکن از الگوی with استفاده کنید تا release حتی هنگام بروز خطا هم انجام شود.
• برای جلوگیری از انتظار بی‌پایان می‌توانید از acquire(timeout=...) یا acquire(blocking=False) استفاده کنید.
• برای ایجاد محدودیت روی تعداد تردها همیشه BoundedSemaphore را ترجیح دهید، چون releaseهای اضافه را با ValueError به شما گزارش می‌کند و از خرابیِ بی‌صدای محدودیت جلوگیری می‌کند.