فهرست مقاله
در آزمون نرم‌افزار مهندسی، لازم نیست برای رسیدن به هر رویداد واقعاً منتظر بمانیم. می‌توان زمان شبیه‌سازی را جلو برد و دید کدام رکورد باید زودتر پردازش شود، در تساوی چه ترتیبی دارد و پس از لغو چه می‌شود. این درس با دادهٔ ساختگی حسگر، یک زمان‌بند قطعیِ کوچک می‌سازد. برنامه فقط رکورد برمی‌گرداند؛ حسگر، ساعت سیستم، نخ هم‌زمان یا تابع اندازه‌گیری را اجرا نمی‌کند. کد کامل، پاسخ دستی و هجده گروه بررسی در اختیار داریم.
۱. زمان شبیه‌سازی را با ساعت واقعی اشتباه نگیریم
زمان رویداد و افق توقف، عدد صحیح غیرمنفی با واحد میکروثانیه‌اند. مبدأ صفر، آغاز همین آزمایش فرضی است؛ تاریخ تقویمی، زمان تهران یا برچسب UTC نیست. مقدار اندازه‌گیری نیز عدد صحیح میلی‌ولت در بازهٔ بستهٔ منفی پنج‌هزار تا مثبت پنج‌هزار است. این دو واحد نقش‌های جدا دارند؛ زمان را از اندازهٔ ولتاژ حدس نمی‌زنیم و میلی‌ولت را بی‌صدا به ولت تغییر نمی‌دهیم.
تابع schedule یک رکورد آینده می‌گیرد. run_until همهٔ رویدادهای فعال با زمان کوچک‌تر یا مساوی افق را برمی‌گرداند و now_us را برابر همان افق می‌گذارد. حتی با صف خالی، زمان مجازی جلو می‌رود. هیچ sleep یا خواندن ساعت دیواری وجود ندارد. رویداد پیش از زمان فعلی و توقف رو به گذشته خطا هستند؛ رویداد دقیقاً در زمان فعلی مجاز است.
در مستندات sched پایتون، تعامل زمان‌بند عمومی با تابع زمان و تأخیر و امکان عقب افتادن از برنامه توضیح داده شده است. ما آن API را اجرا نمی‌کنیم و اولویت عددی جداگانه نداریم. قطعی بودن ترتیب این مدل، تضمین رسیدن به موعد واقعی یا سرعت پردازش روی سخت‌افزار نیست. هدف، روشن کردن سیاست ترتیب و مرز توقف پیش از اتصال به محیط واقعی است.
۲. تودهٔ کمینه و ترتیب ورود در زمان برابر
ساختار heapq عضو کمینه را در ریشه نگه می‌دارد؛ لازم نیست تمام فهرست داخلی مرتب باشد. برای هر ورودی، یک token افزایشی می‌سازیم و چهارتایی زمان، token، شناسهٔ حسگر و میلی‌ولت را وارد توده می‌کنیم. زمان معیار اول است و token یکتا، تساوی زمان را با ترتیب ورود می‌شکند. در نتیجه، اسم حسگر معیار مرتب‌سازیِ تساوی نیست.
برای یک آزمون روشن، سه رویداد در زمان هفت با شناسه‌های Z، A و M به همین ترتیب وارد می‌کنیم. جواب باید Z، A و M بماند، نه ترتیب الفبایی. توده به‌خودی‌خود سیاست پایدارِ دلخواه ما را اختراع نمی‌کند؛ token صریح آن را قابل مشاهده و آزمودن می‌سازد. token تنها در طول عمر همین شیء یکتا می‌ماند و شناسهٔ جهانی یا مجوز دسترسی نیست.
زمان صحیح برای این تمرین مزیت دیگری دارد. مستندات نوع‌های عددی پایتون تفاوت عدد صحیح و شناور را بیان می‌کند. در مثال بزرگ، دو زمانِ توان شصتمِ دو و یک واحد بیشتر، به‌صورت صحیح جدا می‌مانند؛ تبدیل آن‌ها به float در اجرای ما هر دو را برابر می‌کند. این نمایش عددی، دقت فیزیکی حسگر یا ظرفیت شمارندهٔ یک پردازندهٔ مشخص را تعیین نمی‌کند.
۳. لغو منطقی، حذف فوری از توده نیست
مجموعهٔ active فقط tokenهای منتظر را نگه می‌دارد. cancel، token موجود را از این مجموعه برمی‌دارد و True برمی‌گرداند؛ چهارتایی در توده باقی می‌ماند. وقتی آن ورودی به ریشه برسد، run_until آن را کنار می‌گذارد. این لغو تنبل، تفاوت «دیگر اجرا نشود» و «همین حالا از حافظه حذف شود» را روشن می‌کند. لغو ناشناخته، لغو دوباره یا لغوِ پردازش‌شده در سیاست ما False می‌دهد.
در دادهٔ دستی پنج ورودی داریم، اما یکی را لغو می‌کنیم؛ قبل از اجرا تعداد فعال چهار و اندازهٔ فیزیکی توده پنج است. پس pending_count معیار حافظهٔ اشغال‌شده نیست. ورودی لغوشدهٔ دوردست می‌تواند تا افق مناسب در توده بماند. این نمونه فشرده‌سازی توده، سقف حافظه یا نگهداری دائمی ندارد؛ برای حجم زیاد باید سیاست نگهداری داده را جدا طراحی کرد.
لغو در این کد، کار آغازشده را متوقف نمی‌کند؛ اصلاً callback اجرایی نداریم. خروجی صرفاً نشان می‌دهد کدام رکورد طبق برنامه انتخاب شده است. تفاوت این تمرین با صف محدود asyncio و پس‌فشار نیز همین‌جاست: آن درس دربارهٔ تولیدکننده و مصرف‌کننده است، این یکی دربارهٔ انتخاب ترتیب رویدادهای از پیش تعیین‌شده و مرز افق مجازی.
۴. کد کامل و مرجع مستقل برای خط زمانی
فایل sensor_event_scheduler.py را اجرا کنید؛ کتابخانهٔ بیرونی لازم نیست. نسخهٔ اجرای ثبت‌شده Python 3.14.5 است. پنج رویداد به ترتیب ورود، زمان‌های صفر، ده، ده، پنج و بیست دارند. شناسه‌ها S01، S02، S01، S03 و S01 و مقدارها هزار تا هزاروچهارصد میلی‌ولت با گام صد هستند. ورودی چهارم، با token سه و زمان پنج، پیش از اجرا لغو می‌شود.
جواب دستی تا افق ده، tokenهای صفر، یک و دو است؛ دو رویداد زمان ده به ترتیب ورود می‌مانند. زمان فعلی ده و یک رویداد فعال باقی است. ادامه تا بیست، token چهار با ۱۴۰۰ میلی‌ولت را می‌دهد. حذف token سه نباید شماره‌های بعدی را جابه‌جا کند. این پاسخ را جدا از توده نوشته‌ایم تا خطای مشترکِ محاسبه و انتظار پنهان نشود.
"""Deterministic discrete-event sampling records, integer microseconds, no hardware/clock."""
import heapq
import json
import platform
import random
import re
def exact_nonnegative_integer(value):
if type(value) is not int or value < 0:
raise ValueError('time_us')
return value
class SamplingEvents:
def __init__(self):
self.now_us = 0
self._sequence = 0
self._heap = []
self._active = set()
def schedule(self, at_us, sensor_id, millivolts):
exact_nonnegative_integer(at_us)
if at_us < self.now_us:
raise ValueError('past_event')
if type(sensor_id) is not str or re.fullmatch(r'[A-Za-z][A-Za-z0-9_-]{0,15}', sensor_id) is None:
raise ValueError('sensor_id')
if type(millivolts) is not int or not -5000 <= millivolts <= 5000:
raise ValueError('millivolts')
token = self._sequence
self._sequence += 1
heapq.heappush(self._heap, (at_us, token, sensor_id, millivolts))
self._active.add(token)
return token
def cancel(self, token):
if type(token) is not int or token < 0:
raise ValueError('token')
if token not in self._active:
return False
self._active.remove(token)
return True
def run_until(self, stop_us):
exact_nonnegative_integer(stop_us)
if stop_us < self.now_us:
raise ValueError('past_stop')
trace = []
while self._heap and self._heap[0][0] <= stop_us:
at_us, token, sensor_id, millivolts = heapq.heappop(self._heap)
if token not in self._active:
continue # Lazy cancellation: discard tombstone when it reaches the root.
self._active.remove(token)
trace.append({'time_us': at_us, 'token': token,
'sensor_id': sensor_id, 'millivolts': millivolts})
self.now_us = stop_us
return trace
@property
def pending_count(self):
return len(self._active)
@property
def physical_heap_count(self):
return len(self._heap)
def reference_trace(events, canceled_tokens, stop_us):
"""Independent batch oracle: filter and sort records, never calls heapq/SamplingEvents."""
kept = [(at, order, sensor, mv) for order, (at, sensor, mv) in enumerate(events)
if order not in canceled_tokens and at <= stop_us]
kept.sort(key=lambda item: (item[0], item[1]))
return [{'time_us': at, 'token': order, 'sensor_id': sensor, 'millivolts': mv}
for at, order, sensor, mv in kept]
def rejects(call, message):
try:
call()
except ValueError as error:
return str(error) == message
return False
def run_example():
fixture = [(0, 'S01', 1000), (10, 'S02', 1100), (10, 'S01', 1200),
(5, 'S03', 1300), (20, 'S01', 1400)]
scheduler = SamplingEvents()
tokens = [scheduler.schedule(*event) for event in fixture]
cancel_first = scheduler.cancel(tokens[3])
canceled_again = scheduler.cancel(tokens[3])
before = {'pending': scheduler.pending_count, 'physical_heap': scheduler.physical_heap_count}
first = scheduler.run_until(10)
after_first = {'now_us': scheduler.now_us, 'pending': scheduler.pending_count}
last = scheduler.run_until(20)
expected_first = [
{'time_us': 0, 'token': 0, 'sensor_id': 'S01', 'millivolts': 1000},
{'time_us': 10, 'token': 1, 'sensor_id': 'S02', 'millivolts': 1100},
{'time_us': 10, 'token': 2, 'sensor_id': 'S01', 'millivolts': 1200},
]
expected_last = [{'time_us': 20, 'token': 4, 'sensor_id': 'S01', 'millivolts': 1400}]
ties = SamplingEvents()
for name in ('Z', 'A', 'M'):
ties.schedule(7, name, 0)
tie_names = [item['sensor_id'] for item in ties.run_until(7)]
unordered = SamplingEvents()
for t in (9, 1, 6):
unordered.schedule(t, 'S01', 0)
ordered_times = [item['time_us'] for item in unordered.run_until(9)]
invalid = SamplingEvents()
invalid_time_result = all(rejects(lambda value=value: invalid.schedule(value, 'S01', 0), 'time_us')
and rejects(lambda value=value: invalid.run_until(value), 'time_us')
for value in (-1, True, 1.0, '1', None))
past = SamplingEvents()
past.run_until(10)
boundary = SamplingEvents()
boundary.schedule(10, 'S01', 0)
boundary.schedule(11, 'S02', 0)
boundary_trace = boundary.run_until(10)
same = SamplingEvents()
empty_trace = same.run_until(12)
same.schedule(12, 'S01', 0)
same_instant_trace = same.run_until(12)
split, once = SamplingEvents(), SamplingEvents()
for event in fixture:
split.schedule(*event)
once.schedule(*event)
split.cancel(3)
once.cancel(3)
split_trace = split.run_until(0) + split.run_until(7) + split.run_until(10) + split.run_until(20)
whole_trace = once.run_until(20)
rng = random.Random(71)
reference_ok = True
reference_cases = 49
for case in range(reference_cases):
events = [(rng.randrange(21), 'S' + str(index), rng.randrange(-5000, 5001))
for index in range(case % 13)]
canceled = {i for i in range(len(events)) if (i + case) % 3 == 0}
stop = case % 21
candidate = SamplingEvents()
for event in events:
candidate.schedule(*event)
for token in canceled:
candidate.cancel(token)
reference_ok = reference_ok and candidate.run_until(stop) == reference_trace(events, canceled, stop)
huge = SamplingEvents()
base = 2**60
huge.schedule(base + 1, 'A', 1)
huge.schedule(base, 'B', 2)
huge_times = [item['time_us'] for item in huge.run_until(base + 1)]
fields = SamplingEvents()
invalid_fields = all(rejects(lambda value=value: fields.schedule(0, value, 0), 'sensor_id')
for value in ('1bad', '', 'حسگر', 'S' * 17, 1))
invalid_units = all(rejects(lambda value=value: fields.schedule(0, 'S01', value), 'millivolts')
for value in (True, 0.0, '0', -5001, 5001))
bound_tokens = [fields.schedule(0, 'S01', value) for value in (-5000, 5000)]
all_cancel = SamplingEvents()
ct = [all_cancel.schedule(t, 'S01', 0) for t in (2, 9)]
all_cancel.cancel(ct[0])
all_cancel.cancel(ct[1])
all_cancel_trace = all_cancel.run_until(9)
checks = {
'authored_hand_timeline_through_inclusive_ten_microseconds': first == expected_first and after_first == {'now_us': 10, 'pending': 1},
'remaining_event_at_twenty_has_original_insertion_token': last == expected_last and scheduler.now_us == 20 and scheduler.pending_count == 0,
'same_time_uses_insertion_sequence_not_sensor_alphabetical_order': tie_names == ['Z', 'A', 'M'],
'unsorted_insertions_are_popped_by_integer_timestamp': ordered_times == [1, 6, 9],
'lazy_cancellation_removes_active_token_but_keeps_heap_tombstone': before == {'pending': 4, 'physical_heap': 5} and all(item['token'] != 3 for item in first + last),
'cancel_pending_once_then_unknown_or_repeated_cancel_returns_false': cancel_first is True and canceled_again is False and scheduler.cancel(999) is False,
'already_processed_event_cannot_be_canceled': scheduler.cancel(0) is False,
'cancel_token_rejects_bool_float_string_or_negative': all(rejects(lambda value=value: scheduler.cancel(value), 'token') for value in (True, 0.0, '0', -1)),
'event_and_stop_times_are_exact_nonnegative_integers': invalid_time_result,
'no_event_or_run_stop_may_move_simulation_backwards': rejects(lambda: past.schedule(9, 'S01', 0), 'past_event') and rejects(lambda: past.run_until(9), 'past_stop') and past.now_us == 10,
'event_on_stop_is_included_next_microsecond_remains_pending': [x['time_us'] for x in boundary_trace] == [10] and boundary.pending_count == 1 and boundary.run_until(11)[0]['time_us'] == 11,
'empty_run_advances_virtual_clock_without_waiting': empty_trace == [] and same.now_us == 12,
'new_event_at_current_time_is_allowed_and_processed': same_instant_trace == [{'time_us': 12, 'token': 0, 'sensor_id': 'S01', 'millivolts': 0}],
'separate_horizon_runs_equal_one_full_run_for_same_schedule': split_trace == whole_trace == expected_first + expected_last,
'forty_nine_deterministic_schedules_match_independent_filter_sortedlist_oracle': reference_ok,
'integer_timestamps_above_float_precision_preserve_one_microsecond_order': huge_times == [base, base + 1] and float(base) == float(base + 1),
'sensor_exact_integer_millivolt_range_and_failed_insertions_checked': invalid_fields and invalid_units and bound_tokens == [0, 1] and [x['millivolts'] for x in fields.run_until(0)] == [-5000, 5000],
'all_canceled_events_are_skipped_and_tombstones_eventually_removed': all_cancel_trace == [] and all_cancel.pending_count == 0 and all_cancel.physical_heap_count == 0 and all_cancel.now_us == 9,
}
assert len(checks) == 18 and all(type(value) is bool and value for value in checks.values()), checks
return {'python': platform.python_version(), 'dependencies': 'standard library only',
'scope': 'Sequential preloaded finite discrete-event records in integer microseconds; no callbacks, threads, wall clock, sleeping, real sensor or deadline guarantee',
'fixture_events': [list(item) for item in fixture], 'canceled_token': 3, 'before_run': before,
'trace_through_10_us': first, 'state_after_10_us': after_first,
'remaining_trace_through_20_us': last, 'tie_sensor_order': tie_names,
'large_integer_times': huge_times, 'independent_sorted_list_cases': reference_cases,
'checks': checks}
if __name__ == '__main__':
print(json.dumps(run_example(), ensure_ascii=False, indent=2, allow_nan=False))
مرجع دوم، تابع reference_trace است: روی همان فهرستِ ورودی، اعضای لغوشده و بعد از افق را حذف و با جفت زمان و ترتیب ورود مرتب می‌کند؛ نه heapq را فراخوانی می‌کند و نه از SamplingEvents کمک می‌گیرد. چهل‌ونه برنامهٔ کوچک با طول، زمان و لغو متفاوت با این مرجع مقایسه می‌شوند. این مجموعه یک گروه از هجده گروه بررسی است؛ چهل‌ونه مقایسه را به‌عنوان چهل‌ونه آزمون مستقلِ اضافه گزارش نمی‌کنیم.
{
"python": "3.14.5",
"dependencies": "standard library only",
"scope": "Sequential preloaded finite discrete-event records in integer microseconds; no callbacks, threads, wall clock, sleeping, real sensor or deadline guarantee",
"fixture_events": [
[
0,
"S01",
1000
],
[
10,
"S02",
1100
],
[
10,
"S01",
1200
],
[
5,
"S03",
1300
],
[
20,
"S01",
1400
]
],
"canceled_token": 3,
"before_run": {
"pending": 4,
"physical_heap": 5
},
"trace_through_10_us": [
{
"time_us": 0,
"token": 0,
"sensor_id": "S01",
"millivolts": 1000
},
{
"time_us": 10,
"token": 1,
"sensor_id": "S02",
"millivolts": 1100
},
{
"time_us": 10,
"token": 2,
"sensor_id": "S01",
"millivolts": 1200
}
],
"state_after_10_us": {
"now_us": 10,
"pending": 1
},
"remaining_trace_through_20_us": [
{
"time_us": 20,
"token": 4,
"sensor_id": "S01",
"millivolts": 1400
}
],
"tie_sensor_order": [
"Z",
"A",
"M"
],
"large_integer_times": [
1152921504606846976,
1152921504606846977
],
"independent_sorted_list_cases": 49,
"checks": {
"authored_hand_timeline_through_inclusive_ten_microseconds": true,
"remaining_event_at_twenty_has_original_insertion_token": true,
"same_time_uses_insertion_sequence_not_sensor_alphabetical_order": true,
"unsorted_insertions_are_popped_by_integer_timestamp": true,
"lazy_cancellation_removes_active_token_but_keeps_heap_tombstone": true,
"cancel_pending_once_then_unknown_or_repeated_cancel_returns_false": true,
"already_processed_event_cannot_be_canceled": true,
"cancel_token_rejects_bool_float_string_or_negative": true,
"event_and_stop_times_are_exact_nonnegative_integers": true,
"no_event_or_run_stop_may_move_simulation_backwards": true,
"event_on_stop_is_included_next_microsecond_remains_pending": true,
"empty_run_advances_virtual_clock_without_waiting": true,
"new_event_at_current_time_is_allowed_and_processed": true,
"separate_horizon_runs_equal_one_full_run_for_same_schedule": true,
"forty_nine_deterministic_schedules_match_independent_filter_sortedlist_oracle": true,
"integer_timestamps_above_float_precision_preserve_one_microsecond_order": true,
"sensor_exact_integer_millivolt_range_and_failed_insertions_checked": true,
"all_canceled_events_are_skipped_and_tombstones_eventually_removed": true
}
}داده‌های مکمل با نمونهٔ محلی Random و بذر ۷۱ ساخته شده‌اند. راهنمای بازتولید پایتون تغییرپذیری بعضی الگوریتم‌ها میان نسخه‌ها را یادآور می‌شود؛ ثبت نسخه و فایل همچنان لازم است. این تولیدگر ابزار رمزنگاری نیست. برای ذخیرهٔ خروجی انتخاب‌ها می‌توان از گزارش ساختاریافتهٔ رویداد حسگر استفاده کرد؛ خود زمان‌بند هیچ فایل گزارش یا ارتباط شبکه‌ای نمی‌سازد.
۵. چهار پرسش پاسخ‌دار برای آزمون مرزها
پرسش اول: رویداد دقیقاً روی افق توقف چه می‌شود؟ پاسخ: در همان اجرا انتخاب می‌شود، چون مقایسه شامل مساوی است. رویداد یک میکروثانیه بعد منتظر می‌ماند؛ این دو وضعیت در کد جدا آزموده شده‌اند.
پرسش دوم: چرا بعد از لغو یک ورودی، اندازهٔ توده هنوز پنج است؟ پاسخ: فقط عضویت فعال حذف شده است. ورودی لغوشده هنگام رسیدن به ریشه کنار گذاشته می‌شود؛ تعداد فعال چهار است، نه پنج. لغو همهٔ ورودی‌ها و رسیدن به افقشان باید خروجی خالی و تودهٔ خالی بدهد.
پرسش سوم: آیا چند run_until با افق‌های افزایشی، ترتیب را عوض می‌کند؟ پاسخ: برای برنامهٔ ثابت و لغوهای یکسانِ این مثال، اتصال خروجی‌های صفر، هفت، ده و بیست باید با یک اجرای تا بیست برابر باشد. افزودن رویداد تازه میان اجراها، فرضِ برنامهٔ ثابت را تغییر می‌دهد.
پرسش چهارم: آیا عبور هجده بررسی، نمونه‌برداری واقعیِ دقیق را اثبات می‌کند؟ پاسخ: خیر؛ زمان صحیح و ترتیب قطعی، قرارداد همین مدل ترتیبی‌اند. تأخیر سیستم، زمان اجرای عملیات، رقابت نخ‌ها و ساعت سخت‌افزار مدل نشده‌اند. منابع در ۲۴ مهر ۱۴۰۵ خوانده شده‌اند؛ به‌روزرسانی مستنداتِ جاری، این آموزش پایه را به خبر پژوهشی تازه تبدیل نمی‌کند.
منابع و مطالعهٔ بیشتر
- Python 3.14؛ heapq، ترتیب صف اولویت و حذف تنبل
- Python 3.14؛ sched و تفاوت زمان‌بندی با تضمین موعد واقعی
- Python 3.14؛ اعداد صحیح و نوع‌های عددی
- Python 3.14؛ نمونهٔ مستقل Random و محدودیت بازتولید بین نسخه‌ها
بررسی منابع: ۲۴ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.
