Condition Synchronization

1
2
3
4
5
6

wait(condition_variable, mutex):
    1. mutex가 현재 스레드에 의해 소유되어야 함
    2. 현재 스레드를 condition_variable의 대기 큐에 추가
    3. mutex를 원자적으로 해제
    4. 스레드를 블로킹 상태로 전환
    5. 신호 수신 시 mutex를 다시 획득하고 반환

1
2
3
4

signal(condition_variable):
    1. condition_variable의 대기 큐에서 하나의 스레드 선택
    2. 선택된 스레드를 실행 가능 상태로 전환
    3. 스케줄러에게 스레드 실행 요청

1
2
3
4

broadcast(condition_variable):
    1. condition_variable의 대기 큐에 있는 모든 스레드 선택
    2. 모든 스레드를 실행 가능 상태로 전환
    3. 스케줄러에게 다중 스레드 실행 요청

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

import threading
import time
from collections import deque

class ProducerConsumerBuffer:
    def __init__(self, max_size):
        self.buffer = deque()  # 공유 버퍼
        self.max_size = max_size
        
        # 동기화 객체들
        self.mutex = threading.Lock()  # 상호 배제
        self.not_full = threading.Condition(self.mutex)  # 버퍼가 가득차지 않음
        self.not_empty = threading.Condition(self.mutex)  # 버퍼가 비어있지 않음
    
    def produce(self, item):
        """생산자가 아이템을 버퍼에 추가"""
        with self.not_full:  # 락 획득 및 조건 변수 연결
            # 버퍼가 가득찰 때까지 대기
            while len(self.buffer) >= self.max_size:
                print(f"Producer waiting - buffer full ({len(self.buffer)}/{self.max_size})")
                self.not_full.wait()  # 조건부 대기
            
            # 아이템 추가
            self.buffer.append(item)
            print(f"Produced: {item} (buffer size: {len(self.buffer)})")
            
            # 소비자에게 버퍼가 비어있지 않음을 알림
            self.not_empty.notify()
    
    def consume(self):
        """소비자가 버퍼에서 아이템을 취득"""
        with self.not_empty:  # 락 획득 및 조건 변수 연결
            # 버퍼가 비어있을 때까지 대기
            while len(self.buffer) == 0:
                print("Consumer waiting - buffer empty")
                self.not_empty.wait()  # 조건부 대기
            
            # 아이템 소비
            item = self.buffer.popleft()
            print(f"Consumed: {item} (buffer size: {len(self.buffer)})")
            
            # 생산자에게 버퍼가 가득차지 않음을 알림
            self.not_full.notify()
            
            return item

def producer_worker(buffer, producer_id):
    """생산자 스레드 작업 함수"""
    for i in range(5):
        item = f"item_{producer_id}_{i}"
        buffer.produce(item)
        time.sleep(0.1)  # 생산 시간 시뮬레이션

def consumer_worker(buffer, consumer_id):
    """소비자 스레드 작업 함수"""
    for i in range(5):
        item = buffer.consume()
        time.sleep(0.2)  # 소비 시간 시뮬레이션

# 사용 예제
if __name__ == "__main__":
    # 크기 3인 버퍼 생성
    buffer = ProducerConsumerBuffer(max_size=3)
    
    # 스레드 생성
    threads = []
    
    # 생산자 스레드 2개
    for i in range(2):
        t = threading.Thread(target=producer_worker, args=(buffer, i))
        threads.append(t)
        t.start()
    
    # 소비자 스레드 2개  
    for i in range(2):
        t = threading.Thread(target=consumer_worker, args=(buffer, i))
        threads.append(t)
        t.start()
    
    # 모든 스레드 완료 대기
    for t in threads:
        t.join()
    
    print("모든 작업 완료")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import threading

buffer = []
max_size = 5
condition = threading.Condition()

def producer():
    while True:
        with condition:
            while len(buffer) == max_size:
                condition.wait()  # 버퍼가 꽉 차 있으면 대기
            buffer.append(1)
            print("Produced, buffer:", buffer)
            condition.notify()  # 소비자가 대기 중이면 신호

