فهرست مقاله
در یک ابزار مهندسی، توقف پذیرش کار تازه با تمام‌شدن کارهای قبلی یکسان نیست. فرض کنید چند درخواست برای بررسی آفست کالیبراسیون در صف قرار گرفته‌اند؛ بعضی هنوز منتظرند و یک درخواست را مصرف‌کننده برداشته است. هنگام بستن ابزار، باید بدانیم کدام درخواست قابل تحویل می‌ماند، کدام کنار گذاشته می‌شود و چه چیزی هنوز تأیید پایان می‌خواهد. این درس همین مرز را با صف واقعی asyncio و دادهٔ کاملاً ساختگی بررسی می‌کند؛ هیچ کالیبراسیون، حسگر یا سرویس واقعی اجرا نمی‌شود.
۱. بستن صف کار کالیبراسیون چه معنایی دارد؟
هر کار یک زوج شامل شناسه و آفست ساختگی بر حسب میلی‌ولت است. نمونهٔ دستی سه کار با شناسه‌های یک، دو و سه و آفست‌های ده، منفی سه و پنج دارد. مقدار آفست فقط همراه شناسه جابه‌جا می‌شود؛ برنامه از آن نتیجهٔ صحت اندازه‌گیری یا کیفیت کالیبراسیون نمی‌سازد. شناسه‌ها برچسب‌اند و بزرگی عددشان اولویت کار را تعیین نمی‌کند. ترتیب ورودی، ترتیب FIFO صف است.
تابع shutdown_queue_report حداکثر چهار کار، تعداد برداشت‌های قبلی، تعداد تأییدهای قبلی و حالت توقف را می‌گیرد. برداشت و تأیید هر دو پیشوندی‌اند: ابتدا چند کار اول برداشته می‌شوند و سپس چند کار اولِ همین مجموعه تأیید می‌شوند. این محدودیت، قرارداد آزمایش خودمان است؛ Queue عمومی پایتون چنین قالب رکورد یا سقف چهارکاری ندارد. عددهای شمارش باید int دقیق باشند و bool به‌جای تعداد پذیرفته نمی‌شود.
در مرجع صف‌های asyncio، صف برای استفاده در کد async و await تعریف شده و thread-safe نیست. این آزمایش هم در یک حلقهٔ رویداد و بدون دسترسی چندرشته‌ای اجرا می‌شود. مقالهٔ صف محدود حسگر و پس‌فشار پیش‌نیاز تفاوت برداشت و تأیید پایان است. موضوع تازهٔ اینجا چرخهٔ توقف صف است، نه اندازه‌گیری سرعت یا تکرار آزمون پرشدن صف.
۲. توقف آرام؛ ورودی بسته است و دادهٔ موجود باقی می‌ماند
فراخوانی shutdown(immediate=False) پذیرش را می‌بندد. کار تازه با put یا put_nowait پذیرفته نمی‌شود، حتی اگر هنوز جای خالی وجود داشته باشد. در مقابل، کارهایی که قبلاً در صف بوده‌اند با get قابل برداشت می‌مانند. پس از خالی‌شدن صف، برداشت بعدی QueueShutDown می‌سازد. این پایانِ جریان را با QueueEmpty صف باز اشتباه نکنید: اولی وضعیت توقف را می‌گوید و دومی فقط نبود قلم آماده در همان لحظه را.
در نمونهٔ سه‌کاری، مصرف‌کننده کار اول را برداشته ولی تأیید نکرده است. توقف آرام دو کار بعدی را داخل صف نگه می‌دارد. آن دو، به‌ترتیب ورودی تحویل می‌شوند؛ سپس برای هر کارِ هنوز تأییدنشده task_done فراخوانی می‌شود. join تا بسته‌شدن این تعهدهای حسابداری منتظر می‌ماند. برنامهٔ ما پردازش محتوای کار را شبیه‌سازی نمی‌کند؛ تأیید پایان فقط عملی مشخص در آزمایش است.
کد یک تولیدکنندهٔ واقعاً منتظرِ put و یک برداشت‌کنندهٔ واقعاً منتظرِ get را نیز در صف‌های جدا می‌سازد. شروع انتظار با Event اعلام می‌شود و بعد توقف انجام می‌گیرد. این فراخوان‌ها با QueueShutDown ادامه پیدا می‌کنند؛ انتظارشان موفقیت کار اعلام نمی‌شود. هیچ sleep یا حدس مبتنی بر چند میلی‌ثانیه وجود ندارد. دروازهٔ صریح، مرحلهٔ مورد بررسی را مشخص می‌کند، نه سرعت رایانه را.
۳. توقف فوری؛ تخلیهٔ صف با پایان کار یکی نیست
در حالت فوری، کارهای موجود در صف تخلیه می‌شوند و تعهدهای متناظرشان از حسابداری صف کنار می‌روند. اگر هیچ کاری قبلاً برداشته نشده باشد، همان سه کار می‌توانند بدون حتی یک تحویل کنار گذاشته شوند و join تمام شود. بنابراین عبارت «join برگشت» به‌تنهایی گزارش انجام همهٔ درخواست‌ها نیست. شناسه‌های کنارگذاشته‌شده در خروجی این درس، استنباط از ورودی و حالت آزمایش‌اند؛ خود shutdown فهرست شناسهٔ کارهای حذف‌شده را برنمی‌گرداند.
مرز مهم‌تر، کارِ از قبل برداشته‌شده است. اگر کار اول بیرون صف و بدون task_done مانده باشد، تخلیهٔ دو کار دیگر تعهد آن را نمی‌بندد. در این حالت صف خالی است ولی join همچنان منتظر می‌ماند. پس ادعای «توقف فوری همیشه join را فوراً تمام می‌کند» درست نیست. مشاهدهٔ دستیِ کارِ نگه‌داشته‌شده، همین تفاوت را از حالت بدون برداشت جدا می‌کند.
ماژول Queue در برچسب CPython3.14.5 نشان می‌دهد که کاهش حسابداریِ توقف فوری مربوط به کارهای تخلیه‌شده از خود صف است. برنامهٔ آزمایش به شمارنده‌ها یا فهرست‌های خصوصی آن ماژول دست نمی‌زند؛ این فایل فقط منبع مطالعه است. دو منبع فنی هر دو از پروژهٔ پایتون‌اند و توافقشان را اعتبارسنجی با دو تولیدکنندهٔ مستقل معرفی نمی‌کنیم.
۴. کد کامل، مرجع مستقل و خروجی آزمایش
کد را در فایل asyncio_queue_shutdown_retirement.py ذخیره کنید. API توقف در Python3.13 اضافه شده و این برنامه به همان API نیاز دارد. نسخهٔ واقعی مفسر در خروجی ثبت می‌شود؛ صفحهٔ مرجعِ خوانده‌شده نسخهٔ ۳٫۱۴٫۹ و فایل پیاده‌سازی، برچسب ۳٫۱۴٫۵ دارد. تاریخ دسترسی، نسخهٔ سند و نسخهٔ اجرای واقعی سه چیز جدا هستند و تاریخ انتشار اولیهٔ صفحات زنده از این اطلاعات نتیجه نمی‌شود.
"""Public Queue shutdown observations for finite synthetic calibration jobs."""
AUTHOR_MAIN_ENABLED = True
import asyncio
from contextlib import suppress
from dataclasses import FrozenInstanceError, asdict, dataclass, replace
import itertools
import json
import platform
import sys
@dataclass(frozen=True, slots=True)
class ShutdownReport:
immediate: bool
taken_before_shutdown: tuple
acknowledged_before_shutdown: tuple
queued_after_shutdown: int
obtained_after_shutdown: tuple
inferred_retired_without_delivery: tuple
join_finished_at_shutdown: bool
final_join_finished: bool
post_shutdown_put_rejected: bool
empty_shutdown_get_rejected: bool
def _admit(jobs, taken, acknowledged, immediate):
if type(jobs) is not tuple:
raise TypeError('jobs must be an exact tuple')
if len(jobs) > 4:
raise ValueError('at most four synthetic jobs')
seen = set()
for job in jobs:
if type(job) is not tuple or len(job) != 2:
raise TypeError('each job is an exact pair')
identifier, offset = job
if type(identifier) is not int or type(offset) is not int:
raise TypeError('job fields are exact integers, not bool')
if not 0 <= identifier <= 7 or not -20 <= offset <= 20:
raise ValueError('job ID or synthetic offset_mV outside bounds')
if identifier in seen:
raise ValueError('job IDs must be unique')
seen.add(identifier)
if type(taken) is not int or type(acknowledged) is not int:
raise TypeError('prefix counts are exact integers')
if not 0 <= taken <= len(jobs) or not 0 <= acknowledged <= taken:
raise ValueError('invalid taken or acknowledged prefix')
if type(immediate) is not bool:
raise TypeError('immediate must be an exact bool')
async def _begin_join(queue, started):
started.set()
await queue.join()
async def shutdown_queue_report(jobs, taken, acknowledged, immediate):
"""Observe only public Queue/Event/Task APIs, in one caller-owned event loop."""
_admit(jobs, taken, acknowledged, immediate)
queue = asyncio.Queue(maxsize=4)
for job in jobs:
queue.put_nowait(job)
held = tuple(queue.get_nowait() for _ in range(taken))
for _ in range(acknowledged):
queue.task_done()
queue.shutdown(immediate=immediate)
queued_after = queue.qsize()
put_rejected = False
try:
queue.put_nowait((0, 0)) # Fixed public-Queue probe after admission closes.
except asyncio.QueueShutDown:
put_rejected = True
started = asyncio.Event()
joiner = asyncio.create_task(_begin_join(queue, started))
try:
await started.wait()
# The child set readiness immediately before awaiting queue.join().
# It has either completed or suspended there before this observer resumes.
joined_at_shutdown = joiner.done()
returned = []
get_rejected = False
while True:
try:
returned.append(queue.get_nowait())
except asyncio.QueueShutDown:
get_rejected = True
break
for _ in held[acknowledged:] + tuple(returned):
queue.task_done()
await joiner
# Queue.shutdown returns no discarded IDs. This field is explicitly an
# inference from the supplied finite trace and selected mode, not API data.
inferred_retired = jobs[taken:] if immediate else ()
return ShutdownReport(immediate, held, held[:acknowledged], queued_after,
tuple(returned), inferred_retired, joined_at_shutdown,
joiner.done(), put_rejected, get_rejected)
finally:
if not joiner.done():
joiner.cancel()
with suppress(asyncio.CancelledError):
await joiner
def partition_reference(jobs, taken, acknowledged, immediate):
"""Independent index-labelled obligation sets; no asyncio or subject helpers."""
positions = frozenset(range(len(jobs)))
received = frozenset(i for i in positions if i < taken)
done = frozenset(i for i in positions if i < acknowledged)
waiting = positions - received
retired = waiting if immediate else frozenset()
live_obligations = positions - done - retired
returned = waiting - retired
choose = lambda indices: tuple(jobs[i] for i in sorted(indices))
return ShutdownReport(immediate, choose(received), choose(done), len(returned),
choose(returned), choose(retired), not live_obligations,
True, True, True)
async def blocked_operation_probe(immediate):
full, empty = asyncio.Queue(maxsize=1), asyncio.Queue(maxsize=1)
full.put_nowait((1, 10))
put_started, get_started = asyncio.Event(), asyncio.Event()
async def putter():
put_started.set()
try:
await full.put((2, 0))
except asyncio.QueueShutDown:
return 'QueueShutDown'
return 'accepted'
async def getter():
get_started.set()
try:
await empty.get()
except asyncio.QueueShutDown:
return 'QueueShutDown'
return 'received'
tasks = [asyncio.create_task(putter()), asyncio.create_task(getter())]
try:
await put_started.wait()
await get_started.wait()
pending = tuple(not task.done() for task in tasks)
full.shutdown(immediate=immediate)
empty.shutdown(immediate=immediate)
outcomes = tuple([await task for task in tasks])
retained = ()
if not immediate:
retained = (full.get_nowait(),)
full.task_done()
await full.join()
await empty.join()
return pending, outcomes, retained
finally:
for task in tasks:
if not task.done():
task.cancel()
with suppress(asyncio.CancelledError):
await task
async def escalation_probe():
queue = asyncio.Queue()
for job in ((1, 10), (2, -3), (3, 5)):
queue.put_nowait(job)
held = queue.get_nowait()
queue.shutdown()
before = queue.qsize()
queue.shutdown(immediate=True)
started = asyncio.Event()
joiner = asyncio.create_task(_begin_join(queue, started))
try:
await started.wait()
waiting_for_held = not joiner.done()
queue.task_done()
await joiner
queue.shutdown(immediate=True)
queue.shutdown()
return before == 2 and queue.empty() and held == (1, 10) and waiting_for_held and joiner.done()
finally:
if not joiner.done():
joiner.cancel()
with suppress(asyncio.CancelledError):
await joiner
async def admission_probe(profiles):
originals = asyncio.Queue, asyncio.Event, asyncio.create_task
entered = []
def forbidden(*args, **kwargs):
entered.append('factory')
raise AssertionError('invalid declaration reached an asyncio factory')
rejected = 0
asyncio.Queue = asyncio.Event = asyncio.create_task = forbidden
try:
for args in profiles:
try:
await shutdown_queue_report(*args)
except (TypeError, ValueError):
rejected += 1
finally:
asyncio.Queue, asyncio.Event, asyncio.create_task = originals
return rejected == len(profiles) and not entered
async def _run_checks():
jobs = ((1, 10), (2, -3), (3, 5))
graceful = await shutdown_queue_report(jobs, 1, 0, False)
held_immediate = await shutdown_queue_report(jobs, 1, 0, True)
all_retired = await shutdown_queue_report(jobs, 0, 0, True)
empty = [await shutdown_queue_report((), 0, 0, mode) for mode in (False, True)]
family_count = 0
family_ok = True
for n in range(5):
for offsets in itertools.product((-1, 2), repeat=n):
fixture = tuple(enumerate(offsets))
for taken in range(n + 1):
for ack in range(taken + 1):
for mode in (False, True):
actual = await shutdown_queue_report(fixture, taken, ack, mode)
family_ok = family_ok and actual == partition_reference(fixture, taken, ack, mode)
family_count += 1
transform = lambda records: tuple((7 - identifier, -offset) for identifier, offset in records)
renamed = await shutdown_queue_report(transform(jobs), 1, 0, False)
renamed_expected = replace(graceful, taken_before_shutdown=transform(graceful.taken_before_shutdown),
acknowledged_before_shutdown=transform(graceful.acknowledged_before_shutdown),
obtained_after_shutdown=transform(graceful.obtained_after_shutdown),
inferred_retired_without_delivery=transform(graceful.inferred_retired_without_delivery))
blocked = [await blocked_operation_probe(mode) for mode in (False, True)]
boundary = asyncio.Queue()
empty_before = False
try:
boundary.get_nowait()
except asyncio.QueueEmpty:
empty_before = True
boundary.shutdown()
shutdown_after = False
try:
boundary.get_nowait()
except asyncio.QueueShutDown:
shutdown_after = True
excess = asyncio.Queue()
excess.put_nowait((0, 0))
excess.shutdown(immediate=True)
over_ack = False
try:
excess.task_done()
except ValueError:
over_ack = True
hook_calls = []
class IterationTrap:
def __iter__(self):
hook_calls.append('iter')
raise AssertionError('arbitrary iterator consumed')
class IntSubclass(int):
def __int__(self):
hook_calls.append('int')
raise AssertionError('coercion invoked')
class TupleSubclass(tuple):
pass
valid = (((0, 1),), 0, 0, False)
invalid = [
([], 0, 0, False), (IterationTrap(), 0, 0, False),
(TupleSubclass(), 0, 0, False), (([0, 1],), 0, 0, False),
(((0,),), 0, 0, False), (((0, 1, 2),), 0, 0, False),
(((True, 1),), 0, 0, False), (((0, True),), 0, 0, False),
(((IntSubclass(0), 1),), 0, 0, False), (((0, 1.0),), 0, 0, False),
(((0, 1), (1, None)), 0, 0, False),
(valid[0], True, 0, False), (valid[0], 0, True, False),
(valid[0], 0.0, 0, False), (valid[0], 0, 0.0, False),
(valid[0], 0, 0, 0), (valid[0], 0, 0, 1),
(valid[0], 0, 0, None), (valid[0], 0, 0, 'false'),
]
bounds = [
(tuple((i, 0) for i in range(5)), 0, 0, False),
(((-1, 0),), 0, 0, False), (((8, 0),), 0, 0, False),
(((0, -21),), 0, 0, False), (((0, 21),), 0, 0, False),
(((0, 0), (0, 1)), 0, 0, False),
(valid[0], -1, 0, False), (valid[0], 2, 0, False),
(valid[0], 0, -1, False), (valid[0], 0, 1, False),
]
invalid_ok = await admission_probe(invalid)
bounds_ok = await admission_probe(bounds)
maximum = await shutdown_queue_report(((0, -20), (7, 20), (3, -20), (4, 20)), 4, 4, True)
original_factory = asyncio.create_task
created = []
def track(coroutine, *args, **kwargs):
task = original_factory(coroutine, *args, **kwargs)
created.append(task)
return task
asyncio.create_task = track
try:
first = await shutdown_queue_report(jobs, 1, 0, True)
second = await shutdown_queue_report(jobs, 1, 0, True)
finally:
asyncio.create_task = original_factory
frozen = False
try:
first.queued_after_shutdown = 99
except FrozenInstanceError:
frozen = True
checks = {
'manual_graceful_wind_down': graceful == ShutdownReport(False, ((1, 10),), (), 2, ((2, -3), (3, 5)), (), False, True, True, True),
'manual_immediate_retirement_without_processing': all_retired.obtained_after_shutdown == () and all_retired.inferred_retired_without_delivery == jobs and all_retired.join_finished_at_shutdown,
'held_unacknowledged_job_survives_immediate_shutdown': not held_immediate.join_finished_at_shutdown and held_immediate.queued_after_shutdown == 0 and held_immediate.inferred_retired_without_delivery == jobs[1:] and held_immediate.final_join_finished,
'empty_queue_shutdown_boundaries': all(x.queued_after_shutdown == 0 and x.join_finished_at_shutdown and x.final_join_finished for x in empty),
'finite_partition_reference_family_ONE_group': family_count == 702 and family_ok,
'FIFO_payload_and_ID_renaming': renamed == renamed_expected,
'blocked_put_callers_raise_after_both_shutdown_modes': all(x[0][0] and x[1][0] == 'QueueShutDown' for x in blocked),
'blocked_get_callers_raise_after_both_shutdown_modes': all(x[0][1] and x[1][1] == 'QueueShutDown' for x in blocked),
'QueueEmpty_before_shutdown_versus_QueueShutDown_after': empty_before and shutdown_after,
'excess_acknowledgement_rejected_after_retirement': over_ack,
'graceful_then_immediate_escalation_tagged_observation': await escalation_probe(),
'strict_whole_declaration_before_asyncio_factories': invalid_ok,
'finite_caps_and_no_arbitrary_hooks': bounds_ok and not hook_calls and maximum.join_finished_at_shutdown,
'fresh_immutable_reports_and_owned_task_cleanup': first == second and first is not second and frozen and all(task.done() for task in created) and jobs == ((1, 10), (2, -3), (3, 5)),
}
assert len(checks) == 14 and all(type(value) is bool and value for value in checks.values()), checks
return dict(runtime=dict(python=platform.python_version(), implementation=platform.python_implementation(), library='stdlib asyncio', public_shutdown_added='Python3.13'),
contract=dict(job_fields=['job_id', 'synthetic_offset_mV'], max_jobs=4, source_scope='one supplied trace in one event loop; no private Queue state', completion='queue bookkeeping only, not successful or persistent calibration'),
manual_jobs=jobs, graceful=asdict(graceful), immediate_with_held_job=asdict(held_immediate), immediate_without_held_jobs=asdict(all_retired),
finite_family_ONE_group=dict(profiles=family_count, value_alphabet=[-1, 2], lengths=[0, 1, 2, 3, 4], independent_reference='index-labelled obligation-set partition'),
validation=dict(type_profiles=len(invalid), bound_profiles=len(bounds), arbitrary_hooks_entered=len(hook_calls), forbidden_asyncio_factory_entries=0),
checks=checks, groups_total=len(checks), all_passed=all(checks.values()),
limitations=['Retired IDs are inferred from supplied trace/mode; shutdown returns no discarded-job list.', 'No real calibration, durable work, retry, network, thread/process, timing, RSS or service/CI evidence.', 'Selected Event handshakes and owned-task cleanup are not a proof of every cancellation or scheduler interleaving.', 'Repeated shutdown escalation is a selected tagged-runtime observation, not an undocumented cross-implementation promise.'])
def run_example():
return asyncio.run(_run_checks())
if __name__ == '__main__':
assert AUTHOR_MAIN_ENABLED is True, 'Await ROOT registered author-run authorization'
sys.stdout.reconfigure(encoding='utf-8', newline='\n')
print(json.dumps(run_example(), ensure_ascii=True, indent=2))
ناظر join ابتدا آمادگی را اعلام می‌کند و بلافاصله await روی join انجام می‌دهد. مشاهده‌گر بعد از دریافت همین علامت، تمام‌شدن Task را می‌سنجد. اگر تعهدی باقی باشد، انتظار هنوز باز است؛ در غیر این صورت همان وظیفه تمام شده است. پاک‌سازی finally نیز فقط وظیفه‌ای را لغو و جمع می‌کند که همین فراخوان ساخته است. این ترتیب کنترل‌شده، ادعای پوشش همهٔ لغوهای بیرونی یا زمان‌بندهای ممکن نیست.
مرجع مستقل اصلاً asyncio را صدا نمی‌زند. جایگاه‌های ورودی را به کارهای تأییدشده، برداشته‌شدهٔ تأییدنشده و منتظر تقسیم می‌کند. در توقف فوری فقط دستهٔ منتظر کنار می‌رود؛ خالی‌بودن مجموعهٔ تعهدهای زنده، جواب مورد انتظار برای join است. این مسیر، شمارندهٔ خصوصی Queue را بازسازی یا از تابع اصلی پاسخ‌گیری نمی‌کند. خانوادهٔ تعریف‌شده ۷۰۲ پروفایل دارد و تمام آن یک گروه بررسی است.
{
"runtime": {
"python": "3.14.5",
"implementation": "CPython",
"library": "stdlib asyncio",
"public_shutdown_added": "Python3.13"
},
"contract": {
"job_fields": [
"job_id",
"synthetic_offset_mV"
],
"max_jobs": 4,
"source_scope": "one supplied trace in one event loop; no private Queue state",
"completion": "queue bookkeeping only, not successful or persistent calibration"
},
"manual_jobs": [
[
1,
10
],
[
2,
-3
],
[
3,
5
]
],
"graceful": {
"immediate": false,
"taken_before_shutdown": [
[
1,
10
]
],
"acknowledged_before_shutdown": [],
"queued_after_shutdown": 2,
"obtained_after_shutdown": [
[
2,
-3
],
[
3,
5
]
],
"inferred_retired_without_delivery": [],
"join_finished_at_shutdown": false,
"final_join_finished": true,
"post_shutdown_put_rejected": true,
"empty_shutdown_get_rejected": true
},
"immediate_with_held_job": {
"immediate": true,
"taken_before_shutdown": [
[
1,
10
]
],
"acknowledged_before_shutdown": [],
"queued_after_shutdown": 0,
"obtained_after_shutdown": [],
"inferred_retired_without_delivery": [
[
2,
-3
],
[
3,
5
]
],
"join_finished_at_shutdown": false,
"final_join_finished": true,
"post_shutdown_put_rejected": true,
"empty_shutdown_get_rejected": true
},
"immediate_without_held_jobs": {
"immediate": true,
"taken_before_shutdown": [],
"acknowledged_before_shutdown": [],
"queued_after_shutdown": 0,
"obtained_after_shutdown": [],
"inferred_retired_without_delivery": [
[
1,
10
],
[
2,
-3
],
[
3,
5
]
],
"join_finished_at_shutdown": true,
"final_join_finished": true,
"post_shutdown_put_rejected": true,
"empty_shutdown_get_rejected": true
},
"finite_family_ONE_group": {
"profiles": 702,
"value_alphabet": [
-1,
2
],
"lengths": [
0,
1,
2,
3,
4
],
"independent_reference": "index-labelled obligation-set partition"
},
"validation": {
"type_profiles": 19,
"bound_profiles": 10,
"arbitrary_hooks_entered": 0,
"forbidden_asyncio_factory_entries": 0
},
"checks": {
"manual_graceful_wind_down": true,
"manual_immediate_retirement_without_processing": true,
"held_unacknowledged_job_survives_immediate_shutdown": true,
"empty_queue_shutdown_boundaries": true,
"finite_partition_reference_family_ONE_group": true,
"FIFO_payload_and_ID_renaming": true,
"blocked_put_callers_raise_after_both_shutdown_modes": true,
"blocked_get_callers_raise_after_both_shutdown_modes": true,
"QueueEmpty_before_shutdown_versus_QueueShutDown_after": true,
"excess_acknowledgement_rejected_after_retirement": true,
"graceful_then_immediate_escalation_tagged_observation": true,
"strict_whole_declaration_before_asyncio_factories": true,
"finite_caps_and_no_arbitrary_hooks": true,
"fresh_immutable_reports_and_owned_task_cleanup": true
},
"groups_total": 14,
"all_passed": true,
"limitations": [
"Retired IDs are inferred from supplied trace/mode; shutdown returns no discarded-job list.",
"No real calibration, durable work, retry, network, thread/process, timing, RSS or service/CI evidence.",
"Selected Event handshakes and owned-task cleanup are not a proof of every cancellation or scheduler interleaving.",
"Repeated shutdown escalation is a selected tagged-runtime observation, not an undocumented cross-implementation promise."
]
}چهارده گروه نام‌دار، همین خانواده و شاهدهای دستی، بیدارشدن فراخوان‌های منتظر، مرز خطاها، ورودی‌های نامعتبر و پاک‌سازی را در بر می‌گیرند. شناسه‌ها و آفست‌های ورودی هم تغییر نمی‌کنند. آزمونِ تبدیل توقف آرام به فوری و فراخوانی تکراری، مشاهده‌ای برای نسخهٔ انتخاب‌شده است؛ آن را تضمینی فراتر از قرارداد مستندات و پیاده‌سازی خوانده‌شده نمی‌دانیم.
۵. مرزهای آزمون و چهار پرسش پاسخ‌دار
پیش از ساخت صف، Event یا Task، تمام اعلان ورودی بررسی می‌شود؛ حتی خرابیِ رکورد آخر نباید بعد از آغاز کار کشف شود. شناسهٔ تکراری، آفست خارج از بازهٔ منفی بیست تا بیست میلی‌ولت، برداشت بیش از تعداد کارها و تأیید بیش از برداشت رد می‌شوند. این قواعد برای یک مثال قابل بازبینی‌اند، نه اعتبارسنجی عمومی هر دادهٔ مهندسی یا یک مرز امنیتی برای اشیای دلخواه پایتون.
پرسش اول: آیا توقف فوری همیشه join را تمام می‌کند؟ پاسخ: خیر؛ در مثال برداشتِ یک کار و صفر تأیید، همان کارِ نگه‌داشته‌شده هنوز تعهد دارد. پرسش دوم: آیا توقف آرام یک put نهایی را می‌پذیرد؟ پاسخ: خیر؛ پذیرش فوراً بسته می‌شود، هرچند داده‌های قبلی هنوز قابل برداشت‌اند.
پرسش سوم: برای کارهای کنارگذاشته‌شده هم task_done بزنیم؟ پاسخ: خیر؛ توقف فوری تعهد کارهای تخلیه‌شده را کنار گذاشته است. تأیید اضافه می‌تواند ValueError بسازد؛ فقط کارِ قبلاً برداشته‌شده تأیید می‌خواهد. پرسش چهارم: آیا پایان join، کالیبراسیون موفق و ماندگار را ثابت می‌کند؟ پاسخ: خیر؛ حالت بدون برداشت شاهد روشنی است که پایان حسابداری حتی بدون پردازش ممکن است.
اگر کارها از خواننده‌های مستقل یا منابع واقعی بیایند، عمر وظیفه‌ها و انتشار خطا مسئلهٔ دیگری است؛ آزمون TaskGroup و پاک‌سازی خواننده‌ها آن مرز را جدا بررسی می‌کند. اینجا ذخیرهٔ پایدار، بازآزمایی، قطع شبکه، ایمنی چندرشته‌ای و کارایی اندازه‌گیری نمی‌شوند. منابع در یکم آبان ۱۴۰۵ خوانده شده‌اند و درس، آزمایش تألیفیِ پایه است؛ خبر پژوهشی یا شاهد عملکرد یک محصول نیست.
منابع و مطالعهٔ بیشتر
بررسی منابع: ۱ آبان ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.
