1
2
3
4
5
6

| 구분 | 항목 | 설명 | 기술적 근거 |
|------|------|------|-------------|
| 장점 | 데이터 무결성 | 동시 접근에도 값 오염 없음 | 원자적 실행|
| 장점 | 고성능 | Lock 불필요, 경량 동기화 실현 | 하드웨어 직접 명령|
| 장점 | 확장성 | 멀티코어 환경에 적합 | Lock-Free 알고리즘 구현 가능|
| 장점 | 신뢰성 | 병렬 처리시 예측 가능 행동 | Memory Fence, CAS 등|

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

단점
| 구분 | 항목 | 설명 | 해결책 | 대안 기술 |
|------|------|------|--------|----------|
| 단점 | 연산 범위 제한 | 복잡·다중 필드에는 적용 곤란 | Lock, 트랜잭션 활용 | 세마포어 등 SW 동기화|

문제점  
| 구분 | 항목 | 원인 | 영향 | 탐지/진단 | 예방 방법 | 해결 기법 |
|------|------|------|------|-----------|----------|----------|
| 문제점| False Sharing | 캐시라인 공유 | 성능 저하 | 프로파일링 | 변수 정렬 | Lock Striping|
| 문제점| ABA 문제 | CAS 재시도시 동일값 오염 | CAS 실패, 데이터 경쟁 | CAS 체크 | Version Tag | 더블워드 CAS|

1
2
3
4
5
6

| 분류 기준 | 유형 | 설명 |
|-----------|------|------|
| CPU 아키텍처| Hardware Atomic| 단일 명령어 실행|
| 언어/OS | Software Atomic| 래핑된 함수로 원자성 확보|
| 연산 종류 | Load/Store| 단순 입출력|
| 연산 종류 | Read-Modify-Write| CAS, XCHG 등 변경 연산|

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

import threading
from multiprocessing import Value

# Atomic Counter 역할 (병렬 카운트 — 원자적 연산)
class AtomicCounter:
    def __init__(self):
        self.value = Value('i', 0) # 'i': integer 타입
    def increment(self):
        with self.value.get_lock(): # Value 객체 자체 Lock 내부 호출
            self.value.value += 1 # 원자적으로 증가

counter = AtomicCounter()

def worker():
    # 웹 요청 등에서 호출
    for _ in range(1000):
        counter.increment() # 원자적 증가 수행

threads = [threading.Thread(target=worker) for _ in range(10)]
[t.start() for t in threads]
[t.join() for t in threads]

print("최종 합계:", counter.value.value)
# 다수 스레드가 경합하여도 값 정확하게 유지됨

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

import (
    "sync/atomic"
    "sync"
    "testing"
)

func BenchmarkAtomicIncrement(b *testing.B) {
    var counter int64
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            atomic.AddInt64(&counter, 1) // 핵심 원자적 연산
        }
    })
}

1
2
3
4
5

| 항목 | 설명 | 권장사항 |
|------|------|----------|
| 메모리 정렬 | 일부 CPU는 정렬 불량시 원자성 불가 | 변수 alignment 확인|
| 연산 복잡도 | 다중 필드 연산 원자성 보장 어려움 | 다중 필드는 Lock, 트랜잭션 사용|
| 성능 이슈 | CAS 재시도, False Sharing 발생 | 프로파일링, 변수 분리|

1
2
3
4
5

| 전략 | 설명 | 권장사항 |
|------|------|----------|
| Lock-Free 설계 | 단일 필드 작업은 atomic 활용 | Lock-Free 자료구조 사용|
| CAS 최적화 | CAS 충돌 최소화 | Version Tag 적용|
| 캐시 라인 관리 | False Sharing 방지 | 변수 배열/분리|

1
2
3
4
5
6
7

