Python Multiprocessing: CPU 코어 활용 완벽 가이드
Python multiprocessing 모듈을 활용하여 CPU 코어를 최대한 활용하는 방법과 실전 예제를 알아봅니다.
Python Multiprocessing: CPU 코어 활용 완벽 가이드
Python의 GIL(Global Interpreter Lock)은 여러 스레드가 동시에 Python bytecode를 실행하는 것을 제한합니다. 이를 극복하기 위해 multiprocessing 모듈은 각 프로세스가 독립적인 Python 인터프리터를 사용하여 진정한 병렬 처리를 가능하게 합니다.
Environment
이 튜토리얼에서 사용하는 환경은 다음과 같습니다:
$ python --version
Python 3.11.5
$ uname -a
Linux workstation 5.15.0-78-generic #85-Ubuntu SMP x86_64 GNU/Linux
$ nproc
8
$ cat /proc/cpuinfo | grep "model name" | head -1
model name: AMD Ryzen 7 5800X 8-Core ProcessorProblem: CPU 코어를 제대로 활용하지 못하는 코드
다음 코드는 이미지 처리 작업을 병렬로 수행하려고 하지만, 실제로는 하나의 코어만 사용하고 있습니다:
import time
from PIL import Image
import os
def process_image(filepath):
img = Image.open(filepath)
img = img.resize((800, 600))
img = img.convert('RGB')
output_path = f"processed_{os.path.basename(filepath)}"
img.save(output_path)
return output_path
def main():
image_dir = "./images"
files = [os.path.join(image_dir, f) for f in os.listdir(image_dir)
if f.endswith(('.png', '.jpg', '.jpeg'))]
start = time.time()
# This runs sequentially - only one core is used
results = [process_image(f) for f in files]
elapsed = time.time() - start
print(f"Processed {len(results)} images in {elapsed:.2f} seconds")
if __name__ == "__main__":
main()실행 결과:
$ python process_images.py
Processed 500 images in 47.23 seconds
$ top -bn1 | grep "Cpu(s)"
Cpu(s): 12.5%us, 2.1%sy, 0.0%ni, 85.2%id, 0.2%waCPU 사용률이 12.5%에 불과합니다. 이는 8코어 중 1코어만 사용하고 있음을 의미합니다.
Analysis: GIL과 multiprocessing의 관계
Python의 GIL은 CPython에서 메모리 관리를 위한 레이스 컨디션을 방지하지만, CPU 바운드 작업에서는 병목이 됩니다. multiprocessing은 각 프로세스에 별도의 GIL을 할당하여 이 문제를 해결합니다:
import multiprocessing
import os
def get_process_info():
return f"PID: {os.getpid()}, Process Name: {multiprocessing.current_process().name}"
if __name__ == "__main__":
# Create a process pool
with multiprocessing.Pool(processes=4) as pool:
results = pool.map(get_process_info, range(4))
for r in results:
print(r)출력:
PID: 12345, Process Name: SpawnPoolWorker-1
PID: 12346, Process Name: SpawnPoolWorker-2
PID: 12347, Process Name: SpawnPoolWorker-3
PID: 12348, Process Name: SpawnPoolWorker-4각 워커가 별도의 프로세스로 실행되는 것을 확인할 수 있습니다.
Solution: multiprocessing.Pool을 활용한 병렬 처리
import time
from PIL import Image
import os
from multiprocessing import Pool, cpu_count
def process_image(filepath):
img = Image.open(filepath)
img = img.resize((800, 600))
img = img.convert('RGB')
output_path = f"processed_{os.path.basename(filepath)}"
img.save(output_path)
return output_path
def main():
image_dir = "./images"
files = [os.path.join(image_dir, f) for f in os.listdir(image_dir)
if f.endswith(('.png', '.jpg', '.jpeg'))]
# Use all available CPU cores
num_cores = cpu_count()
print(f"Using {num_cores} CPU cores")
start = time.time()
# Process images in parallel
with Pool(processes=num_cores) as pool:
results = pool.map(process_image, files)
elapsed = time.time() - start
print(f"Processed {len(results)} images in {elapsed:.2f} seconds")
if __name__ == "__main__":
main()실행 결과:
$ python process_images_parallel.py
Using 8 CPU cores
Processed 500 images in 6.87 seconds
$ top -bn1 | grep "Cpu(s)"
Cpu(s): 95.3%us, 3.2%sy, 0.0%ni, 1.1%id, 0.4%wa처리 시간이 47초에서 7초로 85% 이상 감소했습니다.
Process Pool의 다양한 활용법
chunksize 최적화
from multiprocessing import Pool
import time
def heavy_computation(n):
total = sum(i * i for i in range(n))
return total
if __name__ == "__main__":
data = list(range(1, 1000001))
# Default chunksize
start = time.time()
with Pool(8) as pool:
result1 = pool.map(heavy_computation, data)
print(f"Default chunksize: {time.time() - start:.2f}s")
# Optimized chunksize
start = time.time()
with Pool(8) as pool:
result2 = pool.map(heavy_computation, data, chunksize=10000)
print(f"Chunksize 10000: {time.time() - start:.2f}s")map_async로 비동기 처리
from multiprocessing import Pool
import time
def fetch_data(url):
# Simulate network request
time.sleep(0.1)
return f"Data from {url}"
if __name__ == "__main__":
urls = [f"http://api.example.com/data/{i}" for i in range(100)]
with Pool(10) as pool:
# Non-blocking call
async_result = pool.map_async(fetch_data, urls)
# Do other work while processing
print("Processing started, doing other work...")
# Get results when needed
results = async_result.get(timeout=60)
print(f"Fetched {len(results)} URLs")shared memory를 위한 Manager
from multiprocessing import Process, Manager
def worker(shared_dict, key, value):
shared_dict[key] = value
if __name__ == "__main__":
with Manager() as manager:
shared_dict = manager.dict()
processes = []
for i in range(5):
p = Process(target=worker, args=(shared_dict, f"key_{i}", i * 10))
processes.append(p)
p.start()
for p in processes:
p.join()
print(dict(shared_dict))Lessons Learned
GIL 이해하기: Python의 GIL은 CPU 바운드 작업에서 병렬성을 제한하지만, I/O 바운드 작업에서는 스레드가 여전히 유용합니다.
적절한 프로세스 수: CPU 코어 수와 동일한 프로세스를 생성하는 것이 항상 최선은 아닙니다. 작업 유형에 따라 조정이 필요합니다.
데이터 직렬화: multiprocessing은 프로세스 간 데이터를 직렬화하여 전달하므로, 큰 데이터 구조의 전달은 오버헤드가 발생할 수 있습니다.
Pool vs Process: 간단한 작업에는 Pool을, 복잡한 워커 로직에는 개별 Process를 사용하는 것이 좋습니다.
환경 변수 상속: multiprocessing.Pool은 부모 프로세스의 환경 변수를 상속하지만, fork 시점에 주의가 필요합니다.
This blog does not accept any external sponsorships, affiliate marketing, or ad revenue.