فهرست مقاله
در آماده‌سازی دادهٔ مهندسی، چند خواننده می‌توانند مستقل کار کنند، اما نتیجه باید به حسگر درست متصل بماند و خطای یک خواننده کارهای معلق را رها نکند. فقط ساختن چند task این قرارداد را کامل نمی‌کند. در این درس، جمع‌آور ولتاژ را با TaskGroup، سقف reader فعال و ورودی‌های ساختگی می‌سازیم. داده‌ها می‌توانند ورودیِ مرحلهٔ بعدی تحلیل یا هوش مصنوعی باشند، ولی حسگر، شبکه یا مدل واقعی در آزمایش نداریم.
۱. ترتیب درخواست، ترتیب پایان و عمر گروه
سه درخواست S1، S2 و S3 به‌ترتیب مقدارهای ۱٫۲۵، ۲٫۵ و منفی ۰٫۵ ولت دارند. می‌خواهیم فهرست خروجی با ترتیب درخواست برگردد، حتی اگر خوانندهٔ دوم و سوم زودتر تمام شوند. برای همین هر worker خانهٔ اندیس خودش را پر می‌کند. ترتیب ورود به reader یا پایان آن را تضمین عمومیِ زمان‌بند معرفی نمی‌کنیم؛ فقط ارتباط شناسه و نتیجه باید در قرارداد ما حفظ شود.
مستندات TaskGroup پایتون آن را محدوده‌ای برای انتظار پایان وظیفه‌های مرتبط تعریف می‌کند. این محدوده، عمر کارهای فرزند را به جمع‌آور پیوند می‌دهد. در مثال موفق، خواننده‌ها با دروازه‌های مشخص آزاد می‌شوند؛ پس می‌توان ترتیب پایان متفاوت را ساخت و سپس خروجی نهایی را با جواب دستی مقایسه کرد. زمان اجرا یا سرعت حسگر معیار موفقیت نیست.
reader باید یک مقدار عددی با قرارداد ولت برگرداند. فهرست شناسه‌ها محدود، بدون تکرار و حساس به بزرگی حروف است؛ هر شناسه یک تا شانزده نویسهٔ ASCII، با حرف آغازین و سپس حرف، رقم یا زیرخط دارد. تبدیل نام، حذف تکرار یا حدس واحد انجام نمی‌شود. اگر سامانهٔ واقعی رکورد دارای واحد برمی‌گرداند، اعتبارسنجی همان رکورد باید پیش از اتکا به عدد طراحی شود.
۲. سقف reader فعال را از تعداد task جدا کنیم
با limit برابر دو، حداکثر دو فراخوانی reader داخل محدودهٔ Semaphore هستند. worker منتظرِ گرفتن جایگاه هنوز reader را آغاز نکرده است. مرجع Semaphore و Event قرارداد acquire و release را توضیح می‌دهد؛ از async with استفاده می‌کنیم تا خروج از محدوده جایگاه را آزاد کند. شمار فعالِ mock فقط پس از ورود موفق زیاد و در finally همان reader کم می‌شود.
برای مشاهدهٔ واقعیِ انتظار، دو reader ابتدا شروع‌شدن را اعلام می‌کنند و پشت Event می‌مانند. در همین نقطه، تعداد آغازشده و فعال دو است؛ سومی آغاز نشده است. سپس یکی آزاد می‌شود و شروع سومی را جدا انتظار می‌کشیم. این تأیید مرحله‌ای، جای حدس زدن بر اساس چند میلی‌ثانیه انتظار را می‌گیرد. Event در اینجا ابزار کنترل آزمایش است؛ ساعت یا دادهٔ زمان حسگر نیست.
کد برای N شناسه، N task و N خانهٔ نتیجه می‌سازد. بنابراین سقف reader فعال، سقف حافظهٔ کل، نرخ درخواست یا تعداد taskهای ساخته‌شده نیست. برای جریان بزرگ، پذیرش مرحله‌ای یا workerهای ثابت نیاز به طراحی دیگری دارد. در این مثال یک حلقهٔ رویداد داریم؛ از آن اجرای موازی پردازنده، عدالت ترتیب ورود یا ظرفیت سخت‌افزار را نتیجه نمی‌گیریم.
۳. کد کامل و مرز خطاهای داده و reader
کد با Python 3.14.5 اجرا شده است. limit باید int دقیق و مثبت و واحد دقیقاً V باشد؛ bool و مقدار نامعتبر رد می‌شوند. اعتبارسنجی شناسه‌ها پیش از ساخت فرزندان انجام می‌شود. بعد از await خواننده، فقط int یا float دقیق و متناهی پذیرفته می‌شود؛ NaN، بی‌نهایت، رشته و bool ولتاژ معتبر نیستند. خوانندهٔ همگامی که عدد غیرقابلawait برگرداند نیز خطای نوع می‌سازد.
"""Original deterministic TaskGroup sensor mock; no sleeps, network or timers."""
import asyncio
import json
import math
import platform
import re
class IncompleteCollectionError(Exception):
pass
async def collect_voltages(sensor_ids, reader, limit=2, unit='V'):
if type(limit) is not int or limit < 1:
raise ValueError('limit must be exact positive int')
if type(unit) is not str or unit != 'V':
raise ValueError('unit must be V')
if type(sensor_ids) not in (list, tuple):
raise TypeError('sensor_ids must be finite list or tuple')
if any(type(sid) is not str or re.fullmatch(r'[A-Za-z][A-Za-z0-9_]{0,15}', sid) is None
for sid in sensor_ids):
raise ValueError('invalid case-sensitive sensor identifier')
if len(set(sensor_ids)) != len(sensor_ids):
raise ValueError('sensor identifiers must be unique')
if not callable(reader):
raise TypeError('reader must be callable and return an awaitable')
results = [None] * len(sensor_ids)
semaphore = asyncio.Semaphore(limit)
async def worker(index, sid):
async with semaphore:
raw = await reader(sid)
if type(raw) not in (int, float):
raise TypeError('reader voltage must be exact int or float')
try:
voltage = float(raw)
except OverflowError as error:
raise ValueError('reader voltage cannot be represented as finite float') from error
if not math.isfinite(voltage):
raise ValueError('reader voltage must be finite')
results[index] = {'sensor_id': sid, 'voltage_V': voltage}
async with asyncio.TaskGroup() as group:
for index, sid in enumerate(sensor_ids):
group.create_task(worker(index, sid))
# A child CancelledError alone is not a TaskGroup failure; reject missing slot.
if any(result is None for result in results):
raise IncompleteCollectionError('a child ended without a validated voltage')
return results
class ControlledReader:
"""Observable fake resource; coordinator releases explicit Event gates."""
def __init__(self, actions):
self.actions = actions
self.gates = {sid: asyncio.Event() for sid in actions}
self.started_counts = {count: asyncio.Event() for count in range(1, len(actions) + 1)}
self.closed_events = {sid: asyncio.Event() for sid in actions}
self.opened = []
self.returned = []
self.closed = set()
self.cancelled = set()
self.active = 0
self.peak = 0
def release_all(self):
for event in self.gates.values():
event.set()
async def __call__(self, sid):
# Called only after semaphore admission; no decrement for a queued worker.
self.opened.append(sid)
self.active += 1
self.peak = max(self.peak, self.active)
self.started_counts[len(self.opened)].set()
try:
await self.gates[sid].wait()
action = self.actions[sid]
if isinstance(action, BaseException):
raise action
self.returned.append(sid)
return action
except asyncio.CancelledError:
self.cancelled.add(sid)
raise
finally:
self.active -= 1
self.closed.add(sid)
self.closed_events[sid].set()
def leaf_names(error):
if isinstance(error, BaseExceptionGroup):
return sorted(name for child in error.exceptions for name in leaf_names(child))
return [type(error).__name__]
async def _await_raises(expected, awaitable):
try:
await awaitable
except expected:
return True
return False
async def _controlled_failure(action, finish_other_first=False):
reader = ControlledReader({'A': action, 'B': 2.0})
task = asyncio.create_task(collect_voltages(['A', 'B'], reader, limit=2))
await reader.started_counts[2].wait() # both readers definitely acquired slots
if finish_other_first:
reader.gates['B'].set()
await reader.closed_events['B'].wait()
reader.gates['A'].set()
errors = []
try:
await task
except ExceptionGroup as error:
errors = leaf_names(error)
return {'errors': errors, 'cancelled': sorted(reader.cancelled),
'closed': sorted(reader.closed), 'returned': reader.returned.copy(), 'active': reader.active}
async def _run():
ids = ['S1', 'S2', 'S3']
reader = ControlledReader({'S1': 1.25, 'S2': 2.5, 'S3': -0.5})
task = asyncio.create_task(collect_voltages(ids, reader, limit=2))
await reader.started_counts[2].wait()
barrier_snapshot = {'active': reader.active, 'started': len(reader.opened), 'peak': reader.peak}
release_first = next(sid for sid in reader.opened if sid != ids[0])
reader.gates[release_first].set()
await reader.closed_events[release_first].wait()
await reader.started_counts[3].wait()
third = next(sid for sid in ids if sid not in reader.returned and sid not in reader.opened[:2])
reader.gates[third].set()
await reader.closed_events[third].wait()
for sid in ids:
reader.gates[sid].set()
ordered = await task
hand = [{'sensor_id': 'S1', 'voltage_V': 1.25}, {'sensor_id': 'S2', 'voltage_V': 2.5},
{'sensor_id': 'S3', 'voltage_V': -0.5}]
serial_reader = ControlledReader({'A': 1.0, 'B': 2.0, 'C': 3.0})
serial_task = asyncio.create_task(collect_voltages(['A', 'B', 'C'], serial_reader, limit=1))
serial_snapshots = []
for count in range(1, 4):
await serial_reader.started_counts[count].wait()
serial_snapshots.append(serial_reader.active)
serial_reader.gates[serial_reader.opened[-1]].set()
await serial_reader.closed_events[serial_reader.opened[-1]].wait()
await serial_task
empty_reader = ControlledReader({})
empty_result = await collect_voltages([], empty_reader)
never_reader = ControlledReader({'A': 1.0})
bad_limits = [await _await_raises(ValueError, collect_voltages(['A'], never_reader, limit=limit))
for limit in (True, 0, -1, 1.5, float('nan'))]
bad_ids = [await _await_raises((TypeError, ValueError), collect_voltages(sids, never_reader))
for sids in ('A', iter(['A']), [''], ['A', 'A'], [True], ['A B'], [None])]
bad_api = [await _await_raises(ValueError, collect_voltages(['A'], never_reader, unit=unit))
for unit in ('mV', b'V', None)]
bad_api.append(await _await_raises(TypeError, collect_voltages(['A'], None)))
runtime_failure = await _controlled_failure(RuntimeError('mock read failed'))
policy_timeout = await _controlled_failure(TimeoutError('injected logical deadline event'))
outer_reader = ControlledReader({'A': 1.0, 'B': 2.0})
outer_task = asyncio.create_task(collect_voltages(['A', 'B'], outer_reader))
await outer_reader.started_counts[2].wait()
outer_task.cancel()
external_cancel = await _await_raises(asyncio.CancelledError, outer_task)
invalid_results = []
for value, expected_error in ((float('nan'), 'ValueError'), (float('inf'), 'ValueError'),
('1', 'TypeError'), (None, 'TypeError'), (2**10000, 'ValueError')):
observation = await _controlled_failure(value)
invalid_results.append(observation['errors'] == [expected_error] and observation['cancelled'] == ['B']
and observation['closed'] == ['A', 'B'] and observation['active'] == 0)
boolean_result = await _controlled_failure(True)
nonawaitable = []
try:
await collect_voltages(['A'], lambda sid: 1.0)
except ExceptionGroup as error:
nonawaitable = leaf_names(error)
self_cancel_reader = ControlledReader({'A': asyncio.CancelledError(), 'B': 2.0})
self_cancel_reader.release_all()
child_cancel = await _await_raises(IncompleteCollectionError,
collect_voltages(['A', 'B'], self_cancel_reader))
partial = await _controlled_failure(RuntimeError('mock after other returned'), finish_other_first=True)
current = asyncio.current_task()
remaining = [pending for pending in asyncio.all_tasks() if pending is not current and not pending.done()]
checks = {
'hand_ordered_three_voltage_results': ordered == hand,
'completion_order_is_separate_from_request_order': reader.returned != ids
and sorted(reader.returned) == sorted(ids),
'two_active_and_third_waiting_at_event_barrier': barrier_snapshot == {'active': 2, 'started': 2, 'peak': 2}
and reader.peak == 2,
'all_success_resources_closed': reader.active == 0 and reader.closed == set(ids) and not reader.cancelled,
'limit_one_event_controlled_serial_admission': serial_snapshots == [1, 1, 1]
and serial_reader.peak == 1 and serial_reader.active == 0,
'empty_collection_calls_no_reader': empty_result == [] and empty_reader.opened == [],
'invalid_limits_rejected_before_reader': all(bad_limits) and never_reader.opened == [],
'invalid_duplicate_identifiers_rejected_before_reader': all(bad_ids) and never_reader.opened == [],
'invalid_units_and_reader_contract_rejected': all(bad_api),
'runtime_failure_groups_and_cancels_sibling': runtime_failure == {
'errors': ['RuntimeError'], 'cancelled': ['B'], 'closed': ['A', 'B'], 'returned': [], 'active': 0},
'injected_timeout_groups_and_cleans_up': policy_timeout == {
'errors': ['TimeoutError'], 'cancelled': ['B'], 'closed': ['A', 'B'], 'returned': [], 'active': 0},
'external_cancellation_propagates_after_cleanup': external_cancel and outer_task.cancelled()
and outer_reader.cancelled == {'A', 'B'} and outer_reader.closed == {'A', 'B'} and outer_reader.active == 0,
'invalid_nonfinite_or_nonvoltage_results_fail_group': all(invalid_results),
'bool_is_not_an_engineering_voltage': boolean_result['errors'] == ['TypeError']
and boolean_result['cancelled'] == ['B'] and boolean_result['active'] == 0,
'nonawaitable_reader_result_is_grouped_type_error': nonawaitable == ['TypeError'],
'child_cancelled_without_value_rejected_as_incomplete': child_cancel
and self_cancel_reader.closed == {'A', 'B'} and self_cancel_reader.returned == ['B']
and self_cancel_reader.cancelled == {'A'} and self_cancel_reader.active == 0,
'prior_reader_effect_not_rolled_back_by_later_failure': partial == {
'errors': ['RuntimeError'], 'cancelled': [], 'closed': ['A', 'B'], 'returned': ['B'], 'active': 0},
'no_pending_test_child_tasks_before_loop_exit': remaining == [],
}
assert len(checks) == 18 and all(type(value) is bool and value for value in checks.values())
return {'runtime': {'python': platform.python_version(), 'libraries': ['stdlib asyncio'],
'wall_clock_measurements': False, 'sleeps_or_real_timers': False},
'contract': {'unit': 'V', 'input': 'finite unique case-sensitive ASCII identifiers; exact positive int active limit',
'output_order': 'request order, not completion order',
'bounds': 'at most limit active reader calls; N tasks/results still allocated',
'timeout_scope': 'injected TimeoutError models policy failure, not a real elapsed deadline'},
'metrics': {'ordered_result': ordered, 'mock_return_order': reader.returned,
'event_barrier_snapshot': barrier_snapshot, 'peak_active_readers': reader.peak,
'runtime_failure': runtime_failure, 'injected_timeout': policy_timeout,
'external_cancel_closed_ids': sorted(outer_reader.closed),
'child_cancel_incomplete_rejected': child_cancel, 'partial_effect_observation': partial,
'pending_child_count_before_exit': len(remaining)},
'checks': checks,
'limitations': ['No real sensors/network/AI model/driver/resource or thread/process execution.',
'Mock Event and immediate finally cleanup test order/lifetime, not elapsed deadline, CPU parallelism or cancellation latency.',
'Active reader limit does not bound all task allocation, total memory, request rate or hardware load.',
'No transaction/rollback/partial external effect reversal; arbitrary readers swallowing cancellation are outside the validated contract.']}
def run_example():
return asyncio.run(_run())
if __name__ == '__main__':
print(json.dumps(run_example(), ensure_ascii=False, indent=2))
خطای معمولِ یک فرزند باعث لغو هم‌گروه‌های باقی‌مانده و انتظار برای پایانشان می‌شود. خطاهای انتخابی در ExceptionGroup دیده می‌شوند؛ فهرست نوع خطاها برای بررسی مرتب می‌شود و به ترتیب داخلیِ گروه اتکا نداریم. در مسیر جداگانه، لغو بیرونیِ خود جمع‌آور با CancelledError عبور می‌کند. آن را به TimeoutError یا خروجی موفق تبدیل نمی‌کنیم.
mock در finally بسته‌شدن منبع ساختگی را ثبت می‌کند و اگر CancelledError بگیرد، پس از پاک‌سازی دوباره آن را عبور می‌دهد. این ثبتِ مجموعه و شمارنده، بستن واقعی پورت، فایل یا درایور نیست. خواننده‌ای که لغو را نادیده بگیرد، خارج از قراردادِ آزمایش‌شده است. از این مثال زمان تضمین‌شده برای توقف یا مقاومت در برابر هر رفتار ناسازگار یک coroutine به دست نمی‌آید.
۴. هجده گروه بررسی با مرجع مشاهده‌پذیر مستقل
در اجرای موفق، mock به ترتیب S2، S3 و S1 مقدار برگردانده، اما خروجی نهایی همچنان S1، S2 و S3 است. شمار اوج reader فعال دو و در پایان صفر است؛ همهٔ شناسه‌های آغازشده در مجموعهٔ بسته‌شده قرار دارند. این دفتر مشاهدهٔ mock از خانه‌های خصوصیِ نتیجهٔ جمع‌آور جداست. مقادیر دستی و این مشاهده‌ها، مرجع رفتارند؛ تابع اصلی پاسخ مورد انتظار خودش را تولید نمی‌کند.
{
"runtime": {
"python": "3.14.5",
"libraries": [
"stdlib asyncio"
],
"wall_clock_measurements": false,
"sleeps_or_real_timers": false
},
"contract": {
"unit": "V",
"input": "finite unique case-sensitive ASCII identifiers; exact positive int active limit",
"output_order": "request order, not completion order",
"bounds": "at most limit active reader calls; N tasks/results still allocated",
"timeout_scope": "injected TimeoutError models policy failure, not a real elapsed deadline"
},
"metrics": {
"ordered_result": [
{
"sensor_id": "S1",
"voltage_V": 1.25
},
{
"sensor_id": "S2",
"voltage_V": 2.5
},
{
"sensor_id": "S3",
"voltage_V": -0.5
}
],
"mock_return_order": [
"S2",
"S3",
"S1"
],
"event_barrier_snapshot": {
"active": 2,
"started": 2,
"peak": 2
},
"peak_active_readers": 2,
"runtime_failure": {
"errors": [
"RuntimeError"
],
"cancelled": [
"B"
],
"closed": [
"A",
"B"
],
"returned": [],
"active": 0
},
"injected_timeout": {
"errors": [
"TimeoutError"
],
"cancelled": [
"B"
],
"closed": [
"A",
"B"
],
"returned": [],
"active": 0
},
"external_cancel_closed_ids": [
"A",
"B"
],
"child_cancel_incomplete_rejected": true,
"partial_effect_observation": {
"errors": [
"RuntimeError"
],
"cancelled": [],
"closed": [
"A",
"B"
],
"returned": [
"B"
],
"active": 0
},
"pending_child_count_before_exit": 0
},
"checks": {
"hand_ordered_three_voltage_results": true,
"completion_order_is_separate_from_request_order": true,
"two_active_and_third_waiting_at_event_barrier": true,
"all_success_resources_closed": true,
"limit_one_event_controlled_serial_admission": true,
"empty_collection_calls_no_reader": true,
"invalid_limits_rejected_before_reader": true,
"invalid_duplicate_identifiers_rejected_before_reader": true,
"invalid_units_and_reader_contract_rejected": true,
"runtime_failure_groups_and_cancels_sibling": true,
"injected_timeout_groups_and_cleans_up": true,
"external_cancellation_propagates_after_cleanup": true,
"invalid_nonfinite_or_nonvoltage_results_fail_group": true,
"bool_is_not_an_engineering_voltage": true,
"nonawaitable_reader_result_is_grouped_type_error": true,
"child_cancelled_without_value_rejected_as_incomplete": true,
"prior_reader_effect_not_rolled_back_by_later_failure": true,
"no_pending_test_child_tasks_before_loop_exit": true
},
"limitations": [
"No real sensors/network/AI model/driver/resource or thread/process execution.",
"Mock Event and immediate finally cleanup test order/lifetime, not elapsed deadline, CPU parallelism or cancellation latency.",
"Active reader limit does not bound all task allocation, total memory, request rate or hardware load.",
"No transaction/rollback/partial external effect reversal; arbitrary readers swallowing cancellation are outside the validated contract."
]
}خطای RuntimeError و خطای TimeoutError در دو سناریوی جدا تزریق می‌شوند. هر دو reader پیش از رخداد خطا، آمادگی را اعلام کرده‌اند؛ بنابراین لغو reader دوم و اجرای finally آن واقعاً مشاهده می‌شود. TimeoutError اینجا رخداد قراردادیِ شکست است؛ هیچ تایمر، sleep، wait_for یا deadline سپری‌شده‌ای فعال نشده است. آزمون، سیاست انتشار خطا را می‌سنجد و تأخیر لغو یا مهلت زمانی واقعی را اندازه نمی‌گیرد.
لغو تنها یک فرزند با CancelledError به‌تنهایی گروه را مانند خطای معمول شکست نمی‌دهد. به همین دلیل پس از خروج گروه، خانهٔ نتیجهٔ خالی بررسی و IncompleteCollectionError خودمان ایجاد می‌شود؛ دادهٔ ناقص موفق اعلام نمی‌شود. همچنین سناریویی داریم که B پیش از خطای A مقدار برگردانده است؛ اثر ثبت‌شدهٔ آن باقی می‌ماند. نداشتن خروجی نهاییِ موفق، بازگردانی اثرهای بیرونی یا تراکنش نیست.
صف محدود حسگر و پس‌فشار مسئلهٔ انتظار put و پایان حسابداری صف را پوشش می‌دهد؛ اینجا readerهای مستقل و عمر گروه مهم‌اند. قرارداد JSON حسگر نیز به اعتبار داده مربوط است. هجده گروه بررسی، از جمله نبود فرزند معلق پیش از خروج حلقهٔ همین آزمون، ثبت شده‌اند؛ پوشش همهٔ زمان‌بندی‌ها یا سرویس واقعی ادعا نمی‌شود.
۵. چهار پرسش پاسخ‌دار برای طراحی آزمون ناهمگام
پرسش اول: پایان S2 پیش از S1 ترتیب خروجی را عوض می‌کند؟ پاسخ: خیر؛ جایگاه هر نتیجه از اندیس درخواست تعیین می‌شود. جواب دستی همچنان S1 با ۱٫۲۵، S2 با ۲٫۵ و S3 با منفی ۰٫۵ ولت است.
پرسش دوم: limit دو یعنی فقط دو task داریم؟ پاسخ: خیر؛ برای هر شناسه task ساخته شده، ولی حداکثر دو reader داخل جایگاه فعال‌اند. حد حافظه یا نرخ درخواست از این عدد نتیجه نمی‌شود.
پرسش سوم: TimeoutError تزریق‌شده، مهلت واقعی را آزموده است؟ پاسخ: خیر؛ فقط انتشار خطا، لغو هم‌گروه و پاک‌سازی کنترل شده‌اند. هیچ مدت سپری‌شده یا آستانهٔ زمانی در آزمون نیست.
پرسش چهارم: چرا برای فرزند لغوشده، کنترل نتیجهٔ ناقص داریم؟ پاسخ: چون CancelledError آن فرزند به‌تنهایی خطای معمول گروه نیست. خانهٔ خالی را با خطای صریح رد می‌کنیم؛ منابع در ۲۴ مهر ۱۴۰۵ خوانده شدند و این رفتار، تضمین سرویس یا سخت‌افزار واقعی نیست.
منابع و مطالعهٔ بیشتر
- Python 3.14؛ TaskGroup و انتشار لغو
- Python 3.14؛ Event و Semaphore
- Python 3.14؛ ExceptionGroup و BaseExceptionGroup
بررسی منابع: ۲۴ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.
