Lock

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

Test-and-Set 의사코드:
function test_and_set(lock_variable):
    old_value = lock_variable
    lock_variable = 1
    return old_value

Compare-and-Swap 의사코드:
function compare_and_swap(variable, expected, new_value):
    if variable == expected:
        variable = new_value
        return true
    return false

 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

import threading
import time

class SpinLock:
    def __init__(self):
        self._locked = threading.Lock()
        self._flag = False
    
    def acquire(self):
        while True:
            with self._locked:
                if not self._flag:
                    self._flag = True
                    return
            # 짧은 백오프로 CPU 사용량 조절
            time.sleep(0.0001)
    
    def release(self):
        with self._locked:
            self._flag = False

# 사용 예시
spin_lock = SpinLock()
shared_resource = 0

def worker(worker_id):
    global shared_resource
    for i in range(1000):
        spin_lock.acquire()
        # Critical Section - 짧은 연산
        temp = shared_resource
        temp += 1
        shared_resource = temp
        spin_lock.release()

 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

import threading
import time
import queue

class MutexExample:
    def __init__(self):
        self.mutex = threading.Lock()
        self.shared_queue = queue.Queue()
        self.processing_count = 0
    
    def producer(self, producer_id):
        for i in range(5):
            item = f"item-{producer_id}-{i}"
            
            # 뮤텍스로 보호되는 Critical Section
            with self.mutex:
                self.shared_queue.put(item)
                self.processing_count += 1
                print(f"Producer {producer_id} added {item}, "
                      f"total items: {self.processing_count}")
            
            time.sleep(0.1)  # 생산 작업 시뮬레이션
    
    def consumer(self, consumer_id):
        while True:
            try:
                # 타임아웃을 통한 안전한 종료
                item = self.shared_queue.get(timeout=1)
                
                with self.mutex:
                    self.processing_count -= 1
                    print(f"Consumer {consumer_id} processed {item}, "
                          f"remaining: {self.processing_count}")
                
                time.sleep(0.2)  # 처리 작업 시뮬레이션
                
            except queue.Empty:
                break

# 시스템 구성 예시
def mutex_system_example():
    system = MutexExample()
    
    # 생산자와 소비자 스레드 생성
    producers = [threading.Thread(target=system.producer, args=(i,)) 
                for i in range(2)]
    consumers = [threading.Thread(target=system.consumer, args=(i,)) 
                for i in range(3)]
    
    # 모든 스레드 시작
    for t in producers + consumers:
        t.start()
    
    # 생산자 완료 대기
    for t in producers:
        t.join()
    
    # 소비자 완료 대기
    for t in consumers:
        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
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

import threading
import time
import random

class ReaderWriterLock:
    def __init__(self):
        self._readers = 0
        self._writers = 0
        self._read_ready = threading.Condition(threading.RLock())
        self._write_ready = threading.Condition(threading.RLock())
    
    def acquire_read(self):
        with self._read_ready:
            while self._writers > 0:
                self._read_ready.wait()
            self._readers += 1
    
    def release_read(self):
        with self._read_ready:
            self._readers -= 1
            if self._readers == 0:
                self._read_ready.notifyAll()
    
    def acquire_write(self):
        with self._write_ready:
            while self._writers > 0 or self._readers > 0:
                self._write_ready.wait()
            self._writers += 1
    
    def release_write(self):
        with self._write_ready:
            self._writers -= 1
            self._write_ready.notifyAll()
        with self._read_ready:
            self._read_ready.notifyAll()

# 공유 데이터베이스 시뮬레이션
class SharedDatabase:
    def __init__(self):
        self.rw_lock = ReaderWriterLock()
        self.data = {"users": 1000, "posts": 5000, "comments": 15000}
    
    def read_data(self, key, reader_id):
        self.rw_lock.acquire_read()
        try:
            value = self.data.get(key, "Not Found")
            print(f"Reader {reader_id} read {key}: {value}")
            time.sleep(0.1)  # 읽기 작업 시뮬레이션
            return value
        finally:
            self.rw_lock.release_read()
    
    def write_data(self, key, value, writer_id):
        self.rw_lock.acquire_write()
        try:
            old_value = self.data.get(key, 0)
            self.data[key] = value
            print(f"Writer {writer_id} updated {key}: {old_value} -> {value}")
            time.sleep(0.2)  # 쓰기 작업 시뮬레이션
        finally:
            self.rw_lock.release_write()

