В чем разница между Mutex и Semaphore?

# Mutex vs Semaphore в многопоточном программировании

Mutex (Mutual Exclusion) и Semaphore — это оба механизма синхронизации потоков, но они работают по-разному и используются для разных целей. Оба предотвращают race conditions, но имеют разную функциональность.

Основное различие

Mutex (Взаимное исключение):
- Двоичный (либо заблокирован, либо нет)
- Может быть занят только одним потоком
- Значение: 0 (заблокирован) или 1 (свободен)

Semaphore (Семафор):
- Счётчик (может иметь несколько значений)
- Может быть занят несколькими потоками (до N)
- Значение: 0, 1, 2, 3, ... N

Mutex

Что такое Mutex

Mutex — это примитив синхронизации, который ограничивает доступ к ресурсу только одному потоку за раз. Это как дверь с одним ключом — когда один человек входит, дверь закрывается, остальные ждут.

Примеры в Python

import threading

# Создание mutex
mutex = threading.Lock()  # threading.Lock() — это просто Lock

# Синхроизованный доступ
shared_counter = 0

def increment_counter():
    global shared_counter
    for _ in range(1000000):
        mutex.acquire()  # Захватить mutex
        try:
            shared_counter += 1
        finally:
            mutex.release()  # Освободить mutex

# Или с контекстным менеджером (рекомендуется)
def increment_counter_safe():
    global shared_counter
    for _ in range(1000000):
        with mutex:  # Автоматический acquire/release
            shared_counter += 1

# Создать несколько потоков
threads = []
for i in range(5):
    t = threading.Thread(target=increment_counter_safe)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(shared_counter)  # 5000000 (корректно без race condition)

Проблема без Mutex (Race Condition)

import threading
import time

shared_counter = 0

def increment_without_mutex():
    global shared_counter
    for _ in range(1000000):
        # БЕЗ защиты
        shared_counter += 1

threads = []
for i in range(5):
    t = threading.Thread(target=increment_without_mutex)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(shared_counter)
# Результат: может быть 2345678 вместо 5000000!
# Race condition: несколько потоков одновременно читают/пишут

Как работает Mutex

Поток 1: ┌─ acquire() ─ УСПЕХ (берёт lock)
          │
          ├─ критическая секция (работает)
          │
          └─ release() ─ (отпускает lock)

Поток 2: (ждёт acquire() из-за занятого lock)
         ┌─ acquire() ─ УСПЕХ (получает lock после потока 1)
         │
         ├─ критическая секция
         │
         └─ release()

Поток 3: (ждёт acquire())
         ┌─ acquire() ─ УСПЕХ
         │
         ├─ критическая секция
         │
         └─ release()

Semaphore

Что такое Semaphore

Semaphore — это счётчик, который контролирует доступ к N ресурсам. Если N=1, то это как Mutex. Если N>1, то несколько потоков могут одновременно работать с ресурсом.

Примеры в Python

import threading

# Создание семафора — максимум 3 потока одновременно
semaphore = threading.Semaphore(3)

def worker(worker_id):
    print(f"Worker {worker_id} ждёт...")
    
    with semaphore:  # acquire() и release() автоматически
        print(f"Worker {worker_id} начал работать")
        time.sleep(2)  # Имитация работы
        print(f"Worker {worker_id} завершил работу")

# Создать 10 потоков, но только 3 могут работать одновременно
threads = []
for i in range(10):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

# Вывод:
# Worker 0 начал работать
# Worker 1 начал работать
# Worker 2 начал работать
# Worker 3 ждёт...
# Worker 4 ждёт...
# ... (после 2 сек)
# Worker 0 завершил работу
# Worker 3 начал работать
# ... и так далее

Практический пример: Pool соединений

import threading
import time
from queue import Queue

class ConnectionPool:
    def __init__(self, max_connections=3):
        self.semaphore = threading.Semaphore(max_connections)
        self.connections = Queue(maxsize=max_connections)
        
        # Создаём max_connections соединений
        for i in range(max_connections):
            self.connections.put(f"Connection-{i}")
    
    def acquire(self):
        """Получить соединение из пула"""
        self.semaphore.acquire()  # Ждём, пока будет свободное
        connection = self.connections.get()  # Берём соединение
        return connection
    
    def release(self, connection):
        """Вернуть соединение в пул"""
        self.connections.put(connection)  # Вернуть соединение
        self.semaphore.release()  # Сигнализировать, что свободно

# Использование
pool = ConnectionPool(max_connections=3)

def database_query(query_id):
    print(f"Query {query_id} ждёт соединение...")
    
    connection = pool.acquire()
    print(f"Query {query_id} получил {connection}")
    
    time.sleep(1)  # Имитация запроса
    
    pool.release(connection)
    print(f"Query {query_id} освободил {connection}")

