Celery 비동기 시스템 개선기: 리소스 기반 동시성 제어

들어가며

이전에 Celery 비동기 시스템 관련해서 다룬 블로그 글인 Celery를 활용한 비동기 작업 처리에서 정의했던 시스템 리소스 관리 최적화 부족 문제와, 후속 글인 Celery 비동기 시스템 개선기: Timeout 오류 해결에 이어서 CellCraft의 비동기 시스템에서 리소스 기반 동시성 제어를 도입한 과정을 정리하고자 합니다.

 

CellCraft는 리소스를 많이 사용하는 추론 작업을 수행할 수 있는 웹 플랫폼으로 분석 작업은 독립된 Docker 컨테이너에서 실행됩니다. 각 작업은 CPU 또는 GPU 리소스를 점유하며, 가용 리소스를 초과하여 동시에 실행되면 서버가 장애가 발생할 수 있는 리스크가 존재합니다. 해당 글에서는 이전에 임시로 적용했던 리소스 관리 방식의 한계와 Redis Lua 세마포어 기반의 슬롯 제어 시스템으로 문제를 해결한 과정을 정리해보고자 합니다.

이전 접근의 한계

첫 번째 블로그 글에서 언급한 임시로 적용했었던 리소스 관리 코드입니다.

# 이전 접근 (2025.03 기준)
CPU_USAGE_LIMIT = 85
MEMORY_USAGE_LIMIT = 90
GPU_USAGE_LIMIT = 90

def check_system_usage():
    while True:
        cpu_usage = psutil.cpu_percent(interval=1)
        memory_usage = psutil.virtual_memory().percent
        gpus = GPUtil.getGPUs()
        gpu_usage = max((gpu.load * 100 for gpu in gpus), default=0)

        if cpu_usage >= CPU_USAGE_LIMIT or memory_usage >= MEMORY_USAGE_LIMIT or gpu_usage >= GPU_USAGE_LIMIT:
            celery_app.control.cancel_consumer("workflow_task")  # 큐 전체 중단
        else:
            celery_app.control.add_consumer("workflow_task")     # 큐 전체 재개

        time.sleep(5)

 

해당 접근은 시스템 리소스 사용률을 주기적으로 확인하여, 임계값을 초과하면 큐 전체를 중단하고 여유가 생기면 재개하는 방식이었습니다. 실제로 개발 과정 중 여러 차례 테스트를 통해 다음과 같은 한계들을 확인했습니다.

 

1. 리소스 사용률 기반 모니터링으로 정확한 제어 불가능

리소스 사용률 모니터링 기반 제어는 현재 부하 상태를 기반으로 대략적으로 제어만 할 뿐 새로운 작업이 얼마나 리소스를 필요로 하는지를 알 수 없어 정확한 판단이 불가능했습니다.

 

2. 큐 전체 중단/재개 로직의 문제점

cancel_consumer와 add_consumer는 큐 전체의 메시지 소비를 중단/재개합니다. 리소스가 정해진 임계값을 초과할 시 모든 대기 작업이 중단되고, 여유가 생기면 대기 중이던 작업이 한꺼번에 시작되어 다시 리소스 초과가 발생하는 현상이 나타날 수 있습니다.

 

3. Race Condition 문제

여러 Worker 프로세스가 동시에 리소스 사용률을 확인하고 각각 작업을 시작하면, 확인 시점과 실행 시점 사이의 간극으로 인해 실제로는 리소스가 부족한데 여러 작업이 동시에 실행될 수 있습니다.

 

4. while True 폴링 방식의 리소스 낭비

5초마다 psutil.cpu_percent(interval=1)을 호출하는 무한 루프가 별도 스레드에서 상시 실행되며, 리소스 관리를 위한 코드 자체가 CPU 리소스를 쓸데없이 낭비하는 상황이었습니다.

슬롯 기반 원자적 제어

이전 접근의 근본적인 한계는 현재 리소스 상태를 관찰하고 반응하는 방식이었다는 점입니다. 관찰 시점과 제어 시점 사이에 간극이 존재하고, 사용률 기반으로는 작업 단위의 판단이 불가능했습니다.

 

