صف محدود asyncio برای دادهٔ حسگر؛ پس‌فشار و آزمون بدون تأخیر ساختگی

در یک زنجیرهٔ دادهٔ مهندسی، دریافت‌کننده و پردازشگر همیشه با آهنگ یکسان جلو نمی‌روند. اگر فقط وظیفه‌های بیشتری بسازیم، محل انتظار داده‌ها ممکن است نامشخص شود. این درس یک تولیدکننده، یک مصرف‌کننده و صفی با ظرفیت دو قرائت می‌سازد تا انتظار، ترتیب تحویل و پایان کار را جدا بررسی کنیم. داده‌ها ساختگی‌اند؛ هدف، طراحی آزمون روشن برای جریان برنامه است و هیچ افزایش سرعت یا اندازه‌گیری شبکه‌ای گزارش نمی‌شود.

۱. قرارداد جریان: سه قرائت و ظرفیت دو قلم

هر قرائت شامل شمارهٔ ترتیب، زمان ساختگی بر حسب میکروثانیه و ولتاژ بر حسب میلی‌ولت است. سه رکوردِ انتخابی، شماره‌های یک تا سه، زمان‌های صفر، هزار و دو هزار و ولتاژهای ۱۰۱۰، ۱۰۲۰ و ۱۰۳۰ دارند. فاصلهٔ درج‌شده در فیلد زمان، مدت اجرای برنامه نیست؛ کد برای شبیه‌سازی آن منتظر نمی‌ماند. همچنین ولتاژها از حسگر یا فایل اندازه‌گیری واقعی نیامده‌اند.

در مرجع Queue پایتون، ظرفیت مثبت و انتظار put هنگام پر بودن صف تعریف شده است. ما maxsize=2 را انتخاب می‌کنیم. این عدد، شمار قلم‌های داخل همین صف را محدود می‌کند؛ اندازهٔ بایتی هر رکورد، دادهٔ نگهداری‌شده در مصرف‌کننده و حافظهٔ دیگر بخش‌های برنامه را محدود نمی‌کند. صف کوچک، به‌تنهایی تضمین حافظهٔ محدود کل سامانه نیست.

برنامه فقط در یک حلقهٔ رویداد و با یک مصرف‌کننده اجرا می‌شود. این ساختار را با دسترسی مستقیم چند رشتهٔ سیستم‌عامل مخلوط نکنید؛ Queue این درس برای آن طراحی نشده است. اگر هنوز به جداسازی تولید داده و مصرف مرحله‌ای آشنا نیستید، مقالهٔ مولدها و بسته‌بندی جریان دادهٔ حسگر نقطهٔ شروع مناسبی است. آن مثال همگام است؛ اینجا موضوع اصلی هماهنگی وظیفه‌های ناهمگام است.

۲. پس‌فشار را با دروازهٔ صریح مشاهده کنیم

تولیدکننده ابتدا دو رکورد را وارد صف می‌کند. سپس اعلام می‌کند که به درج سوم رسیده و برای آن put را فراخوانی می‌کند. مصرف‌کننده هنوز پشت دروازهٔ اجازهٔ برداشت است؛ بنابراین جا باز نشده و درج سوم نمی‌تواند تمام شود. در همین مرحله، آزمون هم پر بودن صف و هم ناتمام بودن تولیدکننده را بررسی می‌کند. این انتظارِ قابل مشاهده، نمونهٔ محدودِ پس‌فشار در همین مسیر است.

دروازه‌ها با Event پایتون ساخته شده‌اند. پرچم آن با set فعال می‌شود و wait امکان انتظار برای همان وضعیت را می‌دهد. به جای اینکه حدس بزنیم بعد از چند میلی‌ثانیه تولیدکننده کجا رسیده، مرحلهٔ لازم را صریح اعلام می‌کنیم. در این کد، اعلام آغاز درج سوم خودش توقف ایجاد نمی‌کند؛ تولیدکننده در put صفِ پر معلق می‌شود و کنترل به وظیفهٔ بررسی می‌رسد.

بعد، فقط اجازهٔ برداشت نخست را می‌دهیم. یک جایگاه آزاد می‌شود و تولیدکننده می‌تواند رکورد سوم را قرار دهد. تا اعلام پایان این درج صبر می‌کنیم؛ سپس تعداد قلم‌های داخل صف دوباره دو است. برداشت‌های باقی‌مانده هنوز با دروازهٔ جدا نگه داشته شده‌اند. این ترتیب عمدی، اثر آزاد شدن جا را از سرعت مصرف بعدی جدا می‌کند؛ آزمون هیچ آستانهٔ زمانی، sleep یا وابستگی به تند بودن رایانه ندارد.