| 카테고리 | Phase | 항목 | 중요도 | 설명 |
|----------|-------|------|--------|------|
| 기초     | 1     | 원자성, 동기화원리  | 필수   | 동시성의 기본 개념|
| 이론     | 2     | 하드웨어명령, RMW 패턴 | 필수   | CPU/메모리 핵심원리|
| 구현     | 5     | atomic 함수/패키지 | 권장   | 실제 Lock-Free 코드|
| 운영     | 6     | 모니터링, 최적화 | 권장   | 성능/이슈 관리|
| 고급     | 7     | 트랜잭셔널 메모리 | 선택   | 최신 환경 튜닝|

 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

# 기본 CAS 패턴 구현 예시
import threading
import time
from typing import TypeVar, Generic, Optional

T = TypeVar('T')

class AtomicReference(Generic[T]):
    """
    원자적 참조 클래스 - CAS 기반 구현
    """
    def __init__(self, initial_value: T):
        self._value = initial_value
        self._lock = threading.Lock()  # 실제로는 하드웨어 CAS 사용
    
    def get(self) -> T:
        """원자적 읽기 - acquire 의미론"""
        with self._lock:
            return self._value
    
    def set(self, new_value: T) -> None:
        """원자적 쓰기 - release 의미론"""
        with self._lock:
            self._value = new_value
    
    def compare_and_swap(self, expected: T, new_value: T) -> bool:
        """
        CAS 연산 구현
        expected와 현재 값이 같으면 new_value로 변경
        """
        with self._lock:
            if self._value == expected:
                self._value = new_value
                return True
            return False
    
    def get_and_set(self, new_value: T) -> T:
        """원자적 교환 - 이전 값 반환"""
        with self._lock:
            old_value = self._value
            self._value = new_value
            return old_value

# 락 프리 스택 구현 예시
class Node:
    def __init__(self, data, next_node=None):
        self.data = data
        self.next = next_node

class LockFreeStack:
    """
    CAS 기반 락 프리 스택 구현
    """
    def __init__(self):
        self.head = AtomicReference(None)
    
    def push(self, data):
        """
        새 노드를 스택 상단에 추가
        ABA 문제 방지를 위한 재시도 루프 포함
        """
        new_node = Node(data)
        while True:
            current_head = self.head.get()  # 현재 헤드 읽기
            new_node.next = current_head     # 새 노드가 현재 헤드를 가리킴
            
            # CAS로 헤드 업데이트 시도
            if self.head.compare_and_swap(current_head, new_node):
                break  # 성공 시 루프 종료
            # 실패 시 재시도 (다른 스레드가 헤드를 변경함)
    
    def pop(self) -> Optional:
        """
        스택 상단 노드 제거 및 반환
        빈 스택 처리 포함
        """
        while True:
            current_head = self.head.get()
            if current_head is None:
                return None  # 빈 스택
            
            next_node = current_head.next
            
            # 헤드를 다음 노드로 변경 시도
            if self.head.compare_and_swap(current_head, next_node):
                return current_head.data
            # 실패 시 재시도

  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

import threading
import time
import random
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import Dict, List
import queue

@dataclass
class RequestMetrics:
    """요청 처리 메트릭 클래스"""
    total_requests: int = 0
    active_requests: int = 0
    completed_requests: int = 0
    error_count: int = 0

class AtomicCounter:
    """
    원자적 카운터 구현
    실제 환경에서는 하드웨어 원자적 연산 사용
    """
    def __init__(self, initial_value: int = 0):
        self._value = initial_value
        self._lock = threading.Lock()  # 시뮬레이션용 - 실제로는 하드웨어 CAS
    
    def increment(self) -> int:
        """원자적 증가 - 새 값 반환"""
        with self._lock:
            self._value += 1
            return self._value
    
    def decrement(self) -> int:
        """원자적 감소 - 새 값 반환"""
        with self._lock:
            self._value -= 1
            return self._value
    
    def get(self) -> int:
        """원자적 읽기"""
        with self._lock:
            return self._value
    
    def add(self, delta: int) -> int:
        """원자적 덧셈 - 새 값 반환"""
        with self._lock:
            self._value += delta
            return self._value