# 시나리오: 읽기 집약적 워크로드
def reader_writer_scenario():
    db = SharedDatabase()
    
    def reader_worker(reader_id):
        for _ in range(5):
            key = random.choice(["users", "posts", "comments"])
            db.read_data(key, reader_id)
            time.sleep(random.uniform(0.1, 0.3))
    
    def writer_worker(writer_id):
        for i in range(2):
            key = random.choice(["users", "posts", "comments"])
            value = random.randint(1000, 9999)
            db.write_data(key, value, writer_id)
            time.sleep(random.uniform(0.5, 1.0))
    
    # 많은 읽기 스레드와 적은 쓰기 스레드
    readers = [threading.Thread(target=reader_worker, args=(i,)) 
               for i in range(5)]
    writers = [threading.Thread(target=writer_worker, args=(i,)) 
               for i in range(2)]
    
    for t in readers + writers:
        t.start()
    
    for t in readers + writers:
        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
26
27
28
29
30
31
32
33
34

import threading
import time
import random

# 세마포어(동시 접근 3개 제한)
semaphore = threading.Semaphore(3)

# 공유 자원 (예: 외부 API 호출, DB 커넥션 등)
def access_shared_resource(thread_id):
    print(f"Thread-{thread_id} 대기 중…")
    
    # 세마포어 획득
    semaphore.acquire()
    print(f"Thread-{thread_id} 공유 자원에 접근")
    
    # 임계 구역(자원 접근)
    time.sleep(random.uniform(1, 2))  # 자원 사용 시뮬레이션
    
    # 세마포어 해제
    print(f"Thread-{thread_id} 자원 사용 종료")
    semaphore.release()