۳. کد کامل، گزارش صف و پانزده بررسی اجراشده

کد را در فایل async_sensor_queue.py ذخیره و با دستور python async_sensor_queue.py اجرا کنید. اجرای ثبت‌شده با پایتون ۳٫۱۴٫۵ و کتابخانهٔ استاندارد انجام شده است. ورودی‌ها ثابت‌اند و انتظار FIFO از همان سه رکورد اولیه ساخته می‌شود. ثبت وضعیت بعد از هر درج یا برداشت موفق، شمار داخل صف را نگه می‌دارد؛ دادهٔ خصوصیِ پیاده‌سازی صف خوانده نمی‌شود.

"""Controlled synthetic queue experiment; no clock-based assertions or I/O benchmark."""
import asyncio
import json
import platform

READINGS = ((1, 0, 1010), (2, 1000, 1020), (3, 2000, 1030))


async def normal_flow():
    queue = asyncio.Queue(maxsize=2)
    third_started, third_done = asyncio.Event(), asyncio.Event()
    allow_first, first_removed = asyncio.Event(), asyncio.Event()
    allow_rest, last_removed = asyncio.Event(), asyncio.Event()
    allow_last_ack = asyncio.Event()
    join_started, join_done = asyncio.Event(), asyncio.Event()
    seen, acknowledgments, trace = [], [], []
    checks = {}

    def snapshot(action):
        trace.append({'action': action, 'queued_items': queue.qsize()})

    async def produce():
        for reading in READINGS[:2]:
            await queue.put(reading)
            snapshot(f'put_{reading[0]}')
        third_started.set()
        await queue.put(READINGS[2])
        snapshot('put_3')
        third_done.set()

    async def consume():
        await allow_first.wait()
        for index in range(len(READINGS)):
            if index == 1:
                await allow_rest.wait()
            reading = await queue.get()
            seen.append(reading)
            snapshot(f'get_{reading[0]}')
            if index == 0:
                first_removed.set()
            if index == 2:
                last_removed.set()
                await allow_last_ack.wait()
            queue.task_done()
            acknowledgments.append(reading[0])

    async def wait_for_completion():
        join_started.set()
        await queue.join()
        join_done.set()

    async with asyncio.TaskGroup() as group:
        producer = group.create_task(produce())
        group.create_task(consume())
        await third_started.wait()
        checks['queue_full_before_consumption'] = queue.full() and queue.qsize() == 2
        checks['third_put_pending_before_get'] = not third_done.is_set() and not producer.done()
        allow_first.set()
        await first_removed.wait()
        await third_done.wait()
        checks['third_put_resumes_after_space_released'] = queue.qsize() == 2
        allow_rest.set()
        await last_removed.wait()
        checks['empty_queue_can_have_unfinished_work'] = queue.empty() and acknowledgments == [1, 2]
        waiter = group.create_task(wait_for_completion())
        await join_started.wait()
        checks['join_pending_until_last_ack'] = not join_done.is_set() and not waiter.done()
        allow_last_ack.set()
        await join_done.wait()

    checks['observed_qsize_never_exceeds_capacity'] = all(0 <= row['queued_items'] <= 2 for row in trace)
    checks['fifo_preserves_complete_readings'] = tuple(seen) == READINGS
    checks['each_reading_delivered_once'] = [reading[0] for reading in seen] == [1, 2, 3]
    checks['all_acknowledgments_received'] = acknowledgments == [1, 2, 3]
    checks['join_completed_after_acknowledgments'] = join_done.is_set() and queue.empty()
    try:
        queue.task_done()
    except ValueError:
        checks['excess_task_done_rejected'] = True
    else:
        checks['excess_task_done_rejected'] = False
    return checks, trace, seen


class SensorFailure(RuntimeError):
    pass


