ظرفیت تماس هم‌زمان در پایتون؛ شبیه‌سازی پذیرش با بازهٔ نیمه‌باز

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

۱. ظرفیت فرضی و واحد اشغال را روشن کنیم

هر درخواست پذیرفته‌شده در مدل، دقیقاً یک جایگاه همسان می‌گیرد. ظرفیت دو یعنی در هر لحظه حداکثر دو درخواست پذیرفته‌شده فعال باشند. زمان شروع و پایان، از قبل در دادهٔ آموزشی اعلام شده‌اند؛ مدت تماس واقعی را پیش‌بینی نکرده‌ایم. زنگ خوردن، مذاکرهٔ رسانه، انتقال و چند شاخهٔ یک تماس در این مدل نیستند. بنابراین «یک جایگاه» را نباید بدون بررسی، معادل یک کانال یا یک نشست نرم‌افزار دیگر گرفت.

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

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

۲. بازهٔ نیمه‌باز و سیاست روشن برای تساوی

بازهٔ اشغال را به صورت [start_ms,end_ms) تعریف می‌کنیم: لحظهٔ شروع داخل است و پایان داخل نیست. پس در زمان ۴۰۰۰، درخواستی که پایانش ۴۰۰۰ است دیگر فعال نیست. برنامه پیش از بررسی هر ورود، همهٔ پایان‌های کوچک‌تر یا مساوی زمان ورود را آزاد می‌کند. اشتباه گرفتن علامت کوچک‌تر با کوچک‌تر یا مساوی، ممکن است تماس بعدی را بی‌دلیل رد کند.

ورودها ابتدا برحسب زمان و سپس شناسه مرتب می‌شوند. در تساوی زمان، ترتیب رشته‌های Unicode شناسه تصمیم می‌گیرد؛ ترتیب قراردادن ردیف‌ها در فایل تعیین‌کننده نیست. این قرارداد برای تکرارپذیری است و عدالت خدمت، اولویت مشتری یا بیشینه کردن تعداد تماس موفق را تضمین نمی‌کند. برای سیاست دیگری باید ترتیب و پاسخ مرجع را با هم تغییر داد؛ تصادفی شدن پنهان ترتیب، سیاست محسوب نمی‌شود.

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

۳. کد کامل با heap و مرجع پیمایش مستقل

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

فایل concurrent_call_capacity.py با Python 3.14.5 و کتابخانهٔ استاندارد اجرا شده است. مرجع کوچک دوم، به جای heap، تمام بازه‌های پذیرفته‌شدهٔ قبلی را پیمایش می‌کند. همچنین اشغال در چند نقطه و مجموع مدت‌ها مستقل کنترل می‌شوند. این مقایسه‌ها همراه پاسخ دستی‌اند؛ توافق دو مسیر مشابه، به تنهایی درستی تعریف مسئله را ثابت نمی‌کند.

"""Offline assumed single-slot call admission; no SIP, network, clocks or PBX."""
import copy
import heapq
import json
import platform


def validate_calls(calls):
    if type(calls) not in (list, tuple):
        raise TypeError('Calls must be an exact list or tuple')
    seen = set()
    for call in calls:
        if type(call) is not tuple or len(call) != 3:
            raise TypeError('Each call must be an exact (id, start_ms, end_ms) tuple')
        identifier, start, end = call
        if type(identifier) is not str or not identifier or identifier.strip() != identifier:
            raise ValueError('A nonempty trimmed string identifier is required')
        if identifier in seen:
            raise ValueError('Call identifiers must be unique')
        seen.add(identifier)
        if type(start) is not int or type(end) is not int:
            raise TypeError('Times must be exact integer milliseconds; bool excluded')
        if not 0 <= start < end:
            raise ValueError('Require 0 <= start_ms < end_ms')


def occupancy(intervals):
    """Integrate half-open intervals using aggregated boundary changes."""
    changes = {}
    for _, start, end in intervals:
        changes[start] = changes.get(start, 0) + 1
        changes[end] = changes.get(end, 0) - 1
    current = peak = area = 0
    previous = 0
    segments = []
    for point in sorted(changes):
        if point > previous:
            segments.append({'start_ms': previous, 'end_ms': point, 'active': current})
            area += (point - previous) * current
        current += changes[point]
        peak = max(peak, current)
        previous = point
    assert current == 0
    return {'peak': peak, 'busy_call_ms': area, 'segments': segments}


