워커를 더 효율적으로 사용하는 워커풀(Worker Pool) 개발기

이미지 처리 작업은 생각보다 리소스를 많이 소모합니다. 특히 사용자가 가지고 있는 고해상도 이미지를 썸네일로 변환하는 과정에서는, 성능뿐만 아니라 메모리 관리까지도 신경 써야 했습니다. 이러한 문제를 해결하기 위해 웹 워커(Web Worker)를 활용한 워커풀(Worker Pool) 클래스를 직접 설계하게 되었는데요, 이 글에서는 그 과정을 정리해보았습니다.

워커를 사용하자

이미지를 썸네일 크기로 전환하는 것은 아주 빠른 작업이긴 하지만 많은 이미지를 처리한다면 사용자와의 상호작용이 생각보다 오랜 시간 차단될 수 있습니다. 그래서 이미지 처리를 메인쓰레드에서 하는 대신 워커에 작업을 맡겨 메인 쓰레드의 부담을 줄이고 사용자의 상호작용을 차단하지 않게 만들 수 있습니다. 그래서 당연히 Web Worker를 사용했죠.

워커는 훌륭하다. 하지만...

워커가 이미지를 썸네일로 변환하기 위해서는 당연히 이미지를 받아야 합니다. 원본 이미지를 말이죠. 그래서 메인 쓰레드가 이미지를 Blob 형태로 워커에 넘겨주었습니다. 그런데 여기서 문제가 발생합니다.

워커도 내부에 대기열이 있다?!

Web Worker는 단일 스레드로 동작하지만, postMessage를 통해 연달아 메시지를 보내면 내부에서 작업 큐처럼 쌓이는 구조를 가지고 있습니다. 사실 이 구조는 브라우저 구현체마다 조금씩 다를 수 있으나, 대체로 메시지가 큐에 쌓이는 등 비동기적으로 메시지를 처리하도록 합니다.

 

Web Worker는 비동기 방식으로 메시지를 처리합니다. 즉, 워커는 큐에 쌓인 메시지를 순차적으로 처리하는 구조입니다. 이때, 메시지로 보내는 데이터가 크면 클수록 메모리 사용량이 증가하게 되는데, 특히 Blob 형태의 대용량 데이터는 메모리 차지 비율이 매우 높습니다. 여기서 문제가 발생하는 지점은 바로 Blob 객체가 큐에 쌓이지만 처리 속도는 그에 미치지 못해 메모리 초과 문제가 발생하는 것입니다.

 

워커가 이 Blob 데이터를 처리할 때 메모리 내에서 복사하여 전달되므로, 대기열에 많은 크기의 Blob이 쌓이면 결국 브라우저 메모리에 이를 처리할 수 없을 정도로 많은 데이터가 쌓일 수 있습니다. 아무리 transfer 객체로 변환한다고 해도 워커의 대기열에 큰 blob 객체가 쌓이는 것은 변함없습니다.

 

쉽게 정리하자면, 워커 인스턴스는 메시지를 수신하기 전까지 어딘가에 데이터를 가지고 있어야 하고, 이때 Blob이 메모리에 쌓이기 때문에 처리 대기 중인 작업들이 누적될수록 메모리가 급격히 상승하게 됩니다. 결국 브라우저가 탭을 강제로 죽이는 상황까지 발생하게 된 것이죠.

워커풀로 해결해 보자

이 문제를 해결하기 위해서는 다음과 같은 전략이 필요했습니다.

1. 작업 대기열을 따로 관리하며 blob 객체는 워커에 전달하기 직전에만 생성합니다.
2. 워커에 메시지를 보내는 것은 워커가 작업을 하지 않고 있을 때입니다.
3. 직접 관리하는 대기열에 더 이상 필요하지 않은 작업은 취소하여 늦게 쌓였지만 긴급한 작업이 빨리 처리될 수 있도록합니다.
4. 워커의 개수는 필요에 따라 유동적으로 조정되어야 합니다.

사실 위에서 언급한 문제를 해결하는 것은 1, 2번까지로 충분합니다. 하지만 직접 대기열을 관리하고 있기도하고 워커가 처리해야할 작업이 실시간으로 바뀌기 때문에(스크롤에 의해 보이는 위치의 사진만 File System API로 가져와 썸네일로 변환) 3번의 작업도 필요했습니다. 4번의 경우는 블로그 포스팅을 위해 새롭게 추가한 기능입니다.

 