async def failure_flow():
    queue = asyncio.Queue(maxsize=1)
    queue.put_nowait(READINGS[0])
    sibling_ready, sibling_cleaned = asyncio.Event(), asyncio.Event()
    never_released = asyncio.Event()
    caught = []

    async def idle_sibling():
        try:
            sibling_ready.set()
            await never_released.wait()
        finally:
            sibling_cleaned.set()

    async def failing_consumer():
        await sibling_ready.wait()
        await queue.get()
        try:
            raise SensorFailure('synthetic_consumer_failure')
        finally:
            # Completion of queue bookkeeping does not mean processing succeeded.
            queue.task_done()

    try:
        async with asyncio.TaskGroup() as group:
            sibling = group.create_task(idle_sibling())
            group.create_task(failing_consumer())
    except* SensorFailure as errors:
        caught.extend(str(error) for error in errors.exceptions)
    await queue.join()
    checks = {
        'consumer_failure_propagated_and_observed': caught == ['synthetic_consumer_failure'],
        'taskgroup_canceled_sibling': sibling.cancelled(),
        'canceled_sibling_finally_ran': sibling_cleaned.is_set(),
        'queue_bookkeeping_completion_is_not_success': queue.empty() and bool(caught),
    }
    return checks, caught


async def run_async():
    checks, trace, readings = await normal_flow()
    failure_checks, errors = await failure_flow()
    checks.update(failure_checks)
    assert all(checks.values()), checks
    return {'python': platform.python_version(), 'model': 'controlled_single_event_loop_bounded_queue',
            'maxsize_items': 2, 'reading_fields': ['sequence', 'synthetic_time_us', 'voltage_mv'],
            'synthetic_readings': readings, 'queue_trace': trace,
            'maximum_observed_queued_items': max(row['queued_items'] for row in trace),
            'timing_assertions_used': False, 'network_or_cpu_benchmark': False,
            'failure_probe_errors': errors, 'checks': checks}


def run_example():
    return asyncio.run(run_async())


if __name__ == '__main__':
    print(json.dumps(run_example(), indent=2))

مسیر عادی، پر شدن صف، انتظار درج سوم، ادامه پس از برداشت، تحویل رکوردها، تأییدهای پایان کار و رفتار join را کنترل می‌کند. مسیر جداگانهٔ خطا نیز انتقال استثنا، لغو وظیفهٔ هم‌گروه و اجرای پاک‌سازی را می‌سنجد. در مجموع پانزده بررسی نام‌دار موفق شده‌اند. این شمار، پوشش کامل همهٔ زمان‌بندی‌ها یا خرابی‌های ممکن را ادعا نمی‌کند؛ آنچه اجرا شده، همین سناریوی کنترل‌شده با حلقهٔ پیش‌فرض است.

رشتهٔ ثبت‌شدهٔ اندازه‌ها برابر 1, 2, 1, 2, 1, 0 است و بیشترین مقدار مشاهده‌شده دو است. سه رکورد با تمام فیلدهایشان در ترتیب اولیه تحویل شده‌اند. شمارهٔ ترتیب برای تشخیص تکرار در همین آزمون هم کنترل می‌شود. این تحویل یک‌باره در حافظه، تضمین «دقیقاً یک‌بار» پس از قطع برق، خرابی فرایند یا ارسال مجدد شبکه‌ای نیست؛ هیچ‌یک از آن رویدادها در مدل وجود ندارد.

۴. صف خالی، task_done و پایان پردازش یکسان نیستند

برای روشن شدن تفاوت، آخرین رکورد را از صف برمی‌داریم ولی تأیید پایانش را نگه می‌داریم. اکنون صف خالی است، اما فقط دو تأیید ثبت شده‌اند. join آغاز می‌شود و هنوز تمام نمی‌شود؛ سپس اجازهٔ تأیید آخر را می‌دهیم و join پایان می‌یابد. در قرارداد task_done و join، شمار کارهای تأییدنشده مهم است. برداشت از صف، خودکار به معنی پایان پردازش نیست.

