동시성 모델 추론 입문

서론

Concurrent inference - 2가지 방법

1. Batched Concurrency



• 하나의 모델 instance에 여러 개의 request를 batch 형태로 만들어서 추론하는 방법이다.
• 이 방법은 batch로 만드는 과정에서 trade-off가 발생한다.
• batch가 만들어지기 전까지 기다리면 latency는 길어지기 때문이다.

1. Multiprocess Concurrency



• 해당 방법은 여러 개의 모델을 띄우고, 각각의 input은 하나의 모델에서 추론하는 방식이다.
• 만약 이 방법이 가능하다면, 1번 방식과 다르게 batch로 만드는 동안 생기는 overhead없이 추론할 수 있다.

어떻게 2개를 혼용할 수 있을까?

간단한 Use Case

1. Message Pass 구조

1. Queue



python multiprocessing에서 지원하는 Queue는 SimpleQueue, Queue, JoinableQueue가 있다.
1. SimpleQueue
SimpleQueue는 non-blocking get() 기능이 없고, polling을 구현하기 위해서 .empty(), .full()을 사용할 수 있다.
2. Queue
simpleQueue와 다르게 put, get 등등 blocking에서 timeout을 걸고 사용할 수 있다.
3. JoinableQueue
Queue에서 제공하는 모든 기능을 다 갖고 있고, .join()을 사용해서 queue안에 있는 item들이 모두 처리될 때까지 blocking할 수 있다. 또한 task_done을 사용해서 queue에 대기 중인 작업이 완료되었음을 알리는 데 사용된다. (이 객체는 아직 안 사용해봐서 정확한 풀이보단 개념으로만 알고 있다는 점 참고)
💡
 Queue 과 JoinableQueue 는 "feeder" thread를 사용해서 pickling 과 queue 구현체 내부에 있는 pipe에 item을 넣기 때문에 만약 제대로 처리가 안된다면 에러가 발생할 것이다.

2. Event



Event란 boolean 변수를 setting, unsetting 시켜서 signal을 보내는 방법이다. Event object에 넘겨진 process는 .set(), .clear() 메소드를 사용해서 bool을 set, unset할 수 있다. 또한 .wait()과 .is_set() 을 사용해서 각각 blocking 과 polling 기능도 사용할 수 있다.

3. Pipe



pipe는 두 개의 프로세스들 사이에서 message를 전송하는데 사용한다. n개의 producer, n개의 multiple consumer들이 하나의 pipe를 사용하면 안된다. pipe는 duplex, simplex 방법이 존재하는데 아래 코드 블럭에 소개된 것처럼 특징이 다르다.

recv, send = mp.Pipe(duplex=False) 
# recv only receives message, send only sends message.

recv, send = mp.Pipe(duplex=True)
# both recv, send can receive and send to each other.

Producer function

1. folder 내부에서 이미지들 읽기

2. queue에 image들 넣기

제일 위 simple use case 그림에 있는 구조 중, D#의 detector들에게 들어가기 전에 queue까지의 과정을 맡는다고 보면 된다.

def read_images_into_q(images_path, queue, event, transform):
    image_list = list(Path(images_path).rglob(f"*.jpg"))

    while len(image_list) > 0:
        if queue.full():
            time.sleep(0.05)
            continue
        else:
            image_path = image_list.pop()
            image = Image.open(image_path)
            image = transform(image)
            queue.put((image, image_path))

    event.set()
    queue.join()

1. image_list는 image_path인 folder안에 존재하는 jpg 파일경로들을 리스트로 저장

2. image_list안에 1개라도 이미지 파일 경로가 있다면 진행
1. queue 안에 공간이 없다면 time.sleep(0.05)
2. 만약 a가 아니라면 image_list안에 index(0)에 있는 파일 경로추출
1. image open
2. image 전처리
3. queue에 전처리된 image와 image_path를 같이 저장

3. event의 bool을 set()으로 지정 → 이미지 관련 처리가 다 끝났음을 명시

4. 모든 item들에 대한 작업이 끝나기 전까지 queue.join()을 통해서 block시킨다.

Consumer function

1. Queue에서 이미지를 읽는다.

2. Detector를 사용해서 이미지 내의 모델 추론을 진행한다.

