《互联网产品的 AI 架构·推理性能与成本工程》2.2 并发上限与背压

offered = 到达率（请求/秒） × P95 服务时间（毫秒） / 1000

concurrency = ceil(offered / targetUtilization)

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

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

export interface ConcurrencyInput {
  /** 到达率，单位请求/秒。 */
  arrivalRps: number;
  /** 单请求 P95 服务时间，单位毫秒。 */
  p95ServiceMs: number;
  /** 允许的平均利用率，必须小于 1，给抖动留余量。 */
  targetUtilization: number;
  minConcurrency?: number;
  maxConcurrency?: number;
}

export interface ConcurrencyPlan {
  /** 用 Little 定律算出的在途请求数，即 Erlang 负载。 */
  offeredLoad: number;
  /** 在目标利用率下需要的并发槽位。 */
  targetConcurrency: number;
  /** 计划并发相对负载的余量。 */
  headroom: number;
}

/**
 * 并发上限不是拍脑袋的固定值。先算 offered = 到达率 × 服务时间，
 * 再按目标利用率放大；利用率越低，同样的负载需要越多槽位来吸收抖动。
 */
export function deriveConcurrencyLimit(input: ConcurrencyInput): ConcurrencyPlan {
  if (!Number.isFinite(input.arrivalRps) || input.arrivalRps < 0) {
    throw new AiError("invalid-request", "backpressure: arrivalRps 必须是非负有限数");
  }
  if (!Number.isFinite(input.p95ServiceMs) || input.p95ServiceMs <= 0) {
    throw new AiError("invalid-request", "backpressure: p95ServiceMs 必须是正有限数");
  }
  if (!(input.targetUtilization > 0 && input.targetUtilization < 1)) {
    throw new AiError("invalid-request", "backpressure: targetUtilization 必须落在 (0, 1)");
  }
  const min = input.minConcurrency ?? 1;
  const max = input.maxConcurrency ?? Number.MAX_SAFE_INTEGER;
  if (!Number.isInteger(min) || min <= 0 || !Number.isInteger(max) || max < min) {
    throw new AiError("invalid-request", "backpressure: 并发上下界非法");
  }
  const offeredLoad = (input.arrivalRps * input.p95ServiceMs) / 1000;
  const raw = Math.ceil(offeredLoad / input.targetUtilization);
  const targetConcurrency = Math.max(min, Math.min(max, Math.max(1, raw)));
  return { offeredLoad, targetConcurrency, headroom: targetConcurrency - offeredLoad };
}

/**
 * 同步信号量：只限制在途数量，不负责排队策略。
 * 等待队列也不设上限时，信号量会把过载转成无限内存；调用方必须先做准入。
 */
export class Semaphore {
  readonly limit: number;
  private active = 0;
  private readonly waiters: Array<() => void> = [];

  constructor(limit: number) {
    if (!Number.isInteger(limit) || limit <= 0) {
      throw new AiError("invalid-request", "backpressure: 信号量上限必须是正整数");
    }
    this.limit = limit;
  }

  get inFlight(): number {
    return this.active;
  }

  get waiting(): number {
    return this.waiters.length;
  }

  tryAcquire(): boolean {
    if (this.active >= this.limit) return false;
    this.active += 1;
    return true;
  }

  /** 等待一个槽位。等待者按调用顺序唤醒，不区分优先级。 */
  acquire(): Promise<void> {
    if (this.active < this.limit) {
      this.active += 1;
      return Promise.resolve();
    }
    return new Promise<void>((resolve) => {
      this.waiters.push(resolve);
    });
  }

  release(): void {
    if (this.active <= 0) {
      throw new AiError("invalid-request", "backpressure: release 次数多于 acquire");
    }
    const next = this.waiters.shift();
    if (next === undefined) {
      this.active -= 1;
      return;
    }
    // 槽位直接转交给等待者，active 不变，避免出现瞬时超卖。
    next();
  }
}

export type PressureLevel = "normal" | "elevated" | "critical";
export type PressureAction = "accept" | "queue" | "degrade" | "reject";

export interface PressureInput {
  inFlight: number;
  limit: number;
  waiting: number;
  oldestWaitMs: number;
  /** 请求还能接受的最长排队时间。 */
  waitBudgetMs: number;
}

export interface PressureDecision {
  level: PressureLevel;
  action: PressureAction;
  reason: "ok" | "queue-building" | "wait-budget-half" | "wait-budget-exceeded" | "queue-full";
  /** 拒绝时回给上游的 Retry-After，单位毫秒。 */
  retryAfterMs: number;
}

/**
 * 背压、降级、拒绝是三件事：
 * - 背压：允许进入有界队列，同时向上游发出变慢信号；
 * - 降级：不再排队，改走更便宜但仍有结果的路径；
 * - 拒绝：明确失败，避免用户等到超过预算才收到超时。
 */
export function decidePressure(input: PressureInput): PressureDecision {
  if (!Number.isInteger(input.limit) || input.limit <= 0) {
    throw new AiError("invalid-request", "backpressure: limit 必须是正整数");
  }
  const queueFull = input.waiting >= Math.max(8, input.limit * 2);
  const hardLimit = Math.max(1, Math.floor(input.waitBudgetMs / 2));
  if (input.waiting > 0 && (input.oldestWaitMs >= input.waitBudgetMs || queueFull)) {
    return {
      level: "critical",
      action: "reject",
      reason: queueFull ? "queue-full" : "wait-budget-exceeded",
      retryAfterMs: Math.max(1, input.waitBudgetMs),
    };
  }
  if (input.inFlight >= input.limit && input.oldestWaitMs >= hardLimit) {
    return { level: "elevated", action: "degrade", reason: "wait-budget-half", retryAfterMs: 0 };
  }
  if (input.inFlight >= input.limit || input.waiting > 0) {
    return { level: "elevated", action: "queue", reason: "queue-building", retryAfterMs: 0 };
  }
  return { level: "normal", action: "accept", reason: "ok", retryAfterMs: 0 };
}

node --test test/perf/scheduling.test.ts

ℹ tests 14
ℹ pass 14
ℹ fail 0