def admit(calls, capacity):
    validate_calls(calls)
    if type(capacity) is not int or capacity < 0:
        raise ValueError('Capacity must be an exact nonnegative integer')
    active_ends = []
    accepted = []
    decisions = []
    for call in sorted(calls, key=lambda item: (item[1], item[0])):
        identifier, start, end = call
        # Half-open [start,end): departures at start release their slot first.
        while active_ends and active_ends[0][0] <= start:
            heapq.heappop(active_ends)
        allowed = len(active_ends) < capacity
        if allowed:
            heapq.heappush(active_ends, (end, identifier))
            accepted.append(call)
        decisions.append({'id': identifier, 'start_ms': start, 'end_ms': end,
                          'accepted': allowed, 'active_after_decision': len(active_ends)})
    stats = occupancy(accepted)
    return {'capacity': capacity, 'accepted_ids': [c[0] for c in accepted],
            'denied_ids': [d['id'] for d in decisions if not d['accepted']],
            'decisions': decisions, 'accepted_occupancy': stats,
            'offered_occupancy': occupancy(calls),
            'horizon_ms': max((c[2] for c in calls), default=0)}


def scan_oracle(calls, capacity):
    """Independent tiny-fixture reference: scan admitted intervals, no heap."""
    accepted = []
    for call in sorted(calls, key=lambda item: (item[1], item[0])):
        active_count = sum(start <= call[1] < end for _, start, end in accepted)
        if active_count < capacity:
            accepted.append(call)
    return accepted


def rejects(call):
    try:
        call()
    except (ValueError, TypeError):
        return True
    return False


