8장: 백프레셔와 흐름 제어

[정상 상태]
생산자 (100 req/s) ──> 큐 ──> 소비자 (120 req/s)
                       [여유]
 
[백프레셔 발생]
생산자 (100 req/s) ──> 큐 ──> 소비자 (50 req/s)
                       [넘침!]

시간  | 요청 유입 | 처리 완료 | 큐 깊이
12:00 |    10     |    10     |    0
12:01 |    50     |    10     |   40
12:02 |   100     |    10     |  130
12:03 |   200     |    10     |  320  ← 위험!

class BoundedBuffer<T> {
  private buffer: T[] = [];
  private readonly maxSize: number;
  private readonly highWaterMark: number;
  private readonly lowWaterMark: number;
  private isPaused = false;
 
  constructor(options: {
    maxSize: number;
    highWaterMark?: number;
    lowWaterMark?: number;
  }) {
    this.maxSize = options.maxSize;
    this.highWaterMark = options.highWaterMark ?? options.maxSize * 0.8;
    this.lowWaterMark = options.lowWaterMark ?? options.maxSize * 0.2;
  }
 
  push(item: T): "ok" | "pause" | "reject" {
    if (this.buffer.length >= this.maxSize) {
      return "reject"; // 버퍼 가득 참
    }
 
    this.buffer.push(item);
 
    if (this.buffer.length >= this.highWaterMark) {
      this.isPaused = true;
      return "pause"; // 생산자에게 속도 줄이라는 신호
    }
 
    return "ok";
  }
 
  pull(): T | undefined {
    const item = this.buffer.shift();
 
    if (
      this.isPaused &&
      this.buffer.length <= this.lowWaterMark
    ) {
      this.isPaused = false;
      // 생산자에게 다시 보내도 된다는 신호
    }
 
    return item;
  }
 
  get depth(): number {
    return this.buffer.length;
  }
}

type DropStrategy = "drop-oldest" | "drop-newest" | "drop-random";
 
class DroppingBuffer<T> {
  private buffer: T[] = [];
  private dropped = 0;
 
  constructor(
    private maxSize: number,
    private strategy: DropStrategy
  ) {}
 
  push(item: T) {
    if (this.buffer.length < this.maxSize) {
      this.buffer.push(item);
      return;
    }
 
    // 버퍼가 가득 찬 경우
    this.dropped++;
 
    switch (this.strategy) {
      case "drop-oldest":
        // 가장 오래된 항목 제거, 새 항목 추가
        this.buffer.shift();
        this.buffer.push(item);
        break;
 
      case "drop-newest":
        // 새 항목을 폐기 (버퍼 유지)
        break;
 
      case "drop-random":
        // 랜덤 위치의 항목을 교체
        const idx = Math.floor(Math.random() * this.maxSize);
        this.buffer[idx] = item;
        break;
    }
  }
 
  getDropCount(): number {
    return this.dropped;
  }
}

class LoadShedder {
  private currentLoad = 0;
  private readonly capacity: number;
  private readonly shedThreshold: number;
 
  constructor(capacity: number, shedThreshold = 0.9) {
    this.capacity = capacity;
    this.shedThreshold = shedThreshold;
  }
 
  shouldAccept(priority: "low" | "normal" | "high"): boolean {
    const utilization = this.currentLoad / this.capacity;
 
    // 여유 있으면 모두 수용
    if (utilization < this.shedThreshold) return true;
 
    // 과부하 시 우선순위에 따라 차단
    switch (priority) {
      case "high":
        return utilization < 0.98; // 98%까지 수용
      case "normal":
        return utilization < 0.95; // 95%까지 수용
      case "low":
        return false; // 즉시 차단
    }
  }
 
  async execute<T>(
    priority: "low" | "normal" | "high",
    fn: () => Promise<T>
  ): Promise<T> {
    if (!this.shouldAccept(priority)) {
      throw new Error("Service overloaded. Please retry later.");
    }
 
    this.currentLoad++;
    try {
      return await fn();
    } finally {
      this.currentLoad--;
    }
  }
}

class TokenBucket {
  private tokens: number;
  private lastRefill: number;
 
  constructor(
    private readonly capacity: number,
    private readonly refillRate: number // tokens per second
  ) {
    this.tokens = capacity;
    this.lastRefill = Date.now();
  }
 
  private refill() {
    const now = Date.now();
    const elapsed = (now - this.lastRefill) / 1000;
    this.tokens = Math.min(
      this.capacity,
      this.tokens + elapsed * this.refillRate
    );
    this.lastRefill = now;
  }
 
  tryConsume(count: number = 1): boolean {
    this.refill();
 
    if (this.tokens >= count) {
      this.tokens -= count;
      return true;
    }
 
    return false;
  }
 
  async waitAndConsume(count: number = 1): Promise<void> {
    while (!this.tryConsume(count)) {
      // 필요한 토큰이 채워질 때까지 대기
      const deficit = count - this.tokens;
      const waitMs = (deficit / this.refillRate) * 1000;
      await new Promise((resolve) =>
        setTimeout(resolve, Math.max(waitMs, 10))
      );
    }
  }
 
  get availableTokens(): number {
    this.refill();
    return Math.floor(this.tokens);
  }
}

