csapi 프로젝트 (2) - celery 적용

https://not-to-be-reset.tistory.com/343 글에 작성한 대로, 모델 다운로드 링크가 만료되고 코드가 동작하지 않아서 다시 포스팅 합니다.
단일 인스턴스 부터 쿠버네티스 클러스터까지 확장해 나가려고 합니다.

개발환경

2023. 3. 22 기준 m1 맥북에서 진행

github 주소

https://github.com/kyeongjun-dev/csapi

---

최종 디렉토리 구조

아래와 같이 파일 이름이 변경됩니다.

tree csapi/
csapi/
├── celery
│   ├── Dockerfile
│   ├── model.h5
│   ├── requirements.txt
│   ├── run.py
│   ├── tasks.py
│   └── test_input.png
└── docker-compose.yml

 

run.py 코드 변경

먼저 run.py 코드를 확인합니다.

import cv2
import numpy as np
from tensorflow.keras.models import load_model
import tensorflow as tf
import sys
from tensorflow.compat.v1 import ConfigProto
from tensorflow.compat.v1 import InteractiveSession
config = ConfigProto()
config.gpu_options.allow_growth = True
session = InteractiveSession(config=config)


f = sys.argv[1]

saved = load_model("./model.h5")

class fashion_tools(object):
    def __init__(self,imageid,model,version=1.1):
        self.imageid = imageid
        self.model   = model
        self.version = version

    def get_dress(self,stack=False):
        """limited to top wear and full body dresses (wild and studio working)"""
        """takes input rgb----> return PNG"""
        name =  self.imageid
        file = cv2.imread(name)
        file = tf.image.resize_with_pad(file,target_height=512,target_width=512)
        rgb  = file.numpy()
        file = np.expand_dims(file,axis=0)/ 255.
        seq = self.model.predict(file)
        seq = seq[3][0,:,:,0]
        seq = np.expand_dims(seq,axis=-1)
        c1x = rgb*seq
        c2x = rgb*(1-seq)
        cfx = c1x+c2x
        dummy = np.ones((rgb.shape[0],rgb.shape[1],1))
        rgbx = np.concatenate((rgb,dummy*255),axis=-1)
        rgbs = np.concatenate((cfx,seq*255.),axis=-1)
        if stack:
            stacked = np.hstack((rgbx,rgbs))
            return stacked
        else:
            return rgbs


    def get_patch(self):
        return None

# running code
api = fashion_tools(f, saved)
image_ = api.get_dress(False)
cv2.imwrite("out.png", image_)

 

이를 기반으로 celery 를 추가하여 tasks.py를 추가합니다.

import cv2
import numpy as np
from tensorflow.keras.models import load_model
import tensorflow as tf

class fashion_tools(object):
    def __init__(self,imageid,model,version=1.1):
        self.imageid = imageid
        self.model   = model
        self.version = version
        
    def get_dress(self,stack=False):
        """limited to top wear and full body dresses (wild and studio working)"""
        """takes input rgb----> return PNG"""
        name =  self.imageid
        file = cv2.imread(name)
        file = tf.image.resize_with_pad(file,target_height=512,target_width=512)
        rgb  = file.numpy()
        file = np.expand_dims(file,axis=0)/ 255.
        seq = self.model.predict(file)
        seq = seq[3][0,:,:,0]
        seq = np.expand_dims(seq,axis=-1)
        c1x = rgb*seq
        c2x = rgb*(1-seq)
        cfx = c1x+c2x
        dummy = np.ones((rgb.shape[0],rgb.shape[1],1))
        rgbx = np.concatenate((rgb,dummy*255),axis=-1)
        rgbs = np.concatenate((cfx,seq*255.),axis=-1)
        if stack:
            stacked = np.hstack((rgbx,rgbs))
            return stacked
        else:
            return rgbs

    def get_patch(self):
        return None

from celery import Celery, current_task

app = Celery('tasks', broker="pyamqp://guest:guest@rabbit:5672//")

@app.task
def api():
    saved = load_model("./model.h5")
    api = fashion_tools("test_input.png", saved)
    image_ = api.get_dress(False)
    print('task ID :', current_task.request.id)
    cv2.imwrite("out.png", image_)
    return current_task.request.id

 

