Что произойдет с потоками, если сервер попал под высокую нагрузку?

Что произойдёт с потоками при высокой нагрузке на сервер

Этот вопрос о деградации системы под нагрузкой. Расскажу подробно, что происходит и как это предотвратить.

1. Первый признак: увеличение latency

Сценарий: Нормальная нагрузка 100 req/sec, потом скачок до 10K req/sec

import threading
import time
from datetime import datetime

request_times = []
lock = threading.Lock()

def handle_request():
    start = time.time()
    # Простая обработка
    time.sleep(0.01)  # 10ms processing
    duration = time.time() - start
    
    with lock:
        request_times.append(duration)

# При нормальной нагрузке
for _ in range(100):
    t = threading.Thread(target=handle_request)
    t.start()

time.sleep(1)
print(f"Средний response time: {sum(request_times)/len(request_times):.3f}s")
# Output: 0.010s (нормально)

# При высокой нагрузке (10K потоков)
request_times.clear()
threads = []
for _ in range(10000):
    t = threading.Thread(target=handle_request)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"Средний response time: {sum(request_times)/len(request_times):.3f}s")
# Output: 2.340s (100x медленнее!)
# Почему? Context switching overhead

2. Истощение ресурсов системы

Что именно исчерпывается:

import resource
import threading
import os

# 1. MEMORY (каждый поток занимает ~8MB)
print(f"Максимум потоков в памяти: ~{resource.getrlimit(resource.RLIMIT_NPROC)[0]//2}")

# Если создать 50K потоков:
threads = []
try:
    for i in range(50000):
        t = threading.Thread(target=lambda: time.sleep(100))
        threads.append(t)
        t.start()
except RuntimeError as e:
    print(f"Ошибка: {e}")
    # RuntimeError: can't start new thread
    # Закончилась память или достигнут лимит потоков ОС

# 2. STACK MEMORY
# Каждый поток имеет свой stack (~1-8 MB)
# 10K потоков = 10-80 GB памяти только на стеки!

# 3. CPU (context switching overhead)
# Если потоков > ядер CPU → constant switching
# На каждый switch: 1-10 µs потеряется
num_threads = 10000  # На 8-ядерном CPU
switch_overhead_per_sec = 10000 * 0.001  # ~10 ms/sec потеряется

# 4. FILE DESCRIPTORS
print(f"Лимит файловых дескрипторов: {resource.getrlimit(resource.RLIMIT_NOFILE)[0]}")
# Типично 1024
# Если каждый поток открывает соединение с БД → быстро исчерпается

3. Каскадный отказ: проблемы в цепочке

Самая опасная ситуация:

# SCENARIO: Microservice architecture при высокой нагрузке

# Service A: API Gateway
# - Получает 10K req/sec вместо 1K
# - Создаёт 10K потоков
# - Память исчерпывается
# - Начинает отказывать

# Service B: User Service (зависит от A)
# - Ждёт ответа от A
# - Потоки накапливаются (потому что ждут)
# - Memory leak
# - Тоже падает

# Service C: Order Service (зависит от B)
# - B не отвечает
# - Потоки зависают
# - Потом падает и это

# Service D: Payment Service (критический)
# - Последний в цепи
# - ПАДАЕТ ВСЯ СИСТЕМА

from requests import Session
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def requests_retry_session(
    retries=3,
    backoff_factor=0.3,
    status_forcelist=(500, 502, 504),
    session=None,
):
    session = session or Session()
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session

# ❌ Неправильно: бесконечные retry'ы
while True:
    try:
        response = requests.get("http://slow-service.com/api")
        break
    except:
        pass  # Бесконечный loop!

# ✅ Правильно: exponential backoff + timeout
session = requests_retry_session()
try:
    response = session.get(
        "http://slow-service.com/api",
        timeout=5  # Максимум 5 секунд!
    )
except Exception:
    # Fallback или ошибка
    pass

4. Thread starvation

Потоки ждут, но никогда не выполняются:

from concurrent.futures import ThreadPoolExecutor
import time

# ❌ Неправильно: слишком мало рабочих потоков
with ThreadPoolExecutor(max_workers=2) as executor:
    # Подаём 10K задач
    futures = []
    for _ in range(10000):
        future = executor.submit(lambda: time.sleep(10))
        futures.append(future)
    
    # Первая задача выполняется
    # Остальные 9998 ждут в очереди
    # Пользователи видят timeout's

# ✅ Правильно: подобрать нужное количество потоков
from multiprocessing import cpu_count

optimal_threads = cpu_count() * 4  # Для I/O bound
with ThreadPoolExecutor(max_workers=optimal_threads) as executor:
    futures = []
    for _ in range(10000):
        future = executor.submit(lambda: time.sleep(10))
        futures.append(future)
    
    # Задачи распределяются более равномерно

5. GIL становится узким местом

# При высокой нагрузке с CPU-bound работой
import threading
import time

def cpu_work():
    result = 0
    for i in range(100_000_000):
        result += i
    return result

# Нормальная нагрузка: 1 поток
start = time.time()
cpu_work()
print(f"1 поток: {time.time() - start:.2f}s")
# Output: 2.50s

# Высокая нагрузка: 8 потоков
start = time.time()
threads = []
for _ in range(8):
    t = threading.Thread(target=cpu_work)
    threads.append(t)
    t.start()

for t in threads:
    t.join()