def run_example():
    calls = [('A', 0, 6000), ('B', 1000, 4000), ('C', 2000, 5000),
             ('D', 4000, 7000), ('E', 4000, 4500), ('F', 6000, 8000),
             ('G', 7000, 9000)]
    before = copy.deepcopy(calls)
    result = admit(calls, 2)
    oracle = scan_oracle(calls, 2)
    by_id = {item['id']: item for item in result['decisions']}
    high = admit(calls, 7)
    zero = admit(calls, 0)
    empty = admit([], 2)
    tie = [('Z', 0, 100), ('A', 0, 100), ('M', 0, 100)]
    chain = [('a', 0, 1), ('b', 1, 2), ('c', 2, 3)]
    probes = (0, 999, 1000, 1999, 2000, 3999, 4000, 4499, 5000,
              5999, 6000, 6999, 7000, 7999, 8000, 8999)
    oracle_counts = [sum(start <= t < end for _, start, end in oracle) for t in probes]
    reported_counts = [next(s['active'] for s in result['accepted_occupancy']['segments']
                            if s['start_ms'] <= t < s['end_ms']) for t in probes]
    bad_calls = [None, {'A': (0, 1)}, [('A', 0, 1), ('A', 1, 2)],
                 [['A', 0, 1]], [('', 0, 1)], [(' A', 0, 1)], [(1, 0, 1)],
                 [('A', -1, 1)], [('A', 1, 1)], [('A', 2, 1)],
                 [('A', False, 1)], [('A', 0, 1.0)], [('A', 0)]]
    checks = {
        'hand_accepted_ids_are_A_B_D_F_G': result['accepted_ids'] == ['A', 'B', 'D', 'F', 'G'],
        'hand_denied_ids_are_C_E': result['denied_ids'] == ['C', 'E'],
        'decisions_follow_start_then_identifier_order': [x['id'] for x in result['decisions']] == list('ABCDEFG'),
        'admitted_peak_is_two_and_never_exceeds_capacity': result['accepted_occupancy']['peak'] == 2
            and all(s['active'] <= 2 for s in result['accepted_occupancy']['segments']),
        'offered_peak_four_is_separate_from_admitted_peak': result['offered_occupancy']['peak'] == 4,
        'hand_busy_area_is_16000_call_milliseconds': result['accepted_occupancy']['busy_call_ms'] == 16000,
        'area_equals_independent_sum_of_admitted_durations': result['accepted_occupancy']['busy_call_ms']
            == sum(end-start for _, start, end in oracle),
        'departure_at_4000_releases_before_D_arrives': by_id['B']['end_ms'] == by_id['D']['start_ms']
            and by_id['D']['accepted'] and by_id['D']['active_after_decision'] == 2,
        'simultaneous_identifier_tie_rule_is_input_order_independent': admit(list(reversed(tie)), 1) == admit(tie, 1)
            and admit(tie, 1)['accepted_ids'] == ['A'],
        'denied_C_does_not_reserve_a_future_slot': not by_id['C']['accepted'] and by_id['D']['accepted']
            and by_id['C']['end_ms'] > by_id['D']['start_ms'],
        'zero_capacity_denies_all_without_occupied_time': zero['accepted_ids'] == []
            and len(zero['denied_ids']) == 7 and zero['accepted_occupancy']['busy_call_ms'] == 0,
        'large_capacity_admits_all_with_hand_offered_area': high['accepted_ids'] == list('ABCDEFG')
            and high['accepted_occupancy']['peak'] == 4 and high['accepted_occupancy']['busy_call_ms'] == 19500,
        'empty_input_has_zero_horizon_and_no_decisions': empty['horizon_ms'] == 0
            and empty['decisions'] == [] and empty['accepted_occupancy']['segments'] == [],
        'adjacent_half_open_intervals_share_one_slot': admit(chain, 1)['accepted_ids'] == ['a', 'b', 'c']
            and admit(chain, 1)['accepted_occupancy']['peak'] == 1,
        'invalid_capacity_types_and_negative_values_rejected': all(rejects(lambda c=c: admit(calls, c))
            for c in (True, -1, 2.0, '2', None)),
        'invalid_identifiers_intervals_and_containers_rejected': all(rejects(lambda c=c: admit(c, 2)) for c in bad_calls),
        'heap_decisions_and_segments_match_scan_oracle': result['accepted_ids'] == [c[0] for c in oracle]
            and oracle_counts == reported_counts == [1, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 1, 1],
        'input_preserved_and_tuple_and_reverse_input_agree': calls == before
            and admit(tuple(calls), 2) == result and admit(list(reversed(calls)), 2) == result,
    }
    assert len(checks) == 18 and all(type(v) is bool for v in checks.values())
    assert all(checks.values()), checks
    return {'python': platform.python_version(),
            'scope': 'Offline authored single-slot admission only; no SIP, PBX, network or load benchmark',
            'contract': {'intervals': '[start_ms,end_ms)', 'time_unit': 'integer milliseconds',
                         'arrival_order': 'start_ms then Unicode string identifier',
                         'departure_before_arrival_at_same_time': True,
                         'denied_requests_never_queue_or_reserve_slots': True},
            'calls': calls, 'capacity_two': result,
            'busy_fraction_of_authored_capacity_horizon': 16000 / (2 * 9000),
            'large_capacity_all_accepted_busy_call_ms': high['accepted_occupancy']['busy_call_ms'],
            'checks': checks}


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

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

۴. پاسخ هفت درخواست و تفاوت قله با اشغال تجمعی

در دادهٔ مسئله، A از صفر تا ۶۰۰۰، B از ۱۰۰۰ تا ۴۰۰۰ و C از ۲۰۰۰ تا ۵۰۰۰ درخواست اشغال دارند. D از ۴۰۰۰ تا ۷۰۰۰، E از ۴۰۰۰ تا ۴۵۰۰، F از ۶۰۰۰ تا ۸۰۰۰ و G از ۷۰۰۰ تا ۹۰۰۰ تعریف شده‌اند. با ظرفیت دو، پاسخ دستی پذیرش A، B، D، F و G و رد C و E است. در زمان ۴۰۰۰، B خارج و D پیش از E بررسی می‌شود.

