[Python] Multi Processing and tqdm

이분 블로그가 진짜 도움이 많이 됐다. 

https://velog.io/@hslim8888/python-%EB%A9%80%ED%8B%B0-%ED%94%84%EB%A1%9C%EC%84%B8%EC%8A%A4%EB%A1%9C-%EC%9E%91%EC%97%85-%EC%86%8D%EB%8F%84-%ED%96%A5%EC%83%81%EC%8B%9C%ED%82%A4%EA%B8%B0

[python] 멀티 프로세스로 작업 속도 향상시키기

 

 

아래는 나의 코드

    def single_process(self, src_img_path, dest_img_path, funcs):
        img = self.get_img(src_img_path)
        for fun in funcs :
            img = getattr(self,fun)(img)
        cv2.imwrite(dest_img_path, img)

    def multiple_process(self,src_list, dest_list, n_funcs):
        with ProcessPoolExecutor(max_workers=self.n_cpu) as executor:
            list(tqdm(executor.map(self.single_process, src_list, dest_list, n_funcs), total=len(src_list)))
            
    def aug_data(self):
        process_types = [[''], ['blur'], ['flip'], ['jitter'], ['blur', 'flip']]

        for src_paths, dest_paths, category in [(self.src_mask_pathes, self.dest_mask_pathes, 'mask'),
                                                (self.src_normal_pathes, self.dest_normal_pathes, 'normal'),
                                                (self.src_incorrect_pathes, self.dest_incorrect_pathes, 'incorrect')]:
            for funcs in process_types:
                suffix = ''.join(funcs)
                mod_dest_paths = [p+f'_{suffix}.jpg' for p in dest_paths]
                self.multiple_process(src_paths, mod_dest_paths, funcs*len(src_paths))

 

 

깨달은점

1. multi processing을 잘 하기 위해서는 single process를 어떻게 구성할건지가 중요한 것 같다.

이번에 코드를 작성하면서 고민한 건, 어떻게 해야 single process를 따로 떼어내서 독립적인 메소드로 쓸 수 있을까 였다.

결론은 "getattr을 사용하자!" 였다. 클래스 안의 서로 다른 전처리 메소드를 어떻게 해야 동등한 single process로 바라볼 수 있을지 고민하였는데, getattr을 사용하면 사용하고 싶은 메소드를 str 매개변수로 보낼 수 있고, self 라는 obj 안의 메소드를 내 마음대로 꺼내어 사용할 수 있게 해주니. single process로 간단히 묶어 정리하는게  매우 편해졌다.

 

2. getattr은 에러메세지를 안내더라.

그러나 왠걸, 에러메세지도 안냈는데 이미지가 생성이 안되던 것이었다. 가장 먼저 확장자를 잘 써주었는지 확인했으나, 잘 써준 상태였었다. 결국 고민하다가, 원래 내가 구현하려 했던 기능들을 쪼개어 다른 파일로 나누어 담아서 코드를 보다 이해가기 쉽게 한 뒤에 디버깅을 돌렸다. 확인해보니 multiprocessing에 단일 함수만 적용하는 경우, ['function'] 처럼 리스트로 들어간게 아니라 'function' 으로 들어갔기 때문에 for문이 str 단위로 쪼개는 바람에 getattr이 애꿎은 'f', 'u', 'n' ... 을 self에서 검색하고 있던 것. 찾는 함수가 obj 안에 없으면 오류를 내지 않고 그냥 넘어간다는 것을 처음 알게 되었다.

이상한 점은 cv2.writer 을 안하고 그냥 뛰어넘던데 그건 왜그런지 모르겠다.

assert가 이래서 필요하구나를 또한 느꼈었다.

 

3. tqdm

with ProcessPoolExecutor(max_workers=self.n_cpu) as executor:
    list(tqdm(executor.map(self.single_process, src_list, dest_list, n_funcs), total=len(src_list)))

tqdm은 반복 가능한 객체를 감싸서 진행 상태를 시각적으로 표시합니다. 하지만 executor.map은 즉시 완료되지 않고, 결과를 나중에 제공하는 미래(future) 객체를 반환합니다. 따라서 tqdm을 executor.map과 직접 사용하는 것은 바람직하지 않습니다. 대신, list() 함수를 사용하여 executor.map의 결과를 명시적으로 리스트로 변환할 수 있습니다. 이렇게 하면 tqdm이 진행 상태를 제대로 표시할 수 있습니다.  - gpt4

 

정리하자면, 여기서 executor.map은 일반적인 반복 가능한 객체를 반환하지 않는다.

작업이 완료된 후 결과를 주게 되는데, 이를 list로 감싸게 되면 tqdm이 완료된 것들을 체크하면서 로딩 바를 만들 수 있게 되나보다.