requirements.txt 파일 변경

기존의 requirements.txt 파일에 아래와 같이 추가합니다.

opencv-python==4.5.1.48
tensorflow==2.11.0
tensorflow-io==0.24.0
importlib-metadata==4.8.3
celery==5.1.2

 

docker compose 적용

아래와 같이 docker-compose.yml 파일을 작성합니다

version: "3"
services:
  worker:
    build:
      context: ./celery
    command: celery -A tasks worker --loglevel=info --max-tasks-per-child 1 -c 1
    expose:
      - "5672"

  rabbit:
    image: rabbitmq:3-management
    expose:
      - "5672"
    ports:
      - "15672:15672"
    environment:
      - RABBITMQ_DEFAULT_USER=guest
      - RABBITMQ_DEFAULT_PASS=guest

 

build와 동시에 docker-compose up을 진행합니다.

docker-compose up --build

 

아래와 같이 ready가 표시되면 성공입니다.

csapi-rabbit-1  | 2023-07-22 05:19:06.515016+00:00 [info] <0.736.0> accepting AMQP connection <0.736.0> (172.18.0.2:37222 -> 172.18.0.3:5672)
csapi-worker-1  | [2023-07-22 05:19:06,517: INFO/MainProcess] Connected to amqp://guest:**@rabbit:5672//
csapi-rabbit-1  | 2023-07-22 05:19:06.516967+00:00 [info] <0.736.0> connection <0.736.0> (172.18.0.2:37222 -> 172.18.0.3:5672): user 'guest' authenticated and granted access to vhost '/'
csapi-rabbit-1  | 2023-07-22 05:19:06.519459+00:00 [info] <0.745.0> accepting AMQP connection <0.745.0> (172.18.0.2:37224 -> 172.18.0.3:5672)
csapi-worker-1  | [2023-07-22 05:19:06,522: INFO/MainProcess] mingle: searching for neighbors
csapi-rabbit-1  | 2023-07-22 05:19:06.521815+00:00 [info] <0.745.0> connection <0.745.0> (172.18.0.2:37224 -> 172.18.0.3:5672): user 'guest' authenticated and granted access to vhost '/'
csapi-rabbit-1  | 2023-07-22 05:19:06.532483+00:00 [info] <0.764.0> accepting AMQP connection <0.764.0> (172.18.0.2:37226 -> 172.18.0.3:5672)
csapi-rabbit-1  | 2023-07-22 05:19:06.536422+00:00 [info] <0.764.0> connection <0.764.0> (172.18.0.2:37226 -> 172.18.0.3:5672): user 'guest' authenticated and granted access to vhost '/'
csapi-worker-1  | [2023-07-22 05:19:07,547: INFO/MainProcess] mingle: all alone
csapi-worker-1  | [2023-07-22 05:19:07,583: INFO/MainProcess] celery@bdb183c81964 ready.

 

celery 동작 테스트

터미널 창을 새롭게 열어서 csapi-worker-1 컨테이너로 접속합니다.

docker exec -it csapi-worker-1 bash

 

python 쉘로 접속합니다

python
Python 3.7.16 (default, May 23 2023, 14:46:23) 
[GCC 10.2.1 20210110] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>>

 

api 함수를 임포트 합니다

from tasks import api
2023-07-22 05:24:59.360443: I tensorflow/core/platform/cpu_feature_guard.cc:193] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN) to use the following CPU instructions in performance-critical operations:  AVX2 FMA
To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags.
2023-07-22 05:24:59.483622: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcudart.so.11.0'; dlerror: libcudart.so.11.0: cannot open shared object file: No such file or directory
2023-07-22 05:24:59.483673: I tensorflow/compiler/xla/stream_executor/cuda/cudart_stub.cc:29] Ignore above cudart dlerror if you do not have a GPU set up on your machine.
2023-07-22 05:25:00.273849: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer.so.7'; dlerror: libnvinfer.so.7: cannot open shared object file: No such file or directory
2023-07-22 05:25:00.273937: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer_plugin.so.7'; dlerror: libnvinfer_plugin.so.7: cannot open shared object file: No such file or directory
2023-07-22 05:25:00.273950: W tensorflow/compiler/tf2tensorrt/utils/py_utils.cc:38] TF-TRT Warning: Cannot dlopen some TensorRT libraries. If you would like to use Nvidia GPU with TensorRT, please make sure the missing libraries mentioned above are installed properly.
>>>

 

