《互联网产品的 AI 架构·推理性能与成本工程》1.2 吞吐、并发与 goodput

并发 ≈ 吞吐 × 平均延迟

// file: src/perf/goodput.ts
// tier: L1
// verify: node --test test/perf/goodput.test.ts

import { AiError } from "../core/errors.ts";
import { percentile } from "../observe/baseline.ts";

/** 压测侧的一条请求：到达时刻与输出长度都由负载模型决定。 */
export interface LoadRequest {
  requestId: string;
  arriveAtMs: number;
  outputTokens: number;
}

export interface ServerConfig {
  workers: number;
  /** 每个请求的固定开销（调度、序列化、连接）。 */
  fixedMs: number;
  /** 每输出一个 token 的基础解码耗时。 */
  perTokenMs: number;
  /** 同一批里每多一个请求增加的单步耗时，模拟带宽与注意力竞争。 */
  perSiblingMs: number;
}

export interface BatchPolicy {
  maxBatch: number;
  /** 攒批窗口：从第一个待处理请求起，最多等这么久。 */
  batchWindowMs: number;
}

export interface Completion {
  requestId: string;
  arriveAtMs: number;
  startAtMs: number;
  doneAtMs: number;
  latencyMs: number;
  outputTokens: number;
  batchSize: number;
}

function fail(message: string): never {
  throw new AiError("invalid-request", `goodput: ${message}`);
}

function checkInteger(value: number, label: string, minimum: number): void {
  if (!Number.isInteger(value) || value < minimum) fail(`${label} 必须是不小于 ${minimum} 的整数`);
}

function validateConfig(config: ServerConfig): void {
  checkInteger(config.workers, "workers", 1);
  for (const [label, value] of [
    ["fixedMs", config.fixedMs],
    ["perTokenMs", config.perTokenMs],
    ["perSiblingMs", config.perSiblingMs],
  ] as const) {
    if (!Number.isFinite(value) || value < 0) fail(`${label} 必须是非负有限数`);
  }
}

/**
 * 批服务耗时由三部分组成：固定开销、最长序列的解码、
 * 以及同一批内并发请求带来的额外单步耗时。
 * 用 max 而不是 sum，是因为解码要等最长的那条序列结束。
 */
export function batchServiceMs(
  config: ServerConfig,
  batch: readonly LoadRequest[],
): number {
  if (batch.length === 0) fail("空批次没有服务耗时");
  const longest = Math.max(...batch.map((item) => item.outputTokens));
  return config.fixedMs + config.perTokenMs * longest + config.perSiblingMs * (batch.length - 1);
}

/**
 * 确定性批调度器：固定请求序列 + 固定 worker 数 => 固定完成时刻。
 * 同一个输入重复运行会得到逐条一致的 Completion，便于做可复现基线。
 */