사실 이러한 구조는 Kotlin의 코루틴과 코루틴이 활용하는 쓰레드 풀(Thread Pool) 구조와 비슷한 면이 있습니다. 애초에 워커풀의 아이디어 자체도 코루틴의 쓰레드 풀이서 개념적인 부분만 가져와 브라우저 환경에서 웹 워커풀을 설계하는 데 응용한 것입니다.

워커풀 구조

queue에서 워커로 작업이 전달될 때 처리기를 거쳐갑니다.

 

간단하게 구조를 설명드리면 다음과 같습니다.

• WorkerPool 클래스는 워커 인스턴스를 여러 개 생성하여 풀로 관리합니다.
• 외부에서 작업 요청이 들어오면, 사용 가능한 워커가 있는 경우 즉시 작업을 할당하고,
• 워커가 모두 바쁘면, 작업은 큐에 저장됩니다.
• 작업이 완료된 워커가 있다면 큐에서 작업을 꺼내 처리하도록 합니다.
• 큐에서 작업을 꺼낼 때는 taskTransform을 통해 큐의 작업을 워커가 처리할 수 있는 형태로 변환합니다.(여기서는 FileHandler -> blob)

다음은 workerpool의 실제 구현입니다. 따로 설명을 추가하면 보기 불편할 것 같아 JSDOC으로 정리하였습니다. 

interface WorkerPoolOptions<T = unknown, U = unknown, V = T> {
  /**
   * Worker 인스턴스를 생성하여 반환하는 팩토리 함수.
   * Vite와 같은 번들러 환경에서 Worker를 올바르게 생성할 수 있도록 도와줍니다.
   */
  workerFactory: () => Worker;
  minPoolSize?: number; // 최소 워커 풀 크기 (기본값: 2)
  maxPoolSize?: number; // 최대 워커 풀 크기 (기본값: 10)
  /**
   * 작업 데이터를 가공하는 함수.
   * 입력 Task의 payload (타입 T)를 받아 실제 워커에 전달할 데이터 (타입 V)를 반환합니다.
   * 이 함수는 동기 또는 비동기일 수 있습니다.
   */
  taskTransform?: (task: Task<T, U>) => V | Promise<V>;
}

type Task<T, U> = {
  id: string;
  payload: T;
  resolve: (result: U) => void;
  reject: (error: unknown) => void;
};

const CORE_COUNT = navigator.hardwareConcurrency || 4;

/**
 * WorkerPool 클래스는 미리 생성한 Worker 인스턴스 풀을 관리하며,
 * 작업 요청이 들어오면 사용 가능한 워커에 작업을 할당하고,
 * 모든 워커가 바쁠 경우 작업을 Map에 저장합니다.
 *
 * 풀 생성 시 옵션 객체를 통해 workerFactory, 최소/최대 풀 크기, taskTransform 함수를 받아 내부에서만 사용하며,
 * 작업 추가 및 완료 시 자동으로 풀 크기를 조절합니다.
 */
export class WorkerPool<T, U, V = T> {
  private pool: Worker[];
  private readonly queue: Map<string, Task<T, U>>;
  private readonly busy: Set<Worker>;
  private readonly taskTransform?: (task: Task<T, U>) => V | Promise<V>;
  private readonly workerFactory: () => Worker;
  private readonly minPoolSize: number;
  private readonly maxPoolSize: number;

  constructor({
    workerFactory,
    minPoolSize = 1,
    maxPoolSize = Math.floor(CORE_COUNT * 0.75),
    taskTransform,
  }: WorkerPoolOptions<T, U, V>) {
    this.workerFactory = workerFactory;
    this.minPoolSize = minPoolSize;
    this.maxPoolSize = maxPoolSize;
    this.taskTransform = taskTransform;
    this.pool = [];
    this.queue = new Map();
    this.busy = new Set();
    const initialSize = Math.min(minPoolSize, maxPoolSize);
    for (let i = 0; i < initialSize; i++) {
      const worker = this.workerFactory();
      worker.onmessage = () => {};
      this.pool.push(worker);
    }
  }

  /**
   * Map에서 첫 번째 작업을 꺼내고 삭제하는 헬퍼 메서드.
   * @returns 작업(Task) 또는 undefined.
   */
  private popQueueTask(): Task<T, U> | undefined {
    for (const [id, task] of this.queue) {
      this.queue.delete(id);
      return task;
    }
    return undefined;
  }

