SimPy网络仿真中同时间戳数据包碰撞检测问题求助

问题描述 #

用SimPy进行网络仿真时，数据包流程为：Generator类（含1.5s延迟）→Slot类（含1.5s延迟）→Wire类→目的地。需要检测来自不同Slot、时间戳相同的数据包在Wire类中的碰撞，但SimPy按事件顺序执行，若两个数据包时间戳均为3，Wire会先处理第一个数据包，再接收第二个，导致无法检测到碰撞。此前通过在Wire中添加1.5s延迟解决了问题，但会给所有数据包带来不必要的传播延迟，希望找到其他可行方案。

相关代码片段 #

Slot类代码 #

class Slot:
    # 省略其他初始化代码
    def run(self):
        yield self.env.timeout(slot_duration)
        # 省略中间逻辑
        self.out.put(packet)  # 将数据包发送到Wire（out属性）

原Wire类代码片段 #

class Wire:
    # 省略初始化代码
    def run(self):
        while True:
            print("before packet yield and wire yield: ",self.env.now)
    
            if first_event:
                yield self.env.timeout(1.5 + self.epsilon)  # 为第一个事件等待额外时间
                print("End of first yield in wire for {} until {}".format(1.5 + self.epsilon, self.env.now))
                first_event = False  # 第一个事件后重置标志
            else:
                yield self.env.timeout(1.5)  # 后续事件等待1.5s
                print("End of yield in wire until {}".format(self.env.now))
    
            num_items_in_store = len(self.store.items)
            print("Number of items in store:", num_items_in_store)
            packet = self.store.items[0]
            packet_store_object = self.store.get() # 未使用yield获取，存在问题
    
            colliding_packets = [] # 重置碰撞列表
            print("poped packet from store:", packet)
    
            if len(self.store.items) >= 1:
                print("now current_time checking store for collision: ",self.env.now)
                # 根据数据包进入Wire的时间检测碰撞
                colliding_packets = [p for p in self.store.items if p.current_time == packet.current_time]

优化方案：使用极小时间窗口收集同时间数据包 #

不需要引入固定的传播延迟，只需在Wire收到第一个数据包后，等待一个极小的时间窗口（如ε=0.001s），让所有同一时间戳的数据包有机会被存入Wire的Store，再统一检测碰撞。

修改后的Wire类示例代码 #

import simpy

class Wire:
    def __init__(self, env, epsilon=0.001):
        self.env = env
        self.store = simpy.Store(env)
        self.epsilon = epsilon
        # 启动Wire的处理流程
        self.env.process(self.run())

    def run(self):
        while True:
            # 等待接收第一个数据包
            current_packet = yield self.store.get()
            # 等待极小窗口，让同时间的其他数据包完成入队
            yield self.env.timeout(self.epsilon)
            
            # 收集所有与当前数据包时间戳相同的碰撞包
            colliding_packets = [p for p in self.store.items if p.current_time == current_packet.current_time]
            colliding_packets.append(current_packet)
            
            # 处理碰撞逻辑
            if len(colliding_packets) > 1:
                print(f"[{self.env.now}] 检测到碰撞！涉及数据包：{[p.id for p in colliding_packets]}")
                # 移除所有碰撞包（根据需求调整处理逻辑，比如重传）
                for pkt in colliding_packets:
                    if pkt in self.store.items:
                        self.store.items.remove(pkt)
            else:
                print(f"[{self.env.now}] 数据包{current_packet.id}无碰撞，正常传输")
                # 此处添加数据包转发到目的地的逻辑
                # 例如：self.destination.put(current_packet)

方案原理 #

SimPy的事件队列按时间戳排序，同一时间戳的事件按提交顺序执行。当Wire收到第一个数据包后，yield self.env.timeout(self.epsilon)会生成一个时间戳为当前时间+ε的事件，而所有同一原始时间戳的数据包put事件仍处于当前时间点，会优先执行并完成入队。等极小窗口超时后，Wire就能收集到所有同时间到达的数据包，从而准确检测碰撞，且不会引入额外的传播延迟。

内容的提问来源于stack exchange，提问作者Hanna