print(f"8 потоков: {time.time() - start:.2f}s")
# Output: 18.50s (!!! медленнее!)

# GIL не пускает потоки параллельно
# Context switching просто убивает performance

6. Практический пример: что видит пользователь

# Нормальное состояние
# Request: 10ms
# Response: 10ms
# Total: 20ms ✅

# Начало перегрузки (100 req/sec вместо 10)
# Request: 10ms
# Queue wait: 50ms (потоки заняты)
# Response: 10ms
# Total: 70ms ⚠️

# Сильная перегрузка (1000 req/sec)
# Request: 10ms
# Queue wait: 5000ms (очень долго)
# Response: 10ms
# Total: 5000ms timeout ❌

# Критическая перегрузка (10000 req/sec)
# Server: не отвечает
# User: Connection refused ❌

7. Как предотвратить деградацию

Вариант 1: Rate Limiting

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()

@app.get("/api/users")
@limiter.limit("100/minute")  # Максимум 100 req/min
def get_users(request: Request):
    return {"users": []}

# Клиенты которые превышают лимит получат 429 Too Many Requests
# Система остаётся стабильна

Вариант 2: Circuit Breaker

from pybreaker import CircuitBreaker

db_breaker = CircuitBreaker(
    fail_max=5,  # После 5 ошибок
    reset_timeout=60  # Ждём 60 сек
)

@db_breaker
def query_database():
    return db.execute("SELECT * FROM users")

# Если БД падает — circuit открывается
# Новые запросы немедленно отказываются (быстро)
# Вместо зависания

Вариант 3: Thread Pool Size Tuning

from concurrent.futures import ThreadPoolExecutor
from multiprocessing import cpu_count

# Для I/O bound (web scraping, API calls)
IO_BOUND_THREADS = cpu_count() * 4

# Для CPU bound (обработка данных)
CPU_BOUND_THREADS = cpu_count()

# Для mixed
MIXED_THREADS = cpu_count() * 2

# ✅ Правильное использование
with ThreadPoolExecutor(max_workers=IO_BOUND_THREADS) as executor:
    futures = [executor.submit(fetch_url, url) for url in urls]
    results = [f.result(timeout=10) for f in futures]  # Тайм-аут!

Вариант 4: Graceful Degradation

@app.get("/api/users/{user_id}")
def get_user(user_id: int):
    try:
        # Пытаемся получить полные данные
        user = db.get_user_with_posts(user_id)
        return {"user": user, "posts": user.posts}
    except TimeoutError:
        # При перегрузке возвращаем только основные данные
        user = cache.get(f"user:{user_id}")
        if user:
            return {"user": user, "posts": []}  # Без постов
        raise

Вариант 5: Асинхронность вместо потоков

# ❌ Потоки — 10K потоков = 80GB памяти
with ThreadPoolExecutor(max_workers=10000) as executor:
    futures = [executor.submit(async_io_task) for _ in range(10000)]

# ✅ Asyncio — 10K задач = ~100MB памяти
import asyncio

async def main():
    tasks = [async_io_task() for _ in range(10000)]
    await asyncio.gather(*tasks)

asyncio.run(main())

# Даже под нагрузкой:
# Asyncio: стабильно, 100MB
# Потоки: памяти нет, система падает

8. Мониторинг и алерты

import psutil
import time

def monitor_system():
    while True:
        # Количество потоков
        thread_count = threading.active_count()
        if thread_count > 1000:
            alert(f"Слишком много потоков: {thread_count}")
        
        # Использование памяти
        memory_percent = psutil.virtual_memory().percent
        if memory_percent > 80:
            alert(f"Критический уровень памяти: {memory_percent}%")
        
        # Нагрузка на CPU
        cpu_percent = psutil.cpu_percent(interval=1)
        if cpu_percent > 90:
            alert(f"Критическая нагрузка CPU: {cpu_percent}%")
        
        # Соединения с БД
        db_connections = len(db.pool._holders)
        if db_connections > db.pool.size * 0.9:
            alert(f"Соединения с БД почти исчерпаны: {db_connections}")
        
        time.sleep(10)

Итоговая таблица: последовательность деградации

┌──────────┬──────────────────┬──────────────┬─────────────┐
│ Stage    │ Нагрузка         │ Что происходит     │ Решение     │
├──────────┼──────────────────┼──────────────┼─────────────┤
│ 1. Normal│ 1K req/sec       │ Всё быстро   │ -           │
│ 2. Warm  │ 5K req/sec       │ Latency ↑ → │ Rate limit  │
│ 3. Hot   │ 10K req/sec      │ Memory ↑ → │ Async      │
│ 4. Crit  │ 50K+ req/sec     │ Потоки ↑ →  │ Circuit     │
│ 5. Dead  │ 100K+ req/sec    │ OOM → crash │ Degradation │
└──────────┴──────────────────┴──────────────┴─────────────┘

Вывод

При высокой нагрузке потоки:

1. Увеличивают latency (context switching)
2. Исчерпывают память (8MB per thread)
3. Создают bottleneck'и (GIL в Python)
4. Могут зависнуть (thread starvation)
5. Вызывают каскадный отказ (cascade failure)

Профилактика:

• Rate limiting
• Circuit breaker'ы
• Asyncio вместо потоков
• Graceful degradation
• Мониторинг и алерты
• Load testing перед production