《互联网产品的 AI 架构·推理性能与成本工程》2.1 队列、优先级与准入

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

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

export type PriorityClass = "interactive" | "standard" | "batch";

/** 数值越小越先被调度。老化只能把等级提到 0，不能造出比交互更高的级别。 */
const CLASS_RANK: Record<PriorityClass, number> = {
  interactive: 0,
  standard: 1,
  batch: 2,
};

export interface SchedulingRequest {
  id: string;
  tenantId: string;
  priority: PriorityClass;
  enqueuedAtMs: number;
  /** 绝对截止时间，等于入队时刻加上业务可接受的等待。 */
  deadlineMs: number;
  estimatedServiceMs: number;
  estimatedCostMicros: number;
}

export interface QueueEntry {
  request: SchedulingRequest;
  sequence: number;
}

export interface PriorityQueueOptions {
  /** 有界容量；满了直接拒绝，而不是无限堆积。 */
  capacity: number;
  /** 每等待多少毫秒提升一级；Infinity 表示关闭老化。 */
  agingStepMs?: number;
  /** 最多提升几级。等级 2 的批量任务最多升到 0，与交互任务同档竞争。 */
  maxAgingLevels?: number;
  /** 对照组：忽略优先级，严格按入队顺序出队。 */
  fifo?: boolean;
}

export interface QueueSnapshot {
  depth: number;
  byClass: Record<PriorityClass, number>;
  oldestWaitMs: number;
}

function requireCount(value: number, name: string): number {
  if (!Number.isInteger(value) || value < 0) {
    throw new AiError("invalid-request", `scheduling: ${name} 必须是非负整数`);
  }
  return value;
}

function validateRequest(request: SchedulingRequest): void {
  if (request.id.trim() === "" || request.tenantId.trim() === "") {
    throw new AiError("invalid-request", "scheduling: id 与 tenantId 不能为空");
  }
  if (!(request.priority in CLASS_RANK)) {
    throw new AiError("invalid-request", `scheduling: 未知优先级 ${String(request.priority)}`);
  }
  requireCount(request.enqueuedAtMs, "enqueuedAtMs");
  requireCount(request.estimatedServiceMs, "estimatedServiceMs");
  requireCount(request.estimatedCostMicros, "estimatedCostMicros");
  if (request.deadlineMs < request.enqueuedAtMs) {
    throw new AiError("invalid-request", "scheduling: deadlineMs 不能早于入队时刻");
  }
}

/**
 * 有效等级 = 基础等级 - 已等待的老化级数。
 * 低优先级等待足够久后会追平高优先级；这是有边界的优先级反转，
 * 用来换取“批量任务不会永远排在队尾”。
 */
export function effectiveRank(
  entry: QueueEntry,
  nowMs: number,
  agingStepMs: number,
  maxAgingLevels: number,
): number {
  const base = CLASS_RANK[entry.request.priority];
  if (!Number.isFinite(agingStepMs) || agingStepMs <= 0) return base;
  const waited = Math.max(0, nowMs - entry.request.enqueuedAtMs);
  const levels = Math.min(maxAgingLevels, Math.floor(waited / agingStepMs));
  return Math.max(0, base - levels);
}

/**
 * 有界优先级队列。出队时扫描全部条目而不是维护堆：
 * 老化会随时间改变顺序，堆只能按固定键排序，扫一遍反而更诚实。
 * 容量是硬上限，常见配置在数千级，线性扫描的成本可以接受。
 */
export class PriorityQueue {
  private readonly capacity: number;
  private readonly agingStepMs: number;
  private readonly maxAgingLevels: number;
  private readonly fifo: boolean;
  private readonly entries: QueueEntry[] = [];
  private sequence = 0;

  constructor(options: PriorityQueueOptions) {
    if (!Number.isInteger(options.capacity) || options.capacity <= 0) {
      throw new AiError("invalid-request", "scheduling: capacity 必须是正整数");
    }
    this.capacity = options.capacity;
    this.agingStepMs = options.agingStepMs ?? Number.POSITIVE_INFINITY;
    this.maxAgingLevels = options.maxAgingLevels ?? 2;
    this.fifo = options.fifo ?? false;
  }

