《Cocos 游戏联机实战》7.1 弱网模拟与网络诊断

// net/weak-net.ts —— 纯 TS，无任何 cc 依赖
import type { Transport } from './transport';

export interface WeakNetConfig {
  latencyMs: number;      // 单程基础延迟
  jitterMs: number;       // 抖动幅度（±）
  lossRate: number;       // 丢包率 0~1
  duplicateRate: number;  // 重复率 0~1
  reorderRate: number;    // 乱序率 0~1（额外等一个基础延迟）
  bandwidthKBps: number;  // 带宽上限，<=0 表示不限
  seed: number;           // 固定随机种子
}

export interface OutageWindow {
  startMs: number;
  durationMs: number;
}

/** mulberry32：32 位状态的小型伪随机数发生器，跨运行完全可复现 */
function mulberry32(seed: number): () => number {
  let a = seed >>> 0;
  return () => {
    a = (a + 0x6d2b79f5) >>> 0;
    let t = a;
    t = Math.imul(t ^ (t >>> 15), t | 1);
    t ^= t + Math.imul(t ^ (t >>> 7), t | 61);
    return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
  };
}

interface Pending {
  at: number;
  seq: number;
  payload: Uint8Array;
}

export class WeakNetworkTransport implements Transport {
  private readonly inner: Transport;
  private readonly cfg: WeakNetConfig;
  private readonly rng: () => number;
  private readonly now: () => number;
  private outbox: Pending[] = [];
  private inbox: Pending[] = [];
  private listeners: Array<(p: Uint8Array) => void> = [];
  private outages: OutageWindow[] = [];
  private seq = 0;
  private nextFreeAt = 0;

  readonly stats = { sent: 0, dropped: 0, delivered: 0, duplicated: 0, reordered: 0 };

  constructor(inner: Transport, cfg: WeakNetConfig, now: () => number) {
    this.inner = inner;
    this.cfg = cfg;
    this.rng = mulberry32(cfg.seed);
    this.now = now;
    inner.onMessage((p) => this.enqueueInbound(p)); // 入站同样过弱网
  }

  send(payload: Uint8Array): void {
    this.stats.sent += 1;
    const t = this.now();
    if (this.inOutage(t) || this.rng() < this.cfg.lossRate) {
      this.stats.dropped += 1;
      return;
    }
    const delay = this.sampleDelay();
    this.outbox.push({ at: this.scheduleAt(payload.byteLength) + delay, seq: this.seq++, payload });
    if (this.rng() < this.cfg.duplicateRate) {
      this.stats.duplicated += 1;
      this.outbox.push({ at: t + delay + 1, seq: this.seq++, payload });
    }
  }

  onMessage(cb: (payload: Uint8Array) => void): void {
    this.listeners.push(cb);
  }

  close(): void {
    this.inner.close();
  }

  /** 由外部时钟驱动：把所有到期消息真正投递出去 */
  advance(): void {
    const t = this.now();
    this.flush(this.outbox, t, (p) => { this.inner.send(p.payload); this.stats.delivered += 1; });
    this.flush(this.inbox, t, (p) => { for (const cb of this.listeners) cb(p.payload); });
  }

  /** 脚本化时间线：在指定时刻插入突发中断 */
  loadTimeline(windows: OutageWindow[]): void {
    this.outages = windows;
  }

  private flush(queue: Pending[], t: number, sink: (p: Pending) => void): void {
    queue.sort((a, b) => a.at - b.at || a.seq - b.seq);
    const keep: Pending[] = [];
    for (const p of queue) {
      if (p.at <= t) sink(p);
      else keep.push(p);
    }
    queue.length = 0;
    queue.push(...keep);
  }

  private sampleDelay(): number {
    const base = this.cfg.latencyMs + (this.rng() * 2 - 1) * this.cfg.jitterMs;
    let extra = 0;
    if (this.rng() < this.cfg.reorderRate) {
      extra = this.cfg.latencyMs;
      this.stats.reordered += 1;
    }
    return Math.max(0, base + extra);
  }

