《互联网产品的 AI 架构·推理性能与成本工程》2.3 多租户公平调度

从「回声」大客户挤占小客户容量的投诉出发,讨论多租户共享推理容量时的加权公平队列、饥饿防护、租户配额与突发额度。文中给出可运行的 TypeScript WFQ 与令牌桶模块,并在三租户确定性样本上给出无防护与 200ms 防护两种配置的真实运行结果,说明公平份额与最坏等待之间的取舍。

大客户一忙,小客户就排队

「回声」把同一组推理容量卖给了多个业务方。某个大客户做活动,请求量突然涨了十倍;几分钟后,几个小客户陆续投诉:平时几十毫秒的接口,现在要等好几秒。

从网关整体看,系统没有故障:容量跑满,错误率不高,平均延迟只是「有所上升」。但从小客户视角看,他们的请求几乎排在所有大客户请求之后。FIFO 只认到达顺序,不认租户;大客户请求多,自然占满队列。

共享容量必须先回答两个问题:每个租户应该拿到多少,以及某个租户最少能拿到多少。前者是权重,后者是饥饿防护。配额则回答第三个问题:某个租户最多能消耗多少。本节按这个顺序展开,文中数字来自 ai-lab 的确定性公平调度模拟,标注为示例样本上的运行结果。

加权公平队列

最简单的多租户做法是按租户轮询:每个租户轮流取一个请求。它保证每个租户机会均等,但无法表达「大客户买了四倍容量」。加权公平队列(WFQ)把轮询推广成比例分配。

每个租户维护一个虚拟完成时间。请求入队时计算:

finish = max(该租户上次 finish, 全局虚拟时间) + 成本 / 权重

每次出队选择 finish 最小的请求。权重越高,同样的成本让 finish 增长越慢,长期获得的服务量就越多。取 max 的作用是防止空闲租户攒下大量信用:一个租户停了一阵再回来,不能凭「以前没用满」一次性插到所有人前面。

WFQ 的公平性是长期的、按成本的。它不保证每个请求的延迟,只保证在持续积压时,各租户的服务量收敛到权重比例。

饥饿防护不是增加容量

权重 8:2:1 意味着低权重租户长期只拿到约 9% 的容量。如果它的请求还很少,单个请求可能要等很多轮才被服务。等待超过阈值后,调度器可以显式覆盖 WFQ 顺序,让「被相对亏待最久」的请求先走。

选择哪个超时请求也有讲究。按到达先后挑,高权重租户永远有更老的请求挡在前面,低权重租户仍然排不上。这里用「等待时长 / 权重」排序:权重 1 的租户等 200ms,和权重 50 的租户等 200ms,前者的相对等待是后者的 50 倍,应当先走。

必须说清楚它的代价:饥饿防护不创造容量,只重新分配等待。下面的实验里,开启防护后低权重租户的最坏等待显著下降,但高权重租户的最坏等待上升,总平均等待几乎不变。防护阈值设得越低,比例公平被破坏得越多。

配额与突发额度

权重解决「怎么分」,配额解决「最多用多少」。配额用令牌桶实现:桶容量是突发额度,补充速率是持续配额。请求入队前先扣令牌,扣不动就拒绝,并返回补满所需的时间。

只有突发没有补充,额度用完就永久拒绝;只有补充没有突发,任何正常峰值都会被误杀。两个参数要一起定:突发额度覆盖短促高峰,补充速率对齐合同里的长期配额。

配额和公平是两件事。配额超限时请求根本不该进队列;公平只在已经进队的请求之间分配顺序。把两者混成一个「限流」开关,会导致超额租户仍然占着队列位置。

实现

下面的模块包含令牌桶、加权公平队列和饥饿防护。它不访问存储,配额状态在进程内;多实例部署时,配额需要换成共享计数器,但调度算法本身不变。

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

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

export interface TenantPolicy {
  tenantId: string;
  /** 相对权重,正整数。权重 8:2:1 表示长期服务量按此比例分配。 */
  weight: number;
  /** 突发额度上限,单位微元;允许短时间超过持续速率。 */
  burstMicros: number;
  /** 持续补充速率,单位微元/毫秒。 */
  refillMicrosPerMs: number;
}