  get depth(): number {
    return this.entries.length;
  }

  enqueue(request: SchedulingRequest): { ok: true; sequence: number } | { ok: false; reason: "queue-full" } {
    validateRequest(request);
    if (this.entries.length >= this.capacity) {
      return { ok: false, reason: "queue-full" };
    }
    const entry: QueueEntry = { request: { ...request }, sequence: this.sequence };
    this.sequence += 1;
    this.entries.push(entry);
    return { ok: true, sequence: entry.sequence };
  }

  /** 取出当前最该执行的请求；同等级按入队顺序先进先出。 */
  dequeue(nowMs: number): SchedulingRequest | undefined {
    if (this.entries.length === 0) return undefined;
    let bestIndex = 0;
    const rankOf = (entry: QueueEntry): number =>
      this.fifo ? 0 : effectiveRank(entry, nowMs, this.agingStepMs, this.maxAgingLevels);
    let bestRank = rankOf(this.entries[0]);
    for (let index = 1; index < this.entries.length; index += 1) {
      const rank = rankOf(this.entries[index]);
      if (
        rank < bestRank ||
        (rank === bestRank && this.entries[index].sequence < this.entries[bestIndex].sequence)
      ) {
        bestRank = rank;
        bestIndex = index;
      }
    }
    const [entry] = this.entries.splice(bestIndex, 1);
    return entry.request;
  }

  snapshot(nowMs: number): QueueSnapshot {
    const byClass: Record<PriorityClass, number> = { interactive: 0, standard: 0, batch: 0 };
    let oldest = 0;
    for (const entry of this.entries) {
      byClass[entry.request.priority] += 1;
      oldest = Math.max(oldest, nowMs - entry.request.enqueuedAtMs);
    }
    return { depth: this.entries.length, byClass, oldestWaitMs: Math.max(0, oldest) };
  }
}

/** 用排队深度与服务时间估算新请求还要等多久，不依赖任何历史分布。 */
export function estimateWaitMs(queueDepth: number, concurrency: number, averageServiceMs: number): number {
  if (queueDepth <= 0) return 0;
  if (!Number.isInteger(concurrency) || concurrency <= 0) {
    throw new AiError("invalid-request", "scheduling: concurrency 必须是正整数");
  }
  return Math.ceil(queueDepth / concurrency) * Math.max(0, averageServiceMs);
}

export type AdmissionRejection = "queue-full" | "budget-exceeded" | "deadline-infeasible";

export interface AdmissionInput {
  request: SchedulingRequest;
  nowMs: number;
  queueDepth: number;
  capacity: number;
  concurrency: number;
  averageServiceMs: number;
  remainingBudgetMicros: number;
}

export interface AdmissionDecision {
  admit: boolean;
  rejection: AdmissionRejection | null;
  estimatedWaitMs: number;
  estimatedCompletionMs: number;
}

/**
 * 准入判据：预算不够、队列已满、或预计完成时间晚于截止时间，都直接拒绝。
 * 拒绝发生在前端，成本只是一次判断；排队后再超时则同时浪费队列槽位与用户等待。
 */
export function decideAdmission(input: AdmissionInput): AdmissionDecision {
  const waitMs = estimateWaitMs(input.queueDepth, input.concurrency, input.averageServiceMs);
  const completionMs = input.nowMs + waitMs + Math.max(0, input.request.estimatedServiceMs);
  const base = { estimatedWaitMs: waitMs, estimatedCompletionMs: completionMs };
  if (input.queueDepth >= input.capacity) {
    return { admit: false, rejection: "queue-full", ...base };
  }
  if (input.request.estimatedCostMicros > input.remainingBudgetMicros) {
    return { admit: false, rejection: "budget-exceeded", ...base };
  }
  if (completionMs > input.request.deadlineMs) {
    return { admit: false, rejection: "deadline-infeasible", ...base };
  }
  return { admit: true, rejection: null, ...base };
}

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

ℹ tests 14
ℹ pass 14
ℹ fail 0