따라서, 세마포어 방식 기반으로 작업이 실행되기 전에 필요한 리소스를 사전 예약하는 방식으로 전환했습니다.

Redis Key 스키마

사용 중인 작업에 의한 리소스 상태를 Redis에 저장하여 관리합니다.

resource:cpu:total      → 전체 CPU 슬롯 수
resource:cpu:used       → 현재 점유된 CPU 슬롯 수
resource:gpu:total      → 전체 GPU 슬롯 수
resource:gpu:used       → 현재 점유된 GPU 슬롯 수
resource:task:{task_id} → Hash: type, slots, plugin_name, started_at

 

예를 들어 CPU 48코어 서버에서 작업당 4슬롯이 사용된다고 가정하면, resource:cpu:total=48 이고 최대 12개 작업이 동시 실행 가능합니다. 각 작업이 시작될 때 4슬롯을 획득하고, 완료 시 반환합니다.

원자적 슬롯 획득/반환 구현

리소스 획득과 반환은 Redis Lua 스크립트로 구현했습니다. Lua 스크립트를 사용한 이유는 Redis에서는 Lua 스크립트를 원자적으로 실행하기 때문입니다. 잔여 슬롯 확인 → 비교 → 슬롯 점유의 흐름에 대한 로직을 별도의 Redis 명령어로 실행하면, 두 Worker가 동시에 리소스 잔여 슬롯을 확인한 뒤 둘 다 점유하는 Race Condition 문제가 발생할 수 있습니다. 하지만 Lua 스크립트 내에서는 중간에 다른 명령이 끼어들 수 없으므로 Race Condition 문제를 해결할 수 있었습니다.

Acquire (슬롯 획득)

local total = tonumber(redis.call('GET', total_key) or 0)
local used  = tonumber(redis.call('GET', used_key) or 0)

if used + requested <= total then
    redis.call('INCRBY', used_key, requested)
    redis.call('HSET', task_key,
        'type', rtype,
        'slots', requested,
        'plugin_name', plugin,
        'started_at', started)
    return 1   -- 성공: 슬롯 획득
end
return 0       -- 실패: 리소스 부족

 

현재 사용량(used)에 요청량(requested)을 더했을 때 총량(total) 이하이면 슬롯을 점유하고, 초과하면 거부합니다. 확인과 점유가 하나의 원자적 연산으로 수행됩니다.

 

Release (슬롯 반환)

local used = tonumber(redis.call('GET', used_key) or 0)
local new_used = used - slots
if new_used < 0 then
    new_used = 0
end
redis.call('SET', used_key, new_used)
redis.call('DEL', task_key)
return new_used

 

슬롯을 반환하고 작업 할당 기록을 삭제합니다.

Celery에서의 리소스 게이팅 흐름

Celery Task 함수에서 Lua 세마포어 로직을 활용하는 실제 흐름입니다.

@shared_task(bind=True, base=MyTask, name="workflow_task:process_data_task", max_retries=None)
def process_data_task(self, ..., resource_type='cpu', resource_slots=4):
    task_id = self.request.id

    # 1) 슬롯 획득 시도
    acquired = acquire_slots(resource_type, resource_slots, task_id, plugin_name)
    if not acquired:
        raise self.retry(countdown=10)  # 10초 후 재시도

    try:
        # 2) Docker 컨테이너에서 분석 작업 실행
        result = snakemakeProcess(targets, snakefile_path, selected_plugin, task_id)
        ...
    finally:
        # 3) 반드시 슬롯 반환
        release_slots(resource_type, resource_slots, task_id)

핵심 로직 및 설정

max_retries=None (재시도 횟수 제한 설정): 리소스 부족 시 작업을 거부하는 것이 아니라, 슬롯이 확보될 때까지 대기합니다.

