Python의 동시성(Concurrency)과 병렬성(Parallelism)
Python에서 동시성(Concurrency)이나 병렬성(Parallelism)을 구현할 때
가장 많이 사용되는 도구가 concurrent.futures의
ThreadPoolExecutorProcessPoolExecutor
이다.
이 둘의 차이를 작업의 병목이 어디에 있는지(CPU vs I/O) 관점에서 이해하지 못하면
- 성능이 거의 안 나오거나
- 불필요하게 자원을 소모하거나
- 심하면 프로세스 폭증 / 메모리 부족으로 프로그램이 죽는 상황까지 발생할 수 있다.
핵심 개념: I/O-bound vs CPU-bound
Executor 선택의 기준은 단 하나다.
“이 작업이 느린 이유가 계산 때문인가, 기다림 때문인가?”
I/O-bound 작업
- CPU는 대부분 놀고 있음
- 네트워크, 디스크, DB 응답을 기다리는 시간이 병목
- I/O 대기 중에는 Python이 GIL을 놓음
예:
- 웹 요청 (
requests,aiohttp) - 파일 읽기/쓰기
- DB 쿼리
sleep, socket 통신
-> ThreadPoolExecutor에 매우 적합
CPU-bound 작업
- CPU가 계속 계산 중
- 대기 없음, 순수 연산이 병목
- Python의 GIL 때문에 스레드 병렬화 불가
예:
- 이미지 처리
- 대규모 반복 계산
- 텍스트 분석, OCR
- 순수 Python 수치 연산
-> ProcessPoolExecutor가 필요
ThreadPoolExecutor vs ProcessPoolExecutor 비교
| 항목 | ThreadPoolExecutor | ProcessPoolExecutor |
|---|---|---|
| 실행 단위 | 스레드(Thread) | 프로세스(Process) |
| 적합한 작업 | I/O-bound | CPU-bound |
| 병목 해결 방식 | I/O 대기 시간 겹치기 | CPU 코어 병렬 사용 |
| 메모리 공간 | 프로세스 내부 공유 | 프로세스별 분리 |
| 직렬화(Pickle) | 필요 없음 | 필요 |
| GIL | 공유됨 (우회 불가) | 분리됨 (우회 가능) |
| 생성 비용 | 가볍고 빠름 | 무겁고 느림 |
ThreadPoolExecutor
적절한 상황: I/O 중심 작업
- 웹 요청
- 파일 입출력
- 데이터베이스 쿼리
- 대기 시간이 긴 작업
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=10) as executor:
results = executor.map(fetch_url, urls)장점
- 가볍고 빠름
- 메모리 공유 가능
- Pickle 필요 없음
주의점
- GIL 때문에 CPU 계산은 병렬화되지 않음
- CPU-bound 작업에 쓰면 거의 효과 없음
ProcessPoolExecutor
적절한 상황: CPU 연산 중심 작업
- 이미지 처리
- 머신러닝 전처리
- 수치 계산
- OCR, 텍스트 분석
- 대규모 반복 연산
from concurrent.futures import ProcessPoolExecutor
with ProcessPoolExecutor(max_workers=4) as executor:
results = executor.map(process_image, image_paths)장점
- 프로세스마다 독립된 GIL → 진짜 병렬 처리
- CPU 코어 수만큼 성능 향상 가능
- CPU-bound 작업에서 ThreadPoolExecutor보다 압도적으로 빠름
주의점
- 프로세스 생성 비용이 큼 (스레드보다 훨씬 무거움)
- 프로세스 간 메모리 공유 불가
- 작업 함수와 인자를 Pickle로 직렬화해야 함
- 데이터 크기가 클수록 직렬화/복사 비용 증가
- 메모리 사용량이 빠르게 증가할 수 있음
Pickle 이슈 (ProcessPoolExecutor의 핵심 함정)
ProcessPoolExecutor는
-> 작업 함수와 인자를 Pickle로 직렬화하여 다른 프로세스에 전달한다.
이로 인해 다음과 같은 제약이 발생한다.
| Pickle 가능 | Pickle 불가 |
|---|---|
| int, str, list, dict | lambda |
| top-level 함수 | 내부 함수 |
| 단순 데이터 구조 | 열린 파일 객체 |
| DB 커넥션 | |
| GPU 세션 | |
| 소켓, 핸들 |
def outer():
def inner(x):
return x * x
# inner는 pickle 불가 → ProcessPoolExecutor에서 오류 발생자주 발생하는 오류 메시지:
AttributeError: Can't pickle local object ...Executor 사용 시 반드시 지켜야 할 원칙
- 동시 실행 수를 반드시 제한해야 함
- Executor는 병목을 해결하는 도구이지, 무제한 병렬 버튼이 아님
- CPU-bound 작업은 CPU 코어 수 이상 늘려도 효과 없음
import asyncio
semaphore = asyncio.Semaphore(4)작업 유형별 추천 전략
| 작업 유형 | 추천 Executor | 이유 |
|---|---|---|
| 웹 크롤링, 파일 입출력 | ThreadPoolExecutor | I/O 대기 중 GIL 해제 |
| 이미지 처리, NLP | ProcessPoolExecutor | CPU 병렬 처리 |
| 모델 추론, GPU 처리 | ThreadPoolExecutor / asyncio | GPU 세션 공유 필요 |
| CPU + I/O 혼합 | asyncio + ProcessPoolExecutor | 역할 분리 필요 |
결과가 섞이지 않을까? (ThreadPoolExecutor / ProcessPoolExecutor)
결론부터
함수 단위로 독립 처리하고, 공유 상태가 없다면 결과는 섞이지 않는다
안전한 패턴 (asyncio + Executor)
results = await asyncio.gather(*[
loop.run_in_executor(pool, wrapper, file)
for file in files
])- 각 wrapper(file)은 독립 실행
- 전역 상태 없음
gather()는 입력 순서대로 결과 반환
-> 병렬 실행이지만 결과는 항상 안정적
섞일 수 있는 위험한 상황
ThreadPoolExecutor나 ProcessPoolExecutor를 사용한다고 해서
결과가 자동으로 섞이거나 꼬이지는 않는다.
문제가 발생하는 대부분의 경우는 공유 상태(shared state) 를 잘못 다룰 때다.
1. 전역 변수 사용 (Race Condition)
여러 스레드 또는 프로세스가
같은 전역 변수나 객체를 동시에 읽고/쓰면 실행 순서가 보장되지 않는다.
위험한 예
counter = 0
def work():
global counter
counter += 1- 여러 갑업이 동시에
counter를 수정 - 실행할 때마다 결과가 달라질 수 있음
- 디버깅이 매우 어려움
해결원칙
- 전역 상태 제거
- 함수는 입력 -> 출력만 담당하도록 설계
- 정말 필요하다면
Lock사용 (권장하지 않음)
2. 파일 저장 이름 충돌
병력 작업이 동일한 파일 이름으로 결과를 저장하면 서로 덮어쓰거나 파일이 손상될 수 있다.
위험한 예
def save_result(data):
with open("result.json", "w") as f:
f.write(data)- 여러 작업이 동시에 같은 파일에 접근
- 결과 유실 또는 파일 깨짐 발생
해결 방법
- 입력마다 고유한 ID 사용
- 파일명에 ID 또는 UUID 포함
save_path = f"output/{file_id}.json"3. 순서에 의존하는 로직
병렬 실행에서는 작업 완료 순서 ≠ 입력 순서다.
입력 순서를 가정한 로직은 쉽게 깨진다.
위험한 예
results = []
def work(x):
results.append(x)- 작업 완료 순서대로 append
- 입력 순서 보장 안됨
해결 방법
- 입력과 결과를 함께 묶어서 반환
- 순서를 명시적으로 관리
def wrapper(idx, value):
return idx, process(value)안전한 병렬 처리 패턴
asyncio + Executor + gather
results = await asyncio.gather(*[
loop.run_in_executor(pool, wrapper, file_id, file)
for file_id, file in files
])- 각 작업은 독립적으로 실행
- 공유 상태 없음
gather()는 입력 순서대로 결과 반환
-> 병렬 실행이지만 결과는 안정적으로 관리됨
실전에서 사용하는 방법
1. 입력 식별자 유지
def wrapper(file_id, file):
return {file_id: process(file)}2. 결과를 dict로 관리
filnal = {}
for item in results:
final.update(item)3. 출력 파일명 충돌 방지
save_path = f"output/{file_id}.json"
끄적끄적