threads = []
for i in range(10):
    t = threading.Thread(target=database_query, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

Таблица сравнения

Виды Semaphore

1. Counting Semaphore (Счётный семафор)

import threading

# N потоков могут одновременно входить в критическую секцию
semaphore = threading.Semaphore(5)  # Максимум 5 потоков

with semaphore:
    # До 5 потоков одновременно здесь
    do_work()

2. Binary Semaphore (Бинарный семафор)

# То же что Mutex, но реализовано как семафор
binary_semaphore = threading.Semaphore(1)  # Как Mutex

with binary_semaphore:
    # Только 1 поток одновременно
    critical_section()

Отличие Mutex vs Semaphore в коде

Mutex — только 1 поток

import threading

mutex = threading.Lock()
counter = 0

def increment():
    global counter
    with mutex:
        # Только 1 поток здесь одновременно
        counter += 1

threads = [threading.Thread(target=increment) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(counter)  # 5 (безопасно)

Semaphore — N потоков

import threading

semaphore = threading.Semaphore(3)  # До 3 потоков одновременно
counter = 0
lock = threading.Lock()  # Для защиты counter при изменении

def work():
    global counter
    with semaphore:
        # До 3 потоков здесь одновременно
        with lock:
            counter += 1

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

print(counter)  # 10 (безопасно)

RLock (Recursive Mutex)

import threading

# Обычный Lock (Mutex)
lock = threading.Lock()

# Попытка захватить дважды из одного потока — deadlock!
def bad_function():
    with lock:
        print("First lock acquired")
        with lock:  # DEADLOCK! Поток ждёт сам себя
            print("This will never execute")

# RLock — рекурсивный Lock, один поток может захватить несколько раз
rlock = threading.RLock()

def good_function():
    with rlock:
        print("First lock acquired")
        with rlock:  # OK! Один поток может захватить множество раз
            print("This will execute")
            with rlock:  # Опять OK!
                print("And this too")

Практические примеры

Пример 1: Потокобезопасный счётчик с Mutex

import threading

class ThreadSafeCounter:
    def __init__(self):
        self._count = 0
        self._mutex = threading.Lock()
    
    def increment(self):
        with self._mutex:
            self._count += 1
    
    def decrement(self):
        with self._mutex:
            self._count -= 1
    
    def get(self):
        with self._mutex:
            return self._count

# Использование
counter = ThreadSafeCounter()
for _ in range(1000000):
    counter.increment()
print(counter.get())  # 1000000

Пример 2: Ограничение одновременных операций с Semaphore

import threading
import time

class ResourcePool:
    """Пул ресурсов с максимум N одновременными пользователями"""
    
    def __init__(self, max_users=3):
        self.semaphore = threading.Semaphore(max_users)
        self.current_users = 0
        self.lock = threading.Lock()
    
    def use_resource(self, user_id, duration):
        self.semaphore.acquire()
        
        with self.lock:
            self.current_users += 1
            print(f"User {user_id} using resource. Current users: {self.current_users}")
        
        time.sleep(duration)  # Используем ресурс
        
        with self.lock:
            self.current_users -= 1
            print(f"User {user_id} released resource. Current users: {self.current_users}")
        
        self.semaphore.release()

# Использование
pool = ResourcePool(max_users=2)

threads = []
for i in range(5):
    t = threading.Thread(target=pool.use_resource, args=(i, 1))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

Когда использовать что

Используй Mutex если:

✓ Нужен исключительный доступ (только 1 поток)
✓ Защита критической секции
✓ Простота — не нужна сложная логика
✓ Производительность важна

Примеры:
- Защита глобальных переменных
- Защита доступа к файлу
- Защита счётчиков

Используй Semaphore если:

✓ Нужно ограничить одновременные операции (N потоков)
✓ Пул ресурсов (соединения, лицензии, лимиты)
✓ Производитель-потребитель паттерн
✓ Контроль пропускной способности

Примеры:
- Пул соединений БД
- Ограничение одновременных загрузок
- Rate limiting
- Контроль доступа к ограниченному ресурсу

Заключение

Mutex:

• Двоичный (1 поток одновременно)
• Предотвращает race conditions
• Простой и быстрый
• Для исключительного доступа

Semaphore:

• Счётчик (N потоков одновременно)
• Контролирует пулы ресурсов
• Более гибкий
• Для ограничения одновременных операций

Практический совет:

• Начни с Mutex (простой и понятный)
• Переходи на Semaphore, если нужна параллельность
• Обе используют with контекстный менеджер в Python для безопасности
• Всегда освобождай ресурсы (используй try/finally или with)