조건 변수 (Condition Variable)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# ❌ 락 없이 조건 검사 → 경쟁 상태 유발
from threading import Condition

cv = Condition()
queue = []

def consumer():
    # 락 없이 검사해서, 신호 손실/경쟁 상태 가능
    if not queue:
        cv.wait()  # RuntimeError (락 없이 wait), 또는 논리적 버그
    item = queue.pop(0)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# ✅ 조건 검사와 wait는 반드시 락 범위 내에서 수행
from threading import Condition, Lock

lock = Lock()
cv = Condition(lock)
queue = []

def consumer():
    with cv:  # cv가 내부적으로 lock을 사용
        while not queue:         # 반드시 while로 재검사
            cv.wait()            # wait()는 락 해제→수면→재획득을 원자적으로 수행
        item = queue.pop(0)      # 임계구역 안에서 안전하게 소비

1
2
3
4

# ❌ 신호를 먼저 보내면, 아직 대기하지 않은 스레드가 신호를 놓침
with cv:
    cv.notify()       # 신호 먼저
    queue.append(1)   # 상태 나중 → 대기 스레드는 영원히 못 깨어날 수 있음

1
2
3
4

# ✅ 공유 상태 변경 후 신호
with cv:
    queue.append(1)   # 상태 변경
    cv.notify()       # 그 다음 신호 → 대기자에게 조건이 보장됨

1
2
3
4
5

# ❌ if 한 번만 검사 → 깨어난 뒤 조건이 여전히 거짓이어도 진행함
with cv:
    if not queue:
        cv.wait()
    item = queue.pop(0)  # 비어있을 수 있음 (경쟁/가짜 깨움/다른 소비자 선점)

1
2
3
4
5

# ✅ while 재검사: 깨어난 뒤에도 반드시 조건 확인
with cv:
    while not queue:
        cv.wait()
    item = queue.pop(0)

1
2
3
4
5

# ❌ 항상 broadcast → Thundering Herd(군집 깨어남) 유발, 경합 증가
with cv:
    produced = [1, 2, 3]  # 1개만 소비 가능해도…
    queue.extend(produced)
    cv.notify_all()       # 불필요하게 모두 깨움

1
2
3
4
5
6

# ✅ 조건을 만족하는 예상 소비자 수만큼 notify
with cv:
    produced = [1, 2, 3]
    queue.extend(produced)
    for _ in range(len(produced)):  # 실제로 소비 가능한 개수만큼
        cv.notify()                  # 필요한 만큼만 깨움 → 경합/컨텍스트 스위칭 절감

1
2
3
4

# ❌ 무한 대기: 신호가 영영 오지 않으면 스레드가 고립
with cv:
    while not queue:
        cv.wait()  # 타임아웃 없음

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# ✅ 타임아웃으로 고립 방지
from time import monotonic

with cv:
    deadline = monotonic() + 2.0  # 2초 내 대기
    while not queue:
        remaining = deadline - monotonic()
        if remaining <= 0:
            break                  # 타임아웃 경로로 빠져나옴
        cv.wait(timeout=remaining)
    if queue:
        item = queue.pop(0)
    else:
        # 타임아웃 대응 로직 (재시도/로그/폴백 등)
        pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# ✅ 외부 취소 이벤트로 빠른 복귀
from threading import Event

cancel = Event()

def consumer():
    with cv:
        while not queue and not cancel.is_set():
            cv.wait(timeout=0.2)  # 짧게 깨어나 cancel 확인
        if cancel.is_set():
            return  # 정중한 종료
        item = queue.pop(0)

1
2
3
4

# ❌ 동일한 공유 상태를 두 개의 락으로 보호 → 순서 꼬임/데드락/레이스
lock1, lock2 = Lock(), Lock()
cv1, cv2 = Condition(lock1), Condition(lock2)
# queue를 lock1/lock2로 번갈아 보호… (금기)

1
2
3
4

# ✅ 공유 상태(queue)는 하나의 락과 대응되는 하나의 Condition 세트로 일관되게 보호
lock = Lock()
cv = Condition(lock)
queue = []

1
2
3

# ❌ 하나의 CV로 “비어있지 않음”과 “가득 참”을 모두 표기 → 신호 충돌/오해
cv = Condition(lock)
not_empty = not_full = cv

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# ✅ 의미가 다른 조건은 조건 변수도 분리
lock = Lock()
not_empty = Condition(lock)  # 소비자 대기
not_full  = Condition(lock)  # 생산자 대기
capacity = 10
queue = []

