用Python实战模拟RDT协议：从1.0到3.0的完整实现与调试技巧

在计算机网络的学习中，可靠数据传输协议（RDT）是一个绕不开的核心概念。但很多人在学习时容易陷入纯理论记忆的困境——记住各种状态转换图却不知道如何落地实现。本文将带你用Python代码完整模拟RDT协议从1.0到3.0的演进过程，通过可运行的代码示例和真实调试经验，让你真正理解协议设计的精妙之处。

1. 环境准备与基础框架搭建

1.1 项目初始化

首先创建一个干净的Python环境（建议3.8+版本），并安装必要的日志模块：

mkdir rdt_simulator && cd rdt_simulator
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate   # Windows

基础代码结构如下：

import logging
import random
from enum import Enum, auto
from dataclasses import dataclass
from typing import Optional

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

class PacketType(Enum):
    DATA = auto()
    ACK = auto()
    NAK = auto()

@dataclass
class Packet:
    seq_num: int
    data: Optional[str]
    checksum: str
    packet_type: PacketType

提示：使用dataclass可以简化数据包的创建和调试输出，Enum则让协议类型更清晰可读

1.2 信道模拟设计

真实网络中的信道需要模拟三种异常情况：

• 比特差错（随机翻转数据位）
• 丢包（随机丢弃数据包）
• 延迟（随机延迟传输）

class NetworkChannel:
    def __init__(self, loss_prob=0.1, corrupt_prob=0.1):
        self.loss_prob = loss_prob
        self.corrupt_prob = corrupt_prob
        
    def transmit(self, packet: Packet) -> Optional[Packet]:
        if random.random() < self.loss_prob:
            logging.warning(f"Packet lost! Seq: {packet.seq_num}")
            return None
            
        if random.random() < self.corrupt_prob:
            corrupted = self._corrupt_packet(packet)
            logging.warning(f"Packet corrupted! Seq: {packet.seq_num}")
            return corrupted
            
        return packet
        
    def _corrupt_packet(self, packet: Packet) -> Packet:
        # 模拟比特翻转
        if packet.data:
            data_list = list(packet.data)
            idx = random.randint(0, len(data_list)-1)
            data_list[idx] = chr(ord(data_list[idx]) ^ 1)
            return Packet(
                seq_num=packet.seq_num,
                data=''.join(data_list),
                checksum='',  # 故意不更新校验和
                packet_type=packet.packet_type
            )
        return packet

2. RDT1.0实现：理想信道假设

2.1 协议设计原理

RDT1.0建立在完美信道的假设上：

• 无比特差错
• 无丢包
• 无延迟

发送方和接收方都只需要单一状态：

• 发送方：等待上层调用 → 发送数据
• 接收方：等待下层调用 → 交付数据

2.2 Python实现代码

class RDTSender1:
    def __init__(self, channel: NetworkChannel):
        self.channel = channel
        
    def rdt_send(self, data: str) -> bool:
        packet = self._make_packet(data)
        self.channel.transmit(packet)
        logging.info(f"Sent data: {data}")
        return True
        
    def _make_packet(self, data: str) -> Packet:
        return Packet(
            seq_num=0,  # 1.0不需要序列号
            data=data,
            checksum=self._checksum(data),
            packet_type=PacketType.DATA
        )
        
    def _checksum(self, data: str) -> str:
        # 简化的校验和计算
        return bin(sum(ord(c) for c in data))[2:]

class RDTReceiver1:
    def rdt_receive(self, packet: Packet) -> Optional[str]:
        if packet.packet_type != PacketType.DATA:
            return None
            
        logging.info(f"Received data: {packet.data}")
        return packet.data

测试用例：

def test_rdt1():
    channel = NetworkChannel(loss_prob=0, corrupt_prob=0)  # 理想信道
    sender = RDTSender1(channel)
    receiver = RDTReceiver1()
    
    data = "Hello RDT1.0!"
    packet = sender._make_packet(data)
    received = receiver.rdt_receive(packet)
    assert received == data

3. RDT2.0实现：处理比特差错

3.1 协议改进要点

RDT2.0引入了：

• 校验和检测（checksum）
• 确认应答（ACK）
• 否定应答（NAK）
• 发送方等待状态

状态机复杂度提升：

• 发送方新增"等待ACK/NAK"状态
• 接收方需要校验数据并返回应答

3.2 关键代码实现

class RDTSender2:
    def __init__(self, channel: NetworkChannel):
        self.channel = channel
        self.current_packet = None
        self.state = "WAIT_CALL"  # 状态机
        
    def rdt_send(self, data: str) -> bool:
        if self.state != "WAIT_CALL":
            return False
            
        packet = self._make_packet(data)
        self.current_packet = packet
        self.channel.transmit(packet)
        self.state = "WAIT_ACK"
        logging.info(f"Sent data, waiting ACK: {data}")
        return True
        
    def handle_response(self, packet: Packet) -> bool:
        if self.state != "WAIT_ACK":
            return False
            
        if packet.packet_type == PacketType.ACK:
            self.state = "WAIT_CALL"
            logging.info("Got ACK, ready for next")
            return True
        elif packet.packet_type == PacketType.NAK:
            logging.warning("Got NAK, resending")
            self.channel.transmit(self.current_packet)
            return False