class WebServerStats:
    """
    웹 서버 통계 수집기
    원자적 연산을 활용한 락 프리 구현
    """
    def __init__(self):
        # 원자적 카운터들 - 실제로는 하드웨어 지원 사용
        self.total_requests = AtomicCounter()      # 총 요청 수
        self.active_requests = AtomicCounter()     # 현재 처리 중인 요청 수
        self.completed_requests = AtomicCounter()  # 완료된 요청 수
        self.error_count = AtomicCounter()         # 에러 수
        
        # 응답 시간 히스토그램 (락 프리 구현)
        self.response_times: Dict[str, AtomicCounter] = {
            "0-100ms": AtomicCounter(),
            "100-500ms": AtomicCounter(),
            "500ms+": AtomicCounter()
        }
    
    def request_started(self) -> None:
        """요청 시작 - 원자적 카운터 업데이트"""
        self.total_requests.increment()     # 총 요청 수 증가
        self.active_requests.increment()    # 활성 요청 수 증가
    
    def request_completed(self, response_time_ms: float, success: bool) -> None:
        """요청 완료 - 원자적 통계 업데이트"""
        self.active_requests.decrement()    # 활성 요청 수 감소
        
        if success:
            self.completed_requests.increment()  # 성공 요청 수 증가
        else:
            self.error_count.increment()         # 에러 수 증가
        
        # 응답 시간 히스토그램 업데이트 (원자적)
        if response_time_ms < 100:
            self.response_times["0-100ms"].increment()
        elif response_time_ms < 500:
            self.response_times["100-500ms"].increment()
        else:
            self.response_times["500ms+"].increment()
    
    def get_metrics(self) -> RequestMetrics:
        """현재 메트릭 스냅샷 조회 - 모든 읽기는 원자적"""
        return RequestMetrics(
            total_requests=self.total_requests.get(),
            active_requests=self.active_requests.get(),
            completed_requests=self.completed_requests.get(),
            error_count=self.error_count.get()
        )
    
    def print_stats(self) -> None:
        """통계 출력"""
        metrics = self.get_metrics()
        print(f"총 요청: {metrics.total_requests}, "
              f"처리 중: {metrics.active_requests}, "
              f"완료: {metrics.completed_requests}, "
              f"에러: {metrics.error_count}")
        
        print("응답 시간 분포:")
        for range_name, counter in self.response_times.items():
            print(f"  {range_name}: {counter.get()}")

def simulate_request_processing(stats: WebServerStats, request_id: int) -> None:
    """
    HTTP 요청 처리 시뮬레이션
    각 스레드에서 독립적으로 실행
    """
    # 요청 시작을 원자적으로 기록
    stats.request_started()
    
    try:
        # 요청 처리 시간 시뮬레이션 (50-1000ms)
        processing_time = random.uniform(0.05, 1.0)
        time.sleep(processing_time)
        
        # 90% 성공률 시뮬레이션
        success = random.random() > 0.1
        
        # 요청 완료를 원자적으로 기록
        stats.request_completed(processing_time * 1000, success)
        
        if success:
            print(f"요청 {request_id} 완료 ({processing_time*1000:.1f}ms)")
        else:
            print(f"요청 {request_id} 실패")
            
    except Exception as e:
        # 예외 발생 시 에러로 기록
        stats.request_completed(0, False)
        print(f"요청 {request_id} 예외: {e}")