이제 api 함수를 celery 테스크로 실행합니다.

api.delay()
<AsyncResult: 4c289a73-1a97-48b1-bf93-a078eb5b7195>

 

docker compose 로그를 확인하면 아래와 같이 테스크가 실행된 것을 확인할 수 있습니다

csapi-rabbit-1  | 2023-07-22 05:26:36.784716+00:00 [info] <0.858.0> accepting AMQP connection <0.858.0> (172.18.0.2:40688 -> 172.18.0.3:5672)
csapi-rabbit-1  | 2023-07-22 05:26:36.787461+00:00 [info] <0.858.0> connection <0.858.0> (172.18.0.2:40688 -> 172.18.0.3:5672): user 'guest' authenticated and granted access to vhost '/'
csapi-worker-1  | [2023-07-22 05:26:36,792: INFO/MainProcess] Task tasks.api[4c289a73-1a97-48b1-bf93-a078eb5b7195] received
csapi-worker-1  | 2023-07-22 05:26:37.173927: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcuda.so.1'; dlerror: libcuda.so.1: cannot open shared object file: No such file or directory
csapi-worker-1  | 2023-07-22 05:26:37.173987: W tensorflow/compiler/xla/stream_executor/cuda/cuda_driver.cc:265] failed call to cuInit: UNKNOWN ERROR (303)
csapi-worker-1  | 2023-07-22 05:26:37.174051: I tensorflow/compiler/xla/stream_executor/cuda/cuda_diagnostics.cc:156] kernel driver does not appear to be running on this host (bdb183c81964): /proc/driver/nvidia/version does not exist
csapi-worker-1  | 2023-07-22 05:26:37.174917: I tensorflow/core/platform/cpu_feature_guard.cc:193] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN) to use the following CPU instructions in performance-critical operations:  AVX2 FMA
csapi-worker-1  | To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags.
1/1 [==============================] - ETA: 0sRNING/ForkPoolWorker-1] 
1/1 [==============================] - 5s 5s/stepNG/ForkPoolWorker-1] 
csapi-worker-1  | 
csapi-worker-1  | 
csapi-worker-1  | [2023-07-22 05:26:46,292: WARNING/ForkPoolWorker-1] task ID :
csapi-worker-1  | [2023-07-22 05:26:46,292: WARNING/ForkPoolWorker-1]  
csapi-worker-1  | [2023-07-22 05:26:46,292: WARNING/ForkPoolWorker-1] 4c289a73-1a97-48b1-bf93-a078eb5b7195
csapi-worker-1  | [2023-07-22 05:26:46,292: WARNING/ForkPoolWorker-1] 
csapi-worker-1  | 
csapi-worker-1  | 
csapi-worker-1  | [2023-07-22 05:26:46,311: INFO/ForkPoolWorker-1] Task tasks.api[4c289a73-1a97-48b1-bf93-a078eb5b7195] succeeded in 9.51734412399992s: '4c289a73-1a97-48b1-bf93-a078eb5b7195'
csapi-rabbit-1  | 2023-07-22 05:26:46.424852+00:00 [warning] <0.764.0> closing AMQP connection <0.764.0> (172.18.0.2:37226 -> 172.18.0.3:5672, vhost: '/', user: 'guest'):
csapi-rabbit-1  | 2023-07-22 05:26:46.424852+00:00 [warning] <0.764.0> client unexpectedly closed TCP connection

 

celery-worker-1 컨테이너에서 python 쉘을 종료시키면, out.png가 생성된것을 확인할 수 있습니다.

exit()
root@bdb183c81964:/csapi# ls
Dockerfile  __pycache__  model.h5  out.png  requirements.txt  run.py  tasks.py  test_input.png

 

out.png 파일을 확인합니다.

docker cp csapi-worker-1:/csapi/out.png ./out.png
Successfully copied 263kB to /home/ubuntu/out.png