export interface FairRequest {
  id: string;
  tenantId: string;
  enqueuedAtMs: number;
  serviceMs: number;
  costMicros: number;
}

interface TenantState {
  policy: TenantPolicy;
  tokens: number;
  lastRefillMs: number;
  virtualFinish: number;
  pending: number;
}

interface FairEntry {
  request: FairRequest;
  finish: number;
  sequence: number;
}

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

/**
 * 令牌桶:burst 决定能扛多长的突发,refill 决定长期可持续速率。
 * 只有 burst 没有 refill,突发额度用完就永久拒绝;只有 refill 没有 burst,
 * 任何一个正常峰值都会被误杀。
 */
export class TokenBucket {
  private readonly burst: number;
  private readonly refillPerMs: number;
  private tokens: number;
  private lastRefillMs: number;

  constructor(burstMicros: number, refillMicrosPerMs: number, startMs = 0) {
    if (!Number.isFinite(burstMicros) || burstMicros < 0) {
      throw new AiError("invalid-request", "fair-scheduling: burstMicros 必须是非负有限数");
    }
    if (!Number.isFinite(refillMicrosPerMs) || refillMicrosPerMs < 0) {
      throw new AiError("invalid-request", "fair-scheduling: refillMicrosPerMs 必须是非负有限数");
    }
    this.burst = burstMicros;
    this.refillPerMs = refillMicrosPerMs;
    this.tokens = burstMicros;
    this.lastRefillMs = startMs;
  }

  available(nowMs: number): number {
    if (nowMs > this.lastRefillMs) {
      const elapsed = nowMs - this.lastRefillMs;
      this.tokens = Math.min(this.burst, this.tokens + elapsed * this.refillPerMs);
      this.lastRefillMs = nowMs;
    }
    return this.tokens;
  }

  tryConsume(costMicros: number, nowMs: number): boolean {
    requireCount(costMicros, "costMicros");
    const available = this.available(nowMs);
    if (costMicros > available) return false;
    this.tokens = available - costMicros;
    return true;
  }

  /** 补满到目标成本还差多少毫秒;没有补充能力时返回 Infinity。 */
  retryAfterMs(costMicros: number, nowMs: number): number {
    const deficit = costMicros - this.available(nowMs);
    if (deficit <= 0) return 0;
    if (this.refillPerMs <= 0) return Number.POSITIVE_INFINITY;
    return Math.ceil(deficit / this.refillPerMs);
  }
}

export interface FairSubmitResult {
  ok: boolean;
  reason: "ok" | "unknown-tenant" | "quota-exhausted";
  retryAfterMs: number;
  remainingMicros: number;
}

export interface FairSnapshot {
  pending: number;
  virtualTime: number;
  tenants: Record<string, { pending: number; tokens: number; virtualFinish: number }>;
}

/**
 * 加权公平队列(WFQ)。每个租户维护一个虚拟完成时间:
 * finish = max(该租户上次 finish, 全局虚拟时间) + 成本 / 权重。
 * 每次取 finish 最小者,长期服务量就收敛到权重比例。
 *
 * 饥饿防护是显式覆盖:等待超过 maxWaitMs 的请求优先于 WFQ 顺序。
 * 它牺牲一点点比例公平,换取低权重租户有界的最坏等待。
 */
export class WeightedFairQueue {
  private readonly tenants = new Map<string, TenantState>();
  private readonly entries: FairEntry[] = [];
  private readonly maxWaitMs: number;
  private sequence = 0;
  private virtualTime = 0;
  private guardPromotions = 0;

  constructor(maxWaitMs = Number.POSITIVE_INFINITY) {
    if (!(maxWaitMs > 0)) {
      throw new AiError("invalid-request", "fair-scheduling: maxWaitMs 必须大于 0");
    }
    this.maxWaitMs = maxWaitMs;
  }

  register(policy: TenantPolicy): void {
    if (policy.tenantId.trim() === "") {
      throw new AiError("invalid-request", "fair-scheduling: tenantId 不能为空");
    }
    if (!Number.isInteger(policy.weight) || policy.weight <= 0) {
      throw new AiError("invalid-request", "fair-scheduling: weight 必须是正整数");
    }
    const existing = this.tenants.get(policy.tenantId);
    if (existing !== undefined) {
      existing.policy = { ...policy };
      return;
    }
    this.tenants.set(policy.tenantId, {
      policy: { ...policy },
      tokens: policy.burstMicros,
      lastRefillMs: 0,
      virtualFinish: 0,
      pending: 0,
    });
  }

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