self.retry(countdown=10) (non-blocking 대기): 리소스 부족 시 Worker를 블로킹하지 않고 메시지를 큐에 반환하여 10초 후 다시 시도합니다. Worker는 그 사이에 다른 작업을 처리할 수 있습니다. 이전 방식에서 while True로 Worker 프로세스를 점유했던 문제를 해결합니다.

finally 블록: 작업의 성공/실패와 관계없이 반드시 슬롯을 반환합니다.

리소스 자동 감지

Celery Worker가 시작될 때 worker_ready 시그널을 통해 호스트 서버의 CPU/GPU 리소스를 자동으로 감지하고 Redis에 기록합니다.

@worker_ready.connect
def on_worker_ready(sender, **kwargs):
    initialize_resource_totals()   # CPU/GPU 감지 → Redis에 total 기록
    cleanup_stale_resources()      # 이전 크래시로 남은 잔여 슬롯 정리

 

initialize_resource_totals()는 os.cpu_count()와 GPUtil을 통해 CPU 코어 수와 GPU 디바이스 수를 감지하되, 별도의 설정 파일을 통한 수동 지정도 가능합니다.

cleanup_stale_resources()는 Worker 재시작 시 이전 크래시로 인해 반환되지 못한 잔여 슬롯을 정리하여 리소스에 대한 deadlock 문제를 방지합니다.

안전장치

장기 실행 작업을 직접 실행해보면서 하나씩 시행착오를 겪으며 엣지케이스를 처리하기 어렵기 때문에 리소스 슬롯이 정상적으로 반환되지 못하는 장애 시나리오를 설정하고, 이에 대해서 시스템의 안정성을 고려한 안전장치 로직을 구성했습니다.

이중 해제 안전장치

Celery Task 함수 코드에서의 finally 블록 외에, Celery Task 라이프사이클 훅인 after_return에 의해서도 Task가 종료된 직후에 슬롯 반환을 시도합니다.

class MyTask(Task):
    def after_return(self, status, retval, task_id, args, kwargs, einfo):
        # retry 상태가 아닌 모든 종료에서 safety-net 해제
        if status != 'RETRY':
            release_slots_by_task_id(task_id)

 

release_slots_by_task_id는 Redis에 저장된 Task의 정보(resource:task:{task_id})를 조회하여 슬롯을 반환합니다. finally 블록이 실행되지 못하는 극단적인 경우(Worker 프로세스 비정상 종료 등)에 대비한 안전장치입니다.

 

finally와 after_return이 순차적으로 실행되므로, 일반적으로는 중복 반환이 발생하지 않습니다. finally에서 release_slots가 호출되면 Lua 스크립트가 슬롯을 반환하면서 Task 키(resource:task:{task_id})를 삭제하고, 이후 after_return에서 release_slots_by_task_id가 해당 키를 조회했을 때 이미 삭제된 상태이므로 반환 시도 자체가 일어나지 않습니다.

음수 방지 로직

Release Lua 스크립트에는 if new_used < 0 then new_used = 0 로직이 포함되어 있습니다. 이는 이중 해제 안전장치와는 별개로, Celery Worker 크래시 후 재시작 시 발생할 수 있는 문제를 방지합니다.

[크래시 전] resource:cpu:used = 20, resource:task:A = {slots: 4}
[Worker 크래시]
    → finally, after_return 모두 실행되지 못함
[Worker 재시작]
    → cleanup_stale_resources() 호출
    → SET resource:cpu:used = 0 (전체 리셋)
    → DEL resource:task:A (태스크 키 삭제)


[크래시 직전에 스케줄링된 release가 지연 실행되면?]
    → release_slots(cpu, 4, task_A) 호출
    → Lua 실행: used(0) - 4 = -4
    → 음수 방지: new_used = 0 (음수 대신 0으로 고정)

 

만약 음수 방지가 없다면 used가 -4가 되어, 이후 acquire에서 used(-4) + 4 = 0 <= 48로 계산되어 실제보다 더 많은 작업을 허용하는 연쇄 오류가 발생할 수 있습니다.

Graceful Degradation: Redis 장애 대응

