《互联网产品的 AI 架构·推理性能与成本工程》7.3 批处理调度与成本模型

限流产能 = 每秒调用数 × 批大小
并发产能 = 并发数 × 批大小 / 单批服务时间

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

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

/** 作业级批处理参数；服务时间口径沿用 1.2，但这里关心的是小时级窗口。 */
export interface BatchPolicy {
  maxBatchSize: number;
  maxWaitMs: number;
  fixedCallMicros: number;
  perItemMicros: number;
  serviceFixedMs: number;
  servicePerItemMs: number;
  callsPerSecond: number;
  concurrency: number;
}

export interface Arrival {
  key: string;
  atMs: number;
}

export interface PlannedBatch {
  items: Arrival[];
  closeAtMs: number;
  serviceMs: number;
}

export interface BatchPlan {
  batches: PlannedBatch[];
  totalCostMicros: number;
}

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

function requireInt(value: number, field: string, min: number): number {
  if (!Number.isInteger(value) || value < min) fail(`${field} must be an integer >= ${min}`);
  return value;
}

function requireNumber(value: number, field: string, min: number): number {
  if (typeof value !== "number" || !Number.isFinite(value) || value < min) {
    fail(`${field} must be a finite number >= ${min}`);
  }
  return value;
}

/** 同一批在“满员”或“窗口到期”时关闭；两个条件都要有，缺一个就会牺牲另一头。 */
export function planBatches(arrivals: readonly Arrival[], policy: BatchPolicy): BatchPlan {
  requireInt(policy.maxBatchSize, "maxBatchSize", 1);
  requireInt(policy.maxWaitMs, "maxWaitMs", 0);
  requireInt(policy.fixedCallMicros, "fixedCallMicros", 0);
  requireInt(policy.perItemMicros, "perItemMicros", 0);
  requireNumber(policy.serviceFixedMs, "serviceFixedMs", 0);
  requireNumber(policy.servicePerItemMs, "servicePerItemMs", 0);
  for (let index = 1; index < arrivals.length; index += 1) {
    if (arrivals[index].atMs < arrivals[index - 1].atMs) fail("arrivals must be sorted by atMs");
  }

  const batches: PlannedBatch[] = [];
  let cursor = 0;
  while (cursor < arrivals.length) {
    const openAtMs = arrivals[cursor].atMs;
    const items: Arrival[] = [];
    let closedBy: "size" | "window" | "end" = "end";
    while (cursor < arrivals.length) {
      if (items.length >= policy.maxBatchSize) {
        closedBy = "size";
        break;
      }
      const next = arrivals[cursor];
      if (items.length > 0 && next.atMs > openAtMs + policy.maxWaitMs) {
        closedBy = "window";
        break;
      }
      items.push(next);
      cursor += 1;
    }
    const lastAtMs = items[items.length - 1].atMs;
    const closeAtMs = closedBy === "window" ? openAtMs + policy.maxWaitMs : lastAtMs;
    const serviceMs = policy.serviceFixedMs + items.length * policy.servicePerItemMs;
    batches.push({ items, closeAtMs, serviceMs });
  }

  return {
    batches,
    totalCostMicros: batches.length * policy.fixedCallMicros + arrivals.length * policy.perItemMicros,
  };
}

export interface BatchStats {
  batches: number;
  items: number;
  averageBatchSize: number;
  p50WaitMs: number;
  p95WaitMs: number;
  maxWaitMs: number;
  unitCostMicros: number;
  totalCostMicros: number;
}

function percentileNearestRank(values: readonly number[], ratio: number): number {
  if (values.length === 0) fail("percentile of empty sample");
  const sorted = [...values].sort((left, right) => left - right);
  const rank = Math.min(sorted.length - 1, Math.max(0, Math.ceil(ratio * sorted.length) - 1));
  return sorted[rank];
}

/** 等待 = 批次关闭时刻 - 条目到达时刻；这是条目在窗口内的排队代价。 */
export function summarizePlan(plan: BatchPlan, windowMs: number): BatchStats {
  if (plan.batches.length === 0) fail("cannot summarize an empty plan");
  requireNumber(windowMs, "windowMs", 1);
  const waits: number[] = [];
  for (const batch of plan.batches) {
    for (const item of batch.items) waits.push(batch.closeAtMs - item.atMs);
  }
  return {
    batches: plan.batches.length,
    items: waits.length,
    averageBatchSize: waits.length / plan.batches.length,
    p50WaitMs: percentileNearestRank(waits, 0.5),
    p95WaitMs: percentileNearestRank(waits, 0.95),
    maxWaitMs: percentileNearestRank(waits, 1),
    unitCostMicros: Math.ceil(plan.totalCostMicros / waits.length),
    totalCostMicros: plan.totalCostMicros,
  };
}

export interface WindowCapacity {
  items: number;
  limitedBy: "rate" | "concurrency";
}