def run_web_server_simulation():
    """
    멀티스레드 웹 서버 시뮬레이션
    원자적 연산을 활용한 통계 수집 테스트
    """
    print("=== 웹 서버 원자적 연산 시뮬레이션 시작 ===")
    
    # 통계 수집기 초기화
    stats = WebServerStats()
    
    # 스레드 풀로 워커 스레드 시뮬레이션
    num_workers = 10
    num_requests = 100
    
    # 통계 모니터링 스레드
    def monitor_stats():
        for i in range(20):  # 20초 동안 모니터링
            time.sleep(1)
            print(f"\n[{i+1}초] ", end="")
            stats.print_stats()
    
    # 모니터링 시작
    monitor_thread = threading.Thread(target=monitor_stats)
    monitor_thread.daemon = True
    monitor_thread.start()
    
    # 워커 스레드로 요청 처리
    with ThreadPoolExecutor(max_workers=num_workers) as executor:
        # 요청들을 병렬로 처리
        futures = []
        for i in range(num_requests):
            future = executor.submit(simulate_request_processing, stats, i)
            futures.append(future)
            
            # 요청 간 간격 시뮬레이션
            time.sleep(0.1)
        
        # 모든 요청 완료 대기
        for future in futures:
            future.result()
    
    # 최종 통계 출력
    print("\n=== 최종 통계 ===")
    stats.print_stats()
    
    # 원자적 연산의 효과 검증
    metrics = stats.get_metrics()
    print(f"\n검증: 총 요청({metrics.total_requests}) = "
          f"완료({metrics.completed_requests}) + 에러({metrics.error_count})")
    
    assert metrics.total_requests == metrics.completed_requests + metrics.error_count
    assert metrics.active_requests == 0  # 모든 요청이 완료되어야 함
    
    print("✅ 모든 통계가 일관성 있게 수집되었습니다!")

if __name__ == "__main__":
    run_web_server_simulation()

  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
258
259
260

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
    "unsafe"
)

// TransactionID 는 트랜잭션 고유 식별자
type TransactionID uint64

// TupleVersion 은 하나의 튜플 버전을 나타냄
type TupleVersion struct {
    xmin    TransactionID  // 생성한 트랜잭션 ID
    xmax    TransactionID  // 삭제한 트랜잭션 ID (0이면 삭제되지 않음)
    data    interface{}    // 실제 데이터
    next    unsafe.Pointer // 다음 버전에 대한 원자적 포인터
}

// TransactionManager 는 트랜잭션 관리자
type TransactionManager struct {
    nextTxnID uint64  // 다음 트랜잭션 ID (원자적 카운터)
    snapshot  uint64  // 현재 스냅샷 (최대 커밋된 트랜잭션 ID)
}

// NewTransactionManager 는 새 트랜잭션 관리자 생성
func NewTransactionManager() *TransactionManager {
    return &TransactionManager{
        nextTxnID: 1,
        snapshot:  0,
    }
}

// BeginTransaction 은 새 트랜잭션 시작 (원자적 ID 할당)
func (tm *TransactionManager) BeginTransaction() TransactionID {
    // 원자적으로 트랜잭션 ID 증가 및 할당
    newTxnID := atomic.AddUint64(&tm.nextTxnID, 1)
    return TransactionID(newTxnID)
}

// CommitTransaction 은 트랜잭션 커밋 (원자적 스냅샷 업데이트)
func (tm *TransactionManager) CommitTransaction(txnID TransactionID) {
    // 원자적으로 커밋된 트랜잭션의 최대값 업데이트
    for {
        currentSnapshot := atomic.LoadUint64(&tm.snapshot)
        if uint64(txnID) <= currentSnapshot {
            break // 이미 더 큰 값이 설정됨
        }
        // CAS로 스냅샷 업데이트 시도
        if atomic.CompareAndSwapUint64(&tm.snapshot, currentSnapshot, uint64(txnID)) {
            break
        }
        // 실패 시 재시도
    }
}

// GetSnapshot 은 현재 가시적인 트랜잭션 범위 반환
func (tm *TransactionManager) GetSnapshot() TransactionID {
    return TransactionID(atomic.LoadUint64(&tm.snapshot))
}

// TupleStore 는 튜플 버전 저장소
type TupleStore struct {
    tuples map[string]unsafe.Pointer  // 튜플 ID별 버전 체인 헤드
    mutex  sync.RWMutex               // 맵 보호용 (실제로는 락프리 해시맵 사용)
}

