queue --- کلاس صف همگام‌شده

کد منبع: Lib/queue.py

---

ماژول queue صف‌هایی با چند تولیدکننده و چند مصرف‌کننده را پیاده‌سازی می‌کند. این ماژول به‌ویژه در برنامه‌نویسی نخی، زمانی که اطلاعات باید به‌صورت ایمن بین چند نخ مبادله شود، مفید است. کلاس Queue در این ماژول، تمام معناهای قفل‌سازی مورد نیاز را پیاده‌سازی می‌کند.

این ماژول سه نوع صف را پیاده‌سازی می‌کند که تنها در ترتیب بازیابی ورودی‌ها تفاوت دارند. در صف FIFO، نخستین وظایف افزوده‌شده، نخستین وظایف بازیابی‌شده هستند. در صف LIFO، آخرین ورودی افزوده‌شده، نخستین ورودی بازیابی‌شده است (مانند یک هیپ عمل می‌کند). در صف اولویت، ورودی‌ها مرتب نگه داشته می‌شوند (با استفاده از ماژول heapq) و ورودی با کم‌ترین مقدار، ابتدا بازیابی می‌شود.

در داخل، آن سه نوع صف از قفل‌ها استفاده می‌کنند تا نخ‌های رقیب را به‌طور موقت مسدود کنند؛ با این حال، آن‌ها برای مدیریت بازورودپذیری (reentrancy) درون یک نخ طراحی نشده‌اند.

علاوه بر این، این ماژول یک نوع صف «ساده» از نوع FIFO، SimpleQueue، را پیاده‌سازی می‌کند که پیاده‌سازی خاص آن، تضمین‌های بیشتری را در ازای قابلیت کمتر ارائه می‌دهد.

ماژول queue کلاس‌ها و استثناهای زیر را تعریف می‌کند:

class queue.Queue(maxsize=0)

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

class queue.LifoQueue(maxsize=0)

سازنده‌ای برای صف LIFO (آخرین ورودی، اولین خروجی) <LIFO (last-in, first-out)>. maxsize یک عدد صحیح است که حد بالای تعداد آیتم‌هایی را که می‌توان در صف قرار داد، تعیین می‌کند. پس از رسیدن به این اندازه، درج تا زمانی که آیتم‌های صف مصرف شوند، مسدود خواهد شد. اگر maxsize کمتر یا مساوی صفر باشد، اندازه صف بی‌نهایت است.

class queue.PriorityQueue(maxsize=0)

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

ابتدا آیتم‌هایی با کم‌ترین مقدار بازیابی می‌شوند (آیتم با کم‌ترین مقدار، آیتمی است که min(entries) آن را بازمی‌گرداند). یک الگوی رایج برای آیتم‌ها، تاپلبه شکل (priority_number, data) است.

اگر عناصر data قابل مقایسه نباشند، می‌توان داده را در کلاسی پوششی قرار داد که آیتم داده را نادیده می‌گیرد و تنها عدد اولویت را مقایسه می‌کند:

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    item: Any=field(compare=False)

class queue.SimpleQueue

سازنده برای یک صف نامحدود FIFO. صف‌های ساده فاقد قابلیت‌های پیشرفته‌ای مانند پیگیری وظایف هستند.

صف‌های ساده نسبت به نوع آیتم‌های خود generic هستند.

اضافه شده در نسخه‌ی 3.7.

exception queue.Empty

استثنایی که هنگام فراخوانی get() به‌صورت غیرمسدودکننده (یا get_nowait()) روی یک شیء Queue که خالی است، پرتاب می‌شود.

exception queue.Full

استثنایی که هنگام فراخوانی put() (یا put_nowait()) به‌صورت غیرمسدودکننده بر روی یک شیء Queue که پر است، پرتاب می‌شود.

exception queue.ShutDown

استثنایی که هنگام فراخوانی put() یا get() روی یک شیء Queue که خاموش شده است، پرتاب می‌شود.

اضافه شده در نسخه‌ی 3.13.

اشیاء صف

اشیای صف (Queue، LifoQueue یا PriorityQueue) متدهای عمومی توصیف‌شده در زیر را ارائه می‌دهند.

Queue.qsize()

اندازه‌ی تقریبی صف را برمی‌گرداند. توجه داشته باشید که qsize() > 0 تضمین نمی‌کند که get() بعدی مسدود نخواهد شد، و qsize() < maxsize نیز تضمین نمی‌کند که put() مسدود نخواهد شد.

Queue.empty()

اگر صف خالی باشد، True و در غیر این صورت False برمی‌گرداند. اگر empty() مقدار True برگرداند، تضمین نمی‌کند که فراخوانی بعدی put() مسدود نخواهد شد. به‌طور مشابه، اگر empty() مقدار False برگرداند، تضمین نمی‌کند که فراخوانی بعدی get() مسدود نخواهد شد.

Queue.full()

اگر صف پر باشد، True و در غیر این صورت False برمی‌گرداند. اگر full() مقدار True را برگرداند، تضمین نمی‌کند که فراخوانی بعدی get() مسدود نخواهد شد. به‌طور مشابه، اگر full() مقدار False را برگرداند، تضمین نمی‌کند که فراخوانی بعدی put() مسدود نخواهد شد.