{
  "python": "3.14.5",
  "scope": "Offline authored single-slot admission only; no SIP, PBX, network or load benchmark",
  "contract": {
    "intervals": "[start_ms,end_ms)",
    "time_unit": "integer milliseconds",
    "arrival_order": "start_ms then Unicode string identifier",
    "departure_before_arrival_at_same_time": true,
    "denied_requests_never_queue_or_reserve_slots": true
  },
  "calls": [
    [
      "A",
      0,
      6000
    ],
    [
      "B",
      1000,
      4000
    ],
    [
      "C",
      2000,
      5000
    ],
    [
      "D",
      4000,
      7000
    ],
    [
      "E",
      4000,
      4500
    ],
    [
      "F",
      6000,
      8000
    ],
    [
      "G",
      7000,
      9000
    ]
  ],
  "capacity_two": {
    "capacity": 2,
    "accepted_ids": [
      "A",
      "B",
      "D",
      "F",
      "G"
    ],
    "denied_ids": [
      "C",
      "E"
    ],
    "decisions": [
      {
        "id": "A",
        "start_ms": 0,
        "end_ms": 6000,
        "accepted": true,
        "active_after_decision": 1
      },
      {
        "id": "B",
        "start_ms": 1000,
        "end_ms": 4000,
        "accepted": true,
        "active_after_decision": 2
      },
      {
        "id": "C",
        "start_ms": 2000,
        "end_ms": 5000,
        "accepted": false,
        "active_after_decision": 2
      },
      {
        "id": "D",
        "start_ms": 4000,
        "end_ms": 7000,
        "accepted": true,
        "active_after_decision": 2
      },
      {
        "id": "E",
        "start_ms": 4000,
        "end_ms": 4500,
        "accepted": false,
        "active_after_decision": 2
      },
      {
        "id": "F",
        "start_ms": 6000,
        "end_ms": 8000,
        "accepted": true,
        "active_after_decision": 2
      },
      {
        "id": "G",
        "start_ms": 7000,
        "end_ms": 9000,
        "accepted": true,
        "active_after_decision": 2
      }
    ],
    "accepted_occupancy": {
      "peak": 2,
      "busy_call_ms": 16000,
      "segments": [
        {
          "start_ms": 0,
          "end_ms": 1000,
          "active": 1
        },
        {
          "start_ms": 1000,
          "end_ms": 4000,
          "active": 2
        },
        {
          "start_ms": 4000,
          "end_ms": 6000,
          "active": 2
        },
        {
          "start_ms": 6000,
          "end_ms": 7000,
          "active": 2
        },
        {
          "start_ms": 7000,
          "end_ms": 8000,
          "active": 2
        },
        {
          "start_ms": 8000,
          "end_ms": 9000,
          "active": 1
        }
      ]
    },
    "offered_occupancy": {
      "peak": 4,
      "busy_call_ms": 19500,
      "segments": [
        {
          "start_ms": 0,
          "end_ms": 1000,
          "active": 1
        },
        {
          "start_ms": 1000,
          "end_ms": 2000,
          "active": 2
        },
        {
          "start_ms": 2000,
          "end_ms": 4000,
          "active": 3
        },
        {
          "start_ms": 4000,
          "end_ms": 4500,
          "active": 4
        },
        {
          "start_ms": 4500,
          "end_ms": 5000,
          "active": 3
        },
        {
          "start_ms": 5000,
          "end_ms": 6000,
          "active": 2
        },
        {
          "start_ms": 6000,
          "end_ms": 7000,
          "active": 2
        },
        {
          "start_ms": 7000,
          "end_ms": 8000,
          "active": 2
        },
        {
          "start_ms": 8000,
          "end_ms": 9000,
          "active": 1
        }
      ]
    },
    "horizon_ms": 9000
  },
  "busy_fraction_of_authored_capacity_horizon": 0.8888888888888888,
  "large_capacity_all_accepted_busy_call_ms": 19500,
  "checks": {
    "hand_accepted_ids_are_A_B_D_F_G": true,
    "hand_denied_ids_are_C_E": true,
    "decisions_follow_start_then_identifier_order": true,
    "admitted_peak_is_two_and_never_exceeds_capacity": true,
    "offered_peak_four_is_separate_from_admitted_peak": true,
    "hand_busy_area_is_16000_call_milliseconds": true,
    "area_equals_independent_sum_of_admitted_durations": true,
    "departure_at_4000_releases_before_D_arrives": true,
    "simultaneous_identifier_tie_rule_is_input_order_independent": true,
    "denied_C_does_not_reserve_a_future_slot": true,
    "zero_capacity_denies_all_without_occupied_time": true,
    "large_capacity_admits_all_with_hand_offered_area": true,
    "empty_input_has_zero_horizon_and_no_decisions": true,
    "adjacent_half_open_intervals_share_one_slot": true,
    "invalid_capacity_types_and_negative_values_rejected": true,
    "invalid_identifiers_intervals_and_containers_rejected": true,
    "heap_decisions_and_segments_match_scan_oracle": true,
    "input_preserved_and_tuple_and_reverse_input_agree": true
  }
}

