본문으로 건너뛰기

[openclaw] OpenConnect Gateway 성능 최적화: 연결 ID 기반 인덱싱 도입

PR 링크: openclaw/openclaw#128198 상태: Merged | 변경: +176 / -11

들어가며

이번 PR은 OpenConnect Gateway 시스템에서 WebSocket 클라이언트 관리 및 통신 방식을 최적화하는 데 중점을 둡니다. 기존에는 모든 연결된 클라이언트를 Set 형태로 관리하여 특정 클라이언트에게 메시지를 전달하거나 관련 정보를 조회할 때 모든 클라이언트를 순회해야 하는 비효율성이 있었습니다. 특히 클라이언트 수가 많아질수록 이러한 순회는 상당한 성능 저하를 야기했습니다. 이 PR은 GatewayClientRegistry라는 새로운 자료구조를 도입하여 연결 ID를 기반으로 클라이언트를 효율적으로 인덱싱함으로써 이러한 문제를 해결합니다.

주요 목표는 다음과 같습니다:

  • 연결 ID 기반 인덱싱: 인증된 게이트웨이 클라이언트를 연결 ID 기준으로 인덱싱합니다.
  • 효율적인 팬아웃: 인덱스를 활용하여 특정 연결 ID 집합에 대한 WebSocket 메시지 팬아웃을 최적화합니다.
  • 성능 향상: 버퍼링된 양 조회, 활성 연결 확인, 세션 뷰어 존재 여부 확인 등의 작업에서 불필요한 순회를 제거합니다.
  • 기존 기능 유지: 기존의 브로드캐스트 방식, 삽입 순서 보장, 소켓 닫힘 필터링 등의 기능은 그대로 유지합니다.

이 글에서는 코드 변경 사항을 상세히 분석하고, 이러한 변경이 왜 성능 향상으로 이어지는지, 그리고 어떤 일반적인 교훈을 얻을 수 있는지 살펴보겠습니다.

코드 분석

1. src/gateway/server/client-registry.ts - 새로운 클라이언트 레지스트리 도입

이 PR의 핵심은 GatewayClientRegistry 클래스의 도입입니다. 이 클래스는 기존 Set<GatewayWsClient>를 상속하면서 연결 ID를 기반으로 클라이언트를 효율적으로 조회할 수 있는 기능을 추가합니다.

주요 변경 사항:

  • #byConnectionId: 연결 ID를 키로, 클라이언트 정보와 삽입 순서를 값으로 가지는 Map을 사용하여 클라이언트 조회 성능을 O(1)으로 만듭니다.
  • #nextOrder: 클라이언트가 추가될 때마다 증가하는 카운터로, 클라이언트의 삽입 순서를 추적합니다.
  • add(client): 클라이언트를 Set에 추가할 때 #byConnectionId 맵에도 등록합니다. 이때 삽입 순서(order)를 기록합니다.
  • delete(client): 클라이언트를 Set에서 제거할 때 #byConnectionId 맵에서도 삭제합니다.
  • clear(): Set과 #byConnectionId 맵을 모두 비웁니다.
  • getByConnectionId(connId): 특정 연결 ID에 해당하는 클라이언트를 O(1) 시간 복잡도로 반환합니다.
  • getByConnectionIds(connIds): 여러 연결 ID에 해당하는 클라이언트들을 조회합니다. 이때 조회된 클라이언트들을 삽입 순서대로 정렬하여 반환합니다. 이는 기존의 삽입 순서 보장 요구사항을 만족시키기 위함입니다.
type IndexedClient = {
  client: GatewayWsClient;
  order: number;
};

export class GatewayClientRegistry extends Set<GatewayWsClient> {
  readonly #byConnectionId = new Map<string, IndexedClient>();
  #nextOrder = 0;

  constructor(clients?: Iterable<GatewayWsClient>) {
    super();
    for (const client of clients ?? []) {
      this.add(client);
    }
  }

  override add(client: GatewayWsClient): this {
    if (!this.has(client)) {
      this.#byConnectionId.set(client.connId, { client, order: this.#nextOrder++ });
    }
    return super.add(client);
  }

  override delete(client: GatewayWsClient): boolean {
    if (!super.delete(client)) {
      return false;
    }
    if (this.#byConnectionId.get(client.connId)?.client === client) {
      this.#byConnectionId.delete(client.connId);
    }
    return true;
  }

  override clear(): void {
    super.clear();
    this.#byConnectionId.clear();
  }

  getByConnectionId(connId: string): GatewayWsClient | undefined {
    return this.#byConnectionId.get(connId)?.client;
  }