{
  "python": "3.14.5",
  "model": "controlled_single_event_loop_bounded_queue",
  "maxsize_items": 2,
  "reading_fields": [
    "sequence",
    "synthetic_time_us",
    "voltage_mv"
  ],
  "synthetic_readings": [
    [
      1,
      0,
      1010
    ],
    [
      2,
      1000,
      1020
    ],
    [
      3,
      2000,
      1030
    ]
  ],
  "queue_trace": [
    {
      "action": "put_1",
      "queued_items": 1
    },
    {
      "action": "put_2",
      "queued_items": 2
    },
    {
      "action": "get_1",
      "queued_items": 1
    },
    {
      "action": "put_3",
      "queued_items": 2
    },
    {
      "action": "get_2",
      "queued_items": 1
    },
    {
      "action": "get_3",
      "queued_items": 0
    }
  ],
  "maximum_observed_queued_items": 2,
  "timing_assertions_used": false,
  "network_or_cpu_benchmark": false,
  "failure_probe_errors": [
    "synthetic_consumer_failure"
  ],
  "checks": {
    "queue_full_before_consumption": true,
    "third_put_pending_before_get": true,
    "third_put_resumes_after_space_released": true,
    "empty_queue_can_have_unfinished_work": true,
    "join_pending_until_last_ack": true,
    "observed_qsize_never_exceeds_capacity": true,
    "fifo_preserves_complete_readings": true,
    "each_reading_delivered_once": true,
    "all_acknowledgments_received": true,
    "join_completed_after_acknowledgments": true,
    "excess_task_done_rejected": true,
    "consumer_failure_propagated_and_observed": true,
    "taskgroup_canceled_sibling": true,
    "canceled_sibling_finally_ran": true,
    "queue_bookkeeping_completion_is_not_success": true
  }
}

در سناریوی خطا، مصرف‌کننده یک رکوردِ جداگانه را می‌گیرد و عمداً SensorFailure ایجاد می‌کند. بخش finally حسابداری همان قلم را با task_done می‌بندد، اما خطا همچنان به TaskGroup می‌رسد. پس join می‌تواند تمام شود و در عین حال پردازش موفق نبوده باشد. اگر پروژه به ذخیرهٔ پایدار نیاز دارد، موفقیت آن باید جدا ثبت شود؛ مقالهٔ آزمون تراکنش دادهٔ حسگر در SQLite به مرز دیگری از این مسئله مربوط است.

مستندات TaskGroup رفتار خطای وظیفه و لغو سایر وظیفه‌های گروه را شرح می‌دهد. در آزمون ما، هم‌گروه از قبل آمادگی‌اش را اعلام کرده است؛ بنابراین می‌توان لغو و اجرای finally آن را واقعاً مشاهده کرد. فقط خطای عمدیِ مشخص با except* دریافت می‌شود. کد سیاست تلاش مجدد، بازگرداندن رکورد خراب یا خاموش‌سازی عمومی یک سرویس تولیدی را پیاده نمی‌کند.

۵. تمرین‌های پاسخ‌دار و حدود استفادهٔ مهندسی

تمرین اول: ظرفیت دو یعنی حداکثر دو رکورد در تمام برنامه داریم؟ پاسخ: خیر؛ فقط صفِ مورد بررسی این محدودیت را دارد. رکورد برداشته‌شده می‌تواند هنوز در مصرف‌کننده باشد. تمرین دوم: آیا خالی بودن صف برای اعلام پایان کافی است؟ پاسخ: خیر؛ آزمون، صف خالی با آخرین تأییدِ نگه‌داشته‌شده را می‌سازد و join همچنان منتظر می‌ماند.

تمرین سوم: چرا برای دیدن انتظار تولیدکننده sleep نگذاشتیم؟ پاسخ: چون زمان سپری‌شده موقعیت دقیق وظیفه را تضمین نمی‌کند؛ پر بودن صف و دروازه‌های مشخص، وضعیت مورد نیاز را قابل بررسی می‌کنند. تمرین چهارم: آیا پایان join در مسیر خطا یعنی ذخیره موفق بوده؟ پاسخ: خیر؛ در این مثال فقط حسابداری صف بسته شده و خطای پردازش جدا مشاهده شده است.

برای افزودن شبکه یا درگاه سریال، باید رفتار قطع ارتباط، محدودیت حجم رکورد، لغو بیرونی و سیاست بازآزمایی را جدا طراحی کنید. همچنین محاسبهٔ سنگین پردازنده از این مثال نتیجهٔ سرعت نمی‌گیرد. منابع در ۲۴ مهر ۱۴۰۵ بازخوانی شده‌اند؛ صفحات مستندات نسخهٔ ۳٫۱۴٫۸ و اجرای واقعی ۳٫۱۴٫۵ جدا ثبت شده‌اند. این مطلب آموزش پایه و آزمایش تألیفی است، نه خبر پژوهشی تازه یا آزمون کارایی یک محصول.

منابع و مطالعهٔ بیشتر

• صف asyncio؛ ظرفیت، FIFO، task_done و join در Python 3.14
• TaskGroup، لغو و زمان‌بندی مشارکتی وظیفه‌ها در Python 3.14
• Event و همگام‌سازی وظیفه‌ها در Python 3.14

بررسی منابع: ۲۴ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.