فهرست مقاله
در پردازش ناهمگام دادهٔ مهندسی، برچسبی برای تشخیص مسیر کار می‌تواند بی‌سروصدا به کار بعدی بچسبد. متغیر سراسریِ معمولی، جایگاه مستقلی برای هر وظیفه نمی‌سازد. در این درس، ContextVar و Task واقعی پایتون را با برچسب‌های ساختگی می‌آزماییم: زمینه چه موقع گرفته می‌شود، تغییر فرزند کجا می‌ماند و توکن چه چیزی را برمی‌گرداند؟ هیچ حسگر، شبکه، حساب کاربری یا اندازه‌گیری زمانی در مثال وجود ندارد.
۱. برچسبِ تشخیصی، دادهٔ نمونه نیست
برچسب‌های A و B فقط فرادادهٔ تشخیصی‌اند؛ ولتاژ، زمان نمونه یا شناسهٔ احرازشده نیستند. مقدار برچسب بی‌واحد است و از نام آن نمی‌توان مالکیت یا مجوز استفاده از داده را نتیجه گرفت. محافظت مورد بحث، جلوگیری از آمیختن اتصالِ برچسب‌ها در محدودهٔ همین کارهاست؛ دربارهٔ محرمانگی، امنیت چندکاربری یا صحت دادهٔ فیزیکی ادعایی نداریم.
متغیر LABEL در سطح ماژول ساخته شده و مقدار پیش‌فرض UNSET دارد. مطابق مرجع ContextVar و Token، get مقدار متناظر با زمینهٔ جاری را می‌خواند و set اتصال همان زمینه را تغییر می‌دهد. شیء ContextVar مشترک است؛ آنچه در زمینه‌ها جدا نگه داشته می‌شود، اتصال متغیر به مقدار است. در مسیر اصلی، مقدارها رشتهٔ تغییرناپذیرند.
تابع آزمایش حداکثر سه برچسب را در tuple دقیق می‌گیرد. هر رشته یک تا دوازده نویسهٔ ASCII دارد، با حرف بزرگ آغاز می‌شود و ادامه‌اش حرف بزرگ، رقم یا زیرخط است. تکرار برچسب مجاز است. ترتیب آزادسازی باید جایگشتی از اندیس‌های صحیح دقیق باشد؛ بولی پذیرفته نیست. این‌ها حدود خوانایی آزمایش‌اند، نه محدودیت ContextVar؛ اعتبارسنجی تابع async هنگام await و پیش از ساخت فرزندان اجرا می‌شود.
پارامتر fail_at یا None است یا یک اندیس صحیح دقیق از همین انتخاب؛ پرچم منطقی و اندیس بیرون محدوده رد می‌شوند. این پارامتر تنها محل خطای ساختگی را تعیین می‌کند. متن برچسب‌ها اعتبارسنجی می‌شود، اما هیچ تبدیل خودکار نام یا اتصال به شناسهٔ واقعی دستگاه نداریم.
۲. زمینه در لحظهٔ ساخت Task کپی می‌شود
ابتدا همهٔ شیءهای coroutine را زیر برچسب CORO_BUILD می‌سازیم. سپس پیش از ساخت Task اول، برچسب را A و پیش از ساخت دومی، B می‌کنیم. طبق قرارداد کامل create_task، وقتی context صریح ندهیم، زمینهٔ جاری کپی می‌شود. پس ساخت شیء coroutine و ساخت Task را دو مرز متفاوت می‌گیریم؛ صرف فراخوانی تابع async، کار را اجرا نمی‌کند.
والد پس از ساخت فرزندان، برچسب خودش را عوض می‌کند. هر فرزند ابتدا آمادگی را اعلام می‌کند و پشت دروازه می‌ماند؛ والد بعد از مشاهدهٔ آمادگی‌ها، مقدار PARENT_AFTER را می‌گذارد و دروازه‌ها را باز می‌کند. خواندن برچسبِ فرزند پس از این مرحله است، ولی همچنان A یا B دیده می‌شود. جواب به حدس دربارهٔ زمان شروع نخستین دستور فرزند وابسته نیست.
مرجع کامل Event رفتار set و wait را مشخص می‌کند. دروازهٔ دوم اجازه می‌دهد هر فرزند پس از تغییر محلی، معلق بماند تا تغییرهای دیگران رخ دهند. نه sleep داریم و نه مهلت سپری‌شده؛ ترتیب موردنیاز را با مشاهدهٔ رخداد می‌سازیم. این کنترل، تضمین عمومی دربارهٔ عدالت زمان‌بند، سرعت اجرای ناهمگام یا رفتار چندریسمانی نمی‌دهد.
۳. تغییرِ فرزند و بازگرداندنِ توکن
فرزند اول پس از دیدن A، مقدار CHILD_0 را می‌گذارد و توکن برگشتی را نگه می‌دارد. بعد از انتظار دوم، همان مقدار محلی را می‌بیند؛ تغییر برچسب والد و فرزند دیگر آن را جایگزین نکرده است. در finally، reset همان توکن را مصرف می‌کند و مقدار پیشین A برمی‌گردد. برای فرزند دوم هم مسیر مشابه با B و CHILD_1 داریم.
در شاهد اصلی، فرزند اول پس از دروازهٔ پایان، ProcessingError ساختگی می‌دهد. finally اجرا و بازگشت A ثبت می‌شود؛ والد بعداً همان شیء خطا را از await وظیفه دریافت می‌کند. خطا پنهان یا موفق اعلام نمی‌شود. این بازگردانی فقط اتصال متغیر زمینه است؛ اثر پردازش بیرونی را عقب نمی‌کشد و تضمین پاک‌سازی در برابر پایان ناگهانی فرایند یا هر لغو خارجی نیست.
توکن به متغیر و زمینهٔ سازنده‌اش وابسته است و دوباره مصرف نمی‌شود. آزمایش جدا در نسخهٔ اجراشده، استفادهٔ دوباره را با RuntimeError و استفاده برای متغیر یا زمینهٔ دیگر را با ValueError مشاهده می‌کند. توکنِ نخستین set همچنین می‌تواند نبودِ اتصال قبلی را نشان دهد. بازگردانی آن، مقدار تازهٔ ثابتی نمی‌نویسد؛ اتصال را حذف می‌کند، تا get یا پیش‌فرض را بدهد یا بدون پیش‌فرض LookupError بسازد.
۴. کد کامل و مرجعِ جدولِ زمینه‌ها
فایل contextvars_task_boundary.py را ذخیره و با دستور python contextvars_task_boundary.py اجرا کنید. اجرای واقعی CPython 3.14.5 و فقط کتابخانهٔ استاندارد دارد؛ مستندات خوانده‌شده نسخهٔ 3.14.8 هستند. نگه‌داشتن ارجاع وظیفه‌ها و await کردن همهٔ آن‌ها در همین مثال صریح است. نمونه‌های شبکه و تأخیر موجود در مستندات، بخشی از برنامهٔ آزمایش‌شدهٔ ما نیستند.
"""Task-local immutable diagnostic labels; no sensor, network or authorization."""
import asyncio
from contextvars import Context, ContextVar, copy_context, Token
from itertools import permutations, product
import json
import platform
import re
LABEL = ContextVar('diagnostic_label', default='UNSET')
BARE = ContextVar('bare_probe')
OTHER = ContextVar('other_probe')
BOX = ContextVar('mutable_box_probe')
class ProcessingError(Exception):
pass
async def trace_contexts(labels, release_order, *, fail_at=None):
"""Own finite profile; validation runs when awaited, before child creation."""
if type(labels) is not tuple or len(labels) > 3:
raise ValueError('labels must be an exact tuple of at most three labels')
if any(type(x) is not str or not 1 <= len(x) <= 12
or re.fullmatch(r'[A-Z][A-Z0-9_]*', x) is None for x in labels):
raise ValueError('labels must be bounded exact ASCII uppercase strings')
n = len(labels)
if (type(release_order) is not tuple or len(release_order) != n
or any(type(i) is not int for i in release_order)
or set(release_order) != set(range(n))):
raise ValueError('release_order must be a permutation of exact indices')
if fail_at is not None and (type(fail_at) is not int or not 0 <= fail_at < n):
raise ValueError('fail_at must be None or an exact valid index')
ready = [asyncio.Event() for _ in labels]
enter = [asyncio.Event() for _ in labels]
changed = [asyncio.Event() for _ in labels]
finish = [asyncio.Event() for _ in labels]
done = [asyncio.Event() for _ in labels]
rows = [{} for _ in labels]
failures = [ProcessingError('synthetic_processing_failure') for _ in labels]
parent_observations = []
tasks = []
outer_token = LABEL.set('CORO_BUILD')
async def worker(i):
ready[i].set()
await enter[i].wait()
rows[i]['inherited'] = LABEL.get()
local_token = LABEL.set('CHILD_' + str(i))
try:
rows[i]['local_before_wait'] = LABEL.get()
changed[i].set()
await finish[i].wait()
rows[i]['local_after_wait'] = LABEL.get()
if i == fail_at:
raise failures[i]
finally:
LABEL.reset(local_token)
rows[i]['restored'] = LABEL.get()
done[i].set()
try:
# All coroutine objects exist under CORO_BUILD, before any task is made.
coroutines = [worker(i) for i in range(n)]
for label, coro in zip(labels, coroutines, strict=True):
LABEL.set(label)
# No explicit context argument: actual Task copies current context.
tasks.append(asyncio.create_task(coro))
LABEL.set('PARENT_READY')
for event in ready:
await event.wait()
LABEL.set('PARENT_AFTER')
for i in release_order:
enter[i].set()
await changed[i].wait()
parent_observations.append(LABEL.get())
# Every child now has a local binding and is suspended at its finish gate.
LABEL.set('PARENT_FINAL')
for i in reversed(release_order):
finish[i].set()
await done[i].wait()
caught = []
error_identity = []
for i, task in enumerate(tasks):
try:
await task
caught.append(None)
error_identity.append(True)
except ProcessingError as error:
caught.append(type(error).__name__)
error_identity.append(error is failures[i])
return {'rows': rows, 'parent_during_children': parent_observations,
'parent_after_children': LABEL.get(), 'caught': caught,
'same_error_objects': error_identity, 'all_child_tasks_done': all(t.done() for t in tasks)}
finally:
LABEL.reset(outer_token)
def table_oracle(labels, release_order, fail_at):
"""Internal immutable fixture table; no contextvars, asyncio or event model."""
return {'rows': [{'inherited': label, 'local_before_wait': 'CHILD_' + str(i),
'local_after_wait': 'CHILD_' + str(i), 'restored': label}
for i, label in enumerate(labels)],
'parent_during_children': ['PARENT_AFTER'] * len(release_order),
'parent_after_children': 'PARENT_FINAL',
'caught': ['ProcessingError' if i == fail_at else None for i in range(len(labels))],
'same_error_objects': [True] * len(labels), 'all_child_tasks_done': True}
def rejects(call, kind):
try:
call()
except kind:
return True
return False
def token_probes():
"""Isolated synchronous fixture, called inside a fresh empty Context."""
missing_before = rejects(BARE.get, LookupError)
t = BARE.set('FIRST')
old_missing = t.old_value is Token.MISSING
BARE.reset(t)
missing_after = rejects(BARE.get, LookupError)
reuse_rejected = rejects(lambda: BARE.reset(t), RuntimeError)
outer = LABEL.set('OUTER')
inner = LABEL.set('INNER')
LABEL.reset(inner)
restored_outer = LABEL.get()
LABEL.reset(outer)
restored_default = LABEL.get()
token = BARE.set('ORIGINAL')
wrong_var = rejects(lambda: OTHER.reset(token), ValueError)
copied = copy_context()
wrong_context = rejects(lambda: copied.run(BARE.reset, token), ValueError)
BARE.reset(token) # Failed foreign resets did not consume the valid token.
box = [1]
box_token = BOX.set(box)
try:
copied_box_context = copy_context()
same_box = copied_box_context.run(BOX.get) is box
copied_box_context.run(lambda: BOX.get().append(2))
shallow_shared = same_box and BOX.get() == [1, 2]
finally:
BOX.reset(box_token)
return {'missing_before_and_after_reset': missing_before and missing_after,
'token_old_value_missing': old_missing, 'token_reuse_rejected': reuse_rejected,
'nested_restore': [restored_outer, restored_default],
'wrong_variable_rejected': wrong_var, 'wrong_context_rejected': wrong_context,
'failed_foreign_resets_leave_original_token_usable': rejects(BARE.get, LookupError),
'copy_has_shared_mutable_value_reference': shallow_shared}
async def direct_await_probe():
"""A direct await creates no new Task/context boundary for this coroutine."""
async def unscoped():
LABEL.set('DIRECT_CHANGE')
await unscoped()
return LABEL.get()
async def _run():
caller_token = LABEL.set('CALLER')
try:
literal = await trace_contexts(('A', 'B'), (1, 0), fail_at=0)
restored_caller = LABEL.get()
hand = {'rows': [
{'inherited': 'A', 'local_before_wait': 'CHILD_0', 'local_after_wait': 'CHILD_0', 'restored': 'A'},
{'inherited': 'B', 'local_before_wait': 'CHILD_1', 'local_after_wait': 'CHILD_1', 'restored': 'B'}],
'parent_during_children': ['PARENT_AFTER', 'PARENT_AFTER'],
'parent_after_children': 'PARENT_FINAL', 'caught': ['ProcessingError', None],
'same_error_objects': [True, True], 'all_child_tasks_done': True}
cases, oracle_ok = 0, True
for n in range(4):
for labels in product(('A', 'B'), repeat=n):
for order in permutations(range(n)):
for fail_at in (None, *range(n)):
actual = await trace_contexts(labels, order, fail_at=fail_at)
oracle_ok = oracle_ok and actual == table_oracle(labels, order, fail_at)
oracle_ok = oracle_ok and LABEL.get() == 'CALLER'
cases += 1
async def invalid_cases():
bad = [([], ()), (('A',) * 4, (0, 1, 2, 3)), ((1,), (0,)),
(('a',), (0,)), (('',), (0,)), (('A' * 13,), (0,)),
(('A B',), (0,)), (('A',), [0]), (('A',), (True,)),
(('A', 'B'), (0, 0)), (('A',), (1,)), (('A',), ())]
for labels, order in bad:
try:
await trace_contexts(labels, order)
except ValueError:
continue
return False
for bad_index in (True, -1, 1, 1.0, '0'):
try:
await trace_contexts(('A',), (0,), fail_at=bad_index)
except ValueError:
continue
return False
return True
invalid_ok = await invalid_cases()
probe = Context().run(token_probes)
direct_token = LABEL.set('BEFORE_DIRECT')
try:
direct = await direct_await_probe()
finally:
LABEL.reset(direct_token)
pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task() and not t.done()]
checks = {
'literal_two_task_context_table_and_deliberate_failure_match': literal == hand,
'task_creation_bindings_not_coroutine_construction_or_parent_resume_bindings': (
[row['inherited'] for row in literal['rows']] == ['A', 'B']),
'child_local_bindings_survive_suspension_while_sibling_and_parent_bindings_change': all(
row['local_before_wait'] == row['local_after_wait'] == 'CHILD_' + str(i)
for i, row in enumerate(literal['rows'])),
'child_exception_finally_restores_its_own_prior_binding_and_preserves_error_identity': (
literal['rows'][0]['restored'] == 'A' and literal['caught'] == ['ProcessingError', None]
and literal['same_error_objects'] == [True, True]),
'parent_binding_is_unchanged_by_child_sets_and_caller_binding_is_restored': (
literal['parent_during_children'] == ['PARENT_AFTER'] * 2
and literal['parent_after_children'] == 'PARENT_FINAL' and restored_caller == 'CALLER'),
'direct_await_unscoped_set_changes_the_same_executing_task_context': direct == 'DIRECT_CHANGE',
'nested_tokens_restore_prior_binding_then_default': probe['nested_restore'] == ['OUTER', 'UNSET'],
'reset_of_first_binding_restores_absence_and_token_missing_marker': (
probe['missing_before_and_after_reset'] and probe['token_old_value_missing']),
'a_consumed_token_cannot_be_used_twice': probe['token_reuse_rejected'],
'foreign_variable_or_context_tokens_rejected_without_consuming_valid_token': (
probe['wrong_variable_rejected'] and probe['wrong_context_rejected']
and probe['failed_foreign_resets_leave_original_token_usable']),
'copying_a_context_does_not_deepcopy_its_mutable_values': probe['copy_has_shared_mutable_value_reference'],
'221_bounded_label_order_failure_cases_match_independent_table_oracle_one_family': oracle_ok and cases == 221,
'bounded_labels_permutations_and_failure_index_reject_invalid_inputs_before_children': invalid_ok,
'all_own_children_awaited_with_no_pending_child_before_loop_exit': pending == [],
}
assert len(checks) == 14 and all(type(v) is bool and v for v in checks.values()), checks
return {'python': platform.python_version(), 'implementation': platform.python_implementation(),
'model': 'immutable_diagnostic_task_context_creation_and_token_restoration',
'actual_stdlib_asyncio_and_contextvars': True, 'sleeps_timers_or_hardware': False,
'literal_scenario': literal, 'caller_binding_after_scenario': restored_caller,
'direct_await_observed': direct, 'synchronous_token_and_copy_probes': probe,
'oracle_case_count_one_group': cases, 'checks': checks}
finally:
LABEL.reset(caller_token)
if __name__ == '__main__':
print(json.dumps(asyncio.run(_run()), indent=2, allow_nan=False))
جدول موردانتظار اصلی دستی است: ورودی‌های ارث‌رسیده A و B، مقدارهای محلی CHILD_0 و CHILD_1 و بازگشت‌های A و B. والد در حین تغییر فرزندان PARENT_AFTER را نگه می‌دارد و پس از آن PARENT_FINAL را می‌خواند. تابع بیرونی نیز سرانجام اتصال CALLER فراخواننده را برمی‌گرداند. خروجی بر اساس اندیس کار مرتب است، نه ادعای ترتیب عمومی پایان وظیفه‌ها.
{
"python": "3.14.5",
"implementation": "CPython",
"model": "immutable_diagnostic_task_context_creation_and_token_restoration",
"actual_stdlib_asyncio_and_contextvars": true,
"sleeps_timers_or_hardware": false,
"literal_scenario": {
"rows": [
{
"inherited": "A",
"local_before_wait": "CHILD_0",
"local_after_wait": "CHILD_0",
"restored": "A"
},
{
"inherited": "B",
"local_before_wait": "CHILD_1",
"local_after_wait": "CHILD_1",
"restored": "B"
}
],
"parent_during_children": [
"PARENT_AFTER",
"PARENT_AFTER"
],
"parent_after_children": "PARENT_FINAL",
"caught": [
"ProcessingError",
null
],
"same_error_objects": [
true,
true
],
"all_child_tasks_done": true
},
"caller_binding_after_scenario": "CALLER",
"direct_await_observed": "DIRECT_CHANGE",
"synchronous_token_and_copy_probes": {
"missing_before_and_after_reset": true,
"token_old_value_missing": true,
"token_reuse_rejected": true,
"nested_restore": [
"OUTER",
"UNSET"
],
"wrong_variable_rejected": true,
"wrong_context_rejected": true,
"failed_foreign_resets_leave_original_token_usable": true,
"copy_has_shared_mutable_value_reference": true
},
"oracle_case_count_one_group": 221,
"checks": {
"literal_two_task_context_table_and_deliberate_failure_match": true,
"task_creation_bindings_not_coroutine_construction_or_parent_resume_bindings": true,
"child_local_bindings_survive_suspension_while_sibling_and_parent_bindings_change": true,
"child_exception_finally_restores_its_own_prior_binding_and_preserves_error_identity": true,
"parent_binding_is_unchanged_by_child_sets_and_caller_binding_is_restored": true,
"direct_await_unscoped_set_changes_the_same_executing_task_context": true,
"nested_tokens_restore_prior_binding_then_default": true,
"reset_of_first_binding_restores_absence_and_token_missing_marker": true,
"a_consumed_token_cannot_be_used_twice": true,
"foreign_variable_or_context_tokens_rejected_without_consuming_valid_token": true,
"copying_a_context_does_not_deepcopy_its_mutable_values": true,
"221_bounded_label_order_failure_cases_match_independent_table_oracle_one_family": true,
"bounded_labels_permutations_and_failure_index_reject_invalid_inputs_before_children": true,
"all_own_children_awaited_with_no_pending_child_before_loop_exit": true
}
}مرجع مستقل از جدول رشته‌ها و اندیس‌ها جواب می‌سازد و از asyncio، ContextVar یا مدل دروازه‌ها استفاده نمی‌کند. برای صفر تا سه کار، برچسب‌های A و B، جایگشت‌های آزادسازی و محل‌های خطای انتخابی، ۲۲۱ حالت در یک خانواده مقایسه می‌شوند. چهارده گروه نام‌دار، مرز ساخت، انتظار محلی، توکن‌ها، نوع ورودی و پایان فرزندان را پوشش می‌دهند؛ هر حالت یا اجرای دوباره، گروه تازه نیست.
در حالت بدون کار، tuple برچسب و ترتیب هر دو تهی‌اند و fail_at باید None باشد. هیچ فرزندی ساخته نمی‌شود، اما بازگردانی اتصال والد همچنان بررسی می‌شود. برچسب‌های یکسان نیز مجازند؛ برابر بودن دو مقدار به معنی مشترک بودن زمینهٔ دو Task نیست.
۵. چهار پرسش دربارهٔ انتظار و اشتراک
پرسش اول: اگر coroutine زیر CORO_BUILD ساخته شود، چرا فرزند A می‌بیند؟ پاسخ: در این مسیر، زمینه در ساخت Task و بدون context صریح گرفته شده است. والد پیش از همان ساخت، A گذاشته؛ تغییر بعدی والد به PARENT_AFTER آن نسخه را جایگزین نکرده است. تغییر محل create_task می‌تواند معنای برچسب را عوض کند، حتی اگر متن coroutine ثابت بماند.
پرسش دوم: await مستقیم هم مرز مستقلی می‌سازد؟ پاسخ: خیر؛ شاهد جدا، coroutineای را مستقیم await می‌کند که بدون reset مقدار DIRECT_CHANGE می‌گذارد. فراخواننده همان تغییر را می‌بیند، چون اجرای آن در Task فعلی ادامه دارد. پوشانندهٔ آزمایش توکن خودش را در finally بازمی‌گرداند؛ افزودن await به‌تنهایی جای آن بازگردانی نیست.
پرسش سوم: کپی زمینه، فهرست داخل آن را هم مستقل می‌کند؟ پاسخ: خیر؛ شاهد همگام copy_context به همان فهرست دسترسی دارد و افزودن ۲، فهرست والد را نیز از یک عضو به دو عضو می‌رساند. کپیِ زمینه عمیق نیست. به همین دلیل مسیر اصلی فقط رشته‌های تغییرناپذیر دارد؛ جدایی اتصال‌ها، جدایی خودکار همهٔ داده‌های تغییرپذیر نیست.
پرسش چهارم: این درس جای مدیریت عمر منابع را می‌گیرد؟ پاسخ: خیر؛ آزمون عمر TaskGroup و آزمون خروج منابع با ExitStack مسئولیت‌های دیگری دارند. اینجا فقط زمینهٔ تشخیصی و بازگردانی کنترل‌شده آزموده شده‌اند. منابع در ۲۷ مهر ۱۴۰۵ خوانده شده‌اند؛ از نتیجه، عملکرد حسگر، مجوز دسترسی یا پوشش همهٔ لغوها نتیجه نمی‌گیریم.
منابع و مطالعهٔ بیشتر
- Python3.14.8؛ ContextVar، توکن و کپی زمینه
- Python3.14.8؛ ساخت Task و پارامتر context
- Python3.14.8؛ قرارداد کامل Event
بررسی منابع: ۲۷ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.