  getByConnectionIds(connIds: ReadonlySet<string>): GatewayWsClient[] {
    const indexed: IndexedClient[] = [];
    for (const connId of connIds) {
      const entry = this.#byConnectionId.get(connId);
      if (entry) {
        indexed.push(entry);
      }
    }
    // Targeted fanout keeps authenticated-client insertion order without
    // walking unrelated sockets.
    if (indexed.length > 1) {
      indexed.sort((a, b) => a.order - b.order);
    }
    return indexed.map((entry) => entry.client);
  }
}

2. src/gateway/server-connection-state.ts - GatewayClientRegistry 사용

기존의 Set<GatewayWsClient>를 사용하던 clients 변수를 새로 도입된 GatewayClientRegistry로 교체합니다.

Before:

// const clients = new Set<GatewayWsClient>();

After:

const clients = new GatewayClientRegistry();

또한, isConnectionActive 함수에서 클라이언트 검색 방식을 변경합니다. 기존에는 clients Set을 순회했지만, 이제는 GatewayClientRegistry의 getByConnectionId 메서드를 사용하여 O(1) 시간에 검색합니다.

Before:

// for (const client of clients) {
//   if (client.connId === connId && !client.invalidated) {
//     return true;
//   }
// }
// return false;

After:

const client = clients.getByConnectionId(connId);
return Boolean(client && !client.invalidated);

3. src/gateway/server-broadcast.ts - 타겟팅된 브로드캐스트 최적화

createGatewayBroadcaster 함수는 메시지 브로드캐스트 로직을 담당합니다. 이 함수에서도 GatewayClientRegistry를 활용하여 성능을 개선합니다.

주요 변경 사항:

  • indexedClients: 브로드캐스터 생성 시 clients가 GatewayClientRegistry 인스턴스인지 확인하고, 맞다면 indexedClients 변수에 할당합니다. 이를 통해 타겟팅된 브로드캐스트 시 인덱스를 활용할 수 있게 됩니다.
  • broadcastInternal 함수 내 루프:
    • targetConnIds가 있고 indexedClients가 있는 경우, params.clients (기존 Set) 대신 indexedClients.getByConnectionIds(targetConnIds)를 사용하여 타겟 클라이언트 목록을 가져옵니다. 이 메서드는 연결 ID를 기반으로 클라이언트를 빠르게 찾고 삽입 순서대로 정렬된 배열을 반환합니다.
    • targetConnIds가 있지만 indexedClients가 없는 경우 (즉, 기존 Set을 사용하는 경우), 기존처럼 targetConnIds.has(c.connId)를 사용하여 필터링합니다. 이는 기존의 호환성을 유지하기 위함입니다.
  • getBufferedAmount 함수:
    • indexedClients가 있는 경우, indexedClients.getByConnectionId(connId)를 사용하여 해당 클라이언트의 bufferedAmount를 O(1) 시간에 조회합니다.
    • 그렇지 않은 경우, 기존처럼 클라이언트를 순회합니다.

Before (broadcastInternal 루프 일부):

// for (const c of params.clients) {
//   if (targetConnIds && !targetConnIds.has(c.connId)) {
//     continue;
//   }
//   // ...
// }

After (broadcastInternal 루프 일부):

const recipients =
  targetConnIds && indexedClients
    ? indexedClients.getByConnectionIds(targetConnIds)
    : params.clients;
for (const c of recipients) {
  // ...
  if (targetConnIds && !indexedClients && !targetConnIds.has(c.connId)) {
    continue;
  }
  // ...
}

Before (getBufferedAmount):

// for (const client of params.clients) {
//   if (client.connId === connId) {
//     return client.socket.bufferedAmount;
//   }
// }
// return undefined;

After (getBufferedAmount):

if (indexedClients) {
  return indexedClients.getByConnectionId(connId)?.socket.bufferedAmount;
}
for (const client of params.clients) {
  if (client.connId === connId) {
    return client.socket.bufferedAmount;
  }
}
return undefined;

4. src/gateway/server-connection-state.test.ts - 새로운 테스트 케이스 추가

GatewayClientRegistry의 도입으로 인해 발생할 수 있는 잠재적인 회귀를 방지하고, 최적화된 동작을 검증하기 위한 새로운 테스트 케이스가 추가되었습니다.

  • bounds targeted delivery and connection lookups to the requested connection: 타겟팅된 전달 및 연결 조회 시, 요청된 연결 ID에만 국한되는지, 그리고 connId getter가 불필요하게 호출되지 않는지 검증합니다. 이 테스트는 PR 설명에서 언급된

참고 자료

⚠️ 알림: 이 분석은 AI가 실제 코드 diff를 기반으로 작성했습니다.

댓글

관련 포스트

PR Analysis 의 다른글