export function simulateBatchScheduler(
  requests: readonly LoadRequest[],
  config: ServerConfig,
  policy: BatchPolicy,
): Completion[] {
  validateConfig(config);
  checkInteger(policy.maxBatch, "maxBatch", 1);
  if (!Number.isFinite(policy.batchWindowMs) || policy.batchWindowMs < 0) {
    fail("batchWindowMs 必须是非负有限数");
  }
  if (requests.length === 0) return [];
  for (const request of requests) {
    if (!Number.isFinite(request.arriveAtMs) || request.arriveAtMs < 0) {
      fail(`${request.requestId}: arriveAtMs 必须是非负有限数`);
    }
    checkInteger(request.outputTokens, `${request.requestId}.outputTokens`, 1);
  }

  const pending = [...requests].sort(
    (left, right) => left.arriveAtMs - right.arriveAtMs || left.requestId.localeCompare(right.requestId),
  );
  const workerFreeAt = Array.from({ length: config.workers }, () => 0);
  const completions: Completion[] = [];
  let cursor = 0;

  while (cursor < pending.length) {
    const openAt = pending[cursor].arriveAtMs;
    const deadline = openAt + policy.batchWindowMs;
    const batch: LoadRequest[] = [];
    while (
      cursor < pending.length &&
      batch.length < policy.maxBatch &&
      (batch.length === 0 || pending[cursor].arriveAtMs <= deadline)
    ) {
      batch.push(pending[cursor]);
      cursor += 1;
    }

    let worker = 0;
    for (let index = 1; index < workerFreeAt.length; index += 1) {
      if (workerFreeAt[index] < workerFreeAt[worker]) worker = index;
    }
    const closeAt = batch[batch.length - 1].arriveAtMs;
    const startAt = Math.max(closeAt, workerFreeAt[worker]);
    const doneAt = startAt + batchServiceMs(config, batch);
    workerFreeAt[worker] = doneAt;

    for (const request of batch) {
      completions.push({
        requestId: request.requestId,
        arriveAtMs: request.arriveAtMs,
        startAtMs: startAt,
        doneAtMs: doneAt,
        latencyMs: doneAt - request.arriveAtMs,
        outputTokens: request.outputTokens,
        batchSize: batch.length,
      });
    }
  }

  return completions.sort(
    (left, right) => left.arriveAtMs - right.arriveAtMs || left.requestId.localeCompare(right.requestId),
  );
}

export interface RunSummary {
  completed: number;
  windowMs: number;
  throughputRps: number;
  tokenThroughput: number;
  metSlo: number;
  goodputRps: number;
  /** 满足延迟 SLO 的请求比例，这才是“有效吞吐”的可比口径。 */
  goodputRatio: number;
  p50LatencyMs: number;
  p95LatencyMs: number;
  peakConcurrency: number;
}

/** 峰值并发：到达 +1、完成 -1；同一时刻先算完成，避免把交接算成两倍。 */
export function peakConcurrency(completions: readonly Completion[]): number {
  const events: Array<{ atMs: number; delta: number }> = [];
  for (const item of completions) {
    events.push({ atMs: item.arriveAtMs, delta: 1 });
    events.push({ atMs: item.doneAtMs, delta: -1 });
  }
  events.sort((left, right) => left.atMs - right.atMs || left.delta - right.delta);
  let current = 0;
  let peak = 0;
  for (const event of events) {
    current += event.delta;
    peak = Math.max(peak, current);
  }
  return peak;
}

/**
 * 统计窗口取「最早到达」到「最晚完成」。
 * 吞吐与 goodput 必须共享同一个窗口，否则两个数字不可比。
 */
export function summarizeRun(completions: readonly Completion[], sloMs: number): RunSummary {
  if (completions.length === 0) fail("没有完成记录，无法汇总");
  if (!Number.isFinite(sloMs) || sloMs <= 0) fail("sloMs 必须是正有限数");
  const firstArrival = Math.min(...completions.map((item) => item.arriveAtMs));
  const lastDone = Math.max(...completions.map((item) => item.doneAtMs));
  const windowMs = Math.max(1, lastDone - firstArrival);
  const met = completions.filter((item) => item.latencyMs <= sloMs);
  const tokens = completions.reduce((sum, item) => sum + item.outputTokens, 0);
  return {
    completed: completions.length,
    windowMs,
    throughputRps: (completions.length * 1000) / windowMs,
    tokenThroughput: (tokens * 1000) / windowMs,
    metSlo: met.length,
    goodputRps: (met.length * 1000) / windowMs,
    goodputRatio: met.length / completions.length,
    p50LatencyMs: percentile(completions.map((item) => item.latencyMs), 0.5),
    p95LatencyMs: percentile(completions.map((item) => item.latencyMs), 0.95),
    peakConcurrency: peakConcurrency(completions),
  };
}

node --test test/perf/goodput.test.ts

ℹ tests 7
ℹ pass 7
ℹ fail 0
ℹ duration_ms 140.059458