Какие инструменты использовать для работы с CPU-bound задачей?

Инструменты для работы с CPU-bound задачами в Python

CPU-bound задачи — это вычисления, которые активно используют процессор (факториалы, обработка больших данных, криптография). Python имеет Global Interpreter Lock (GIL), что усложняет параллелизм, но есть несколько решений.

1. Multiprocessing (Рекомендуется)

Это основной инструмент для CPU-bound задач в Python. Каждый процесс имеет собственный интерпретатор и GIL:

from multiprocessing import Pool, Process
import time

def cpu_bound_task(n):
    """Вычисление факториала — типичная CPU-bound задача"""
    result = 1
    for i in range(1, n + 1):
        result *= i
    return result

# Способ 1: Pool для параллельной обработки данных
if __name__ == '__main__':
    with Pool(processes=4) as pool:
        numbers = [5000, 10000, 15000, 20000]
        results = pool.map(cpu_bound_task, numbers)
        print(results)

# Способ 2: Process для независимых задач
if __name__ == '__main__':
    processes = []
    for i in range(4):
        p = Process(target=cpu_bound_task, args=(10000,))
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()
    
    print("Все процессы завершены")

Преимущества:

• Обходит GIL
• Истинный параллелизм на многоядерных системах
• Простая для использования Pool API

Недостатки:

• Overhead на создание процессов
• Сложнее обмениваться данными между процессами
• Требует больше памяти

2. Threading (НЕ рекомендуется для CPU-bound)

Треды НЕ помогают при CPU-bound задачах из-за GIL:

from threading import Thread
import time

def cpu_task():
    result = sum(i * i for i in range(10**8))
    return result

# ❌ Это будет работать медленнее одного потока!
if __name__ == '__main__':
    start = time.time()
    
    threads = []
    for _ in range(4):
        t = Thread(target=cpu_task)
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    print(f"Время: {time.time() - start:.2f}с")  # МЕДЛЕННО!

ОД нако threading полезен для I/O-bound задач (сетевые запросы, чтение файлов).

3. asyncio (НЕ рекомендуется для CPU-bound)

Asyncio — для I/O-bound, не CPU-bound:

# ❌ Неправильно для CPU-bound
import asyncio

async def cpu_task():
    # Это будет блокировать весь event loop!
    result = sum(i * i for i in range(10**8))
    return result

# Если очень нужно, используй to_thread (Python 3.9+)
async def better_approach():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, cpu_task)
    return result

4. Cython (Для оптимизации)

Переписание критичного кода на Cython даёт 10-100x ускорение:

# calculate.pyx
cpdef long factorial(int n):
    cdef long result = 1
    cdef int i
    for i in range(1, n + 1):
        result *= i
    return result

# main.py
from calculate import factorial

result = factorial(5000)  # Значительно быстрее!

5. NumPy (Для числовых вычислений)

Если работаете с матрицами/массивами, NumPy на C намного быстрее:

import numpy as np
from multiprocessing import Pool
import time

# ❌ Медленно: Python цикл
def slow_sum(n):
    return sum(i * i for i in range(n))

# ✅ Быстро: NumPy операции
def fast_sum(n):
    arr = np.arange(n)
    return np.sum(arr ** 2)

start = time.time()
result = fast_sum(10**7)
print(f"NumPy: {time.time() - start:.4f}с")  # ~0.01s

start = time.time()
result = slow_sum(10**7)
print(f"Python: {time.time() - start:.4f}с")  # ~0.5s

6. Ray (Распределённые вычисления)

Для распределённых CPU-bound задач:

import ray

ray.init(num_cpus=4)

@ray.remote
def cpu_bound_task(n):
    return sum(i * i for i in range(n))

# Параллельное выполнение
futures = [cpu_bound_task.remote(10**6) for _ in range(4)]
results = ray.get(futures)

ray.shutdown()

7. Joblib (Для машинного обучения)

Популярен в ecosystems scikit-learn:

from joblib import Parallel, delayed

def train_model(data, model_id):
    # CPU-bound задача
    return f"Model {model_id} trained"

# Параллельное выполнение
results = Parallel(n_jobs=4)(
    delayed(train_model)(data, i) for i in range(4)
)

Сравнение производительности

import time
from multiprocessing import Pool
from threading import Thread

def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# Однопоточное выполнение
start = time.time()
for _ in range(4):
    fibonacci(35)
print(f"Sequential: {time.time() - start:.2f}s")

# Многопроцессное (правильно для CPU-bound)
if __name__ == '__main__':
    start = time.time()
    with Pool(4) as p:
        p.map(fibonacci, [35, 35, 35, 35])
    print(f"Multiprocessing: {time.time() - start:.2f}s")  # Значительно быстрее!

Рекомендации по выбору

Для CPU-bound используй Multiprocessing!