def producer(x):
    with not_full:
        while len(queue) == capacity:
            not_full.wait()
        queue.append(x)
        not_empty.notify()  # 소비자를 정확히 깨움

def consumer():
    with not_empty:
        while not queue:
            not_empty.wait()
        item = queue.pop(0)
        not_full.notify()   # 생산자를 정확히 깨움

1
2

# ❌ 특정 스레드만 반복적으로 유리 → 기아 상태
# 무분별한 notify_one()과 편향된 조건 갱신

1
2
3
4

# ✅ 조건 재검사 + 균형 잡힌 신호 정책
# - 깨어난 스레드는 while 재검사
# - 생산/소비 시 각각 반대쪽 조건에 공정하게 notify
# - 필요 시 주기적으로 notify_all()로 막힌 큐 해소

1
2
3
4
5
6

# ❌ 임계 구역에서 무거운 연산/IO 수행
with cv:
    while not queue:
        cv.wait()
    item = queue.pop(0)
    do_heavy_io(item)  # 락 잡은 상태로 오래 실행 → 경합 급증

1
2
3
4
5
6
7

# ✅ 데이터만 안전하게 꺼낸 뒤, 락 밖에서 무거운 작업 수행
with cv:
    while not queue:
        cv.wait()
    item = queue.pop(0)

do_heavy_io(item)  # 락 해제 후 실행 → 대기 시간/경합 감소

  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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257

import threading
import time
from collections import deque
from statistics import mean, median

# =========================
# 공통 메트릭 수집 도우미
# =========================

class BenchMetrics:
    """버퍼 내부의 대기/신호 관련 카운터를 수집한다."""
    def __init__(self):
        self.lock = threading.Lock()
        self.waits_on_full = 0    # put 시 버퍼가 가득 차서 기다린 횟수
        self.waits_on_empty = 0   # get 시 버퍼가 비어서 기다린 횟수

    def inc_full(self):
        with self.lock:
            self.waits_on_full += 1

    def inc_empty(self):
        with self.lock:
            self.waits_on_empty += 1

    def snapshot(self):
        with self.lock:
            return dict(waits_on_full=self.waits_on_full,
                        waits_on_empty=self.waits_on_empty)

# =========================
# 공통 인터페이스
# =========================

class BufferIFace:
    """put/get 인터페이스만 강제. 구현은 세 가지 스타일로 제공."""
    def put(self, item):
        raise NotImplementedError
    def get(self):
        raise NotImplementedError

# =========================
# 1) Mesa Style (표준) 구현
# - 스퓨리어스 웨이크업 & Mesa 의미론 대응: while 재검사
# =========================

class MesaBuffer(BufferIFace):
    def __init__(self, capacity=64, metrics: BenchMetrics | None = None):
        self.cap = capacity
        self.q = deque()
        self.lock = threading.Lock()
        self.cv = threading.Condition(self.lock)
        self.metrics = metrics or BenchMetrics()

    def put(self, item):
        with self.cv:  # CV는 항상 락과 페어
            while len(self.q) >= self.cap:
                self.metrics.inc_full()
                # wait는 원자적으로 unlock→sleep→relock 수행
                self.cv.wait()
            self.q.append(item)
            # 상태 변화(비어있지 않음) 알림: 보통 하나만 깨우는 것이 효율적
            self.cv.notify()

    def get(self):
        with self.cv:
            while not self.q:
                self.metrics.inc_empty()
                self.cv.wait()
            item = self.q.popleft()
            # 상태 변화(가득참→여유) 알림
            self.cv.notify()
            return item

# =========================
# 2) Hoare Style (교육용 시뮬레이션)
# - 실제 파이썬 런타임은 Mesa 시맨틱. urgent 큐를 둬서 "즉시 양도" 느낌을 흉내낸다.
# - 논리: signaler는 urgent 큐로 이동해 대기, waiter가 일을 마치면 urgent.notify()로 깨워 제어권 회수
# =========================