def consumer():
    while True:
        with condition:
            while len(buffer) == 0:
                condition.wait() # 버퍼 비었으면 대기
            buffer.pop(0)
            print("Consumed, buffer:", buffer)
            condition.notify()  # 생산자가 대기 중이면 신호

threading.Thread(target=producer).start()
threading.Thread(target=consumer).start()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

import threading
import time
from collections import deque

buffer = deque()
buffer_size = 5
lock = threading.Lock()
condition = threading.Condition(lock)

def producer():
    while True:
        with condition:
            while len(buffer) >= buffer_size:
                condition.wait()  # 큐가 가득 찼다면 대기
            item = time.time()
            buffer.append(item)
            print(f"[Producer] Produced {item}")
            condition.notify()  # 소비자 깨우기
        time.sleep(1)

def consumer():
    while True:
        with condition:
            while not buffer:
                condition.wait()  # 큐가 비었다면 대기
            item = buffer.popleft()
            print(f"[Consumer] Consumed {item}")
            condition.notify()  # 생산자 깨우기
        time.sleep(2)

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)

t1.start()
t2.start()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

import threading
import time
from queue import deque

class JobQueue:
    def __init__(self):
        self.queue = deque()
        self.lock = threading.Lock()
        self.condition = threading.Condition(self.lock)

    def add_job(self, job):
        with self.condition:
            self.queue.append(job)
            print(f"[Dispatcher] Job added: {job}")
            self.condition.notify()

    def get_job(self):
        with self.condition:
            while not self.queue:
                self.condition.wait()
            job = self.queue.popleft()
            return job

class Worker(threading.Thread):
    def __init__(self, job_queue, worker_id):
        super().__init__()
        self.job_queue = job_queue
        self.worker_id = worker_id
        self.daemon = True

    def run(self):
        while True:
            job = self.job_queue.get_job()
            print(f"[Worker-{self.worker_id}] Processing job: {job}")
            time.sleep(1)  # simulate work

job_queue = JobQueue()
workers = [Worker(job_queue, i) for i in range(3)]
for w in workers:
    w.start()

# Dispatcher: simulate job addition
for i in range(10):
    job_queue.add_job(f"Task-{i}")
    time.sleep(0.5)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

class Worker(threading.Thread):
    def __init__(self, task_queue, condition):
        super().__init__()
        self.task_queue = task_queue
        self.condition = condition
    def run(self):
        while True:
            with self.condition:
                while not self.task_queue:
                    self.condition.wait()
                task = self.task_queue.pop(0)  # 조건 충족 시 작업 처리
                print("Processing:", task)

task_queue = []
condition = threading.Condition()
workers = [Worker(task_queue, condition) for _ in range(3)]
for w in workers: w.start()

# Main thread: 작업 추가 및 신호
def add_task(task):
    with condition:
        task_queue.append(task)
        condition.notify()

add_task("Job1")
add_task("Job2")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

import threading
import time
import json
from enum import Enum
from typing import Dict, Set

class NodeState(Enum):
    """노드 상태 정의"""
    NORMAL = "normal"
    MIGRATING = "migrating"
    IMPORTING = "importing"
    COMPLETED = "completed"

