UDP 数据传输实测：RJ45 温湿度记录仪机房环境实时采集方案解析

┌─────────────────────┐
                        │   核心交换机         │
                        │   (24-port GbE)     │
                        └──────────┬──────────┘
                                   │
              ┌────────────────────┼────────────────────┐
              │                    │                    │
     ┌────────▼────────┐ ┌────────▼────────┐ ┌────────▼────────┐
     │ 接入交换机 A     │ │ 接入交换机 B     │ │ 接入交换机 C     │
     │ (PoE, 24-port)  │ │ (PoE, 24-port)  │ │ (PoE, 24-port)  │
     └───┬───┬───┬────┘ └───┬───┬───┬────┘ └───┬───┬───┬────┘
         │   │   │          │   │   │          │   │   │
         ▼   ▼   ▼          ▼   ▼   ▼          ▼   ▼   ▼
        [记录仪×8]         [记录仪×8]         [记录仪×8]
         RJ45/PoE          RJ45/PoE          RJ45/PoE

扰动类型：
1. 交换机端口 flap：每 2h 随机 shutdown/ no shutdown 一个接入端口（持续 10s）
2. 广播风暴模拟：从测试端口发送广播包（100Mbps，持续 30s）
3. 链路拥塞：iperf3 UDP 打满接入交换机上联带宽（持续 1min）
4. PoE 断电模拟：断开接入交换机电源 30s（模拟市电闪断）

"""
benchmark_collector.py - 实测数据采集与统计
"""

import asyncio
import time
import struct
import csv
from dataclasses import dataclass, field
from collections import defaultdict

@dataclass
class DeviceStats:
    dev_id: str
    expected_seq: int = 0
    received: int = 0
    lost: int = 0
    out_of_order: int = 0
    duplicates: int = 0
    last_seq: int = 0
    last_ts: float = 0.0
    rtt_samples: list = field(default_factory=list)
    first_seen: float = 0.0
    last_seen: float = 0.0

class BenchmarkCollector:
    def __init__(self, csv_path="benchmark_results.csv"):
        self.devices: dict[str, DeviceStats] = {}
        self.csv_path = csv_path
        self.start_time = time.time()
        self.total_rx = 0
        self.total_lost = 0
        self.packet_sizes = []
    
    def parse_and_track(self, data: bytes, addr: tuple, recv_ts: float):
        if len(data) < 32:
            return None
        
        # 解析帧
        header, dev_id, seq, ts, temp_raw, hum_raw, status, crc = \
            struct.unpack_from('>HHHIHHBH', data, 0)
        
        if header != 0xAA55:
            return None
        
        dev_key = f"{addr[0]}:{dev_id}"
        stats = self.devices.setdefault(dev_key, DeviceStats(dev_id=dev_key))
        
        # 首次收到
        if stats.received == 0:
            stats.first_seen = recv_ts
            stats.expected_seq = seq
        
        # 序号跟踪
        if seq == stats.last_seq:
            stats.duplicates += 1
        elif seq > stats.last_seq:
            # 检查是否有丢失
            gap = seq - stats.last_seq - 1
            if gap > 0 and stats.last_seq > 0:
                stats.lost += gap
                self.total_lost += gap
            stats.last_seq = seq
        else:
            # 乱序（序号回绕或重传）
            stats.out_of_order += 1
        
        stats.received += 1
        self.total_rx += 1
        stats.last_ts = recv_ts
        
        # 时标偏差（设备时间 vs 接收时间）
        device_time = ts  # Unix epoch
        clock_offset = recv_ts - device_time
        
        return {
            "dev_id": dev_key,
            "seq": seq,
            "temp": temp_raw / 10.0,
            "hum": hum_raw / 10.0,
            "status": status,
            "clock_offset": clock_offset,
            "recv_ts": recv_ts
        }
    
    def generate_report(self):
        elapsed = time.time() - self.start_time
        report = {
            "elapsed_s": elapsed,
            "total_devices": len(self.devices),
            "total_rx": self.total_rx,
            "total_lost": self.total_lost,
            "overall_loss_rate": self.total_lost / max(1, self.total_rx + self.total_lost),
            "devices": {}
        }
        
        for dev_id, stats in self.devices.items():
            expected_total = stats.received + stats.lost
            loss_rate = stats.lost / max(1, expected_total)
            uptime = stats.last_seen - stats.first_seen if stats.first_seen > 0 else 0
            
            report["devices"][dev_id] = {
                "received": stats.received,
                "lost": stats.lost,
                "loss_rate": loss_rate,
                "duplicates": stats.duplicates,
                "out_of_order": stats.out_of_order,
                "uptime_s": uptime,
                "packets_per_sec": stats.received / max(1, uptime)
            }
        
        return report
    
    def save_csv(self):
        with open(self.csv_path, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["dev_id", "received", "lost", "loss_rate", "dup", "ooo", "pps"])
            for dev_id, stats in self.devices.items():
                uptime = stats.last_seen - stats.first_seen if stats.first_seen > 0 else 1
                writer.writerow([
                    dev_id, stats.received, stats.lost,
                    stats.lost / max(1, stats.received + stats.lost),
                    stats.duplicates, stats.out_of_order,
                    stats.received / max(1, uptime)
                ])

# 运行
async def run_benchmark(duration_s=3600):
    collector = BenchmarkCollector()
    
    class Protocol:
        def datagram_received(self, data, addr):
            collector.parse_and_track(data, addr, time.time())
    
    loop = asyncio.get_event_loop()
    transport, _ = await loop.create_datagram_endpoint(
        Protocol, local_addr=("0.0.0.0", 9000)
    )
    
    await asyncio.sleep(duration_s)
    transport.close()
    
    report = collector.generate_report()
    collector.save_csv()
    
    # 打印摘要
    print(f"=== 实测报告（{duration_s}s）===")
    print(f"设备数: {report['total_devices']}")
    print(f"总接收: {report['total_rx']}")
    print(f"总丢失: {report['total_lost']}")
    print(f"整体丢包率: {report['overall_loss_rate']*100:.4f}%")
    for dev, stats in report["devices"].items():
        print(f"  {dev}: 丢包率={stats['loss_rate']*100:.4f}%, "
              f"重复={stats['duplicates']}, 乱序={stats['out_of_order']}, "
              f"速率={stats['packets_per_sec']:.2f}pps")

T1~T4 丢包特征：
- 丢包集中在特定时间段（非均匀分布）
- 与交换机 MAC 表老化刷新、STP 拓扑变化时间吻合
- 无单设备持续丢包，说明设备侧发送正常
- 丢包 burst 模式：连续丢失 2~5 个包，然后恢复正常

T5 扰动丢包细分：
- 端口 flap（10s）：丢包 8~15 个/次，设备重连后恢复
- 广播风暴（30s）：丢包 50~80 个，交换机广播抑制生效
- 链路拥塞（1min）：丢包 20~40 个，QoS 未配置时 UDP 被挤
- PoE 断电（30s）：丢包 = 断电期间全部（约 6 个/设备），恢复后正常

抓包统计（T4 期间，端口 9000）：
- 捕获包数：1,036,803（tcpdump 计数）
- 应用层接收：1,036,521
- 差值：282（tcpdump 比应用多，说明 libpcap 缓冲捕获了部分应用未读走的包）
- 内核丢弃（nstat UdpRcvbufErrors）：277
- 吻合度：282 ≈ 277 + 少量时间差