class HoareSimBuffer(BufferIFace):
    def __init__(self, capacity=64, metrics: BenchMetrics | None = None):
        self.cap = capacity
        self.q = deque()
        self.lock = threading.Lock()
        self.waiters = threading.Condition(self.lock)  # 대기자용
        self.urgent = threading.Condition(self.lock)   # 신호자 복귀 대기열
        self.metrics = metrics or BenchMetrics()

    def put(self, item):
        with self.lock:
            while len(self.q) >= self.cap:
                self.metrics.inc_full()
                self.waiters.wait()
            self.q.append(item)
            # 조건 충족 → waiter를 깨우고, "즉시 양도" 시뮬:
            self.waiters.notify()
            # signaler는 urgent 큐에서 잠깐 대기하여 waiter가 먼저 달리게 한다.
            # (임계영역 길이에 따라 오히려 손해일 수 있음 — 교육용!)
            self.urgent.wait(timeout=0)  # 0으로 스핀 없음, 즉시 리턴 (양도 힌트 효과만)

    def get(self):
        with self.lock:
            while not self.q:
                self.metrics.inc_empty()
                self.waiters.wait()
            item = self.q.popleft()
            self.waiters.notify()
            # waiter가 일을 마쳤으니 signaler 깨움(제어권 복원 힌트)
            self.urgent.notify()
            return item

# =========================
# 3) Predicate-based Style
# - wait_for(predicate)로 프레디킷 캡슐화 + 내부 while 재검사 자동화
# =========================

class PredicateBuffer(BufferIFace):
    def __init__(self, capacity=64, metrics: BenchMetrics | None = None):
        self.cap = capacity
        self.q = deque()
        self.lock = threading.Lock()
        self.cv = threading.Condition(self.lock)
        self.metrics = metrics or BenchMetrics()

    def _not_full(self):
        return len(self.q) < self.cap

    def _not_empty(self):
        return len(self.q) > 0

    def put(self, item):
        with self.cv:
            # 실패(타임아웃 없음 가정) 시 False 반환이지만 여기서는 성공할 때까지 대기
            while not self._not_full():
                self.metrics.inc_full()
                self.cv.wait()
            self.q.append(item)
            self.cv.notify()

    def get(self):
        with self.cv:
            while not self._not_empty():
                self.metrics.inc_empty()
                self.cv.wait()
            item = self.q.popleft()
            self.cv.notify()
            return item

# =========================
# 벤치마크 러너
# =========================

def run_benchmark(BufferCls, *,
                  capacity=64,
                  num_producers=2,
                  num_consumers=2,
                  items_per_producer=10,
                  producer_think_s=0.0,
                  consumer_think_s=0.0):
    """
    동일 인자 하에 BufferCls 구현을 벤치마크한다.
    - 각 아이템은 (t0, payload) 형태로 생산되어 소비 시 지연을 측정한다.
    - 소비자마다 목표 소비 개수를 배분하여 자연 종료하도록 한다.
    """
    metrics = BenchMetrics()
    buf = BufferCls(capacity=capacity, metrics=metrics)
    total_items = num_producers * items_per_producer

    # 소비자별 목표 개수 배분(마지막 소비자가 나머지 처리)
    base = total_items // num_consumers
    shares = [base] * num_consumers
    shares[-1] += total_items - base * num_consumers

    start_barrier = threading.Barrier(num_producers + num_consumers)
    latencies = []            # 전체 지연(ms) 수집
    lat_lock = threading.Lock()

    def record_latency_ms(t0):
        with lat_lock:
            latencies.append((time.perf_counter() - t0) * 1000.0)

    def producer(pid):
        start_barrier.wait()
        for i in range(items_per_producer):
            t0 = time.perf_counter()
            buf.put((t0, (pid, i)))  # payload: (producer_id, seq)
            if producer_think_s:
                time.sleep(producer_think_s)

    def consumer(cid, quota):
        start_barrier.wait()
        for _ in range(quota):
            t0, payload = buf.get()
            record_latency_ms(t0)
            if consumer_think_s:
                time.sleep(consumer_think_s)

    producers = [threading.Thread(target=producer, args=(p,)) for p in range(num_producers)]
    consumers = [threading.Thread(target=consumer, args=(c, shares[c])) for c in range(num_consumers)]

    t_begin = time.perf_counter()
    for th in producers + consumers:
        th.start()
    for th in producers + consumers:
        th.join()
    t_end = time.perf_counter()

    elapsed = t_end - t_begin
    m = metrics.snapshot()

    # 지연 통계
    lat_ms = latencies if latencies else [0.0]
    stats = {
        "total_items": total_items,
        "elapsed_s": elapsed,
        "throughput_items_per_s": total_items / elapsed if elapsed > 0 else float('inf'),
        "latency_ms_avg": mean(lat_ms),
        "latency_ms_p50": median(lat_ms),
        "latency_ms_max": max(lat_ms),
        "waits_on_full": m["waits_on_full"],
        "waits_on_empty": m["waits_on_empty"],
    }
    return stats