  get promotions(): number {
    return this.guardPromotions;
  }

  /** 先扣配额,再进入 WFQ;配额不足时连队列都不进。 */
  submit(request: FairRequest, nowMs: number): FairSubmitResult {
    const state = this.tenants.get(request.tenantId);
    if (state === undefined) {
      return { ok: false, reason: "unknown-tenant", retryAfterMs: 0, remainingMicros: 0 };
    }
    requireCount(request.costMicros, "costMicros");
    if (nowMs > state.lastRefillMs) {
      const elapsed = nowMs - state.lastRefillMs;
      state.tokens = Math.min(
        state.policy.burstMicros,
        state.tokens + elapsed * state.policy.refillMicrosPerMs,
      );
      state.lastRefillMs = nowMs;
    }
    if (state.tokens < request.costMicros) {
      const deficit = request.costMicros - state.tokens;
      return {
        ok: false,
        reason: "quota-exhausted",
        retryAfterMs:
          state.policy.refillMicrosPerMs <= 0
            ? Number.POSITIVE_INFINITY
            : Math.ceil(deficit / state.policy.refillMicrosPerMs),
        remainingMicros: state.tokens,
      };
    }
    state.tokens -= request.costMicros;

    const start = Math.max(state.virtualFinish, this.virtualTime);
    const costUnits = Math.max(1, request.costMicros);
    const finish = start + costUnits / state.policy.weight;
    state.virtualFinish = finish;
    state.pending += 1;
    this.entries.push({ request: { ...request }, finish, sequence: this.sequence });
    this.sequence += 1;
    return { ok: true, reason: "ok", retryAfterMs: 0, remainingMicros: state.tokens };
  }

  dequeue(nowMs: number): FairRequest | undefined {
    if (this.entries.length === 0) return undefined;
    let bestIndex = 0;
    const overdue = this.entries.findIndex((entry) => nowMs - entry.request.enqueuedAtMs >= this.maxWaitMs);
    if (overdue >= 0) {
      this.guardPromotions += 1;
      // 超时请求之间不能按到达先后挑,否则权重高的租户永远有更老的请求挡在前面。
      // 按“等待时长 / 权重”排序,被相对亏待最久的租户先走。
      const score = (entry: FairEntry): number => {
        const weight = this.tenants.get(entry.request.tenantId)?.policy.weight ?? 1;
        return (nowMs - entry.request.enqueuedAtMs) / weight;
      };
      bestIndex = overdue;
      let bestScore = score(this.entries[overdue]);
      for (let index = overdue + 1; index < this.entries.length; index += 1) {
        const candidate = this.entries[index];
        if (nowMs - candidate.request.enqueuedAtMs < this.maxWaitMs) continue;
        const candidateScore = score(candidate);
        if (
          candidateScore > bestScore ||
          (candidateScore === bestScore && candidate.sequence < this.entries[bestIndex].sequence)
        ) {
          bestScore = candidateScore;
          bestIndex = index;
        }
      }
    } else {
      for (let index = 1; index < this.entries.length; index += 1) {
        const candidate = this.entries[index];
        const current = this.entries[bestIndex];
        if (
          candidate.finish < current.finish ||
          (candidate.finish === current.finish && candidate.sequence < current.sequence)
        ) {
          bestIndex = index;
        }
      }
    }
    const [entry] = this.entries.splice(bestIndex, 1);
    const state = this.tenants.get(entry.request.tenantId);
    if (state !== undefined) {
      state.pending = Math.max(0, state.pending - 1);
      state.virtualFinish = Math.max(state.virtualFinish, entry.finish);
    }
    this.virtualTime = Math.max(this.virtualTime, entry.finish);
    return entry.request;
  }

