[python] Multiprocessing (feat. ray)

멀티 프로세싱 관련 총정리

ray

import os

ray.init(num_cpus=os.cpu_count())

@ray.remote
def print_current_datetime():
	time.sleep(0.3)
	current_datetime = datetime.datetime.now()
	print(current_datetime)
	return current_datetime

# async
futures = [print_current_datetime.remote() for i in range(8)]

# sync
ray.get(futures)

multiprocessing

Process, No manager

import time
import multiprocessing
import os


def heavy_work(name):
    result = 0
    for i in range(4000000):
        result += i
    print("%s done" % name)


if __name__ == "__main__":
    # CPU 개수 확인
    num_cpu = os.cpu_count()
    start = time.time()
    procs = []
    for i in range(num_cpu):
        p = multiprocessing.Process(
            target=heavy_work, args=(i,)
        )  # 인자가 1개일 경우에는 Tuple 형태로 만들어서 넘겨줌
        procs.append(p)

    for p in procs:
        p.start()  # 프로세스 시작

    for p in procs:
        p.join()  # 프로세스가 모두 종료될 때까지 대기

    end = time.time()

    print("수행시간: %f 초" % (end - start))

6 done
2 done
7 done
1 done
0 done
4 done
3 done
5 done
수행시간: 0.220689 초

Process, Manager

import time
import multiprocessing
import os


def heavy_work(name, return_dict):
    result = 0
    for i in range(4000000):
        result += i
    return_dict[name] = result
    print(f"{name} done")


if __name__ == "__main__":
    # CPU 개수 확인
    num_cpu = os.cpu_count()
    start = time.time()

    # 리턴값을 모아줄 객체 생성
    manager = multiprocessing.Manager()
    return_dict = manager.dict()
    procs = []

    for i in range(num_cpu):
        p = multiprocessing.Process(target=heavy_work, args=(i, return_dict))
        procs.append(p)

    for p in procs:
        p.start()  # 프로세스 시작

    for p in procs:
        p.join()  # 프로세스가 모두 종료될 때까지 대기

    end = time.time()

    for k, v in return_dict.items():
        print(f"{k}: {v}")
    print("수행시간: %f 초" % (end - start))

0 done
4 done
1 done
2 done
6 done
5 done
7 done
3 done
0: 7999998000000
4: 7999998000000
1: 7999998000000
2: 7999998000000
6: 7999998000000
5: 7999998000000
7: 7999998000000
3: 7999998000000
수행시간: 0.262772 초

Pool

import multiprocessing


def calculate(a, b, c):
    return a * b + c


if __name__ == "__main__":
    # 입력 값 리스트 생성
    input_list = [(1, 2, 3), (4, 5, 6), (7, 8, 9), (10, 11, 12)]

    # 프로세스 풀 생성
    pool = multiprocessing.Pool(processes=4)

    # 병렬 처리를 위한 map 메서드 호출 (map은 인자1개, stamap은 1개 이상)
    output_list = pool.starmap(calculate, input_list)

    # 프로세스 풀 종료
    pool.close()
    pool.join()

    print(output_list)
    # [5, 26, 65, 143]

Pool, tqdm

import time
from multiprocessing import Pool
from tqdm import tqdm


def func(x):
    time.sleep(1)
    return


if __name__ == "__main__":
    pool = Pool()
    total = 10
    with tqdm(total=total) as pbar:
        for _ in tqdm(pool.imap_unordered(func, range(total))):
            pbar.update(1)
    pool.close()
    pool.join()

---

참고 자료

• https://tempdev.tistory.com/27?category=845382
• https://zzsza.github.io/mlops/2021/01/03/python-ray/