class LLMRateLimiter {
  private requestBucket: TokenBucket;
  private tokenBucket: TokenBucket;
  private queue: Array<{
    resolve: () => void;
    tokensNeeded: number;
  }> = [];
 
  constructor(config: {
    requestsPerMinute: number;
    tokensPerMinute: number;
    burstMultiplier?: number;
  }) {
    const burst = config.burstMultiplier ?? 1.2;
 
    this.requestBucket = new TokenBucket(
      Math.ceil(config.requestsPerMinute * burst),
      config.requestsPerMinute / 60
    );
 
    this.tokenBucket = new TokenBucket(
      Math.ceil(config.tokensPerMinute * burst),
      config.tokensPerMinute / 60
    );
  }
 
  async acquire(estimatedTokens: number): Promise<void> {
    // 요청 토큰 소비
    await this.requestBucket.waitAndConsume(1);
 
    // 토큰 소비 (입력 토큰 기준 추정)
    await this.tokenBucket.waitAndConsume(estimatedTokens);
  }
 
  reportActualUsage(actualTokens: number, estimatedTokens: number) {
    // 추정치와 실제 사용량의 차이를 보정
    const diff = actualTokens - estimatedTokens;
    if (diff > 0) {
      // 실제가 더 많았으면 추가 소비
      this.tokenBucket.tryConsume(diff);
    }
    // 실제가 더 적었으면 자연 보충에 맡김
  }
 
  getStatus(): {
    availableRequests: number;
    availableTokens: number;
  } {
    return {
      availableRequests: this.requestBucket.availableTokens,
      availableTokens: this.tokenBucket.availableTokens,
    };
  }
}

interface QueueMetrics {
  depth: number;
  enqueueRate: number; // items/sec
  dequeueRate: number; // items/sec
  avgWaitTime: number; // ms
  maxWaitTime: number; // ms
  dropCount: number;
}
 
class QueueMonitor {
  private metrics: Map<string, QueueMetrics> = new Map();
  private alerts: Array<{
    queueName: string;
    condition: (m: QueueMetrics) => boolean;
    action: (m: QueueMetrics) => void;
  }> = [];
 
  registerAlert(
    queueName: string,
    condition: (m: QueueMetrics) => boolean,
    action: (m: QueueMetrics) => void
  ) {
    this.alerts.push({ queueName, condition, action });
  }
 
  update(queueName: string, metrics: QueueMetrics) {
    this.metrics.set(queueName, metrics);
 
    // 알림 조건 확인
    for (const alert of this.alerts) {
      if (
        alert.queueName === queueName &&
        alert.condition(metrics)
      ) {
        alert.action(metrics);
      }
    }
  }
}
 
// 사용 예시
const monitor = new QueueMonitor();
 
// 추론 큐 깊이가 100을 초과하면 경고
monitor.registerAlert(
  "inference-queue",
  (m) => m.depth > 100,
  (m) => {
    console.warn(
      `추론 큐 깊이 경고: ${m.depth}, 평균 대기: ${m.avgWaitTime}ms`
    );
    // 자동 스케일 아웃 트리거
    triggerAutoScale("inference-servers");
  }
);
 
// 평균 대기 시간이 10초를 초과하면 부하 차단 활성화
monitor.registerAlert(
  "inference-queue",
  (m) => m.avgWaitTime > 10000,
  () => {
    enableLoadShedding("low-priority");
  }
);

class OverloadProtection {
  private rateLimiter: LLMRateLimiter;
  private loadShedder: LoadShedder;
  private queueMonitor: QueueMonitor;
  private circuitBreaker: CircuitBreaker;
 
  async handleRequest(
    request: InferenceRequest
  ): Promise<InferenceResponse> {
    // 1. 서킷 브레이커 확인
    if (this.circuitBreaker.isOpen()) {
      throw new Error("Service temporarily unavailable");
    }
 
    // 2. 부하 차단 확인
    if (!this.loadShedder.shouldAccept(request.priority)) {
      throw new Error("Server overloaded, try again later");
    }
 
    // 3. 레이트 리미팅
    await this.rateLimiter.acquire(request.estimatedTokens);
 
    try {
      const response = await this.executeInference(request);
      this.circuitBreaker.recordSuccess();
      return response;
    } catch (error) {
      this.circuitBreaker.recordFailure();
      throw error;
    }
  }
}
 
// 서킷 브레이커: 연속 실패 시 일정 시간 요청 차단
class CircuitBreaker {
  private failures = 0;
  private lastFailure = 0;
  private state: "closed" | "open" | "half-open" = "closed";
 
  constructor(
    private threshold: number = 5,
    private resetTimeMs: number = 30000
  ) {}
 
  isOpen(): boolean {
    if (this.state === "open") {
      // 리셋 시간이 지났으면 half-open으로 전환
      if (Date.now() - this.lastFailure > this.resetTimeMs) {
        this.state = "half-open";
        return false;
      }
      return true;
    }
    return false;
  }
 
  recordSuccess() {
    this.failures = 0;
    this.state = "closed";
  }
 
  recordFailure() {
    this.failures++;
    this.lastFailure = Date.now();
    if (this.failures >= this.threshold) {
      this.state = "open";
    }
  }
}