  /** 令牌桶式限速：返回这条消息最早可以离开发送队列的时刻 */
  private scheduleAt(bytes: number): number {
    const rate = this.cfg.bandwidthKBps;
    if (rate <= 0) return this.now();
    const at = Math.max(this.now(), this.nextFreeAt);
    this.nextFreeAt = at + (bytes / 1024 / rate) * 1000;
    return at;
  }

  private enqueueInbound(payload: Uint8Array): void {
    const t = this.now();
    if (this.inOutage(t) || this.rng() < this.cfg.lossRate) {
      this.stats.dropped += 1;
      return;
    }
    this.inbox.push({ at: t + this.sampleDelay(), seq: this.seq++, payload });
  }

  private inOutage(t: number): boolean {
    return this.outages.some((w) => t >= w.startMs && t < w.startMs + w.durationMs);
  }
}

import { describe, expect, it } from 'vitest';
import { WeakNetworkTransport, type WeakNetConfig } from './weak-net';
import type { Transport } from './transport';

class RecordingTransport implements Transport {
  times: number[] = [];
  private readonly now: () => number;
  constructor(now: () => number) { this.now = now; }
  send(): void { this.times.push(this.now()); }
  onMessage(): void {}
  close(): void {}
}

function run(cfg: WeakNetConfig, count: number) {
  let t = 0;
  const now = () => t;
  const inner = new RecordingTransport(now);
  const net = new WeakNetworkTransport(inner, cfg, now);
  for (let i = 0; i < count; i++) {
    net.send(new Uint8Array(16));
    t += 1;
    net.advance();
  }
  // 继续按 1ms 推进，把在途包收干；一次性跳 1000ms 会把到达时刻算偏
  for (let k = 0; k < 200; k++) { t += 1; net.advance(); }
  return { net, inner };
}

const base: WeakNetConfig = {
  latencyMs: 50, jitterMs: 10, lossRate: 0, duplicateRate: 0,
  reorderRate: 0, bandwidthKBps: 0, seed: 7,
};

describe('WeakNetworkTransport', () => {
  it('实测丢包率接近配置值', () => {
    const { net } = run({ ...base, lossRate: 0.3, seed: 42 }, 10000);
    const rate = net.stats.dropped / net.stats.sent;
    expect(rate).toBeGreaterThan(0.28);
    expect(rate).toBeLessThan(0.32);
  });

  it('延迟分布落在设定区间内', () => {
    const { inner } = run(base, 2000);
    const delays = inner.times.map((arriveAt, i) => arriveAt - i); // 第 i 个包在 t=i 发出
    const mean = delays.reduce((a, b) => a + b, 0) / delays.length;
    expect(mean).toBeGreaterThan(48);
    expect(mean).toBeLessThan(52);
    expect(Math.min(...delays)).toBeGreaterThanOrEqual(40);
    expect(Math.max(...delays)).toBeLessThanOrEqual(60);
  });

  it('同一 seed 两次运行结果完全一致', () => {
    const cfg = { ...base, lossRate: 0.2, duplicateRate: 0.05, reorderRate: 0.1, seed: 1234 };
    expect(run(cfg, 500).net.stats).toEqual(run(cfg, 500).net.stats);
  });
});

cd mini2d && npx vitest run net/weak-net.test.ts

import { _decorator, Component, Label, director } from 'cc';
const { ccclass, property } = _decorator;

export interface NetStats {
  rttMs: number;
  jitterMs: number;
  lossRate: number;
  retransmits: number;
  reconcileCount: number;
}

@ccclass('NetDiagOverlay')
export class NetDiagOverlay extends Component {
  @property(Label)
  label: Label | null = null;

  private stats: NetStats | null = null;
  private lastFrame = 0;

  setStats(s: NetStats): void {
    this.stats = s;
  }

  update(): void {
    const frame = director.getTotalFrames();
    if (frame - this.lastFrame < 30 || !this.label || !this.stats) return;
    this.lastFrame = frame;
    const s = this.stats;
    this.label.string =
      `RTT ${s.rttMs.toFixed(0)}ms · jitter ${s.jitterMs.toFixed(0)}ms · ` +
      `loss ${(s.lossRate * 100).toFixed(1)}% · rb ${s.reconcileCount}`;
  }
}