如何在集群训练期间统计多机器间的网络收发字节数？

方案1：基于Linux /proc 文件系统的精准统计（无需额外权限） #

Linux系统下，/proc/net/tcp和/proc/net/udp文件记录了所有TCP/UDP连接的详细统计，包括每个连接的发送/接收字节数。你可以定时读取这些文件，过滤出集群内IP对应的连接，累加统计数据。

实现步骤： #

1. 预先定义集群内所有机器的IP列表，比如CLUSTER_IPS = {"192.168.1.10", "192.168.1.11", "192.168.1.12", "192.168.1.13"}
2. 定时（比如200ms）读取/proc/net/tcp和/proc/net/udp，解析每行的本地IP、远程IP、发送字节数（tx_bytes）、接收字节数（rx_bytes）
3. 只保留远程IP在集群列表内的连接，累加对应方向的字节数

核心代码示例： #

import time
import socket
import struct

CLUSTER_IPS = {"192.168.1.10", "192.168.1.11", "192.168.1.12", "192.168.1.13"}
LOCAL_IP = socket.gethostbyname(socket.gethostname())
training_running = True  # 训练运行标志位，需根据实际训练逻辑设置

def hex_to_ip(hex_ip):
    return socket.inet_ntoa(struct.pack("<L", int(hex_ip, 16)))

def get_cluster_traffic():
    traffic = {ip: {"tx": 0, "rx": 0} for ip in CLUSTER_IPS}
    # 处理TCP连接
    with open("/proc/net/tcp", "r") as f:
        lines = f.readlines()[1:]
        for line in lines:
            parts = line.split()
            local_ip_hex, _ = parts[1].split(":")
            remote_ip_hex, _ = parts[2].split(":")
            remote_ip = hex_to_ip(remote_ip_hex)
            if remote_ip not in CLUSTER_IPS:
                continue
            rx_bytes = int(parts[9])
            tx_bytes = int(parts[10])
            traffic[remote_ip]["tx"] += tx_bytes
            traffic[remote_ip]["rx"] += rx_bytes
    # 处理UDP连接
    with open("/proc/net/udp", "r") as f:
        lines = f.readlines()[1:]
        for line in lines:
            parts = line.split()
            local_ip_hex, _ = parts[1].split(":")
            remote_ip_hex, _ = parts[2].split(":")
            remote_ip = hex_to_ip(remote_ip_hex)
            if remote_ip not in CLUSTER_IPS:
                continue
            rx_bytes = int(parts[9])
            tx_bytes = int(parts[10])
            traffic[remote_ip]["tx"] += tx_bytes
            traffic[remote_ip]["rx"] += rx_bytes
    return traffic

def monitor_thread():
    start_time = time.time()
    prev_traffic = get_cluster_traffic()
    total_traffic = {ip: {"tx": 0, "rx": 0} for ip in CLUSTER_IPS}
    while training_running:
        time.sleep(0.2)
        current_traffic = get_cluster_traffic()
        # 计算增量
        for ip in CLUSTER_IPS:
            total_traffic[ip]["tx"] += current_traffic[ip]["tx"] - prev_traffic[ip]["tx"]
            total_traffic[ip]["rx"] += current_traffic[ip]["rx"] - prev_traffic[ip]["rx"]
        prev_traffic = current_traffic
    duration = time.time() - start_time
    generate_report(total_traffic, duration)

优点： #

• 无需root权限，直接读取系统文件即可
• 精准过滤集群内IP，完全排除外部流量（如SSH）
• 性能开销极低，适合后台长期运行

方案2：修复Scapy权限问题，实现抓包过滤 #

你遇到的PermissionError是因为抓包需要特殊权限，可通过两种方式解决：

权限修复： #

1. 临时方案：用root运行脚本（sudo python your_script.py）
2. 永久方案：给Python二进制文件添加网络权限：

sudo setcap cap_net_raw,cap_net_admin+eip /usr/bin/python3

优化抓包逻辑（只过滤集群内IP）： #

from scapy.all import sniff, IP
import socket
import time

CLUSTER_IPS = {"192.168.1.10", "192.168.1.11", "192.168.1.12", "192.168.1.13"}
LOCAL_IP = socket.gethostbyname(socket.gethostname())
training_running = True
traffic_stats = {ip: {"tx": 0, "rx": 0} for ip in CLUSTER_IPS}

def packet_callback(packet):
    if IP not in packet:
        return
    src_ip = packet[IP].src
    dst_ip = packet[IP].dst
    packet_size = len(packet)
    
    if src_ip == LOCAL_IP and dst_ip in CLUSTER_IPS:
        traffic_stats[dst_ip]["tx"] += packet_size
    elif dst_ip == LOCAL_IP and src_ip in CLUSTER_IPS:
        traffic_stats[src_ip]["rx"] += packet_size

def monitor():
    start_time = time.time()
    # 只捕获集群内IP的数据包，降低性能开销
    filter_expr = f"ip src ({' or '.join(CLUSTER_IPS)}) or ip dst ({' or '.join(CLUSTER_IPS)})"
    sniff(iface="eth0", prn=packet_callback, store=False, filter=filter_expr, stop_filter=lambda x: not training_running)
    duration = time.time() - start_time
    generate_report(traffic_stats, duration)