Queue.put(item, block=True, timeout=None)

item را در صف قرار می‌دهد. اگر آرگومان اختیاری block درست باشد و timeout برابر None باشد (پیش‌فرض)، در صورت لزوم تا زمانی که یک جایگاه آزاد در دسترس قرار گیرد، مسدود می‌شود. اگر timeout یک عدد مثبت باشد، حداکثر timeout ثانیه مسدود می‌شود و اگر در آن مدت جایگاه آزادی در دسترس قرار نگرفت، استثنای Full را پرتاب می‌کند. در غیر این صورت (block نادرست است)، اگر یک جایگاه آزاد بلافاصله در دسترس باشد، یک آیتم را در صف قرار می‌دهد، وگرنه استثنای Full را پرتاب می‌کند (در این حالت timeout نادیده گرفته می‌شود).

اگر صف خاموش‌شده باشد، ShutDown را پرتاب می‌کند.

Queue.put_nowait(item)

معادل با put(item, block=False).

Queue.get(block=True, timeout=None)

یک آیتم را از صف حذف و برمی‌گرداند. اگر آرگومان‌های اختیاری به این صورت باشند که block درست باشد و timeout برابر None باشد (پیش‌فرض)، در صورت لزوم تا زمانی که یک آیتم در دسترس باشد مسدود می‌شود. اگر timeout یک عدد مثبت باشد، حداکثر به مدت timeout ثانیه مسدود می‌شود و اگر هیچ آیتمی در آن مدت در دسترس نباشد، استثنای Empty را پرتاب می‌کند. در غیر این صورت (یعنی وقتی block نادرست باشد)، اگر یک آیتم بلافاصله در دسترس باشد آن را برمی‌گرداند، وگرنه استثنای Empty را پرتاب می‌کند (در این حالت timeout نادیده گرفته می‌شود).

پیش از 3.0 در سیستم‌های POSIX، و برای تمام نسخه‌ها در ویندوز، اگر block درست باشد و timeout برابر None باشد، این عملیات وارد یک انتظار غیرقابل‌وقفه برای یک قفل زیربنایی می‌شود. این بدان معناست که هیچ استثنایی نمی‌تواند رخ دهد، و به‌ویژه، یک SIGINT موجب KeyboardInterrupt نخواهد شد.

اگر صف خاموش شده باشد و خالی باشد، یا اگر صف به‌صورت فوری خاموش شده باشد، ShutDown را پرتاب می‌کند.

Queue.get_nowait()

معادل get(False).

دو متد برای پشتیبانی از پیگیری این موضوع ارائه شده است که آیا وظایف در صف قرار گرفته به‌طور کامل توسط نخ‌های مصرف‌کننده‌ی دِیمِن (daemon) پردازش شده‌اند یا خیر.

Queue.task_done()

نشان می‌دهد که وظیفه‌ای که پیش‌تر در صف قرار گرفته، کامل شده است. توسط نخ‌های مصرف‌کننده‌ی صف استفاده می‌شود. به ازای هر get() که برای دریافت یک وظیفه استفاده می‌شود، فراخوانی متعاقب task_done() به صف اطلاع می‌دهد که پردازش روی آن وظیفه کامل شده است.

اگر یک join() در حال حاضر مسدود شده باشد، هنگامی که همه‌ی آیتم‌ها پردازش شده باشند، ادامه می‌یابد (به این معنا که یک فراخوانی task_done() برای هر آیتمی که با put() در صف قرار داده شده بود، دریافت شده باشد).

اگر بیش از تعداد آیتم‌های قرار داده‌شده در صف فراخوانی شود، یک ValueError پرتاب می‌کند.

Queue.join()

مسدود می‌شود تا همه‌ی آیتم‌های موجود در صف دریافت و پردازش شده باشند.

تعداد وظایف ناتمام هر بار که یک آیتم به صف اضافه شود، افزایش می‌یابد. این تعداد هر بار که یک نخ مصرف‌کننده task_done() را فراخوانی می‌کند، کاهش می‌یابد تا نشان دهد آن آیتم دریافت شده و تمام کار روی آن کامل شده است. هنگامی که تعداد وظایف ناتمام به صفر برسد، join() از حالت مسدود خارج می‌شود.

انتظار برای تکمیل وظیفه

مثالی از نحوه‌ی انتظار برای تکمیل شدن وظایف در صف:

import threading
import queue

q = queue.Queue()

def worker():
    while True:
        item = q.get()
        print(f'Working on {item}')
        print(f'Finished {item}')
        q.task_done()

# Turn-on the worker thread.
threading.Thread(target=worker, daemon=True).start()

# Send thirty task requests to the worker.
for item in range(30):
    q.put(item)

# Block until all tasks are done.
q.join()
print('All work completed')

پایان دادن به صف‌ها

هنگامی که دیگر نیازی به آن‌ها نیست، اشیاء Queue می‌توانند تا خالی شدن به‌تدریج متوقف شوند یا بی‌درنگ با یک خاموشی سخت (hard shutdown) خاتمه یابند.