/** 窗口产能取限流与并发两个上限的较小值，不能只按并发算。 */
export function windowCapacity(policy: BatchPolicy, windowMs: number, batchSize: number): WindowCapacity {
  requireInt(batchSize, "batchSize", 1);
  requireNumber(windowMs, "windowMs", 1);
  const serviceMs = policy.serviceFixedMs + batchSize * policy.servicePerItemMs;
  const byRate = policy.callsPerSecond * batchSize;
  const byConcurrency = (policy.concurrency * 1_000 * batchSize) / serviceMs;
  const throughputPerSecond = Math.min(byRate, byConcurrency);
  return {
    items: Math.floor((throughputPerSecond * windowMs) / 1_000),
    limitedBy: byRate <= byConcurrency ? "rate" : "concurrency",
  };
}

/** 单位成本 =（每批固定开销 + 条目成本）/ 批大小；固定开销永远摊不干净。 */
export function unitCostMicros(policy: BatchPolicy, batchSize: number): number {
  requireInt(batchSize, "batchSize", 1);
  return Math.ceil((policy.fixedCallMicros + batchSize * policy.perItemMicros) / batchSize);
}

export interface MissedWindowInput {
  remainingItems: number;
  /** 窗口结束后每小时还能处理多少条。 */
  capacityPerHour: number;
  hoursLeft: number;
  /** 必须当天产出、可以改走在线路径的条数。 */
  criticalItems: number;
  allowLate: boolean;
  maxLateHours: number;
  allowPartialPublish: boolean;
  onlineUnitMicros: number;
  carryOverUnitMicros: number;
}

export interface WindowAction {
  action: "finish" | "extend" | "escalate-online" | "carry-over" | "hold";
  items: number;
  micros: number;
  detail: string;
}

export interface MissedWindowPlan {
  actions: WindowAction[];
  published: number;
  unpublished: number;
  lateHours: number;
  extraMicros: number;
}

/**
 * 错过窗口的降级顺序：先用剩余产能、再申请延期、再升级关键子集、
 * 最后才顺延或扣住。任何一步都不能把旧结果伪装成新结果发布。
 */
export function planMissedWindow(input: MissedWindowInput): MissedWindowPlan {
  requireInt(input.remainingItems, "remainingItems", 0);
  requireInt(input.capacityPerHour, "capacityPerHour", 0);
  requireNumber(input.hoursLeft, "hoursLeft", 0);
  requireInt(input.criticalItems, "criticalItems", 0);
  requireNumber(input.maxLateHours, "maxLateHours", 0);
  requireInt(input.onlineUnitMicros, "onlineUnitMicros", 0);
  requireInt(input.carryOverUnitMicros, "carryOverUnitMicros", 0);

  const capacityLeft = Math.floor(input.capacityPerHour * input.hoursLeft);
  if (input.remainingItems <= capacityLeft) {
    return {
      actions: [{ action: "finish", items: input.remainingItems, micros: 0, detail: "剩余产能足够" }],
      published: input.remainingItems,
      unpublished: 0,
      lateHours: 0,
      extraMicros: 0,
    };
  }

  const shortfall = input.remainingItems - capacityLeft;
  const requiredLateHours = Math.ceil(shortfall / Math.max(1, input.capacityPerHour));
  if (input.allowLate && requiredLateHours <= input.maxLateHours) {
    return {
      actions: [{ action: "extend", items: input.remainingItems, micros: 0, detail: `延期 ${requiredLateHours} 小时` }],
      published: input.remainingItems,
      unpublished: 0,
      lateHours: requiredLateHours,
      extraMicros: 0,
    };
  }

  const critical = Math.min(input.criticalItems, input.remainingItems);
  const carryOver = Math.max(0, input.remainingItems - capacityLeft - critical);
  const actions: WindowAction[] = [];
  if (capacityLeft > 0) {
    actions.push({ action: "finish", items: capacityLeft, micros: 0, detail: "窗口结束前已完成" });
  }
  if (critical > 0) {
    actions.push({
      action: "escalate-online",
      items: critical,
      micros: critical * input.onlineUnitMicros,
      detail: "关键子集改走在线，换时效",
    });
  }
  const carryAction = input.allowPartialPublish ? "carry-over" : "hold";
  if (carryOver > 0) {
    actions.push({
      action: carryAction,
      items: carryOver,
      micros: carryOver * input.carryOverUnitMicros,
      detail: input.allowPartialPublish ? "其余顺延到下一窗口" : "不允许部分发布，整批扣住",
    });
  }

  return {
    actions,
    published: input.allowPartialPublish ? capacityLeft + critical : critical,
    unpublished: input.allowPartialPublish ? carryOver : input.remainingItems - critical,
    lateHours: 0,
    extraMicros: critical * input.onlineUnitMicros + carryOver * input.carryOverUnitMicros,
  };
}

node --test --test-name-pattern="批大小与等待窗口|放大批大小|单条批次|窗口产能|单位成本|剩余产能|允许延期|错过窗口|不允许部分|乱序到达" test/perf/batch.test.ts

ℹ tests 10
ℹ suites 0
ℹ pass 10
ℹ fail 0
ℹ duration_ms 147.464042