قلهٔ اشغال پذیرفته‌شده دو است؛ اگر همهٔ بازه‌های درخواستی را بدون محدودیت فعال فرض کنیم، قله چهار می‌شود. این قلهٔ دوم، سناریوی فرضیِ همهٔ درخواست‌هاست، نه بار مشاهده‌شدهٔ سرور. مجموع زمان اشغال پذیرفته‌شده ۱۶۰۰۰ تماس‌میلی‌ثانیه است: مدت‌های ۶۰۰۰، ۳۰۰۰، ۳۰۰۰، ۲۰۰۰ و ۲۰۰۰ را جمع کرده‌ایم. ضرب مدت هر قطعه در شمار فعال همان جواب را می‌دهد.

برای افق انتخاب‌شدهٔ صفر تا ۹۰۰۰، ظرفیت‌زمان دو ضرب‌در ۹۰۰۰ است و سهم اشغال مدل، هشت‌نهم می‌شود. این نسبت به افق وابسته است و مصرف CPU یا کیفیت مکالمه نیست. دو درخواست ردشده از هفت نیز فقط نسبت شمارش همین نمونه‌اند؛ احتمال انسداد شبکه یا نتیجهٔ فرمول ارلانگ نیستند. توزیع ورود، تماس مجدد، هزینهٔ کدک و افت شبکه در این داده وجود ندارند.

۵. چهار تمرین پاسخ‌دار و مرز انتقال به سامانهٔ واقعی

تمرین اول: چرا D در زمان ۴۰۰۰ پذیرفته می‌شود؟ پاسخ: بازهٔ B در همان لحظه تمام شده و دیگر فعال نیست؛ C نیز قبلاً رد شده و جایگاهی رزرو نکرده است. پس تنها A فعال می‌ماند و برای D یک جایگاه آزاد داریم.

تمرین دوم: اگر ترتیب D و E در فایل عوض شود، E پذیرفته می‌شود؟ پاسخ: خیر؛ مرتب‌سازی زمان و شناسه همچنان D را جلو می‌گذارد. تغییر سیاست اولویت، با جابه‌جایی ظاهری ردیف فرق دارد. این ترتیب آموزشی لزوماً سیاست مناسب بهره‌برداری نیست.

تمرین سوم: ظرفیت هفت چه نتیجه‌ای می‌دهد؟ پاسخ: همهٔ درخواست‌ها پذیرفته می‌شوند، ولی قلهٔ واقعی همین مجموعه چهار و مجموع اشغال ۱۹۵۰۰ تماس‌میلی‌ثانیه است. پذیرش همه، به معنی هم‌زمان بودن همهٔ هفت درخواست نیست؛ سه بازهٔ مجاور نیز با ظرفیت یک قابل پذیرش‌اند.

تمرین چهارم: آیا رد ظرفیت در این کد، پاسخ SIP با کد 503 را ثابت می‌کند؟ پاسخ: خیر؛ بخش ۲۱٫۵٫۴ RFC 3261 برای آن پاسخ معنای پروتکلی دارد و تعمیرات یا نگهداری سامانه نیز ممکن است مطرح باشد. این کد پیامی دریافت نکرده است. برای ادامه، معماری تلفن هوشمند و طراحی مسیر تلفن گویا را به نیاز واقعی وصل کنید. منابع در ۲۴ مهر ۱۴۰۵ خوانده شده‌اند؛ آزمون بار و ارزیابی واقعی جدا لازم‌اند.

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

• Python 3.14؛ heapq، کوچک‌ترین عضو و اولویت در تساوی
• مستندات رسمی Asterisk؛ تفکیک سقف تماس و کنترل بار و حافظه
• RFC 3261؛ معنای پاسخ 503 در بخش ۲۱٫۵٫۴

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