# =========================
# 실행 예시
# =========================

def pretty(name, stats):
    print(f"\n=== {name} ===")
    print(f"items={stats['total_items']}, elapsed={stats['elapsed_s']:f}s, "
          f"throughput={stats['throughput_items_per_s']:f}/s")
    print(f"latency(ms): avg={stats['latency_ms_avg']:f}, "
          f"p50={stats['latency_ms_p50']:f}, max={stats['latency_ms_max']:f}")
    print(f"waits: on_full={stats['waits_on_full']}, on_empty={stats['waits_on_empty']}")

if __name__ == "__main__":
    # 벤치마크 파라미터
    CAP = 64
    NP, NC = 4, 4
    ITEMS_PER_PROD = 500
    P_THINK, C_THINK = 0.0, 0.0

    s1 = run_benchmark(MesaBuffer, capacity=CAP, num_producers=NP, num_consumers=NC,
                       items_per_producer=ITEMS_PER_PROD,
                       producer_think_s=P_THINK, consumer_think_s=C_THINK)
    pretty("Mesa (표준 while 재검사)", s1)

    s2 = run_benchmark(HoareSimBuffer, capacity=CAP, num_producers=NP, num_consumers=NC,
                       items_per_producer=ITEMS_PER_PROD,
                       producer_think_s=P_THINK, consumer_think_s=C_THINK)
    pretty("Hoare (교육용 시뮬레이션)", s2)

    s3 = run_benchmark(PredicateBuffer, capacity=CAP, num_producers=NP, num_consumers=NC,
                       items_per_producer=ITEMS_PER_PROD,
                       producer_think_s=P_THINK, consumer_think_s=C_THINK)
    pretty("Predicate(wait_for)", s3)

 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

import threading, time, random
from collections import deque

class BoundedBuffer:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.q = deque()
        self.lock = threading.Lock()
        self.not_empty = threading.Condition(self.lock)
        self.not_full = threading.Condition(self.lock)

    def put(self, item, timeout=None):
        with self.not_full:
            # while 재검사로 스푸리어스 웨이크업/레이스 방지
            if not self.not_full.wait_for(lambda: len(self.q) < self.capacity, timeout=timeout):
                raise TimeoutError("put timed out")
            self.q.append(item)
            # 상태 변경 후 신호
            self.not_empty.notify()

    def get(self, timeout=None):
        with self.not_empty:
            if not self.not_empty.wait_for(lambda: len(self.q) > 0, timeout=timeout):
                raise TimeoutError("get timed out")
            item = self.q.popleft()
            self.not_full.notify()
            return item

def producer(buf: BoundedBuffer, pid: int):
    for i in range(20):
        item = (pid, i)
        buf.put(item, timeout=5)
        # 관측성: 간단한 로그
        print(f"[P{pid}] produced {item} (size={len(buf.q)})")
        time.sleep(random.random() * 0.05)

def consumer(buf: BoundedBuffer, cid: int):
    for _ in range(20):
        item = buf.get(timeout=5)
        print(f"[C{cid}] consumed {item} (size={len(buf.q)})")
        time.sleep(random.random() * 0.08)

if __name__ == "__main__":
    buf = BoundedBuffer(capacity=10)
    threads = [threading.Thread(target=producer, args=(buf, p)) for p in range(2)] + \
              [threading.Thread(target=consumer, args=(buf, c)) for c in range(2)]
    for t in threads: t.start()
    for t in threads: t.join()

 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

import threading
import time
from collections import deque

queue = deque()
condition = threading.Condition()

def consumer():
    while True:
        with condition:
            while not queue:
                condition.wait()  # 조건이 만족될 때까지 대기
            item = queue.popleft()
            print(f"Consumed: {item}")

def producer():
    for i in range(5):
        time.sleep(1)
        with condition:
            queue.append(i)
            print(f"Produced: {i}")
            condition.notify()

threading.Thread(target=consumer, daemon=True).start()
threading.Thread(target=producer).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
 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

import threading
import time
import random
from collections import deque
from datetime import datetime