// NewTupleStore 는 새 튜플 저장소 생성
func NewTupleStore() *TupleStore {
    return &TupleStore{
        tuples: make(map[string]unsafe.Pointer),
    }
}

// InsertTuple 은 새 튜플 삽입 (원자적 버전 체인 업데이트)
func (ts *TupleStore) InsertTuple(tupleID string, data interface{}, txnID TransactionID) {
    newVersion := &TupleVersion{
        xmin: txnID,
        xmax: 0,  // 삭제되지 않음
        data: data,
        next: nil,
    }
    
    ts.mutex.Lock()
    defer ts.mutex.Unlock()
    
    // 기존 버전 체인 헤드 가져오기
    oldHead := atomic.LoadPointer(&ts.tuples[tupleID])
    newVersion.next = oldHead
    
    // 원자적으로 새 버전을 헤드로 설정
    atomic.StorePointer(&ts.tuples[tupleID], unsafe.Pointer(newVersion))
}

// UpdateTuple 은 튜플 업데이트 (기존 버전 무효화 + 새 버전 추가)
func (ts *TupleStore) UpdateTuple(tupleID string, newData interface{}, txnID TransactionID, snapshot TransactionID) bool {
    ts.mutex.Lock()
    defer ts.mutex.Unlock()
    
    headPtr := ts.tuples[tupleID]
    if headPtr == nil {
        return false  // 튜플이 존재하지 않음
    }
    
    // 현재 가시적인 버전 찾기
    current := (*TupleVersion)(atomic.LoadPointer(&headPtr))
    for current != nil {
        // 트랜잭션 가시성 검사 (원자적 읽기)
        xmin := TransactionID(atomic.LoadUint64((*uint64)(&current.xmin)))
        xmax := TransactionID(atomic.LoadUint64((*uint64)(&current.xmax)))
        
        if xmin <= snapshot && (xmax == 0 || xmax > snapshot) {
            // 가시적인 버전 발견 - 업데이트 수행
            
            // 1. 기존 버전을 논리적으로 삭제 (원자적 xmax 설정)
            atomic.StoreUint64((*uint64)(&current.xmax), uint64(txnID))
            
            // 2. 새 버전 생성
            newVersion := &TupleVersion{
                xmin: txnID,
                xmax: 0,
                data: newData,
                next: unsafe.Pointer(current),  // 기존 버전 체인 유지
            }
            
            // 3. 원자적으로 새 버전을 헤드로 설정
            atomic.StorePointer(&ts.tuples[tupleID], unsafe.Pointer(newVersion))
            return true
        }
        
        // 다음 버전으로 이동
        current = (*TupleVersion)(atomic.LoadPointer(&current.next))
    }
    
    return false  // 업데이트할 가시적인 버전이 없음
}

// ReadTuple 은 튜플 읽기 (스냅샷 격리)
func (ts *TupleStore) ReadTuple(tupleID string, snapshot TransactionID) (interface{}, bool) {
    ts.mutex.RLock()
    defer ts.mutex.RUnlock()
    
    headPtr := ts.tuples[tupleID]
    if headPtr == nil {
        return nil, false
    }
    
    // 버전 체인을 순회하며 가시적인 버전 찾기
    current := (*TupleVersion)(atomic.LoadPointer(&headPtr))
    for current != nil {
        // 원자적으로 트랜잭션 ID 읽기
        xmin := TransactionID(atomic.LoadUint64((*uint64)(&current.xmin)))
        xmax := TransactionID(atomic.LoadUint64((*uint64)(&current.xmax)))
        
        // 스냅샷 가시성 검사
        if xmin <= snapshot && (xmax == 0 || xmax > snapshot) {
            return current.data, true
        }
        
        // 다음 버전으로 이동 (원자적 포인터 읽기)
        current = (*TupleVersion)(atomic.LoadPointer(&current.next))
    }
    
    return nil, false  // 가시적인 버전이 없음
}