Redis가 장애 상태일 때 리소스 게이팅이 불가능하기 때문에 작업이 실행 불가능해집니다. 하지만 가용성을 우선하여, Redis 장애 시 게이팅 없이 작업 실행을 허용하도록 구성했습니다.

def acquire_slots(resource_type, slots, task_id, plugin_name=''):
    try:
        client = get_redis_client()
        result = client.eval(_LUA_ACQUIRE, ...)
        return int(result) == 1
    except redis.RedisError as e:
        logger.warning(f"Redis error: {e}. Falling back to ungated execution.")
        return True   # Redis 장애 시 → 게이팅 없이 실행 허용

 

리소스 제한 없이 작업이 실행되어 리소스 과부하 가능성이 있지만, 만약 새로 시작한 작업이 리소스 과부하를 발생시킨다면 CPU만 사용하는 작업의 경우에는 해당 작업이 OOM Kill로 인해 중단될 것이고 GPU를 사용하는 작업의 경우에는 CUDA Out of Memory 에러으로 인해 해당 작업이 중단될 뿐이지, 모든 작업이 중단되는 것은 아니기 때문에 적절한 선택이라고 판단했습니다.

 

다만, Redis 장애 중에 게이팅 없이 실행된 작업들은 Redis에 슬롯 정보가 기록되지 않으므로, Redis가 복구된 이후에도 해당 작업들이 여전히 실행 중이라면 슬롯 카운터에 반영되지 않은 상태로 처리됩니다. 해당 상황에서 새로운 작업이 acquire_slots를 통해 슬롯을 확보하고 실행되면, 실제로는 리소스가 부족한데 게이팅을 통과하여 과부하가 발생할 수 있는 리스크가 남아있습니다.

 

이에 대해서는 Redis Persistence(AOF) 활성화를 통해 Redis 재시작 시 슬롯 데이터를 복원하거나, Task 키에 TTL을 설정하고 실행 중인 태스크가 주기적으로 TTL을 갱신하는 Heartbeat 방식을 도입하여 개선하는 것을 고려하고 있습니다.

결과

기준	이전	현재
제어 방식	시스템 사용률(%) 관찰 후 반응	작업 단위 슬롯 사전 예약
제어 단위	큐 전체 중단/재개	개별 작업별 acquire/release
Race Condition	방지 불가	Redis Lua 원자적 연산으로 방지
리소스 부족 시	큐 전체 중단 → 모든 작업 대기	해당 작업만 큐에 반환 → 10초 후 재시도
Worker 점유	while True 폴링 (상시 점유)	self.retry (non-blocking)
리소스 해제	해제 메커니즘 없음	finally + after_return 이중 안전장치
장애 대응	없음	Redis 장애 시 graceful fallback
가시성	print 출력만	Redis 기반 리소스 모니터링 API 제공

마치며

이전 접근에서 가장 아쉬웠던 점은 리소스 관리를 단순히 지속적인 사용률 모니터링으로 제어하도록 접근했다는 것입니다. 시스템 사용률을 관찰하고 임계값을 넘으면 큐를 중단하는 방식은, 관찰과 제어 사이의 간극, 작업 단위 제어 불가, Race Condition 등 수많은 한계가 존재합니다. 사실 돌이켜보면 백엔드 엔지니어링에 대한 이해가 많이 부족했던 것 같습니다.

 

결과적으로 Redis Lua 세마포어로 전환하면서 작업 실행 전에 리소스를 예약하는 방식으로 바꿨고, 원자적 연산을 통해 Race Condition을 방지하면서도 non-blocking 재시도로 Celery Worker를 가용 가능한 리소스가 있을 때만 실행하도록 구현할 수 있었습니다. 또한 장애 상황을 고려한 안전 장치 로직을 통해 안정적으로 운영할 수 있도록 구성하는 것도 중요함을 경험할 수 있었습니다. 현재도 완벽한 비동기 시스템은 아니지만 하나씩 개선점을 찾아가며 발전하고 있는 것 같아서 뿌듯하네요 ㅎㅎ..