class RedisClusterSync:
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.state = NodeState.NORMAL
        self.keys_storage: Dict[str, str] = {}  # 키-값 저장소
        
        # 동기화 객체들
        self.state_mutex = threading.Lock()
        self.migration_complete = threading.Condition(self.state_mutex)
        self.import_ready = threading.Condition(self.state_mutex)
        
        # 마이그레이션 관련 상태
        self.migrating_keys: Set[str] = set()
        self.pending_imports: Dict[str, str] = {}
        
    def start_key_migration(self, target_node: str, keys_to_migrate: Set[str]):
        """키 마이그레이션 시작 (소스 노드)"""
        with self.state_mutex:
            print(f"[{self.node_id}] 키 마이그레이션 시작: {len(keys_to_migrate)}개 키 -> {target_node}")
            
            # 상태를 마이그레이션으로 변경
            self.state = NodeState.MIGRATING
            self.migrating_keys = keys_to_migrate.copy()
            
            # 다른 노드들이 준비될 때까지 대기
            while self.state != NodeState.COMPLETED:
                print(f"[{self.node_id}] 마이그레이션 완료 대기 중…")
                self.migration_complete.wait(timeout=1.0)
            
            print(f"[{self.node_id}] 마이그레이션 완료 확인됨")
    
    def prepare_key_import(self, source_node: str, incoming_keys: Dict[str, str]):
        """키 임포트 준비 (타겟 노드)"""
        with self.state_mutex:
            print(f"[{self.node_id}] 키 임포트 준비: {len(incoming_keys)}개 키 <- {source_node}")
            
            # 상태를 임포팅으로 변경
            self.state = NodeState.IMPORTING
            self.pending_imports = incoming_keys.copy()
            
            # 임포트 준비 완료 신호
            self.import_ready.notify_all()
    
    def execute_key_transfer(self, target_node_sync):
        """실제 키 전송 실행"""
        with self.state_mutex:
            # 마이그레이션할 키들을 준비
            keys_to_transfer = {}
            for key in self.migrating_keys:
                if key in self.keys_storage:
                    keys_to_transfer[key] = self.keys_storage[key]
            
            print(f"[{self.node_id}] 키 전송 실행: {len(keys_to_transfer)}개")
        
        # 타겟 노드에 키 전송 (실제로는 네트워크를 통해)
        target_node_sync.prepare_key_import(self.node_id, keys_to_transfer)
        
        # 타겟 노드가 준비될 때까지 대기
        with target_node_sync.import_ready:
            while target_node_sync.state != NodeState.IMPORTING:
                print(f"[{self.node_id}] 타겟 노드 임포트 준비 대기…")
                target_node_sync.import_ready.wait(timeout=0.5)
        
        # 키 전송 완료 처리
        self._complete_migration(keys_to_transfer.keys())
        target_node_sync._complete_import()
    
    def _complete_migration(self, transferred_keys):
        """마이그레이션 완료 처리 (소스 노드)"""
        with self.state_mutex:
            # 전송된 키들을 로컬에서 제거
            for key in transferred_keys:
                if key in self.keys_storage:
                    del self.keys_storage[key]
                    print(f"[{self.node_id}] 키 삭제: {key}")
            
            # 상태를 완료로 변경
            self.state = NodeState.COMPLETED
            self.migrating_keys.clear()
            
            # 마이그레이션 완료 신호
            self.migration_complete.notify_all()
            print(f"[{self.node_id}] 마이그레이션 완료 - 남은 키: {len(self.keys_storage)}")
    
    def _complete_import(self):
        """임포트 완료 처리 (타겟 노드)"""
        with self.state_mutex:
            # 펜딩된 키들을 로컬 저장소에 추가
            for key, value in self.pending_imports.items():
                self.keys_storage[key] = value
                print(f"[{self.node_id}] 키 임포트: {key} = {value}")
            
            # 상태를 완료로 변경
            self.state = NodeState.COMPLETED
            self.pending_imports.clear()
            
            print(f"[{self.node_id}] 임포트 완료 - 총 키: {len(self.keys_storage)}")
    
    def add_key(self, key: str, value: str):
        """키-값 추가 (테스트용)"""
        with self.state_mutex:
            self.keys_storage[key] = value
            print(f"[{self.node_id}] 키 추가: {key} = {value}")
    
    def get_state_info(self):
        """현재 상태 정보 반환"""
        with self.state_mutex:
            return {
                "node_id": self.node_id,
                "state": self.state.value,
                "keys_count": len(self.keys_storage),
                "migrating_keys": len(self.migrating_keys),
                "pending_imports": len(self.pending_imports)
            }

