当整个互联网开始堵车

from dataclasses import dataclass


@dataclass
class TcpState:
    cwnd: float = 1.0
    ssthresh: float = 12.0


def simulate_tcp(
    capacity=20,
    rounds=30,
    initial_ssthresh=12,
    loss_mode="timeout"
):
    """
    按RTT聚合的教学版TCP拥塞窗口模拟。

    参数：
        capacity:
            简化路径可以容纳的窗口阈值，单位MSS

        rounds:
            模拟的RTT轮数

        initial_ssthresh:
            初始慢启动阈值

        loss_mode:
            "timeout" -> 超时后cwnd=1
            "fast"    -> 简化快速恢复，cwnd约减半

    不模拟真实ACK、TCP序号、RTO计时、队列及SACK。
    """

    if capacity <= 0:
        raise ValueError("容量阈值必须为正")

    if loss_mode not in ("timeout", "fast"):
        raise ValueError("不支持的loss_mode")

    state = TcpState(
        cwnd=1.0,
        ssthresh=float(initial_ssthresh)
    )

    history = []

    for rtt in range(rounds):

        old_window = state.cwnd
        old_threshold = state.ssthresh

        if old_window > capacity:
            # 教学约定：窗口超过固定阈值，
            # 则触发一次拥塞事件。
            state.ssthresh = max(
                2.0,
                old_window / 2.0
            )

            if loss_mode == "timeout":
                # 超时恢复：重置窗口
                state.cwnd = 1.0
                action = "TIMEOUT"

            else:
                # 简化的窗口减半模型
                state.cwnd = max(
                    1.0,
                    old_window / 2.0
                )
                action = "FAST_LOSS"

        elif old_window < state.ssthresh:
            # 慢启动：一轮近似翻倍
            state.cwnd = min(
                old_window * 2.0,
                state.ssthresh
            )
            action = "SLOW_START"

        else:
            # 拥塞避免：每轮近似加1 MSS
            state.cwnd = old_window + 1.0
            action = "CONGESTION_AVOIDANCE"

        history.append({
            "rtt": rtt,
            "cwnd": old_window,
            "ssthresh": old_threshold,
            "action": action,
            "next_cwnd": state.cwnd,
        })

    return history


history = simulate_tcp(
    capacity=20,
    rounds=30,
    initial_ssthresh=12,
    loss_mode="timeout"
)

print(
    f"{'RTT':>3} "
    f"{'cwnd':>7} "
    f"{'ssthresh':>9} "
    f"{'action':>20} "
    f"{'next':>7}"
)

for row in history:
    print(
        f"{row['rtt']:>3} "
        f"{row['cwnd']:>7.1f} "
        f"{row['ssthresh']:>9.1f} "
        f"{row['action']:>20} "
        f"{row['next_cwnd']:>7.1f}"
    )


# 验证最开始的慢启动
assert history[0]["cwnd"] == 1
assert history[0]["next_cwnd"] == 2

assert history[1]["cwnd"] == 2
assert history[1]["next_cwnd"] == 4

assert history[2]["cwnd"] == 4
assert history[2]["next_cwnd"] == 8

# 验证窗口最终会进入拥塞避免
assert any(
    row["action"] == "CONGESTION_AVOIDANCE"
    for row in history
)

# 验证拥塞后会发生超时恢复
assert any(
    row["action"] == "TIMEOUT"
    for row in history
)

print("全部基础测试通过")

if cwnd > capacity:

def jain_fairness(values):
    """
    Jain公平指数。
    两条连接获得相同资源份额时为1。
    """
    total = sum(values)
    squared = sum(x*x for x in values)

    if squared == 0:
        return 1.0

    return total*total / (len(values)*squared)


def simulate_two_flows(
    initial_a=18.0,
    initial_b=2.0,
    capacity=20.0,
    rounds=40
):
    """
    两条具有相同RTT的AIMD连接。

    不拥塞：A和B各增加1
    拥塞：A和B各乘以0.5

    用于观察公平性变化，不是完整TCP模拟。
    """

    a = float(initial_a)
    b = float(initial_b)

    history = []

    for rtt in range(rounds + 1):

        history.append({
            "rtt": rtt,
            "a": a,
            "b": b,
            "fairness": jain_fairness([a, b]),
        })

        if a+b > capacity:
            a *= 0.5
            b *= 0.5
        else:
            a += 1
            b += 1

    return history


history = simulate_two_flows()

print(
    "初始公平指数:",
    round(history[0]["fairness"], 4)
)

print(
    "最终公平指数:",
    round(history[-1]["fairness"], 4)
)

print()

print("RTT  Window-A  Window-B  Fairness")

for row in history[::5]:
    print(
        f"{row['rtt']:>3} "
        f"{row['a']:>8.2f} "
        f"{row['b']:>8.2f} "
        f"{row['fairness']:>8.4f}"
    )

assert history[-1]["fairness"] > history[0]["fairness"]

print("AIMD公平性实验通过")