// 시뮬레이션 실행 함수
func runMVCCSimulation() {
    fmt.Println("=== MVCC 원자적 연산 시뮬레이션 시작 ===")
    
    tm := NewTransactionManager()
    store := NewTupleStore()
    
    var wg sync.WaitGroup
    
    // 동시 트랜잭션 시뮬레이션
    numTxns := 10
    numOpsPerTxn := 5
    
    for i := 0; i < numTxns; i++ {
        wg.Add(1)
        go func(txnNum int) {
            defer wg.Done()
            
            // 트랜잭션 시작 (원자적 ID 할당)
            txnID := tm.BeginTransaction()
            snapshot := tm.GetSnapshot()
            
            fmt.Printf("트랜잭션 %d 시작 (ID: %d, Snapshot: %d)\n", 
                txnNum, txnID, snapshot)
            
            // 여러 연산 수행
            for j := 0; j < numOpsPerTxn; j++ {
                tupleID := fmt.Sprintf("tuple_%d", j%3)  // 3개 튜플에 대해 연산
                
                if j%2 == 0 {
                    // 읽기 연산
                    data, found := store.ReadTuple(tupleID, snapshot)
                    if found {
                        fmt.Printf("  TXN%d: 읽기 %s = %v\n", txnID, tupleID, data)
                    } else {
                        fmt.Printf("  TXN%d: 읽기 %s = 없음\n", txnID, tupleID)
                    }
                } else {
                    // 쓰기 연산
                    newData := fmt.Sprintf("데이터_TXN%d_OP%d", txnID, j)
                    
                    // 기존 튜플이 있으면 업데이트, 없으면 삽입
                    if _, found := store.ReadTuple(tupleID, snapshot); found {
                        success := store.UpdateTuple(tupleID, newData, txnID, snapshot)
                        if success {
                            fmt.Printf("  TXN%d: 업데이트 %s = %s\n", txnID, tupleID, newData)
                        }
                    } else {
                        store.InsertTuple(tupleID, newData, txnID)
                        fmt.Printf("  TXN%d: 삽입 %s = %s\n", txnID, tupleID, newData)
                    }
                }
                
                // 연산 간 지연 시뮬레이션
                time.Sleep(time.Millisecond * 10)
            }
            
            // 트랜잭션 커밋 (원자적 스냅샷 업데이트)
            tm.CommitTransaction(txnID)
            fmt.Printf("트랜잭션 %d 커밋됨\n", txnNum)
            
        }(i)
        
        // 트랜잭션 시작 간격
        time.Sleep(time.Millisecond * 50)
    }
    
    // 모든 트랜잭션 완료 대기
    wg.Wait()
    
    // 최종 상태 확인
    fmt.Println("\n=== 최종 상태 ===")
    finalSnapshot := tm.GetSnapshot()
    fmt.Printf("최종 스냅샷: %d\n", finalSnapshot)
    
    // 모든 튜플의 최종 상태 출력
    for i := 0; i < 3; i++ {
        tupleID := fmt.Sprintf("tuple_%d", i)
        data, found := store.ReadTuple(tupleID, finalSnapshot)
        if found {
            fmt.Printf("최종 %s = %v\n", tupleID, data)
        } else {
            fmt.Printf("최종 %s = 없음\n", tupleID)
        }
    }
    
    fmt.Println("✅ MVCC 시뮬레이션 완료!")
}

func main() {
    runMVCCSimulation()
}

 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

// 원자적 연산 모니터링 예시 (Go)
package monitoring

import (
    "sync/atomic"
    "time"
    "log"
)

// AtomicOperationMetrics 원자적 연산 성능 메트릭
type AtomicOperationMetrics struct {
    casAttempts   uint64  // 총 CAS 시도 횟수
    casSuccesses  uint64  // CAS 성공 횟수
    casFailures   uint64  // CAS 실패 횟수
    totalLatency  uint64  // 총 지연시간 (나노초)
    maxRetries    uint64  // 최대 재시도 횟수
    contentions   uint64  // 경합 발생 횟수
}