# 10개의 스레드 실행
threads = []
for i in range(10):
    t = threading.Thread(target=access_shared_resource, args=(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
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

import ctypes
import os
import time
import threading

# Linux syscall 번호 (x86_64 기준)
SYS_futex = 202  # __NR_futex
FUTEX_WAIT = 0
FUTEX_WAKE = 1

libc = ctypes.CDLL("libc.so", use_errno=True)

# ctypes를 통한 futex syscall 함수 정의
def futex(uaddr, futex_op, val, timeout=None):
    uaddr_p = ctypes.pointer(ctypes.c_int(uaddr[0]))
    return libc.syscall(SYS_futex,
                        uaddr_p,
                        futex_op,
                        val,
                        ctypes.c_void_p(0))

class FutexLock:
    def __init__(self):
        self._state = [0]  # 공유 메모리 (값이 0이면 락 없음, 1이면 락 중)
        self._lock = threading.Lock()

    def acquire(self):
        while True:
            with self._lock:
                if self._state[0] == 0:
                    self._state[0] = 1
                    return

            # 실패 시 사용자 공간에서 실패 → 커널로 들어가 대기
            futex(self._state, FUTEX_WAIT, 1)
    
    def release(self):
        with self._lock:
            self._state[0] = 0
        futex(self._state, FUTEX_WAKE, 1)

# 테스트 예시
futex_lock = FutexLock()
shared_value = 0

def worker(tid):
    global shared_value
    for _ in range(10):
        futex_lock.acquire()
        temp = shared_value
        time.sleep(0.01)  # 시뮬레이션용
        shared_value = temp + 1
        futex_lock.release()
    print(f"Thread-{tid} done.")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]

for t in threads:
    t.start()

for t in threads:
    t.join()

print(f"최종 shared_value: {shared_value}")

 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

lock_a = threading.Lock()
lock_b = threading.Lock()

def safe_task():
    while True:
        acquired_a = lock_a.acquire(timeout=0.5)
        if not acquired_a:
            continue  # backoff 가능

        acquired_b = lock_b.acquire(timeout=0.5)
        if not acquired_b:
            lock_a.release()
            time.sleep(0.1)  # backoff
            continue

        try:
            # 임계 구역 작업
            print("Lock A and B acquired")
        finally:
            lock_b.release()
            lock_a.release()
        break

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

import threading
import queue

task_queue = queue.Queue()
lock = threading.Lock()

def worker():
    while True:
        task = task_queue.get()
        with lock:  # FIFO 큐 처리 기반
            print(f"Processing {task}")
        task_queue.task_done()

for i in range(5):
    threading.Thread(target=worker, daemon=True).start()

for i in range(20):
    task_queue.put(f"Task-{i}")

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

import threading
import random
import time

lock = threading.Lock()

def livelock_safe_worker(name):
    backoff = 0.01
    while True:
        if lock.acquire(blocking=False):
            try:
                print(f"{name} acquired lock")
                break
            finally:
                lock.release()
        else:
            time.sleep(random.uniform(0, backoff))
            backoff = min(backoff * 2, 1)  # 최대 1초까지 증가

for i in range(3):
    threading.Thread(target=livelock_safe_worker, args=(f"Thread-{i}",)).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

public class PriorityInversionDemo {
    static final Object lock = new Object();

    public static void main(String[] args) {
        Thread low = new Thread(() -> {
            synchronized (lock) {
                try {
                    Thread.sleep(3000);  // 락을 오래 보유
                } catch (InterruptedException ignored) {}
            }
        }, "LowPriority");

        Thread high = new Thread(() -> {
            synchronized (lock) {
                System.out.println("High priority thread got lock");
            }
        }, "HighPriority");

        low.setPriority(Thread.MIN_PRIORITY);
        high.setPriority(Thread.MAX_PRIORITY);

        low.start();
        Thread.yield();  // low가 먼저 락을 잡게 유도
        high.start();
    }
}

1
2
3
4
5
6
7
8
9

import threading
NUM_SHARDS = 8
locks = [threading.Lock() for _ in range(NUM_SHARDS)]
data = [{} for _ in range(NUM_SHARDS)]

def access_data(key, value):
    shard = hash(key) % NUM_SHARDS
    with locks[shard]:
        data[shard][key] = value

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

import threading

session_cache = {}
cache_lock = threading.Lock()

def get_user_session(user_id):
    with cache_lock:
        if user_id in session_cache:
            return session_cache[user_id]

    # 캐시 미스 처리 (락 해제 후 DB 접근)
    session = load_from_db(user_id)

    with cache_lock:
        session_cache[user_id] = session
    return session

  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

import threading
import time
import queue
from contextlib import contextmanager

class DatabaseConnection:
    """데이터베이스 연결을 시뮬레이션하는 클래스"""
    def __init__(self, connection_id):
        self.connection_id = connection_id
        self.is_active = False
    
    def execute_query(self, query):
        self.is_active = True
        print(f"Connection {self.connection_id} executing: {query}")
        time.sleep(0.1)  # 쿼리 실행 시뮬레이션
        result = f"Result from connection {self.connection_id}"
        self.is_active = False
        return result
    
    def __str__(self):
        return f"DBConnection({self.connection_id})"

class ThreadSafeConnectionPool:
    """스레드 안전한 연결 풀 구현"""
    
    def __init__(self, pool_size=5):
        # Lock을 통한 동시성 제어
        self._lock = threading.RLock()  # 재진입 가능한 Lock 사용
        self._condition = threading.Condition(self._lock)
        
        # 연결 풀 초기화
        self._available_connections = queue.Queue()
        self._all_connections = []
        self._active_connections = set()
        
        # 풀 통계
        self._total_requests = 0
        self._active_requests = 0
        self._max_active = 0
        
        # 연결 생성
        for i in range(pool_size):
            conn = DatabaseConnection(f"conn_{i}")
            self._available_connections.put(conn)
            self._all_connections.append(conn)
        
        print(f"Connection pool initialized with {pool_size} connections")
    
    @contextmanager
    def get_connection(self, timeout=5):
        """
        컨텍스트 매니저를 통한 안전한 연결 획득/반환
        Lock을 사용하여 스레드 안전성 보장
        """
        connection = None
        start_time = time.time()
        
        try:
            # Lock 획득 및 연결 할당
            with self._condition:
                self._total_requests += 1
                
                # 사용 가능한 연결이 있을 때까지 대기
                while self._available_connections.empty():
                    elapsed = time.time() - start_time
                    if elapsed >= timeout:
                        raise TimeoutError(f"Connection timeout after {timeout}s")
                    
                    print(f"Thread {threading.current_thread().name} waiting for connection…")
                    self._condition.wait(timeout - elapsed)
                
                # 연결 획득
                connection = self._available_connections.get()
                self._active_connections.add(connection)
                self._active_requests += 1
                self._max_active = max(self._max_active, self._active_requests)
                
                print(f"Thread {threading.current_thread().name} acquired {connection}")
            
            # Critical Section: 연결 사용
            yield connection
            
        finally:
            # 연결 반환 (예외 발생 시에도 실행됨)
            if connection:
                with self._condition:
                    self._active_connections.remove(connection)
                    self._available_connections.put(connection)
                    self._active_requests -= 1
                    
                    print(f"Thread {threading.current_thread().name} returned {connection}")
                    
                    # 대기 중인 스레드들에게 알림
                    self._condition.notify()
    
    def get_statistics(self):
        """풀 사용 통계 반환"""
        with self._lock:
            return {
                'total_connections': len(self._all_connections),
                'available_connections': self._available_connections.qsize(),
                'active_connections': len(self._active_connections),
                'total_requests': self._total_requests,
                'max_concurrent_active': self._max_active
            }

# 웹 서버 요청 시뮬레이션
def simulate_web_request(pool, request_id, queries):
    """웹 요청을 시뮬레이션하는 워커 함수"""
    thread_name = threading.current_thread().name
    
    try:
        # 연결 풀에서 연결 획득 (Lock으로 보호됨)
        with pool.get_connection(timeout=3) as connection:
            print(f"Request {request_id} started by {thread_name}")
            
            # 여러 쿼리 실행
            for query in queries:
                result = connection.execute_query(query)
                print(f"Request {request_id}: {result}")
                
                # 쿼리 간 처리 시간
                time.sleep(0.05)
        
        print(f"Request {request_id} completed successfully")
        
    except TimeoutError as e:
        print(f"Request {request_id} failed: {e}")
    except Exception as e:
        print(f"Request {request_id} error: {e}")

def demonstrate_connection_pool():
    """연결 풀 데모 실행"""
    # 작은 풀 크기로 경합 상황 시뮬레이션
    pool = ThreadSafeConnectionPool(pool_size=3)
    
    # 다양한 웹 요청 시뮬레이션
    requests = [
        (1, ["SELECT * FROM users", "UPDATE user_stats SET last_login=NOW()"]),
        (2, ["SELECT * FROM products WHERE category='electronics'"]),
        (3, ["INSERT INTO orders (user_id, total) VALUES (123, 99.99)"]),
        (4, ["SELECT COUNT(*) FROM orders WHERE status='pending'"]),
        (5, ["DELETE FROM temp_sessions WHERE created < NOW() - INTERVAL 1 HOUR"]),
        (6, ["SELECT * FROM user_preferences WHERE user_id=456"]),
        (7, ["UPDATE inventory SET quantity=quantity-1 WHERE product_id=789"]),
    ]
    
    # 동시 요청 처리를 위한 스레드 생성
    threads = []
    for request_id, queries in requests:
        thread = threading.Thread(
            target=simulate_web_request,
            args=(pool, request_id, queries),
            name=f"Worker-{request_id}"
        )
        threads.append(thread)
    
    print("Starting concurrent web requests simulation…")
    print("-" * 60)
    
    # 모든 요청 스레드 시작
    for thread in threads:
        thread.start()
        time.sleep(0.1)  # 약간의 시작 지연으로 경합 상황 생성
    
    # 모든 요청 완료 대기
    for thread in threads:
        thread.join()
    
    print("-" * 60)
    print("All requests completed")
    
    # 최종 통계 출력
    stats = pool.get_statistics()
    print("\nConnection Pool Statistics:")
    for key, value in stats.items():
        print(f"  {key}: {value}")

if __name__ == "__main__":
    demonstrate_connection_pool()

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

import redis
import time

r = redis.Redis(host='redis-server', port=6379)
lock = r.lock('cron_job_lock', timeout=60)

if lock.acquire(blocking=False):
    try:
        # 단일 노드만 이 블록 실행!
        print("Job started")
        # 작업 처리
        time.sleep(10)
    finally:
        lock.release()
else:
    print("Lock held by another node. Skipping job.")

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

import redis
from redlock import Redlock

dlm = Redlock([
    {"host": "redis1.example.com", "port": 6379},
    {"host": "redis2.example.com", "port": 6379},
    {"host": "redis3.example.com", "port": 6379},
])

lock = dlm.lock("distributed:cron:lock", 10000)  # 10초 TTL

if lock:
    try:
        print("Lock acquired, running job…")
        run_hourly_job()
    finally:
        dlm.unlock(lock)
else:
    print("Another instance is running. Exiting.")