def simulate_cluster_migration():
    """클러스터 키 마이그레이션 시뮬레이션"""
    # 노드 생성
    node1 = RedisClusterSync("node1")
    node2 = RedisClusterSync("node2")
    node3 = RedisClusterSync("node3")
    
    # 초기 데이터 설정
    for i in range(10):
        node1.add_key(f"key_{i}", f"value_{i}")
    
    # 마이그레이션할 키 선택
    keys_to_migrate = {"key_0", "key_1", "key_2", "key_3", "key_4"}
    
    print("\n=== 마이그레이션 시작 ===")
    
    # 마이그레이션 스레드들
    def migration_task():
        node1.start_key_migration("node2", keys_to_migrate)
    
    def transfer_task():
        time.sleep(0.1)  # 약간의 지연
        node1.execute_key_transfer(node2)
    
    # 스레드 실행
    migration_thread = threading.Thread(target=migration_task)
    transfer_thread = threading.Thread(target=transfer_task)
    
    migration_thread.start()
    transfer_thread.start()
    
    migration_thread.join()
    transfer_thread.join()
    
    print("\n=== 마이그레이션 완료 후 상태 ===")
    print(f"Node1: {node1.get_state_info()}")
    print(f"Node2: {node2.get_state_info()}")

# 사용 예제
if __name__ == "__main__":
    simulate_cluster_migration()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

import threading
from collections import deque
import time
import random

class BoundedBlockingQueue:
    """고정 크기 큐: 두 조건 변수(notFull, notEmpty)로 상태를 제어"""
    def __init__(self, capacity: int):
        if capacity <= 0:
            raise ValueError("capacity must be > 0")
        self.capacity = capacity
        self.q = deque()
        self.lock = threading.Lock()
        self.notEmpty = threading.Condition(self.lock)
        self.notFull = threading.Condition(self.lock)

    def put(self, item, timeout: float | None = None):
        """큐가 가득 차면 notFull 조건에서 대기. timeout 지원."""
        end = None if timeout is None else time.monotonic() + timeout
        with self.notFull:  # lock도 함께 획득
            while len(self.q) >= self.capacity:
                if timeout is None:
                    self.notFull.wait()
                else:
                    remaining = end - time.monotonic()
                    if remaining <= 0:
                        raise TimeoutError("put timeout")
                    self.notFull.wait(timeout=remaining)
            self.q.append(item)
            # 상태 변경 직후 알림
            self.notEmpty.notify()  # 하나만 깨워도 충분한 패턴

    def take(self, timeout: float | None = None):
        """큐가 비었으면 notEmpty 조건에서 대기. timeout 지원."""
        end = None if timeout is None else time.monotonic() + timeout
        with self.notEmpty:
            while not self.q:
                if timeout is None:
                    self.notEmpty.wait()
                else:
                    remaining = end - time.monotonic()
                    if remaining <= 0:
                        raise TimeoutError("take timeout")
                    self.notEmpty.wait(timeout=remaining)
            item = self.q.popleft()
            # 상태 변경 직후 알림
            self.notFull.notify()
            return item

    def size(self) -> int:
        with self.lock:
            return len(self.q)

# ---- 데모: 생산자/소비자 ----
def producer(q: BoundedBlockingQueue, pid: int, n: int):
    for i in range(n):
        item = f"P{pid}-{i}"
        q.put(item, timeout=2.0)
        print(f"[Producer {pid}] put {item}, size={q.size()}")
        time.sleep(random.uniform(0.05, 0.2))

def consumer(q: BoundedBlockingQueue, cid: int, n: int):
    for _ in range(n):
        item = q.take(timeout=2.0)
        print(f"[Consumer {cid}] took {item}, size={q.size()}")
        time.sleep(random.uniform(0.1, 0.3))

if __name__ == "__main__":
    q = BoundedBlockingQueue(capacity=5)
    producers = [threading.Thread(target=producer, args=(q, 1, 10)),
                 threading.Thread(target=producer, args=(q, 2, 10))]
    consumers = [threading.Thread(target=consumer, args=(q, 1, 10)),
                 threading.Thread(target=consumer, args=(q, 2, 10))]

    for t in producers + consumers:
        t.start()
    for t in producers + consumers:
        t.join()

    print("Done. final size:", q.size())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

import logging
import time
from contextlib import contextmanager