// RecordCASAttempt CAS 시도 기록
func (m *AtomicOperationMetrics) RecordCASAttempt(success bool, retries uint64, latency time.Duration) {
    atomic.AddUint64(&m.casAttempts, 1)
    atomic.AddUint64(&m.totalLatency, uint64(latency.Nanoseconds()))
    
    if success {
        atomic.AddUint64(&m.casSuccesses, 1)
    } else {
        atomic.AddUint64(&m.casFailures, 1)
    }
    
    // 최대 재시도 횟수 업데이트 (락 프리)
    for {
        current := atomic.LoadUint64(&m.maxRetries)
        if retries <= current {
            break
        }
        if atomic.CompareAndSwapUint64(&m.maxRetries, current, retries) {
            break
        }
    }
    
    if retries > 0 {
        atomic.AddUint64(&m.contentions, 1)
    }
}

// GetMetrics 현재 메트릭 스냅샷 반환
func (m *AtomicOperationMetrics) GetMetrics() map[string]interface{} {
    attempts := atomic.LoadUint64(&m.casAttempts)
    successes := atomic.LoadUint64(&m.casSuccesses)
    failures := atomic.LoadUint64(&m.casFailures)
    totalLatency := atomic.LoadUint64(&m.totalLatency)
    maxRetries := atomic.LoadUint64(&m.maxRetries)
    contentions := atomic.LoadUint64(&m.contentions)
    
    var avgLatency float64
    if attempts > 0 {
        avgLatency = float64(totalLatency) / float64(attempts) / 1000000  // ms
    }
    
    var successRate float64
    if attempts > 0 {
        successRate = float64(successes) / float64(attempts) * 100
    }
    
    var contentionRate float64
    if attempts > 0 {
        contentionRate = float64(contentions) / float64(attempts) * 100
    }
    
    return map[string]interface{}{
        "cas_attempts":      attempts,
        "cas_success_rate":  successRate,
        "avg_latency_ms":    avgLatency,
        "max_retries":       maxRetries,
        "contention_rate":   contentionRate,
    }
}

// LogMetrics 주기적 메트릭 로깅
func (m *AtomicOperationMetrics) LogMetrics() {
    metrics := m.GetMetrics()
    log.Printf("원자적 연산 메트릭: %+v", metrics)
}

  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

// Rust를 사용한 고성능 원자적 카운터 최적화 예시
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};

// 샤딩된 원자적 카운터 - 경합 감소를 위한 설계
struct ShardedAtomicCounter {
    shards: Vec<AtomicUsize>,
    num_shards: usize,
}

impl ShardedAtomicCounter {
    fn new(num_shards: usize) -> Self {
        let mut shards = Vec::with_capacity(num_shards);
        for _ in 0..num_shards {
            shards.push(AtomicUsize::new(0));
        }
        
        Self {
            shards,
            num_shards,
        }
    }
    
    // 스레드별로 다른 샤드 사용 - 경합 최소화
    fn increment(&self) -> usize {
        let thread_id = thread_local_thread_id();
        let shard_index = thread_id % self.num_shards;
        
        // Relaxed 순서로 성능 최적화 (순서가 중요하지 않은 카운터)
        self.shards[shard_index].fetch_add(1, Ordering::Relaxed)
    }
    
    // 전체 카운트 계산 - 모든 샤드 합계
    fn total(&self) -> usize {
        self.shards
            .iter()
            .map(|shard| shard.load(Ordering::Relaxed))
            .sum()
    }
    