  snapshot(nowMs: number): FairSnapshot {
    const tenants: FairSnapshot["tenants"] = {};
    for (const [tenantId, state] of this.tenants) {
      if (nowMs > state.lastRefillMs) {
        const elapsed = nowMs - state.lastRefillMs;
        state.tokens = Math.min(
          state.policy.burstMicros,
          state.tokens + elapsed * state.policy.refillMicrosPerMs,
        );
        state.lastRefillMs = nowMs;
      }
      tenants[tenantId] = {
        pending: state.pending,
        tokens: state.tokens,
        virtualFinish: state.virtualFinish,
      };
    }
    return { pending: this.entries.length, virtualTime: this.virtualTime, tenants };
  }
}

WeightedFairQueue.submit 先扣配额再入队,配额不足时连队列都不进。dequeue 先检查是否有请求超过 maxWaitMs,有则按「等待/权重」选择,否则按 finish 最小选择。两个分支都保持确定性,测试可以逐条复现。

测试与运行结果

测试覆盖 WFQ 份额、饥饿防护、令牌桶突发与补充、未知租户与配额拒绝。执行:

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

本机一次运行的真实输出摘要如下,耗时随机器变化:

ℹ tests 14
ℹ pass 14
ℹ fail 0

公平性样本有三个租户,权重 8:2:1;A 每 10ms 到达、B 每 30ms、C 每 150ms,单请求服务 10ms,单槽位容量 100 请求/秒。统计窗口取前 3000ms,避免尾部排空把份额拉平。结果如下,均为示例样本上的运行结果:

配置A 窗口份额B 窗口份额C 窗口份额B 最坏等待A 最坏等待防护提升次数
无防护74.67%18.67%6.67%1720ms960ms0
200ms 防护63.33%30.33%6.33%310ms1210ms363

无防护时 A 与 B 的窗口服务量是 224 与 56,比值正好 4.00,对应 8:2 的权重。C 的份额低于 9.1%,是因为它在窗口内只提交了 20 条请求,需求本身不足,不是被挤占。

开启 200ms 防护后,B 的最坏等待从 1720ms 降到 310ms,代价是 B 的份额从 18.67% 升到 30.33%,A 从 74.67% 降到 63.33%,A 与 B 的比值降到约 2.09。总平均等待仍是 615.7ms,没有变化。这再次说明防护只重新分配等待,不增加容量。

配额的行为可以直接用小样本验证:突发额度 3000 微元、补充速率 10 微元/毫秒时,连续三条 1000 微元的请求被接受,第四条被拒绝并返回 100ms 后重试;等待 100ms 后,同样的请求被接受。

小结

多租户调度要同时回答三个问题:按什么比例分、最少给多少、最多能用多少。WFQ 用虚拟完成时间实现比例公平;饥饿防护用有界覆盖换取低权重租户的最坏等待;令牌桶把长期配额与短时突发分开。三者都只改变「谁先走」,不改变总容量。

下一章从调度转向上下文:同样一次调用,输入越长,成本与延迟如何变化,哪些 token 可以省,哪些省了会伤质量。

阅读导航:上一节:2.2 并发上限与背压 · 下一节:3.1 上下文窗口的成本结构 。

常见错误与边界

  • 用 FIFO 服务多租户。请求多的租户自然占满队列,小租户被饿死。
  • 认为轮询就是公平。轮询忽略权重,也忽略请求成本差异。
  • WFQ 不设虚拟时间下限。空闲租户攒信用,回来后一次性抢占。
  • 按请求数而不是成本计权。长请求会让实际资源分配偏离权重。
  • 饥饿防护阈值设得过低。比例公平被破坏,高权重租户的最坏等待上升。
  • 以为防护能降低总等待。它只重新分配等待,不增加容量。
  • 配额只有突发没有补充。额度用完后的请求永久失败。
  • 配额只有补充没有突发。正常峰值会被误判为超额。
  • 超额请求仍进入队列。队列位置被占,公平调度无从谈起。
  • 单实例内存配额直接上多实例。每个实例各算各的,实际配额被放大数倍。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《互联网产品的 AI 架构·平台化与治理》附录 C 平台化评审清单
  2. 《互联网产品的 AI 架构·平台化与治理》附录 B 能力目录与配额字段速查
  3. 《互联网产品的 AI 架构·平台化与治理》附录 A 治理角色与职责速查