3. output을 만들기 위해 후처리를 진행

4. file에 output 결과를 작성한다.

def detect_objects(queue, event, detector, device, lock, output_path):
    file = open(output_path.as_posix(), "a")
    detector.eval().to(device)
    while not (event.is_set() and queue.empty()):
        try:
            image, image_path = queue.get(block=True, timeout=0.1)
        except Empty:
            continue

        with torch.no_grad():
            image = [image.to(device)]
            output = detector(image)[0]
        queue.task_done()
        handle_output(image_path, output, lock, file) # Points 3. & 4.
    file.close()

1. file 경로를 받아서 열어놓는다.

2. 만약 queue안에 task가 존재한다면 아래의 과정을 반복한다.
1. queue에서 image와 image 경로를 get한다.
2. image를 input을 넣어줄 device를 호출하고, detector에 넣어서 추론한다.
3. output을 추출하여 handle_output함수에 넘겨준다.

3. 모두 다 종료되면 file을 닫는다.

def handle_output(path, output, lock, file):
    filter_output(output)
    output_string = get_output_string(path, output)
    lock.acquire()
    file.write(output_string)
    file.flush()
    lock.release()

1. image에 존재하는 class의 label을 confidence를 기반으로 filter시킨다. 이때 filter_output을 사용한다.

2. 이후 file에 쓰기 위해 output_string을 통해 저장한다.

3. lock으로 다른 프로세스에서 file에 접근하지 못하도록 막는다.

4. 이후 write 후 flush를 사용해서 file 입출력 처리에 에러를 관리한다.

5. 이후 file에 대한 lock을 release시킨다.

caller function

def caller(device, images_path, output_path, detector_count, qsize):
    # Initialize sync structures
    queue = mp.JoinableQueue(qsize)
    event = mp.Event()
    lock = mp.Lock()

    # Initialize processes
    reader_process = mp.Process(
        target=read_images_into_q,
        args=(images_path, queue, event, transform)
    )
    detector_processes = [mp.Process(target=detect_objects,\
                args=(queue, event, get_detector(),\
                    device, lock, output_path))\
            for i in range(detector_count)]

    # Starting processes
    reader_process.start()
    [dp.start() for dp in detector_processes]

    # Waiting for processes to complete
    [dp.join() for dp in detector_processes]
    reader_process.join()

    # Closing the queue
    queue.close()

1. queue, lock을 multiprocessing에서 제공하는 객체들을 사용해서 선언한다.

2. target과 args를 정리해서 reader와 detector 프로세스를 초기화해준다.

3. reader쪽 process를 실행(단일 producer)

4. detector쪽 process들을 start()를 진행

5. join을 사용해서 해당 process들이 끝날때까지 기다린다.

6. 마지막으로 queue에 대한 작업이 끝났다면 close()를 호출해서 정리한다.

추가 내용

1. Message Passing & Synchronization : 중요
spawn 방식에서 mp.Process 생성자를 통해서 다른 프로세스로 전달되는 argument들은 serialize가 가능한 object여야 한다.

SimpleQueue를 사용하면 serialize하는데 추가 thread를 사용하지 않아서 좋지만, 제한된 기능만 제공한다는 것을 감안해야 한다.

→ 특히 queue.qsize()도 SimpleQueue에서는 불가능한데 모든 OS에서 제공을 안한다고 한다.
또한 ConnectionResetError, FileNotFoundError, BrokenPipe 등과 같은 에러는 feeder thread에서 발생하는데 이런 문제들이 발생한다면 joinalbeQueue를 사용하는 것도 권장된다.

결론

참고 자료

1. concurrent inference
Concurrent Inference
Getting multiple models to run on the GPU at the same time using PyTorch.

https://medium.com/swlh/concurrent-inference-e2f438469214

2. pytorch multiprocessing in window
Yet another multiprocessing demonstration, for PyTorch in Windows!
Hey folks, I figured this might be an excellent time to take a hard look at a barebone implementation of multi-processing for computer vision/deep learning models in PyTorch. There is a data-parallel method and not-so-clear documentation of distributed data-parallel with a lack of support for the Wi

https://www.linkedin.com/pulse/yet-another-multiprocessing-demonstration-pytorch-windows-jose/