Queue.shutdown(immediate=False)

یک نمونه Queue را در حالت خاموشی قرار دهید.

صف دیگر نمی‌تواند رشد کند. فراخوانی‌های آینده به put()، ShutDown را پرتاب خواهند کرد. فراخوانی‌کنندگان مسدودشده‌ی کنونی put() از انسداد خارج خواهند شد و ShutDown را در نخ پیش‌تر مسدودشده پرتاب خواهند کرد.

اگر immediate نادرست باشد (پیش‌فرض)، می‌توان صف را به‌صورت عادی با فراخوانی‌های get() برای استخراج وظایفی که از قبل بارگذاری شده‌اند، تخلیه کرد.

و اگر task_done() برای هر وظیفه‌ی باقی‌مانده فراخوانی شود، یک join() در انتظار به‌طور عادی رفع انسداد خواهد شد.

همین که صف خالی شد، فراخوانی‌های بعدی به get()، ShutDown را پرتاب خواهند کرد.

اگر immediate مقدار true داشته باشد، صف بلافاصله خاتمه داده می‌شود. صف تخلیه می‌شود تا کاملاً خالی شود و تعداد وظایف تمام‌نشده به اندازه تعداد وظایف تخلیه‌شده کاهش می‌یابد. اگر تعداد وظایف تمام‌نشده صفر باشد، فراخوانندگان join() از حالت مسدود خارج می‌شوند. همچنین، فراخوانندگان مسدودشده‌ی get() از حالت مسدود خارج می‌شوند و ShutDown را پرتاب خواهند کرد، زیرا صف خالی است.

هنگام استفاده از join() در صورتی که immediate روی true تنظیم‌شده باشد، احتیاط کنید. این کار حتی زمانی که هیچ کاری روی وظایف انجام‌نشده باشد، join را از حالت مسدود خارج می‌کند و ناوردایی معمول برای پیوستن به یک صف را نقض می‌کند.

اضافه شده در نسخه‌ی 3.13.

اشیای SimpleQueue

اشیای SimpleQueue متدهای عمومی توصیف‌شده در زیر را ارائه می‌دهند.

SimpleQueue.qsize()

اندازه‌ی تقریبی صف را برمی‌گرداند. توجه داشته باشید که qsize() > 0 تضمین نمی‌کند که get() بعدی مسدود نخواهد شد.

SimpleQueue.empty()

اگر صف خالی باشد، True و در غیر این صورت False را برمی‌گرداند. اگر empty() مقدار False را برگرداند، تضمینی وجود ندارد که فراخوانی بعدی به get() مسدود نشود.

SimpleQueue.put(item, block=True, timeout=None)

item را در صف قرار می‌دهد. این متد هرگز مسدود نمی‌شود و همیشه با موفقیت انجام می‌شود (به‌جز خطاهای احتمالی سطح پایین مانند ناتوانی در تخصیص حافظه). آرگومان‌های اختیاری block و timeout نادیده گرفته می‌شوند و تنها برای سازگاری با Queue.put() ارائه شده‌اند.

این متد یک پیاده‌سازی C دارد که بازورودپذیر (reentrant) دارد. یعنی یک فراخوانی put() یا get() می‌تواند توسط یک فراخوانی put() دیگر در همان نخ، بدون ایجاد بن‌بست (deadlock) یا خراب کردن وضعیت داخلی صف، قطع شود. این موضوع آن را برای استفاده در تخریب‌کننده‌ها (destructors)، مانند متدهای __del__ یا کال‌بک‌های weakref مناسب می‌سازد.

SimpleQueue.put_nowait(item)

معادل put(item, block=False) است و برای سازگاری با Queue.put_nowait() ارائه شده است.

SimpleQueue.get(block=True, timeout=None)

یک آیتم را از صف حذف و برمی‌گرداند. اگر آرگومان‌های اختیاری به این صورت باشند که block درست باشد و timeout برابر None باشد (پیش‌فرض)، در صورت لزوم تا زمانی که یک آیتم در دسترس باشد مسدود می‌شود. اگر timeout یک عدد مثبت باشد، حداکثر به مدت timeout ثانیه مسدود می‌شود و اگر هیچ آیتمی در آن مدت در دسترس نباشد، استثنای Empty را پرتاب می‌کند. در غیر این صورت (یعنی وقتی block نادرست باشد)، اگر یک آیتم بلافاصله در دسترس باشد آن را برمی‌گرداند، وگرنه استثنای Empty را پرتاب می‌کند (در این حالت timeout نادیده گرفته می‌شود).

SimpleQueue.get_nowait()

معادل get(False).

همچنین ملاحظه نمائید

کلاس multiprocessing.Queue

یک کلاس صف برای استفاده در زمینه‌ی چندفرایندی (multi-processing) به‌جای چندنخی (multi-threading).

collections.deque یک پیاده‌سازی جایگزین از صف‌های نامحدود است که عملیات اتمی سریع append() و popleft() را بدون نیاز به قفل‌گذاری فراهم می‌کند و از اندیس‌دهی نیز پشتیبانی می‌کند.