  /**
   * 워커에 transfarable 데이터를 전송합니다.
   * @param payload 워커가 받을 작업을 래핑한 객체
   */
  private getTransferables(payload: V): Transferable[] {
    if (this.isTransferable(payload)) {
      return [payload];
    }

    if (
      typeof payload === "object" &&
      payload !== null &&
      "transferables" in payload &&
      Array.isArray((payload as { transferables: Transferable[] }).transferables)
    ) {
      return [...(payload as { transferables: Transferable[] }).transferables];
    }

    return [];
  }

  private isTransferable(payload: any): payload is Transferable {
    return (
      payload instanceof ArrayBuffer ||
      payload instanceof ImageBitmap ||
      (typeof OffscreenCanvas !== "undefined" && payload instanceof OffscreenCanvas)
    );
  }

  /**
   * 사용 가능한 워커에 작업을 실행합니다.
   * @param worker 작업을 수행할 워커.
   * @param task 실행할 작업.
   */
  private async runTask(worker: Worker, task: Task<T, U>) {
    this.busy.add(worker);

    const handleMessage = (e: MessageEvent<U>) => {
      worker.removeEventListener("message", handleMessage);
      this.busy.delete(worker);
      task.resolve(e.data);
      this.checkQueue();
      this.updatePoolSize();
    };

    const handleError = (err: ErrorEvent) => {
      worker.removeEventListener("error", handleError);
      this.busy.delete(worker);
      task.reject(err);
      this.checkQueue();
      this.updatePoolSize();
    };

    worker.addEventListener("message", handleMessage, { once: true });
    worker.addEventListener("error", handleError, { once: true });

    try {
      const payloadToSend = await Promise.resolve(
        this.taskTransform ? this.taskTransform(task) : (task.payload as unknown as V)
      );
      const transferables = this.getTransferables(payloadToSend);
      worker.postMessage(payloadToSend, transferables);
    } catch (error) {
      this.busy.delete(worker);
      task.reject(error);
      this.checkQueue();
      this.updatePoolSize();
    }
  }

  /**
   * 대기 중인 작업이 있으면 사용 가능한 워커에 작업을 할당합니다.
   */
  private checkQueue() {
    if (this.queue.size === 0) return;
    const availableWorker = this.pool.find((worker) => !this.busy.has(worker));
    if (availableWorker) {
      const task = this.popQueueTask();
      if (task) {
        this.runTask(availableWorker, task);
      }
    }
  }

  /**
   * 작업 추가 또는 완료 시 자동으로 풀 크기를 조절합니다.
   * - 대기 작업이 있는 경우: 현재 큐 크기의 1.5배 만큼 추가하여 최대 풀 크기를 초과하지 않도록 합니다.
   * - 대기 작업이 없으면, 최소 풀 크기까지 idle 워커를 종료합니다.
   */
  private updatePoolSize() {
    if (this.queue.size > 0 && this.pool.length < this.maxPoolSize) {
      const targetSize = Math.min(this.pool.length + Math.max(1, Math.ceil(this.queue.size * 1.5)), this.maxPoolSize);
      const toAdd = targetSize - this.pool.length;
      for (let i = 0; i < toAdd; i++) {
        const worker = this.workerFactory();
        worker.onmessage = () => {};
        this.pool.push(worker);
      }
    } else if (this.queue.size === 0 && this.pool.length > this.minPoolSize) {
      for (let i = this.pool.length - 1; i >= 0 && this.pool.length > this.minPoolSize; i--) {
        const worker = this.pool[i];
        if (!this.busy.has(worker)) {
          worker.terminate();
          this.pool.splice(i, 1);
        }
      }
    }
  }

  /**
   * 작업을 실행합니다.
   * 사용 가능한 워커가 없으면 작업을 큐(Map)에 저장하고, 자동으로 풀 크기를 조절합니다.
   * @param id 고유 작업 식별자.
   * @param payload 워커에 전달할 원본 데이터 (타입 T).
   * @returns 워커의 처리 결과를 반환하는 Promise (타입 U).
   */
  public execute(payload: T, id: string = this.getUniqueId()): Promise<U> {
    return new Promise((resolve, reject) => {
      const task: Task<T, U> = { id, payload, resolve, reject };
      const availableWorker = this.pool.find((worker) => !this.busy.has(worker));
      if (availableWorker) {
        this.runTask(availableWorker, task);
      } else {
        this.queue.set(id, task);
        this.updatePoolSize();
      }
    });
  }