class LogBuffer:
    """조건 변수를 활용한 로그 버퍼"""
    
    def __init__(self, max_size=100):
        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)
        
        self.running = True
    
    def produce_log(self, log_message):
        """로그 메시지 생성 (Producer)"""
        with self.not_full:  # 뮤텍스 자동 획득
            # Mesa 스타일: while 루프로 조건 재검사
            while len(self.buffer) >= self.max_size and self.running:
                print(f"Buffer full, producer waiting…")
                self.not_full.wait()  # 버퍼에 공간 생길 때까지 대기
            
            if self.running:
                # 타임스탬프와 함께 로그 저장
                timestamped_log = {
                    'timestamp': datetime.now(),
                    'message': log_message,
                    'thread_id': threading.get_ident()
                }
                self.buffer.append(timestamped_log)
                print(f"Produced: {log_message}")
                
                # 소비자에게 데이터 준비됨을 알림
                self.not_empty.notify()
    
    def consume_log(self):
        """로그 메시지 소비 (Consumer)"""
        with self.not_empty:
            # 버퍼가 비어있으면 대기
            while len(self.buffer) == 0 and self.running:
                print("Buffer empty, consumer waiting…")
                self.not_empty.wait()
            
            if self.running and self.buffer:
                log_entry = self.buffer.popleft()
                print(f"Consumed: {log_entry['message']} "
                      f"at {log_entry['timestamp']}")
                
                # 생산자에게 공간 생겼음을 알림
                self.not_full.notify()
                return log_entry
        
        return None
    
    def shutdown(self):
        """시스템 종료"""
        with self.mutex:
            self.running = False
            # 모든 대기 중인 스레드 깨우기
            self.not_full.notify_all()
            self.not_empty.notify_all()

class LogProducer(threading.Thread):
    """로그 생성 스레드"""
    
    def __init__(self, buffer, producer_id, log_count=10):
        super().__init__()
        self.buffer = buffer
        self.producer_id = producer_id
        self.log_count = log_count
    
    def run(self):
        for i in range(self.log_count):
            # 실제 로그 생성 시뮬레이션
            log_message = f"Producer-{self.producer_id} Log-{i}"
            self.buffer.produce_log(log_message)
            
            # 가변적인 로그 생성 간격
            time.sleep(random.uniform(0.1, 0.5))
        
        print(f"Producer-{self.producer_id} finished")

class LogConsumer(threading.Thread):
    """로그 처리 스레드"""
    
    def __init__(self, buffer, output_file="logs.txt"):
        super().__init__()
        self.buffer = buffer
        self.output_file = output_file
        self.processed_count = 0
    
    def run(self):
        with open(self.output_file, 'w') as f:
            while self.buffer.running or len(self.buffer.buffer) > 0:
                log_entry = self.buffer.consume_log()
                if log_entry:
                    # 파일에 로그 기록
                    f.write(f"{log_entry['timestamp']}: "
                           f"{log_entry['message']}\n")
                    f.flush()  # 즉시 디스크에 쓰기
                    self.processed_count += 1
                    
                    # 처리 시간 시뮬레이션
                    time.sleep(random.uniform(0.05, 0.2))
        
        print(f"Consumer processed {self.processed_count} logs")

# 시스템 실행 예제
def main():
    """로그 수집 시스템 메인"""
    # 공유 버퍼 생성
    log_buffer = LogBuffer(max_size=10)
    
    # 여러 생산자 스레드 생성
    producers = []
    for i in range(3):
        producer = LogProducer(log_buffer, i, log_count=5)
        producers.append(producer)
    
    # 단일 소비자 스레드 생성
    consumer = LogConsumer(log_buffer)
    
    print("Starting log collection system…")
    
    # 모든 스레드 시작
    consumer.start()
    for producer in producers:
        producer.start()
    
    # 모든 생산자 완료까지 대기
    for producer in producers:
        producer.join()
    
    print("All producers finished, shutting down…")
    
    # 버퍼 정리 후 소비자 종료
    time.sleep(1)  # 마지막 로그 처리 시간 확보
    log_buffer.shutdown()
    consumer.join()
    
    print("Log collection system shutdown complete")

if __name__ == "__main__":
    main()

 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

import threading
import time
import queue

log_queue = queue.Queue()
condition = threading.Condition()

def log_collector():
    for i in range(3):
        with condition:
            log_queue.put(f"log message {i}")
            print("[Collector] 로그 추가")
            condition.notify()
        time.sleep(1)

def log_processor():
    while True:
        with condition:
            while log_queue.empty():
                condition.wait()
            log = log_queue.get()
            print(f"[Processor] 로그 처리: {log}")

threading.Thread(target=log_processor, daemon=True).start()
threading.Thread(target=log_collector).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
 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

# pip install prometheus-client
import threading, time, queue, contextlib, random
from prometheus_client import Counter, Histogram, Gauge, start_http_server

