فهرست مقاله
در انتقال دادهٔ مهندسی، یک بار خواندن از جریان بایت الزاماً یک پیام کامل نمی‌دهد. ممکن است نصف سربرگ برسد یا چند پیام پشت سر هم در یک قطعهٔ دریافت قرار گیرند. در این درس، قرارداد کوچک پیشوند طول را می‌نویسیم و آن را با قطعه‌های ساختگی می‌آزماییم. کد هیچ سوکت، سرور یا اتصال شبکه‌ای ایجاد نمی‌کند؛ تمرکز بر مرزبندی پیام در لایهٔ کاربرد است و ادعایی دربارهٔ سرعت، قابلیت اطمینان شبکه یا آماده بودن یک محصول ندارد.
۱. جریان TCP با فهرست پیام‌های کاربرد یکی نیست
بخش تقسیم‌بندی RFC 9293 توضیح می‌دهد که مرز بخش‌های TCP با مرز خواندن و نوشتن دادهٔ کاربرد رابطهٔ تضمین‌شده‌ای ندارد. پس برنامه باید قرارداد پیام خودش را داشته باشد. قطعه‌هایی که در آزمون ما به feed داده می‌شوند نیز فقط برش‌هایی از بایت‌ها هستند؛ آن‌ها را بسته‌های واقعی TCP یا رخدادهای ثبت‌شدهٔ یک شبکه معرفی نمی‌کنیم.
سه محمولهٔ نمونه انتخاب کرده‌ایم: دنبالهٔ هگز 01 00 ff، یک محمولهٔ خالی و دنبالهٔ 02 10. این بایت‌ها برای تمرین جریان دادهٔ حسگر هستند، اما شماره، زمان یا ولتاژ از آن‌ها استخراج نمی‌شود. قالب محتوای حسگر، واحدها و معنای فیلدها باید در قرارداد جدا تعریف شوند. اینجا تنها کمیت عددیِ مربوط به محموله، طول آن بر حسب بایت است.
در Socket Programming HOWTO پایتون، هم دریافت ناقص پیشوند طول و هم خوانده شدن آغاز پیام بعدی مطرح می‌شوند. آزمایش آفلاین ما این دو وضعیت را قابل تکرار می‌کند. ترتیب بایت‌های ورودی حفظ می‌شود و چیزی برای مرتب کردن بسته‌ها نداریم؛ پیش از این مرحله فرض کرده‌ایم قطعه‌ها همان جریان مرتبِ تحویل‌شده به کاربرد را تشکیل می‌دهند.
۲. قرارداد چهار بایت طول و سقف محموله
پیش از هر محموله، یک عدد صحیح بدون علامتِ چهاربایتی قرار می‌دهیم که طول محموله را می‌گوید؛ طول خودِ سربرگ در آن حساب نمی‌شود. قالب !I در struct پایتون اندازهٔ استاندارد و ترتیب بایت شبکه را صریح می‌کند. برای طول سه، چهار بایت آغازین 00 00 00 03 هستند. این سربرگِ قرارداد ماست، نه سربرگ TCP.
سقف پیش‌فرض محموله شانزده بایت است. این عدد را عمداً کوچک انتخاب کرده‌ایم تا مرزها خوانا باشند؛ نه MSS است، نه MTU و نه اندازهٔ مناسب برای همهٔ داده‌های حسگر. طول صفر نیز مجاز است و یک پیام خالی می‌سازد. رمزگذار برای محمولهٔ بزرگ‌تر از سقف خطا می‌دهد؛ داده را بی‌صدا کوتاه نمی‌کند. رمزگشا هم پس از کامل شدن طولِ بیش از سقف، منتظر بدنه نمی‌ماند. در اتصال دو سمت، توافق دربارهٔ طول مجاز اهمیت دارد؛ اگر فرستنده محموله‌ای بزرگ‌تر از سقف گیرنده تولید کند، آن پیام در گیرنده رد می‌شود.
وضعیت ناتمام فقط سربرگ یا محمولهٔ جاری را نگه می‌دارد و کل قطعهٔ ورودی به بافر داخلی افزوده نمی‌شود. با این حال، سقف یک پیام، سقف حافظهٔ کل برنامه نیست: خودِ قطعهٔ ورودی و فهرست پیام‌های کاملِ خروجی می‌توانند بزرگ‌تر باشند. اگر هزاران پیام کوچک در یک feed قرار گیرند، مصرف‌کننده باید تعداد و حجم خروجی را هم مدیریت کند. این کد محدودیت نرخ یا امنیت کامل پروتکل را تأمین نمی‌کند.
۳. کد کامل رمزگشای مرحله‌ای و هجده بررسی
کد را با نام tcp_framing.py ذخیره و دستور python tcp_framing.py را اجرا کنید. اجرای ثبت‌شده با پایتون ۳٫۱۴٫۵ و کتابخانهٔ استاندارد انجام شده است. feed فقط bytes دریافت می‌کند و پیام‌های کاملِ همان فراخوانی را برمی‌گرداند. اگر داده هنوز کافی نباشد، وضعیت لازم برای فراخوانی بعدی حفظ می‌شود؛ انتظار واقعی یا زمان‌سنج در برنامه وجود ندارد.
"""Incremental application framing over synthetic byte chunks; no socket or network."""
import json
import platform
import struct
HEADER = struct.Struct('!I')
MAX_PAYLOAD = 16
PAYLOADS = (b'\x01\x00\xff', b'', b'\x02\x10')
class FrameError(ValueError):
pass
def check_limit(limit):
if type(limit) is not int or not 1 <= limit <= 2**32 - 1:
raise ValueError('Positive uint32 payload limit required')
def encode_frame(payload, limit=MAX_PAYLOAD):
check_limit(limit)
if type(payload) is not bytes:
raise TypeError('Opaque bytes payload required')
if len(payload) > limit:
raise FrameError('Payload exceeds the application limit')
return HEADER.pack(len(payload)) + payload
class FrameDecoder:
def __init__(self, limit=MAX_PAYLOAD):
check_limit(limit)
self.limit = limit
self.header = bytearray()
self.payload = bytearray()
self.expected = None
self.failed = False
self.closed = False
@property
def buffered_bytes(self):
return len(self.header) + len(self.payload)
def _check_open(self):
if self.failed or self.closed:
raise RuntimeError('Decoder failed or finished; create a new instance')
def _fail(self, message):
self.failed = True
self.header.clear()
self.payload.clear()
self.expected = None
raise FrameError(message)
def feed(self, chunk):
self._check_open()
if type(chunk) is not bytes:
raise TypeError('Byte chunk required')
view = memoryview(chunk)
position = 0
messages = []
while position < len(view):
if self.expected is None:
take = min(HEADER.size - len(self.header), len(view) - position)
self.header.extend(view[position:position + take])
position += take
if len(self.header) < HEADER.size:
break
length = HEADER.unpack(self.header)[0]
self.header.clear()
if length > self.limit:
self._fail('Announced payload exceeds the application limit')
if length == 0:
messages.append(b'')
continue
self.expected = length
take = min(self.expected - len(self.payload), len(view) - position)
self.payload.extend(view[position:position + take])
position += take
if len(self.payload) == self.expected:
messages.append(bytes(self.payload))
self.payload.clear()
self.expected = None
return messages
def finish(self):
self._check_open()
if self.header or self.expected is not None:
self._fail('EOF inside a header or payload')
self.closed = True
def rejects(call, exception):
try:
call()
except exception:
return True
return False
def decode_chunks(chunks):
decoder = FrameDecoder()
messages = []
pending_sizes = []
for chunk in chunks:
messages.extend(decoder.feed(chunk))
pending_sizes.append(decoder.buffered_bytes)
decoder.finish()
return tuple(messages), pending_sizes
def run_example():
wire = b''.join(encode_frame(payload) for payload in PAYLOADS)
full, _ = decode_chunks([wire])
bytewise, pending = decode_chunks([wire[i:i+1] for i in range(len(wire))])
splits = [decode_chunks([wire[:cut], wire[cut:]])[0]
for cut in range(len(wire) + 1)]
oversize = FrameDecoder()
oversize_rejected = rejects(lambda: oversize.feed(HEADER.pack(MAX_PAYLOAD + 1)), FrameError)
short_header = FrameDecoder()
short_header.feed(HEADER.pack(3)[:3])
short_body = FrameDecoder()
short_body.feed(encode_frame(PAYLOADS[0])[:-1])
mixed_error = FrameDecoder()
mixed_rejected = rejects(lambda: mixed_error.feed(encode_frame(b'ok') + HEADER.pack(17)), FrameError)
delivered_first = FrameDecoder()
already_delivered = delivered_first.feed(encode_frame(PAYLOADS[0]))
later_rejected = rejects(lambda: delivered_first.feed(HEADER.pack(17)), FrameError)
empty = FrameDecoder()
empty_result = empty.feed(b'')
empty.finish()
type_probe = FrameDecoder()
wrong_chunk_type = rejects(lambda: type_probe.feed('abc'), TypeError)
recovered_after_type_error = type_probe.feed(encode_frame(b'ok'))
type_probe.finish()
checks = {
'network_uint32_header_known_bytes': encode_frame(PAYLOADS[0])[:4].hex() == '00000003',
'coalesced_chunk_decodes_three_messages': full == PAYLOADS,
'one_byte_chunks_preserve_messages': bytewise == PAYLOADS,
'all_eighteen_single_split_positions_match': len(splits) == 18 and all(x == PAYLOADS for x in splits),
'embedded_zero_byte_is_payload_not_terminator': full[0] == b'\x01\x00\xff',
'zero_length_frame_is_an_empty_message': full[1] == b'',
'payload_exactly_at_limit_accepted': decode_chunks([encode_frame(b'x' * 16)])[0] == (b'x' * 16,),
'encoder_rejects_oversized_payload': rejects(lambda: encode_frame(b'x' * 17), FrameError),
'oversize_prefix_rejected_without_waiting_for_body': oversize_rejected and oversize.failed and oversize.buffered_bytes == 0,
'poisoned_decoder_rejects_further_use': rejects(lambda: oversize.feed(wire), RuntimeError) and rejects(oversize.finish, RuntimeError),
'partial_header_eof_rejected': rejects(short_header.finish, FrameError) and short_header.failed,
'partial_payload_eof_rejected': rejects(short_body.finish, FrameError) and short_body.failed,
'empty_feed_does_not_mean_eof': empty_result == [] and empty.closed,
'finished_decoder_rejects_new_data': rejects(lambda: empty.feed(wire), RuntimeError),
'invalid_types_and_limits_rejected': all((
rejects(lambda: encode_frame('abc'), TypeError),
wrong_chunk_type and recovered_after_type_error == [b'ok'],
rejects(lambda: FrameDecoder(True), ValueError),
rejects(lambda: FrameDecoder(0), ValueError))),
'mixed_valid_and_invalid_feed_raises_without_returning_a_list': mixed_rejected and mixed_error.failed,
'previous_successful_feed_is_not_rolled_back': later_rejected and already_delivered == [PAYLOADS[0]],
'observed_pending_state_within_one_header_or_payload': max(pending) <= max(HEADER.size, MAX_PAYLOAD),
}
assert all(checks.values()), checks
return {'python': platform.python_version(), 'model': 'synthetic_incremental_length_prefix_framing',
'header_bytes': HEADER.size, 'header_format': '!I', 'payload_limit_bytes': MAX_PAYLOAD,
'payloads_hex': [payload.hex() for payload in PAYLOADS],
'encoded_stream_bytes': len(wire), 'encoded_stream_hex': wire.hex(),
'single_split_cases': len(splits), 'bytewise_chunk_count': len(wire),
'pending_bytes_after_each_bytewise_feed': pending,
'socket_or_network_used': False, 'checks': checks}
if __name__ == '__main__':
print(json.dumps(run_example(), indent=2))
در این قرارداد، feed(b'') فقط یعنی این بار بایتی اضافه نکرده‌ایم؛ پایان جریان نیست. فراخوانی finish، اعلام صریح پایان ورودیِ آزمون است و تنها در مرز کامل پیام موفق می‌شود. این API آموزشی را با رفتار recv واقعی اشتباه نگیرید؛ دریافت تهی از سوکت معنای پایان دریافت دارد و لایهٔ اتصال باید آن رخداد را به سیاست پایان پروتکل وصل کند.
هر بار ابتدا سربرگ تا چهار بایت کامل می‌شود، سپس بدنه تا طول اعلام‌شده جمع می‌شود. بعد از ساخت پیام، وضعیت بدنه پاک و پردازش بایت‌های باقی‌ماندهٔ همان قطعه ادامه پیدا می‌کند. بنابراین یک feed می‌تواند هیچ، یک یا چند پیام برگرداند. این رفتار با نوشتن حلقه‌ای که هر قطعهٔ دریافت را یک پیام فرض کند متفاوت است؛ آن فرض، وابستگی پنهان به اندازهٔ برش‌ها ایجاد می‌کند.
۴. خروجی اجرا، قطعه‌بندی‌ها و وضعیت خطا
سه پیام، در مجموع پنج بایت محموله و دوازده بایت سربرگ دارند؛ جریان نهایی هفده بایت است. کد هم دریافت همهٔ آن در یک قطعه و هم هفده قطعهٔ تک‌بایتی را بررسی می‌کند. علاوه بر این، همهٔ هجده محل برش تکی، از ابتدا تا انتها، آزموده می‌شوند. این هجده حالتِ برش با هجده بررسی نام‌دار یکی نیستند؛ گزارش، هر دو شمار را جدا نگه می‌دارد.
{
"python": "3.14.5",
"model": "synthetic_incremental_length_prefix_framing",
"header_bytes": 4,
"header_format": "!I",
"payload_limit_bytes": 16,
"payloads_hex": [
"0100ff",
"",
"0210"
],
"encoded_stream_bytes": 17,
"encoded_stream_hex": "000000030100ff00000000000000020210",
"single_split_cases": 18,
"bytewise_chunk_count": 17,
"pending_bytes_after_each_bytewise_feed": [
1,
2,
3,
0,
1,
2,
0,
1,
2,
3,
0,
1,
2,
3,
0,
1,
0
],
"socket_or_network_used": false,
"checks": {
"network_uint32_header_known_bytes": true,
"coalesced_chunk_decodes_three_messages": true,
"one_byte_chunks_preserve_messages": true,
"all_eighteen_single_split_positions_match": true,
"embedded_zero_byte_is_payload_not_terminator": true,
"zero_length_frame_is_an_empty_message": true,
"payload_exactly_at_limit_accepted": true,
"encoder_rejects_oversized_payload": true,
"oversize_prefix_rejected_without_waiting_for_body": true,
"poisoned_decoder_rejects_further_use": true,
"partial_header_eof_rejected": true,
"partial_payload_eof_rejected": true,
"empty_feed_does_not_mean_eof": true,
"finished_decoder_rejects_new_data": true,
"invalid_types_and_limits_rejected": true,
"mixed_valid_and_invalid_feed_raises_without_returning_a_list": true,
"previous_successful_feed_is_not_rolled_back": true,
"observed_pending_state_within_one_header_or_payload": true
}
}اگر طول نامعتبر یا پایان ناقص دیده شود، رمزگشا وارد حالت خطای پایانی می‌شود، وضعیت ناتمام را پاک می‌کند و دیگر قابل ادامه نیست. بازیابی با نمونهٔ تازه انجام می‌شود؛ جست‌وجوی خودکار نقطهٔ همگام‌سازی نداریم. finish موفق نیز نمونه را می‌بندد. در مقابل، خطای نوع آرگومان یک خطای برنامه‌نویسی است و سیاست خراب شدن جریان را فعال نمی‌کند. این تفاوت، بخشی از قرارداد صریح API است.
اگر یک feed ابتدا پیام کامل و سپس طول نامعتبر داشته باشد، استثنا رخ می‌دهد و فهرستی از همان فراخوانی برنمی‌گردد. پیام‌هایی که فراخوانی‌های موفقِ قبلی برگردانده‌اند، پس گرفته نمی‌شوند. این رفتار را هم می‌آزماییم تا مصرف‌کننده، تراکنش عمومی یا بازگردانی خودکار فرض نکند. پس از مرزبندی، اعتبار محتوای رکورد هنوز جداست؛ نوع‌گذاری و اعتبارسنجی دادهٔ حسگر با dataclass آن مسئولیت را توضیح می‌دهد.
۵. چهار تمرین پاسخ‌دار و اتصال مرحله‌های بعدی
تمرین اول: اگر یک بایت از سربرگ برسد، آیا پیام آماده است؟ پاسخ: خیر؛ ابتدا چهار بایتِ طول و سپس بدنهٔ لازم باید کامل شوند. تمرین دوم: بایت صفر داخل محموله چه اثری دارد؟ پاسخ: جداکننده نیست؛ طول صریح داریم و محمولهٔ 01 00 ff بدون حذف آن بازسازی می‌شود.
تمرین سوم: چرا پیام خالی با پایان ناقص فرق دارد؟ پاسخ: پیام خالی یک سربرگ کامل با طول صفر دارد؛ پایان ناقص، سربرگ یا بدنهٔ وعده‌داده‌شده را کامل نکرده است. تمرین چهارم: آیا سقف شانزده بایت یعنی کل برنامه فقط شانزده بایت مصرف می‌کند؟ پاسخ: خیر؛ این فقط سقف محمولهٔ هر پیام است و ورودی، خروجی و ساختارهای دیگر جدا هستند.
پس از شناخت مرز پیام، می‌توان محتوای آن را طبق طرح داده بررسی و سپس وارد صف کرد؛ مقالهٔ صف محدود asyncio برای دادهٔ حسگر مرحلهٔ صف و پس‌فشار را جدا می‌سنجد. منابع این درس در ۲۴ مهر ۱۴۰۵ خوانده شده‌اند؛ مستندات ۳٫۱۴٫۸ و اجرای ۳٫۱۴٫۵ جدا ثبت شده‌اند. هیچ آزمایش سوکت، رشته، رمزنگاری یا کارایی شبکه در این آموزش انجام نشده است.
منابع و مطالعهٔ بیشتر
- Socket Programming HOWTO در Python3.14؛ مرزبندی پیام و دریافت ناقص
- struct در Python3.14؛ اندازه و ترتیب بایت شبکه
- RFC9293؛ سرویس جریان بایتی TCP و جدایی مرز بخش‌ها
بررسی منابع: ۲۴ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.
