import threading
from threading import RLock
class ThreadSafeCounter:
def __init__(self):
self._value = 0
self._lock = RLock()
def increment(self):
with self._lock: # 첫 번째 획득
self._value += 1
self.validate() # 같은 락을 다시 획득해야 함
def validate(self):
with self._lock: # 재진입! 카운트 증가
if self._value < 0:
raise ValueError("Invalid counter value")
def get_and_double(self):
with self._lock: # 첫 번째 획득
current = self._value
self.increment() # increment()가 다시 락 획득
return current * 2
# 실행 흐름:
# counter.get_and_double() 호출
# → RLock 획득 (카운트: 1)
# → increment() 호출
# → RLock 재획득 (카운트: 2)
# → validate() 호출
# → RLock 재획득 (카운트: 3)
# → validate 완료, release (카운트: 2)
# → increment 완료, release (카운트: 1)
# → get_and_double 완료, release (카운트: 0, 락 해제)
import threading
from threading import RLock
class RecursiveProcessor:
def __init__(self):
self._lock = RLock()
self._data = []
def process_recursive(self, items, depth=0):
with self._lock: # 재귀 호출마다 같은 락 획득
print(f"Processing at depth {depth}: {items}")
self._data.append(f"depth_{depth}")
if depth < 3 and items:
# 재귀 호출 - 같은 락을 다시 획득
self.process_recursive(items[1:], depth + 1)
def get_results(self):
with self._lock: # 다른 메서드에서도 같은 락 사용
return self._data.copy()
# 실행 시:
# process_recursive([1,2,3], 0) 호출
# → RLock 획득 (카운트: 1)
# → process_recursive([2,3], 1) 호출
# → RLock 재획득 (카운트: 2)
# → process_recursive([3], 2) 호출
# → RLock 재획득 (카운트: 3)
# → process_recursive([], 3) 호출
# → RLock 재획득 (카운트: 4)
# ← RLock 해제 (카운트: 3)
# ← RLock 해제 (카운트: 2)
# ← RLock 해제 (카운트: 1)
# ← RLock 해제 (카운트: 0, 완전 해제)
import threading
import time
from threading import Lock, RLock
class DeadlockDemo:
def __init__(self, use_rlock=True):
self._lock = RLock() if use_rlock else Lock()
def outer_method(self):
print("Outer method starting...")
with self._lock:
print("Outer method got lock")
time.sleep(0.1)
self.inner_method() # 같은 락을 다시 획득 시도
print("Outer method completing...")
def inner_method(self):
print("Inner method starting...")
with self._lock: # RLock: 성공, Lock: 데드락!
print("Inner method got lock")
time.sleep(0.1)
print("Inner method completing...")
# RLock 사용 - 정상 동작
rlock_demo = DeadlockDemo(use_rlock=True)
threading.Thread(target=rlock_demo.outer_method).start()
# Lock 사용 - 데드락 발생
lock_demo = DeadlockDemo(use_rlock=False)
# threading.Thread(target=lock_demo.outer_method).start() # 주석 처리 (데드락 방지)
성능 비교 코드
import threading
import time
from threading import Lock, RLock
def performance_test():
iterations = 1000000
# Regular Lock 테스트
regular_lock = Lock()
start_time = time.time()
for _ in range(iterations):
with regular_lock:
pass # 빈 critical section
regular_time = time.time() - start_time
# RLock 테스트
reentrant_lock = RLock()
start_time = time.time()
for _ in range(iterations):
with reentrant_lock:
pass # 빈 critical section
rlock_time = time.time() - start_time
print(f"Regular Lock: {regular_time:.4f}초")
print(f"RLock: {rlock_time:.4f}초")
print(f"오버헤드: {((rlock_time - regular_time) / regular_time * 100):.2f}%")
# 일반적으로 RLock은 Regular Lock보다 10-30% 느림