  private getUniqueId() {
    return `${Date.now()}-${Math.random().toString(36).slice(2, 9)}`;
  }
  /**
   * 지정된 id를 이용해 큐에서 작업을 제거합니다.
   * @param id 제거할 작업의 고유 식별자.
   * @returns 제거 성공 여부.
   */
  public removeTaskById(id: string): boolean {
    return this.queue.delete(id);
  }

  /**
   * 워커 풀에 있는 모든 워커를 종료하고, 큐와 내부 상태를 초기화합니다.
   */
  public terminate() {
    this.pool.forEach((worker) => worker.terminate());
    this.pool = [];
    this.queue.clear();
    this.busy.clear();
  }
}

사용은?

어떻게 구현하였는 지도 중요하지만 어떻게 사용하는 지도 중요합니다.workerPoool을 활용하여 기존의 썸네일 생성 워커를 그대로 사용하려면 다음과 같이 사용할 수 있습니다.

workerPool 생성

import { ThumbnailWorkerResult } from "./thumbnailWorker";
import { WorkerPool } from "./workerPool";

function workerFactory(): Worker {
  // vite에서의 예시
  return new Worker(new URL("./worker", import.meta.url), {
    type: "module",
  });
}

const workerPoolExample = new WorkerPool<FileSystemFileHandle, ThumbnailWorkerResult, File>({
  workerFactory,
  minPoolSize: 4,
  maxPoolSize: 8,
  taskTransform: async (task) => {
    const file = await task.payload.getFile();
    return file;
  },
});

export default workerPoolExample;

React에서 사용하기

React에서 사용하기 위한 커스텀 훅입니다. 여기서는 화면에 보일 때 작업을 시작하고 화면에서 사라지면 작업을 취소하는 일을 하기 위해, 컴포넌트에서 ref를 직접 특정 요소에 붙여주어야 합니다. 그리고 reCreateUrl의 경우는 외부에서 URL.revokeObjectURL 을 사용하여서 url은 존재하지만 실제 해당 url에는 blob 객체가 없는 경우를 대비한 함수입니다. 만약 컴포넌트 외부에서 URL.revokeObjectURL를 사용한다면 꼭 reCreateUrl을 같이 써주어야 합니다.

import { useEffect, useRef, useState } from "react";
import workerPoolExample from "./workerPoolusingExample";

/**
 *
 * @param file 작업의 대상이 될 파일
 * @returns
 *
 * ref: 이 대상이 화면에 보이면 작업 시작
 *
 * thumbUrl: 썸네일 크기로 줄여진 blob의 url
 *
 * reCreateUrl: LRU 캐시 사용 등으로 실제 url은 revoke 될 경우 다시 url을 생성할 때 필요
 */
export function useWorkerPool(file: FileSystemFileHandle) {
  const [thumbUrl, setThumbUrl] = useState<string, null>(null);
  const containerRef = useRef<HTMLDivElement>(null);
  const cancelRef = useRef<() => void>(() => {});

  useEffect(() => {
    if (thumbUrl) return;

    let observer: IntersectionObserver | null = null;

    if (containerRef.current) {
      observer = new IntersectionObserver((entries) => {
        const entry = entries[0];
        if (entry.isIntersecting) {
          workerPoolExample.execute(file, file.name).then(({ promise }) => {
            cancelRef.current = workerPoolExample.removeTaskById(file.name);
            promise
              // 기존 워커가 처리 완료된 blob을 반환한다고 가정
              .then((blob: Blob) => {
                setThumbUrl(URL.createObjectURL(blob));
              })
              // 기존 워커가 에러 시 에러 메시지를 반환한다고 가정
              .catch((err: string) => {
                console.error("Thumbnail task error:", err);
              });
          });
        } else {
          thumbUrl ?? URL.revokeObjectURL(thumbUrl);
          cancelRef.current();
        }
      });
      observer.observe(containerRef.current);
    }

    return () => {
      observer?.disconnect();
      cancelRef.current();
    };
  }, [thumbUrl, file]);

  const reCreateUrl = () => {
    thumbUrl ?? URL.revokeObjectURL(thumbUrl);
    setThumbUrl(null);
  };

  return { ref: containerRef, thumbUrl, reCreateUrl };
}