# 동기화 전용 로거 설정
sync_logger = logging.getLogger('sync_monitor')
sync_logger.setLevel(logging.INFO)

class SyncMonitor:
    def __init__(self):
        self.metrics = {
            'lock_acquisitions': 0,
            'wait_times': [],
            'contentions': 0,
            'deadlock_detections': 0
        }
    
    @contextmanager
    def monitor_lock_acquisition(self, lock_name):
        """락 획득 모니터링 컨텍스트 매니저"""
        start_time = time.time()
        try:
            sync_logger.info(f"Lock acquisition attempt: {lock_name}")
            yield
            
            acquisition_time = time.time() - start_time
            self.metrics['lock_acquisitions'] += 1
            self.metrics['wait_times'].append(acquisition_time)
            
            if acquisition_time > 0.01:  # 10ms 이상이면 경합으로 간주
                self.metrics['contentions'] += 1
                sync_logger.warning(f"Lock contention detected: {lock_name}, time: {acquisition_time:f}s")
            
        except Exception as e:
            sync_logger.error(f"Lock acquisition failed: {lock_name}, error: {e}")
            raise
        finally:
            sync_logger.info(f"Lock released: {lock_name}")
    
    def get_performance_report(self):
        """성능 리포트 생성"""
        if not self.metrics['wait_times']:
            return "No data available"
        
        avg_wait = sum(self.metrics['wait_times']) / len(self.metrics['wait_times'])
        max_wait = max(self.metrics['wait_times'])
        contention_rate = (self.metrics['contentions'] / self.metrics['lock_acquisitions']) * 100
        
        return f"""
        동기화 성능 리포트:
        - 총 락 획득 횟수: {self.metrics['lock_acquisitions']}
        - 평균 대기 시간: {avg_wait:f}초
        - 최대 대기 시간: {max_wait:f}초
        - 경합 비율: {contention_rate:f}%
        - 데드락 탐지 횟수: {self.metrics['deadlock_detections']}
        """

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

import threading
import time
from typing import Optional

class AdaptiveSpinLock:
    """적응적 스핀락 구현"""
    
    def __init__(self, max_spin_count: int = 1000):
        self._lock = threading.Lock()
        self._max_spin_count = max_spin_count
        self._contention_history = []  # 경합 히스토리
        
    def acquire(self, timeout: Optional[float] = None) -> bool:
        """적응적 락 획득"""
        start_time = time.time()
        spin_count = self._calculate_optimal_spin_count()
        
        # 1단계: 스핀 락 시도
        for _ in range(spin_count):
            if self._lock.acquire(blocking=False):
                self._record_success(time.time() - start_time)
                return True
            time.sleep(0.000001)  # 1마이크로초 대기
        
        # 2단계: 블로킹 락 시도
        remaining_time = timeout - (time.time() - start_time) if timeout else None
        if self._lock.acquire(blocking=True, timeout=remaining_time):
            self._record_contention(time.time() - start_time)
            return True
        
        return False
    
    def release(self):
        """락 해제"""
        self._lock.release()
    
    def _calculate_optimal_spin_count(self) -> int:
        """최적 스핀 카운트 계산"""
        if not self._contention_history:
            return self._max_spin_count // 2
        
        # 최근 경합 히스토리 기반으로 적응
        recent_contentions = self._contention_history[-10:]
        avg_contention = sum(recent_contentions) / len(recent_contentions)
        
        if avg_contention < 0.001:  # 1ms 미만
            return self._max_spin_count  # 더 많이 스핀
        elif avg_contention < 0.010:  # 10ms 미만
            return self._max_spin_count // 2  # 중간 정도 스핀
        else:
            return self._max_spin_count // 10  # 적게 스핀
    
    def _record_success(self, duration: float):
        """성공적인 획득 기록"""
        pass  # 성공 시 특별한 기록 없음
    
    def _record_contention(self, duration: float):
        """경합 발생 기록"""
        self._contention_history.append(duration)
        if len(self._contention_history) > 100:
            self._contention_history.pop(0)  # 히스토리 크기 제한