[Celery] python에서 celery 연동할 때 고려할 점

python 기반 웹 서버(Django, flask)를 다루면서 비동기 worker를 할당할 때 celery를 자주 사용했다.(+RabbitMQ or Redis) 사용하면서 마주쳤던 문제를 적어보려고 한다.

1. @app.task로 할당할 함수를 다른 파일로 분리한다.

• 예를 들어, a+b를 리턴하는 plus 함수가 있다고 하자. Django에서 views.py에서 plus 함수 결과를 리턴하는 API를 B라고 하자. B 안에 plus 함수 매커니즘을 넣고 B를 @app.task로 지정하면 celery가 실행되지 않는다. 명령어 자체가 실행할 파이썬 파일을 지정한 다음 함수를 불러내서 delay 함수를 적용하는 방식이라 다른 파일로 분리해야한다.

@app.task(name="get_keyword")
@api_view(['POST'])
def get_keyword(request):
    
    diary_keyword = kkma.nouns.delay(diary_contents)

    while True:
        if diary_keyword.ready() == False:
            time.sleep(5)
            print("    delay...    ")
            continue
        else :
            print(diary_keyword)
            print(diary_keyword.ready())
            return Response({
                "resuilt" :"성공"
            }, status=status.HTTP_201_CREATED)

또 다른 예시로, 여기서는 kkma를 API안에서 실행 후 API 자체를 @app.task로 지정해서 post 요청을 보내도 celery가 실행되지 않았다.

@app.task(name="decode")
def decode():
    kkma = Kkma()
    a = kkma.nouns("새해가 밝았습니다 제주도에 눈이 와요 어젠 목도리만 둘렀는데.. 날씨가 왕왕 많이 바뀌네요")
    return a

kkma()를 이용하는 함수를 decode.py의 decode 함수로 분리 후 @app.task 설정을 하고

@api_view(['POST'])
def get_keyword(request):
    diary_keyword = decode.delay()

    while True:
        if diary_keyword.ready() == False:
            time.sleep(5)
            print("    delay...    ")
            continue
        else :
            print(diary_keyword.get())
            print(diary_keyword.ready())
            return Response({
                "result" :"성공"
            }, status=status.HTTP_201_CREATED)

API에서 decode 함수를 호출하니 성공했다.

2. task 함수에서 용량이 큰 객체나 파일을 return하면 안된다.

• 메모리가 커지면 worker가 동작하다가 실패한다. 실제로 AI 모듈 실행 후 이미지를 반환하는 함수를 worker로 지정했었는데, 약 30KB인 이미지를 반환하니 결과가 나오지 않았다.

• 결과 파일을 bucket에(AWS S3, Google Storage 등) 저장 후 url을 불러오는 방식으로 변경하면 해결 가능하다.

3. task 함수에서 사용하는 라이브러리와 celery의 연동성을 찾아본다.

• 첫번째로, Kkma() 등 자바기반 konlpy tag를 처음 init하는 경우 celery의 쓰레드 안에서 jvm_init()이 실행되는 문제가 있었다. 설정된 jdk를(로컬이든 docker에 환경변수에 설정된 jdk든) 불러오고 Kkma를 실행해야하는데 jvm_init()이 실행되어 설정이 초기화되어 실행시키지 못했다.

• 이 문제는 Kkma()를 celery가 실행하는 함수 내에서 실행하여 해결했다. 이렇게 되면 jvm_init()과 상관없이 초기 설정을 한 상태에서 Kkma()를 이용할 수 있다. 다만, 함수를 호출할 때마다 설정을 하고 함수를 실행해야하므로 비효율적이다.

  • 다른 해결방안으로는 Kkma() 대신 mecab을 사용하는 것이다. mecab을 사용하니 위와 같은 문제는 발생하지 않았다.

• 두번째로, pytorch를 이용하는 함수를 celery worker에 할당했을 때 생겼다. exited with code 247이 뜨고 강제로 종료되어 이유를 찾아보니 celery worker에서 pool 옵션의 초기값인 prefork가 pytorch를 사용하면 동작하지 않았기 때문이다.

  • 이 문제는 celery를 실행할 때 worker pool을 solo로 변경해서 해결했다.

# celery 실행 명령어
celery -A config.celery worker --loglevel=info --pool=solo

참고
https://github.com/konlpy/konlpy/issues/65

celery를 사용한 프로젝트
https://github.com/2022-SeongNam-Pre-Internship-TeamE/PhoToon
https://github.com/2022SiliconValley-Team-ForV/TTS
https://github.com/2023-SV-winter-bootcamp-G-Diary/G-Diary