동시성 이해하기: 공장 모델과 GIL
공장 생산 라인을 생각해보자. 단일 스레드는 한 명의 작업자가 순차적으로 모든 공정을 처리하는 것과 같다. 다중 스레드는 여러 작업자가 각자의 작업을 동시에 수행하는 것이다. 그러나 Python에는 전역 인터프리터 잠금(GIL)이라는 특수한 규칙이 있어서, 특정 시점에 오직 하나의 작업자만 핵심 공정(CPU)을 사용할 수 있다.
import threading
import time
class FactoryWorker:
def __init__(self, task_name, duration):
self.task_name = task_name
self.duration = duration
def execute(self):
worker_name = threading.current_thread().name
print(f"[{worker_name}] {self.task_name} 시작")
time.sleep(self.duration)
print(f"[{worker_name}] {self.task_name} 완료")
tasks = [
FactoryWorker("부품 조립", 2),
FactoryWorker(" 품질 검사", 1),
FactoryWorker("포장 작업", 1.5)
]
workers = [threading.Thread(target=task.execute) for task in tasks]
for worker in workers:
worker.start()
for worker in workers:
worker.join()
print("모든 생산 공정 종료")
GIL 메커니즘 심층 분석
Python의 전역 인터프리터锁은 메모리 관리 안전성을 위해 존재한다. 객체 참조 횟수를 추적하는 참조 카운팅 방식이 원자적 연산이 불가능하기 때문에, 이锁이 필수적이다. 그 결과로 다음과 같은 특성이 나타난다:
- 계산 집약적인 작업: 스레드 간 lock 경쟁으로 인해 오히려 성능 저하 발생
- I/O 집약적인 작업: I/O 대기 중에는 GIL이 해제되므로 동시성 이점 획득
import threading
import time
def compute_heavy():
result = 0
for _ in range(8000000):
result += (result * 2) % 1000
return result
# 단일 스레드 측정
single_start = time.perf_counter()
compute_heavy()
compute_heavy()
single_elapsed = time.perf_counter() - single_start
print(f"단일 스레드 소요 시간: {single_elapsed:.4f}초")
# 다중 스레드 측정
multi_start = time.perf_counter()
thread_a = threading.Thread(target=compute_heavy)
thread_b = threading.Thread(target=compute_heavy)
thread_a.start()
thread_b.start()
thread_a.join()
thread_b.join()
multi_elapsed = time.perf_counter() - multi_start
print(f"다중 스레드 소요 시간: {multi_elapsed:.4f}초")
print(f"성능 향상 비율: {single_elapsed/multi_elapsed:.2f}x")
스레드 안전성 구현方案
- 기본 mutual exclusion
import threading
stock_count = 100
inventory_lock = threading.Lock()
def update_inventory(change):
global stock_count
with inventory_lock:
current = stock_count
time.sleep(0.001) # 문맥 전환 유발
stock_count = current + change
print(f"재고 변동: {change}, 현재: {stock_count}")
threads = []
for delta in [10, -20, 15, -5]:
threads.append(threading.Thread(target=update_inventory, args=(delta,)))
for t in threads:
t.start()
for t in threads:
t.join()
print(f"최종 재고: {stock_count}")
- 재진입 locking
import threading
class ResourceManager:
def __init__(self):
self.reentrant_lock = threading.RLock()
self.resource_a = 0
self.resource_b = 0
def process_a(self, value):
with self.reentrant_lock:
self.resource_a += value
if value > 50:
self.process_b(value // 2)
def process_b(self, value):
with self.reentrant_lock:
self.resource_b += value
self.process_a(self.resource_b)
manager = ResourceManager()
manager.process_a(100)
- 조건 변수를 활용한 조율
import threading
import queue
message_queue = queue.Queue()
processing_done = threading.Event()
def message_consumer():
while not processing_done.is_set():
try:
message = message_queue.get(timeout=0.5)
print(f"수신 처리: {message}")
message_queue.task_done()
except queue.Empty:
continue
def message_producer():
messages = ["데이터 A", "데이터 B", "데이터 C", "종료 신호"]
for msg in messages:
message_queue.put(msg)
processing_done.set()
producer = threading.Thread(target=message_producer)
consumer = threading.Thread(target=message_consumer)
producer.start()
consumer.start()
producer.join()
processing_done.set()
consumer.join()
스레드 풀 구현 최적화
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
class DataFetcher:
@staticmethod
def fetch(url_id):
time.sleep(0.8) # 네트워크 지연 시뮬레이션
return {"id": url_id, "status": "success"}
urls = list(range(1, 11))
results = []
with ThreadPoolExecutor(max_workers=4) as executor:
future_mapping = {executor.submit(DataFetcher.fetch, url): url for url in urls}
for future in as_completed(future_mapping):
original_id = future_mapping[future]
try:
data = future.result()
results.append(data)
print(f"URL {original_id} 처리 완료")
except Exception as error:
print(f"URL {original_id} 오류: {error}")
print(f"총 처리 건수: {len(results)}")
성능 최적화 전략
작업 특성에 따른 최적 접근법:
I/O 집약적 워크로드의 경우:
- 스레드 풀 크기를 CPU 코어 수의 5배로 설정
- 비동기 I/O(asyncio)와 병행 사용 검토
계산 집약적 워크로드의 경우:
- multiprocessing 모듈로 전환
- 핵심 알고리즘을 Cython으로 컴파일
스레드 상태 모니터링:
import threading
def monitor_threads():
active_threads = threading.enumerate()
print(f"활성 스레드 수: {len(active_threads)}")
for t in active_threads:
status = "실행 중" if t.is_alive() else "대기"
print(f" - {t.name}: {status}")
monitor_threads()
Python 동시성 모듈의 진화
Python 3.2부터 도입된 concurrent.futures는 스레드/프로세스 관리의 추상화 수준을 높였다:
from concurrent.futures import ProcessPoolExecutor, as_completed
import math
def prime_check(number):
if number < 2:
return False
for i in range(2, int(math.sqrt(number)) + 1):
if number % i == 0:
return False
return True
def process_range(start, end):
primes = []
for num in range(start, end):
if prime_check(num):
primes.append(num)
return primes
ranges = [(1, 5000), (5001, 10000), (10001, 15000)]
with ProcessPoolExecutor(max_workers=3) as executor:
futures = [executor.submit(process_range, r[0], r[1]) for r in ranges]
total_primes = 0
for future in as_completed(futures):
result = future.result()
total_primes += len(result)
print(f"구간 처리 완료: {len(result)}개의 소수")
print(f"총 소수 개수: {total_primes}")
일반적인 문제 해결 가이드
데드_LOCK 발생 사례
import threading
import time
resource_alpha = threading.Lock()
resource_beta = threading.Lock()
def operation_alpha():
with resource_alpha:
time.sleep(0.5)
with resource_beta:
print("알파 작업 완료")
def operation_beta():
with resource_beta:
time.sleep(0.5)
with resource_alpha:
print("베타 작업 완료")
# 해결 방법: Lock 확보 순서를统一
해결 방안:
- Lock 확보 순서를 항상 동일한 순서로 유지
- RLock 대신 LockHierarchy 사용
- 타임아웃 механизм 적용
스레드 누수 감지
import threading
import weakref
active_refs = set()
original_start = threading.Thread.start
def patched_start(self):
active_refs.add(weakref.ref(self))
original_start(self)
threading.Thread.start = patched_start
def check_leaks():
alive = sum(1 for ref in active_refs if ref() is not None)
print(f"현재 활성 스레드 수: {alive}")
return alive