    // 적응적 백오프를 사용한 CAS 루프
    fn increment_with_backoff(&self, shard_index: usize) -> usize {
        let mut backoff = 1;
        let max_backoff = 1024;
        
        loop {
            let current = self.shards[shard_index].load(Ordering::Relaxed);
            
            // CAS 시도
            match self.shards[shard_index].compare_exchange_weak(
                current,
                current + 1,
                Ordering::Release,  // 성공 시 release
                Ordering::Relaxed,  // 실패 시 relaxed
            ) {
                Ok(prev) => return prev,
                Err(_) => {
                    // 백오프 - CPU 사이클 낭비 방지
                    for _ in 0..backoff {
                        std::hint::spin_loop();  // CPU pause 명령어
                    }
                    
                    // 지수적 백오프 (상한 적용)
                    backoff = std::cmp::min(backoff * 2, max_backoff);
                }
            }
        }
    }
}

// 스레드 로컬 ID 시뮬레이션
thread_local! {
    static THREAD_ID: usize = {
        use std::sync::atomic::{AtomicUsize, Ordering};
        static COUNTER: AtomicUsize = AtomicUsize::new(0);
        COUNTER.fetch_add(1, Ordering::Relaxed)
    };
}

fn thread_local_thread_id() -> usize {
    THREAD_ID.with(|id| *id)
}

// 성능 벤치마크 예시
fn benchmark_atomic_counters() {
    println!("=== 원자적 카운터 성능 최적화 벤치마크 ===");
    
    let num_threads = 8;
    let operations_per_thread = 1_000_000;
    let num_shards = num_threads; // 스레드 수와 동일한 샤드 수
    
    // 1. 기본 원자적 카운터
    println!("\n1. 기본 원자적 카운터 테스트");
    let basic_counter = Arc::new(AtomicUsize::new(0));
    let start = Instant::now();
    
    let mut handles = vec![];
    for _ in 0..num_threads {
        let counter = basic_counter.clone();
        let handle = thread::spawn(move || {
            for _ in 0..operations_per_thread {
                counter.fetch_add(1, Ordering::Relaxed);
            }
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    let basic_duration = start.elapsed();
    println!("기본 카운터 - 시간: {:?}, 값: {}", 
        basic_duration, basic_counter.load(Ordering::Relaxed));
    
    // 2. 샤딩된 카운터
    println!("\n2. 샤딩된 카운터 테스트");
    let sharded_counter = Arc::new(ShardedAtomicCounter::new(num_shards));
    let start = Instant::now();
    
    let mut handles = vec![];
    for _ in 0..num_threads {
        let counter = sharded_counter.clone();
        let handle = thread::spawn(move || {
            for _ in 0..operations_per_thread {
                counter.increment();
            }
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    let sharded_duration = start.elapsed();
    println!("샤딩된 카운터 - 시간: {:?}, 값: {}", 
        sharded_duration, sharded_counter.total());
    
    // 성능 비교
    let improvement = basic_duration.as_nanos() as f64 / sharded_duration.as_nanos() as f64;
    println!("\n성능 개선: {:.2}x 빨라짐", improvement);
    
    // 3. 메모리 순서별 성능 비교
    println!("\n3. 메모리 순서별 성능 비교");
    test_memory_ordering_performance();
}

fn test_memory_ordering_performance() {
    let counter = Arc::new(AtomicUsize::new(0));
    let operations = 10_000_000;
    
    // Relaxed 순서
    let start = Instant::now();
    for _ in 0..operations {
        counter.fetch_add(1, Ordering::Relaxed);
    }
    let relaxed_duration = start.elapsed();
    
    counter.store(0, Ordering::Relaxed);
    
    // SeqCst 순서
    let start = Instant::now();
    for _ in 0..operations {
        counter.fetch_add(1, Ordering::SeqCst);
    }
    let seqcst_duration = start.elapsed();
    
    println!("Relaxed: {:?}", relaxed_duration);
    println!("SeqCst: {:?}", seqcst_duration);
    
    let overhead = seqcst_duration.as_nanos() as f64 / relaxed_duration.as_nanos() as f64;
    println!("SeqCst 오버헤드: {:.2}x", overhead);
}

fn main() {
    benchmark_atomic_counters();
}