优点： #

• 能精确统计每个数据包的大小，适合细粒度监控
• 可额外过滤特定协议（如TCP/UDP）

缺点： #

• 需要root权限或调整系统权限
• 高流量场景下，抓包会有一定性能开销

方案3：基于eBPF的高效内核态监控（适合高流量集群） #

如果集群流量极大，抓包或读取/proc的开销不可接受，可以用eBPF工具在内核态直接过滤统计，性能几乎无损耗。

核心示例（用bpfcc）： #

from bcc import BPF
import socket
import struct
import time

CLUSTER_IPS = {"192.168.1.10", "192.168.1.11", "192.168.1.12", "192.168.1.13"}
LOCAL_IP = socket.gethostbyname(socket.gethostname())
LOCAL_IP_INT = struct.unpack("<L", socket.inet_aton(LOCAL_IP))[0]
CLUSTER_IP_INTS = [struct.unpack("<L", socket.inet_aton(ip))[0] for ip in CLUSTER_IPS]
training_running = True

# eBPF内核程序
bpf_text = """
#include <uapi/linux/ip.h>
#include <uapi/linux/skb.h>

BPF_HASH(tx_stats, u32, u64);
BPF_HASH(rx_stats, u32, u64);

int count_packets(struct __sk_buff *skb) {
    struct iphdr *ip = bpf_hdr_pointer(skb);
    u32 src_ip = ip->saddr;
    u32 dst_ip = ip->daddr;
    u64 size = skb->len;
    
    u32 local_ip = LOCAL_IP_INT;
    u32 cluster_ips[] = CLUSTER_IP_INTS;
    int num_ips = sizeof(cluster_ips)/sizeof(cluster_ips[0]);
    
    for (int i=0; i<num_ips; i++) {
        if (src_ip == local_ip && dst_ip == cluster_ips[i]) {
            tx_stats.increment(cluster_ips[i], size);
            return 0;
        }
        if (dst_ip == local_ip && src_ip == cluster_ips[i]) {
            rx_stats.increment(cluster_ips[i], size);
            return 0;
        }
    }
    return 0;
}
"""

# 替换占位符
bpf_text = bpf_text.replace("LOCAL_IP_INT", str(LOCAL_IP_INT))
bpf_text = bpf_text.replace("CLUSTER_IP_INTS", "{" + ", ".join(map(str, CLUSTER_IP_INTS)) + "}")

b = BPF(text=bpf_text)
b.attach_xdp(dev="eth0", fn=b.load_func("count_packets", BPF.XDP))

def get_stats():
    tx = {}
    rx = {}
    for ip_int, count in b["tx_stats"].items():
        ip = socket.inet_ntoa(struct.pack("<L", ip_int.value))
        tx[ip] = count.value
    for ip_int, count in b["rx_stats"].items():
        ip = socket.inet_ntoa(struct.pack("<L", ip_int.value))
        rx[ip] = count.value
    return tx, rx

def monitor():
    start_time = time.time()
    total_tx = {ip:0 for ip in CLUSTER_IPS}
    total_rx = {ip:0 for ip in CLUSTER_IPS}
    while training_running:
        time.sleep(0.2)
        current_tx, current_rx = get_stats()
        for ip in CLUSTER_IPS:
            total_tx[ip] += current_tx.get(ip, 0)
            total_rx[ip] += current_rx.get(ip, 0)
        # 重置统计避免重复累加
        b["tx_stats"].clear()
        b["rx_stats"].clear()
    duration = time.time() - start_time
    generate_report({ip: {"tx": total_tx[ip], "rx": total_rx[ip]} for ip in CLUSTER_IPS}, duration)

优点： #

• 内核态过滤统计，性能开销极小
• 适合高并发、大流量的训练场景

缺点： #

• 需要root权限
• 学习成本较高，需了解eBPF基本语法

生成目标格式的报告 #

无论采用哪种方案，都可以用以下代码生成你需要的表格：

def generate_report(total_traffic, duration):
    machines = sorted(CLUSTER_IPS)
    # 打印表头
    print("+---+" + "+------+------+"*len(machines))
    print("|   |" + "".join([f"      {m}      |" for m in machines]))
    print("+---+" + "+------+------+"*len(machines))
    print("|   |" + "".join(["  TX  |  RX  |" for _ in machines]))
    print("+---+" + "+------+------+"*len(machines))
    
    # 打印每行数据
    for src in machines:
        row = [f"| {src} |"]
        for dst in machines:
            if src == dst:
                row.append("  --  |  --  |")
            else:
                tx = total_traffic[dst]["tx"]
                rx = total_traffic[dst]["rx"]
                row.append(f" {tx:4d} | {rx:4d} |")
        print("".join(row))
        print("+---+" + "+------+------+"*len(machines))
    print(f"\n训练通信时长: {duration:.2f} 秒")

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