REQS_WAIT = Histogram("pool_wait_seconds", "Time waiting for a connection")
REQS_TIMEOUT = Counter("pool_acquire_timeout_total", "Acquire timeouts")
POOL_ACTIVE = Gauge("pool_active_connections", "Active connections")
POOL_IDLE = Gauge("pool_idle_connections", "Idle connections")

class Connection:
    def __init__(self, id): self.id = id
    def healthy(self): return True
    def close(self): pass

class ConnectionPool:
    def __init__(self, min_size=2, max_size=10, check_interval=5.0):
        self.min = min_size; self.max = max_size
        self.idle = queue.deque()
        self.active = set()
        self.size = 0
        self.mu = threading.Lock()
        self.not_empty = threading.Condition(self.mu)  # idle>0
        self.not_full = threading.Condition(self.mu)   # size<max
        self._closed = False
        self._start(min_size)
        self._hc = threading.Thread(target=self._health_checker,
                                    args=(check_interval,), daemon=True)
        self._hc.start()

    def _start(self, n):
        for _ in range(n):
            self._create_locked()

    def _create_locked(self):
        assert self.size < self.max
        conn = Connection(self.size + 1)
        self.size += 1
        self.idle.append(conn)
        POOL_IDLE.set(len(self.idle))
        POOL_ACTIVE.set(len(self.active))

    @contextlib.contextmanager
    def acquire(self, timeout=3.0):
        start = time.perf_counter()
        conn = self._acquire(timeout)
        try:
            yield conn
        finally:
            self.release(conn)
        REQS_WAIT.observe(time.perf_counter() - start)

    def _acquire(self, timeout):
        with self.mu:
            end = time.time() + timeout if timeout else None
            # 대기조건: idle==0 && size==max
            while len(self.idle) == 0 and self.size == self.max and not self._closed:
                remaining = end - time.time() if end else None
                if remaining is not None and remaining <= 0:
                    REQS_TIMEOUT.inc()
                    raise TimeoutError("acquire timed out")
                self.not_full.wait(remaining)

            if self._closed: raise RuntimeError("pool closed")

            # idle이 있다면 사용
            if self.idle:
                conn = self.idle.popleft()
            else:
                # idle이 없고 size<max면 생성
                if self.size < self.max:
                    self._create_locked()
                    conn = self.idle.popleft()
                else:
                    # 방어적: 재검사 루프가 보장하지만 안전망
                    REQS_TIMEOUT.inc()
                    raise TimeoutError("acquire contention")

            self.active.add(conn)
            POOL_IDLE.set(len(self.idle)); POOL_ACTIVE.set(len(self.active))
            # 상태 변경 후 신호: idle 감소 → not_empty는 생략, not_full은 다른 waiters가 유의
            return conn

    def release(self, conn):
        with self.mu:
            if conn in self.active:
                self.active.remove(conn)
                self.idle.append(conn)
                POOL_IDLE.set(len(self.idle)); POOL_ACTIVE.set(len(self.active))
                # idle 증가 → idle을 기다리던 소비자를 깨움
                self.not_empty.notify()
                self.not_full.notify()  # size<max 대기자에게도 힌트
            else:
                # 이중 반환 방지
                pass

    def _health_checker(self, interval):
        while not self._closed:
            time.sleep(interval)
            with self.mu:
                # 비건전 커넥션 제거 및 보충
                evicted = 0
                self.idle = queue.deque([c for c in self.idle if c.healthy()])
                # size 재계산은 간단화: unhealthy 없다고 가정 (데모)
                while self.size < self.min:
                    self._create_locked()
                # 상태가 좋아졌을 수 있으니 최소한으로 깨움 (broadcast 지양)
                if len(self.idle) > 0:
                    self.not_empty.notify()
                if self.size < self.max:
                    self.not_full.notify()

    def close(self):
        with self.mu:
            self._closed = True
            # 대기자 종료
            self.not_empty.notify_all()
            self.not_full.notify_all()
            # 리소스 정리
            for c in list(self.idle): c.close()
            for c in list(self.active): c.close()

# 데모 실행
if __name__ == "__main__":
    start_http_server(8000)
    pool = ConnectionPool(min_size=2, max_size=5)

    def worker(i):
        for _ in range(30):
            try:
                with pool.acquire(timeout=1.0) as c:
                    time.sleep(random.uniform(0.01, 0.05))
                    print(f"[W{i}] got conn {c.id}")
            except TimeoutError:
                print(f"[W{i}] timeout")
                time.sleep(0.02)

    threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
    for t in threads: t.start()
    for t in threads: t.join()
    pool.close()