class RDTReceiver2:
    def __init__(self, channel: NetworkChannel):
        self.channel = channel
        
    def rdt_receive(self, packet: Packet) -> Optional[Packet]:
        if packet.packet_type != PacketType.DATA:
            return None
            
        if self._verify_checksum(packet):
            logging.info(f"Data verified: {packet.data}")
            return self._make_ack()
        else:
            logging.warning("Data corrupted, sending NAK")
            return self._make_nak()
            
    def _verify_checksum(self, packet: Packet) -> bool:
        return packet.checksum == bin(sum(ord(c) for c in packet.data))[2:]

注意：实际实现时需要处理信道丢包情况，这里简化了演示

4. RDT3.0实现：处理丢包问题

4.1 定时器机制设计

RDT3.0的核心改进：

• 发送方增加超时重传定时器
• 每个数据包独立计时
• 接收方需要处理重复数据包

import time
from threading import Timer

class RDTSender3:
    def __init__(self, channel: NetworkChannel, timeout=2.0):
        self.channel = channel
        self.timeout = timeout
        self.timer = None
        self.current_seq = 0
        self.buffer = None
        
    def _start_timer(self):
        self.timer = Timer(self.timeout, self._timeout_handler)
        self.timer.start()
        
    def _timeout_handler(self):
        logging.error(f"Timeout! Resending seq={self.current_seq}")
        self.channel.transmit(self.buffer)
        self._start_timer()
        
    def rdt_send(self, data: str) -> bool:
        packet = Packet(
            seq_num=self.current_seq,
            data=data,
            checksum=self._checksum(data),
            packet_type=PacketType.DATA
        )
        self.buffer = packet
        self.channel.transmit(packet)
        self._start_timer()
        logging.info(f"Sent seq={self.current_seq}, data={data}")
        return True
        
    def handle_ack(self, ack_packet: Packet):
        if ack_packet.seq_num == self.current_seq:
            self.timer.cancel()
            self.current_seq ^= 1  # 切换0/1序列
            logging.info(f"ACK received for seq={ack_packet.seq_num}")

4.2 接收方去重处理

class RDTReceiver3:
    def __init__(self):
        self.expected_seq = 0
        
    def rdt_receive(self, packet: Packet) -> Optional[Packet]:
        if not packet.data or packet.packet_type != PacketType.DATA:
            return None
            
        if packet.seq_num == self.expected_seq:
            logging.info(f"New data received: {packet.data}")
            self.expected_seq ^= 1
            return self._make_ack(packet.seq_num)
        else:
            logging.warning(f"Duplicate packet seq={packet.seq_num}")
            return self._make_ack(packet.seq_num ^ 1)  # 返回上一个ACK

5. 调试技巧与常见问题

5.1 日志分析要点

建议记录以下关键信息：

• 数据包序列号变化
• 状态机转换过程
• 定时器启动/取消事件
• 校验和验证结果

示例日志配置：

def setup_logging():
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    
    # 控制台输出
    ch = logging.StreamHandler()
    ch.setLevel(logging.INFO)
    
    # 文件日志
    fh = logging.FileHandler('rdt_debug.log')
    fh.setLevel(logging.DEBUG)
    
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    ch.setFormatter(formatter)
    fh.setFormatter(formatter)
    
    logger.addHandler(ch)
    logger.addHandler(fh)

5.2 常见问题排查表

问题现象	可能原因	解决方案
发送方无限重传	ACK丢失或超时设置过短	增加超时时间，检查ACK传输
接收方重复处理数据	序列号未正确切换	检查序列号翻转逻辑
校验和总是失败	字符编码处理错误	统一使用UTF-8编码
性能低下	停等协议固有缺陷	考虑实现滑动窗口

5.3 单元测试建议

关键测试场景：

import unittest

class TestRDT3(unittest.TestCase):
    def setUp(self):
        self.channel = NetworkChannel(loss_prob=0.3, corrupt_prob=0.2)
        self.sender = RDTSender3(self.channel)
        self.receiver = RDTReceiver3()
        
    def test_sequence_alternation(self):
        # 测试序列号正确切换
        pass
        
    def test_duplicate_handling(self):
        # 测试重复包处理
        pass
        
    def test_timeout_recovery(self):
        # 测试超时恢复机制
        pass

在实际项目中，我建议先用RDT1.0版本跑通基础流程，然后逐步添加2.0和3.0的特性。每次只测试一个新增功能，可以大大降低调试难度。另外，可视化工具如Wireshark的日志分析模式也很适合用来观察协议数据流。